Adopt chock v0.9.0 and agentseam 0.3.0

.agents/hooks.json

@@ -57,6 +57,40 @@
           }
         ]
       }
+    ],
+    "Stop": [
+      {
+        "hooks": [
+          {
+            "type": "command",
+            "command": "\"/usr/local/bin/python3\" \".chock/bin/antigravity.py\" --gate \".chock/compiled/block-invisible-unicode/stop/gate.json\""
+          }
+        ]
+      },
+      {
+        "hooks": [
+          {
+            "type": "command",
+            "command": "\"/usr/local/bin/python3\" \".chock/bin/antigravity.py\" --gate \".chock/compiled/block-wildcard-agent-permissions/stop/gate.json\""
+          }
+        ]
+      },
+      {
+        "hooks": [
+          {
+            "type": "command",
+            "command": "\"/usr/local/bin/python3\" \".chock/bin/antigravity.py\" --gate \".chock/compiled/pin-github-actions/stop/gate.json\""
+          }
+        ]
+      },
+      {
+        "hooks": [
+          {
+            "type": "command",
+            "command": "\"/usr/local/bin/python3\" \".chock/bin/antigravity.py\" --gate \".chock/compiled/scan-secrets/stop/gate.json\""
+          }
+        ]
+      }
     ]
   }
 }

.chock/bin/antigravity.py

@@ -1,4 +1,4 @@
-# Generated by agentseam 0.2.1 -- bundle("antigravity"). Do not hand-edit, except the
+# Generated by agentseam 0.3.0 -- bundle("antigravity"). Do not hand-edit, except the
 # HANDLER block below (between "agentseam handler >>>" and "<<< agentseam handler"),
 # which is exactly what this file leaves for you to fill in.
 #
@@ -15,14 +15,13 @@
 
 import os as _chock_os
 import shlex as _chock_shlex
-import shutil as _chock_shutil
 import subprocess as _chock_subprocess
 from datetime import datetime as _chock_datetime, timezone as _chock_timezone
 from pathlib import Path as _chock_Path
 import warnings as _warnings
 
 # ------------------------------------------------------------------------------
-# contract (agentseam 0.2.1)
+# contract (agentseam 0.3.0)
 
 """Canonical event vocabulary, normalized envelope, and decision type."""
 
@@ -310,7 +309,19 @@ def hj_parse(cfg, raw, wire=None):
 _TRANSFORM_MISSING_INPUT = "transform_missing_input"
 
 def hj_reverse(cfg):
-    """Canonical event -> wire name: the naive inverse, then the entry's pinned overrides."""
+    """Canonical event -> wire name: the naive inverse, then the entry's pinned overrides.
+
+    One wire name per canonical event, which is a real limit and not an oversight (R3, gap 4).
+    Cursor is where it bites: `pre_tool` pins to `preToolUse`, and Cursor honours `ask` only at
+    `beforeShellExecution` / `beforeMCPExecution`. So installing at `pre_tool` forecloses `ask`
+    before dispatch is ever reached -- the runtime degrade to `deny` is honest about it, but the
+    install already chose. Deny-style policies, which is all that ships today, are unaffected.
+
+    The day an ask-style Cursor policy exists, this map has to select by decision dialect
+    (deny -> `preToolUse`, ask -> `beforeShellExecution`) and become one-to-many. Recorded here
+    rather than built, so the resolution is not re-litigated from scratch; the three facts it
+    rests on are pinned in tests/test_cursor_ask_dialect.py.
+    """
     reverse = {}
     for name, canonical in cfg["events"].items():
         if canonical != UNKNOWN:
@@ -602,6 +613,10 @@ def degrade(decision, event):
 # >>> agentseam handler >>>
 GUARD_VIOLATION = 1
 
+GUARD_ASK_EXIT = 3
+
+PYTHON_SUFFIX = '.py'
+
 _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash')
 
 GATE_LOG_ENV = 'CHOCK_GATE_LOG'
@@ -614,6 +629,8 @@ def degrade(decision, event):
 
 GUARD_CLEAN = 'clean'
 
+GUARD_ASKED = 'asked'
+
 GUARD_UNCHECKED = 'unchecked'
 
 GUARD_ERRORED = 'errored'
@@ -641,41 +658,56 @@ def find_bash(guard: _chock_Path) -> str | None:
             return candidate
     return None
 
+def find_interpreter(guard: _chock_Path) -> str | None:
+    """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash."""
+    if guard.suffix == PYTHON_SUFFIX:
+        return sys.executable or None
+    return find_bash(guard)
+
 def run_guard(guard: _chock_Path, command: str) -> str:
-    """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not."""
+    """`GUARD_BLOCKED` / `GUARD_ASKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not."""
+    return run_guard_detailed(guard, command)[0]
+
+def run_guard_detailed(guard: _chock_Path, command: str) -> tuple[str, str]:
+    """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user."""
     try:
         args = _chock_shlex.split(command)
     except ValueError:
         print('chock: could not parse command (unbalanced quotes), not checked', file=sys.stderr)
-        return GUARD_UNCHECKED
+        return (GUARD_UNCHECKED, '')
     if not args:
-        return GUARD_UNCHECKED
-    bash = find_bash(guard)
-    if bash is None:
-        print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr)
-        return GUARD_UNCHECKED
+        return (GUARD_UNCHECKED, '')
+    interpreter = find_interpreter(guard)
+    if interpreter is None:
+        print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr)
+        return (GUARD_UNCHECKED, '')
     try:
         env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command}
-        proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False)
+        proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False)
     except _chock_subprocess.TimeoutExpired:
         print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr)
-        return GUARD_ERRORED
+        return (GUARD_ERRORED, '')
     except (OSError, UnicodeError) as exc:
         print(f'chock: guard could not run, not checked: {exc}', file=sys.stderr)
-        return GUARD_ERRORED
+        return (GUARD_ERRORED, '')
+    output = ((proc.stderr or '') + (proc.stdout or '')).strip()
+    first_line = output.splitlines()[0].strip() if output else ''
     if proc.returncode == GUARD_VIOLATION:
         sys.stderr.write(proc.stdout or '')
         sys.stderr.write(proc.stderr or '')
-        if not ((proc.stdout or '') + (proc.stderr or '')).strip():
+        if not output:
             print(f'chock: blocked by {_chock_Path(guard).name} (guard gave no reason)', file=sys.stderr)
-        return GUARD_BLOCKED
+        return (GUARD_BLOCKED, first_line)
+    if proc.returncode == GUARD_ASK_EXIT:
+        sys.stderr.write(proc.stdout or '')
+        sys.stderr.write(proc.stderr or '')
+        return (GUARD_ASKED, first_line)
     if proc.returncode != 0:
-        detail = (proc.stderr or proc.stdout or '').strip().splitlines()
-        print(f'chock: guard exited {proc.returncode}, not checked' + (f': {detail[0][:120]}' if detail else ''), file=sys.stderr)
-        return GUARD_ERRORED
-    return GUARD_CLEAN
+        print(f'chock: guard exited {proc.returncode}, not checked' + (f': {first_line[:120]}' if first_line else ''), file=sys.stderr)
+        return (GUARD_ERRORED, '')
+    return (GUARD_CLEAN, '')
 
-def log_outcome(guard: _chock_Path, tool: str, *, blocked: bool) -> None:
+def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None:
     """Append one outcome record. Best effort: never raises, never changes the verdict."""
     try:
         if _chock_os.environ.get(GATE_LOG_ENV) == '0':
@@ -695,7 +727,7 @@ def log_outcome(guard: _chock_Path, tool: str, *, blocked: bool) -> None:
         log_path = log_dir / 'gate-events.jsonl'
         if log_path.exists() and log_path.stat().st_size > _LOG_MAX_BYTES:
             log_path.replace(log_dir / 'gate-events.1.jsonl')
-        record = {'ts': _chock_datetime.now(_chock_timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'), 'policy_id': guard.parent.parent.name, 'surface': 'pre-tool-use', 'event': 'tool_use', 'kind': guard.stem, 'tool': tool, 'verdict': 'block' if blocked else 'allow'}
+        record = {'ts': _chock_datetime.now(_chock_timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'), 'policy_id': guard.parent.parent.name, 'surface': 'pre-tool-use', 'event': 'tool_use', 'kind': guard.stem, 'tool': tool, 'verdict': verdict}
         with log_path.open('a', encoding='utf-8') as fh:
             fh.write(json.dumps(record, ensure_ascii=False) + '\n')
     except Exception:
@@ -706,23 +738,176 @@ def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | N
     guard = guard_path_from_argv(argv)
     if guard is None or not guard.exists():
         return None
-    verdict = run_guard(guard, command)
-    if verdict in (GUARD_BLOCKED, GUARD_CLEAN):
-        log_outcome(guard, tool, blocked=verdict == GUARD_BLOCKED)
+    verdict, message = run_guard_detailed(guard, command)
+    logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'}
+    if verdict in logged:
+        log_outcome(guard, tool, verdict=logged[verdict])
     if verdict == GUARD_BLOCKED:
         return (VERDICT_DENY, f'Blocked by chock policy: {guard.stem}')
+    if verdict == GUARD_ASKED:
+        return (VERDICT_ESCALATE, f'chock policy {guard.stem} asks before this runs: {message}' if message else f'chock policy {guard.stem} asks for confirmation before this runs (guard gave no reason).')
     if verdict == GUARD_ERRORED:
         return (VERDICT_ESCALATE, f"chock could not check this command: the {guard.stem} guard did not complete (see this hook's stderr). Approving runs it unchecked.")
     return None
 
+GATE_FLAG = '--gate'
 
-def handle(event):
+_GATE_TIMEOUT_SECONDS = 30
+
+_GATE_DEPTH_TO_CHOCK = 3
+
+_RUNNER_PARTS = ('bin', 'gate.py')
+
+_GIT = 'git'
+
+_DELETED = 'D'
+
+_RENAMED = 'R'
+
+GATE_BLOCKED = 'blocked'
+
+GATE_CLEAN = 'clean'
+
+GATE_ERRORED = 'errored'
+
+VERDICT_DENY = 'deny'
+
+def gate_path_from_argv(argv):
+    """The `--gate <path>` argument a vendored runtime was invoked with, or None."""
+    if GATE_FLAG in argv:
+        index = argv.index(GATE_FLAG)
+        if index + 1 < len(argv):
+            return _chock_Path(argv[index + 1])
+    return None
+
+def runner_for(gate):
+    """The vendored gate runner beside this compiled gate, or None when it is not there."""
+    parents = gate.resolve().parents
+    if len(parents) <= _GATE_DEPTH_TO_CHOCK:
+        return None
+    runner = parents[_GATE_DEPTH_TO_CHOCK].joinpath(*_RUNNER_PARTS)
+    return runner if runner.exists() else None
+
+def writes_from_event(event):
+    """The one file this tool call would write, or none when it carries no file text."""
+    path = getattr(event, 'path', None)
+    content = getattr(event, 'content', None)
+    if not path or not isinstance(content, str):
+        return {}
+    return {str(path): content}
+
+def changed_paths(repo_root):
+    """Every uncommitted path in the worktree. Outside a repository there is nothing to list."""
+    try:
+        proc = _chock_subprocess.run([_GIT, '-C', str(repo_root), 'status', '--porcelain=v1', '--untracked-files=all', '-z'], capture_output=True, text=True, timeout=_GATE_TIMEOUT_SECONDS, check=False)
+    except (OSError, _chock_subprocess.SubprocessError):
+        return []
+    if proc.returncode != 0:
+        return []
+    fields = [field for field in (proc.stdout or '').split('\x00') if field]
+    paths = []
+    skip_next = False
+    for field in fields:
+        if skip_next:
+            skip_next = False
+            continue
+        status, path = (field[:2], field[3:])
+        skip_next = status.startswith(_RENAMED)
+        if _DELETED in status or not path:
+            continue
+        paths.append(path)
+    return paths
+
+def writes_from_worktree(repo_root):
+    """What this turn actually left on disk, however it was written.
+
+    The write path sees only writes it recognises; a shell heredoc carries no file argument.
+    Reading final state is what makes that stop mattering, so this deliberately does not care
+    which tool produced the bytes.
+    """
+    writes = {}
+    for path in changed_paths(repo_root):
+        try:
+            writes[path] = _chock_Path(repo_root, path).read_text(encoding='utf-8')
+        except (OSError, UnicodeDecodeError):
+            continue
+    return writes
+
+def run_gate(gate, writes, event):
+    """Ask the vendored runner. Returns (outcome, message) and never decides for itself."""
+    runner = runner_for(gate)
+    if runner is None:
+        return (GATE_ERRORED, 'the vendored gate runner is not installed beside this gate')
+    try:
+        proc = _chock_subprocess.run([sys.executable, str(runner), 'run', '--gate', str(gate), '--event', event], input=json.dumps({'writes': writes}), capture_output=True, text=True, timeout=_GATE_TIMEOUT_SECONDS, check=False)
+    except (OSError, _chock_subprocess.SubprocessError) as exc:
+        return (GATE_ERRORED, str(exc))
+    if proc.returncode == 0:
+        return (GATE_CLEAN, '')
+    if proc.returncode == 1:
+        return (GATE_BLOCKED, (proc.stderr or '').strip())
+    return (GATE_ERRORED, (proc.stderr or '').strip())
+
+_EVENT_ARG = {'pre_tool': 'pre-tool-use', 'stop': 'stop'}
+
+PRE_TOOL = 'pre_tool'
+
+def root_for(gate):
+    """The repository this compiled gate belongs to, or None when the layout is not that."""
+    parents = gate.resolve().parents
+    if len(parents) <= _GATE_DEPTH_TO_CHOCK:
+        return None
+    return parents[_GATE_DEPTH_TO_CHOCK].parent
+
+def writes_for(event, gate):
+    """What this event puts under judgement: the call's own text, or what the turn left behind."""
+    if event.event == PRE_TOOL:
+        return writes_from_event(event)
+    if (event.raw or {}).get('stop_hook_active'):
+        return {}
+    root = root_for(gate)
+    return writes_from_worktree(root) if root is not None else {}
+
+def evaluate_gate(argv, event):
+    """The decision this event earns from a compiled gate, or None when it has nothing to say."""
+    gate = gate_path_from_argv(argv)
+    name = _EVENT_ARG.get(getattr(event, 'event', ''))
+    if gate is None or name is None or (not gate.exists()):
+        return None
+    writes = writes_for(event, gate)
+    if not writes:
+        return None
+    outcome, message = run_gate(gate, writes, name)
+    if outcome == GATE_BLOCKED:
+        return (VERDICT_DENY, message or f'Blocked by chock policy: {gate.parent.parent.name}')
+    if outcome == GATE_ERRORED:
+        return (VERDICT_DENY, f'chock could not check this write: {message}. Refusing rather than reporting an allow it never established.')
+    return None
+
+
+def _judge(event):
     if event.event == "pre_tool" and event.command:
         verdict = evaluate(sys.argv[1:], event.command, event.tool or "")
         if verdict is not None:
             outcome, reason = verdict
             return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason)
+    gated = evaluate_gate(sys.argv[1:], event)
+    if gated is not None:
+        return Decision.deny(gated[1])
     return None
+
+
+def handle(event):
+    # A door that cannot decide refuses. An exception escaping here would exit the hook
+    # with a traceback, which every client reads as a non-blocking error: fail-open, with
+    # the reason on a stderr nobody watches. The refusal carries the reason instead.
+    try:
+        return _judge(event)
+    except Exception as exc:  # noqa: BLE001 -- any failure here must refuse, never fall through
+        return Decision.deny(
+            "chock could not check this call (%s: %s). Refusing rather than reporting an "
+            "allow it never established." % (type(exc).__name__, exc)
+        )
 # <<< agentseam handler <<<