weichen
5618cfc8cf
Layer advisory Cursor hooks and skills on the PI executor, shrink the test-report rule to the validation contract, and fix MCP relative paths.
102 lines
3.2 KiB
Python
102 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""sessionStart / stop: audit + optional fail-open inventory (no context injection)."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
import _lib as lib # noqa: E402
|
|
|
|
HOOK = "audit-session"
|
|
INVENTORY_TIMEOUT = 15
|
|
|
|
|
|
def _run_inventory(root: Path) -> dict:
|
|
out: dict = {"cleanup_report": None, "docs_drift": None, "errors": []}
|
|
if os.environ.get("CURSOR_HOOKS_NO_INVENTORY", "").strip() in ("1", "true", "TRUE", "yes"):
|
|
out["skipped"] = "CURSOR_HOOKS_NO_INVENTORY"
|
|
return out
|
|
agentctl = root / "scripts" / "agentctl.py"
|
|
if not agentctl.is_file():
|
|
out["errors"].append("agentctl.py missing")
|
|
return out
|
|
for label, args in (
|
|
("cleanup_report", ["cleanup", "report"]),
|
|
("docs_drift", ["docs", "drift"]),
|
|
):
|
|
try:
|
|
r = subprocess.run(
|
|
["python3", "-B", str(agentctl), *args],
|
|
cwd=str(root),
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=INVENTORY_TIMEOUT,
|
|
check=False,
|
|
)
|
|
payload = None
|
|
for stream in (r.stdout, r.stderr):
|
|
for line in (stream or "").splitlines():
|
|
line = line.strip()
|
|
if line.startswith("{"):
|
|
try:
|
|
payload = json.loads(line)
|
|
break
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if payload is not None:
|
|
break
|
|
out[label] = {
|
|
"exit_code": r.returncode,
|
|
"payload": payload,
|
|
"stdout_tail": (r.stdout or "")[-2000:],
|
|
"stderr_tail": (r.stderr or "")[-1000:],
|
|
}
|
|
except subprocess.TimeoutExpired:
|
|
out["errors"].append(f"{label}: timeout>{INVENTORY_TIMEOUT}s")
|
|
except OSError as exc:
|
|
out["errors"].append(f"{label}: {exc}")
|
|
return out
|
|
|
|
|
|
def main() -> None:
|
|
raw = lib.read_stdin()
|
|
lib.maybe_probe(raw, HOOK)
|
|
data = lib.parse_json_loose(raw) or {}
|
|
event = str(
|
|
lib.first_key(data, "hook_event_name", "event", "event_name", "type") or "unknown"
|
|
)
|
|
root = lib.repo_root()
|
|
if "start" in event.lower() or event == "unknown":
|
|
inv = _run_inventory(root)
|
|
inv_path = None
|
|
try:
|
|
inv_dir = lib.inventory_dir()
|
|
inv_dir.mkdir(parents=True, exist_ok=True)
|
|
ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
|
path = inv_dir / f"inventory-{ts}.json"
|
|
path.write_text(json.dumps(inv, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
inv_path = str(path)
|
|
except OSError:
|
|
pass
|
|
lib.audit(
|
|
HOOK,
|
|
"SESSION_AUDIT",
|
|
event=event,
|
|
inventory_path=inv_path,
|
|
inventory_errors=inv.get("errors"),
|
|
)
|
|
else:
|
|
lib.audit(HOOK, "SESSION_AUDIT", event=event)
|
|
|
|
lib.emit({})
|
|
raise SystemExit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|