From 466d3f671a995c2f1b21fca9d262e668360bdf21 Mon Sep 17 00:00:00 2001 From: Jothi Mani Rajendran <250249270+jothimani-rajendran@users.noreply.github.com> Date: Mon, 19 Oct 2026 15:30:24 +0000 Subject: [PATCH 1/2] ci: add the CI / check workflow the release check reads (REL-1c) chock-release check 2 reads ci.yml job `check` on every pinner repo and refuses the release when it is absent. This repo had no workflow. The job installs chock from the single pin in README.md (refusing when there is none or more than one distinct pin), runs chock check (with --base on pull requests) and chock sync --check. Co-Authored-By: Claude Sonnet 5.5 Signed-off-by: Jothi Mani Rajendran <250249270+jothimani-rajendran@users.noreply.github.com> --- .github/workflows/ci.yml | 51 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 51 insertions(+) create mode 100644 .github/workflows/ci.yml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..287cce4 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,51 @@ +# The release check (chock-release, check 2) reads this file's job `check`: keep the name. +# chock is installed from the pin in README.md so the README stays the single source. +name: CI + +on: + pull_request: + push: + branches: [main] + +permissions: + contents: read + +jobs: + check: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 + with: + persist-credentials: false + fetch-depth: 0 + + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7 + with: + python-version: "3.12" + + - name: Install chock from the README pin + run: | + pins="$(grep -E -o 'pip install "chock @ git\+https://github\.com/open-coder-ai/chock@[0-9a-f]{40}"' README.md | sort -u)" + if [ -z "$pins" ]; then + echo "README.md has no 'pip install \"chock @ git+https://github.com/open-coder-ai/chock@<40-hex>\"' line" >&2 + exit 1 + fi + if [ "$(printf '%s\n' "$pins" | wc -l)" -ne 1 ]; then + echo "README.md carries more than one distinct chock pin:" >&2 + printf '%s\n' "$pins" >&2 + exit 1 + fi + bash -e -c "$pins" + + - name: chock check (pull request) + if: github.event_name == 'pull_request' + env: + BASE_SHA: ${{ github.event.pull_request.base.sha }} + run: chock check --repo . --base "$BASE_SHA" + + - name: chock check (push) + if: github.event_name != 'pull_request' + run: chock check --repo . + + - name: Compiled files match the policies + run: chock sync --repo . --check From 3105bb83317476d82f293d15ae34a000335f1b6b Mon Sep 17 00:00:00 2001 From: Jothi Mani Rajendran <250249270+jothimani-rajendran@users.noreply.github.com> Date: Mon, 19 Oct 2026 21:31:20 -0400 Subject: [PATCH 2/2] regen: compiled tree under the README's engine pin Signed-off-by: Jothi Mani Rajendran <250249270+jothimani-rajendran@users.noreply.github.com> --- .chock/bin/claude_code.py | 1451 ++++++++++++++++++++++++++++++++++--- .chock/bin/launch.sh | 21 + 2 files changed, 1371 insertions(+), 101 deletions(-) create mode 100644 .chock/bin/launch.sh diff --git a/.chock/bin/claude_code.py b/.chock/bin/claude_code.py index e4c6207..41a7f48 100755 --- a/.chock/bin/claude_code.py +++ b/.chock/bin/claude_code.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.1 -- bundle("claude_code"). Do not hand-edit, except the +# Generated by agentseam 0.3.5 -- bundle("claude_code"). 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. # @@ -9,21 +9,29 @@ from __future__ import annotations +import contextlib +import io import json _json = json +import os import sys +import hashlib as _chock_hashlib +import fnmatch as _chock_fnmatch +import re as _chock_re import os as _chock_os import shlex as _chock_shlex import shutil as _chock_shutil import subprocess as _chock_subprocess +import time as _chock_time from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +from pathlib import PurePosixPath as _chock_PurePosixPath, PureWindowsPath as _chock_PureWindowsPath import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.1) +# contract (agentseam 0.3.5) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -201,7 +209,7 @@ def tool_input_of(raw): """The tool's arguments as a dict, decoding the JSON-string form some vendors send.""" if isinstance(raw, dict): return raw - if isinstance(raw, str) and raw[:1] == "{": + if isinstance(raw, str) and raw.lstrip()[:1] == "{": try: parsed = _json.loads(raw) except (ValueError, RecursionError): @@ -562,6 +570,8 @@ def hj_respond(cfg, decision, event, wire=None): _WINDOWS_KEYS = ("commandWindows", "windows") +_MATCHER_EVENTS = (PRE_TOOL, POST_TOOL, TOOL_FAILURE) + def _hook_dict(cfg, command): entry = {"type": "command", "command": command} for key, value in cfg["hook_entry"].get("entry_extra", {}).items(): @@ -578,20 +588,21 @@ def _flat_list_wrapper(hook_entry, reverse, canonical_events, command, matcher): if not name: continue rule = {"event": name, "command": command} - if matcher and hook_entry["matcher"]: + if matcher and hook_entry["matcher"] and ev in _MATCHER_EVENTS: rule["matcher"] = matcher rules.append(rule) return rules def _cursor_wrapper(cfg, reverse, canonical_events, command, *, fail_closed): gates = cfg["verdicts"]["answer_events"] + allow_silent = cfg["verdicts"].get("allow_silent_events", ()) hooks = {} for ev in canonical_events: name = reverse.get(ev) if not name: continue entry = {"command": command} - if fail_closed and name in gates: + if fail_closed and name in gates and name not in allow_silent: entry["failClosed"] = True hooks.setdefault(name, []).append(entry) return {"version": 1, "hooks": hooks} @@ -616,7 +627,7 @@ def _default_wrapper(cfg, reverse, canonical_events, command, matcher): if not name: continue entry = {"hooks": [_hook_dict(cfg, command)]} - if matcher and hook_entry["matcher"]: + if matcher and hook_entry["matcher"] and ev in _MATCHER_EVENTS: entry["matcher"] = matcher hooks.setdefault(name, []).append(entry) if hook_entry.get("group"): @@ -646,7 +657,7 @@ def hook_entry_config(cfg, canonical_events, command, matcher=None, *, fail_clos AGENT = "claude_code" -VENDOR = {'agent': 'claude_code', 'claims': {'client_types': (None, 'claude_code'), 'event_key': ('hook_event_name',), 'mode': 'marker', 'notes': 'prompt_id rejects only when looks_like_claude_code(raw) is also false; a real Claude Code payload may carry prompt_id and must still be accepted (matrix-notes.json: fixed 2026-08-27).', 'reject_markers': ('turn_id', 'project_path', 'timestamp'), 'reject_markers_unless_probe': {'looks_like_claude_code': ('prompt_id',)}}, 'config_format': 'json', 'config_path': '.claude/settings.json', 'display': 'Claude Code', 'events': {'FileChanged': 'file_changed', 'InstructionsLoaded': 'instructions_loaded', 'PostToolUse': 'post_tool', 'PostToolUseFailure': 'tool_failure', 'PreCompact': 'pre_compact', 'PreToolUse': 'pre_tool', 'SessionEnd': 'session_end', 'SessionStart': 'session_start', 'Stop': 'stop', 'SubagentStart': 'subagent_start', 'SubagentStop': 'subagent_stop', 'UserPromptSubmit': 'prompt_submit'}, 'evidence': {'claims': {'basis': 'live-run', 'date': '2026-09-07', 'test': 'tests/test_examples.py::test_each_payload_is_claimed_by_its_own_adapter'}, 'config_path': {'basis': 'live-run', 'date': '2026-09-07', 'test': 'tests/test_vendor_config.py::test_config_path_agrees_with_matrix'}, 'events': {'basis': 'live-run', 'date': '2026-09-07', 'test': 'tests/test_examples.py::test_each_payload_parses_to_the_event_it_is_filed_under'}, 'family': {'basis': 'live-run', 'date': '2026-09-07', 'test': 'tests/test_golden_fixtures.py::test_wire_output_matches_the_frozen_fixture'}, 'fields': {'basis': 'live-run', 'date': '2026-09-07', 'test': 'tests/test_vendor_config.py::test_entries_match_recount'}, 'hook_entry': {'basis': 'live-run', 'date': '2026-09-07', 'test': 'tests/test_golden_fixtures.py::test_hook_config_matches_the_frozen_fixture_on_both_matcher_paths'}, 'repo_root_token': {'basis': 'vendor-docs', 'date': '2026-09-01', 'test': 'tests/test_vendor_config.py::test_repo_root_token_is_recorded_only_where_primary_sourced'}, 'tools': {'basis': 'live-run', 'date': '2026-09-07', 'test': 'tests/test_vendor_config.py::test_entries_match_recount'}, 'verdicts': {'basis': 'live-run', 'date': '2026-09-07', 'test': 'tests/test_golden_fixtures.py::test_wire_output_matches_the_frozen_fixture'}}, 'family': 'hook_json', 'fields': {'command': ('tool_input.command',), 'content': ('tool_input.content', 'tool_input.new_string', 'tool_input.new_source', 'content', 'tool_input.edits[].new_string'), 'cwd': ('cwd',), 'output': ('tool_output',), 'path': ('tool_input.file_path', 'tool_input.path', 'tool_input.notebook_path', 'file_path'), 'prompt': ('prompt',), 'session_id': ('session_id',), 'tool': ('tool_name',), 'tool_use_id': ('tool_use_id',)}, 'hook_entry': {'matcher': True, 'wrapper': 'hooks_map'}, 'needs_trust': False, 'repo_root_token': '${CLAUDE_PROJECT_DIR}', 'tools': {'shell': ('Bash',), 'write': ('Write', 'Edit', 'MultiEdit', 'NotebookEdit')}, 'verdicts': {'answer_events': ('PreToolUse', 'Stop', 'UserPromptSubmit'), 'bare_allow': 'silent', 'context_events': ('SessionStart', 'UserPromptSubmit'), 'context_source': 'context', 'degrade_notes': {'escalate': 'confirmation requested; this event cannot prompt, so it blocks', 'transform': 'input rewrite requested; this event cannot modify input, so it blocks'}, 'echo': 'reverse_map', 'gates': {'PreToolUse': {'grammar': 'G2', 'honours_escalate': True, 'honours_transform': True}, 'Stop': {'grammar': 'G1', 'honours_escalate': False, 'honours_transform': False}, 'UserPromptSubmit': {'grammar': 'G1', 'honours_escalate': False, 'honours_transform': False}}, 'note_style': 'suffix', 'reason_defaults': {'deny_gate': 'blocked', 'escalate_gate': 'confirmation required'}, 'transform_grammar': 'hook_specific_updated_input', 'vocabulary': ('allow', 'ask', 'block', 'deny'), 'vocabulary_basis': 'verified', 'words': {'block': 'block', 'deny': 'deny', 'escalate': 'ask', 'transform': 'allow', 'vouch': 'allow'}}} +VENDOR = {'agent': 'claude_code', 'claims': {'client_types': (None, 'claude_code'), 'event_key': ('hook_event_name',), 'mode': 'marker', 'notes': 'prompt_id rejects only when looks_like_claude_code(raw) is also false; a real Claude Code payload may carry prompt_id and must still be accepted (matrix-notes.json: fixed 2026-08-27).', 'reject_markers': ('turn_id', 'project_path', 'timestamp'), 'reject_markers_unless_probe': {'looks_like_claude_code': ('prompt_id',)}}, 'config_format': 'json', 'config_path': '.claude/settings.json', 'display': 'Claude Code', 'events': {'FileChanged': 'file_changed', 'InstructionsLoaded': 'instructions_loaded', 'PostToolUse': 'post_tool', 'PostToolUseFailure': 'tool_failure', 'PreCompact': 'pre_compact', 'PreToolUse': 'pre_tool', 'SessionEnd': 'session_end', 'SessionStart': 'session_start', 'Stop': 'stop', 'SubagentStart': 'subagent_start', 'SubagentStop': 'subagent_stop', 'UserPromptSubmit': 'prompt_submit'}, 'evidence': {'claims': {'basis': 'live-run', 'date': '2026-09-07', 'test': 'tests/test_examples.py::test_each_payload_is_claimed_by_its_own_adapter'}, 'config_path': {'basis': 'live-run', 'date': '2026-09-07', 'test': 'tests/test_vendor_config.py::test_config_path_agrees_with_matrix'}, 'events': {'basis': 'live-run', 'date': '2026-09-07', 'test': 'tests/test_examples.py::test_each_payload_parses_to_the_event_it_is_filed_under'}, 'family': {'basis': 'live-run', 'date': '2026-09-07', 'test': 'tests/test_golden_fixtures.py::test_wire_output_matches_the_frozen_fixture'}, 'fields': {'basis': 'live-run', 'date': '2026-09-07', 'test': 'tests/test_vendor_config.py::test_entries_match_recount'}, 'hook_entry': {'basis': 'live-run', 'date': '2026-09-07', 'test': 'tests/test_golden_fixtures.py::test_hook_config_matches_the_frozen_fixture_on_both_matcher_paths'}, 'repo_root_token': {'basis': 'vendor-docs', 'date': '2026-09-01', 'test': 'tests/test_vendor_config.py::test_repo_root_token_is_recorded_only_where_primary_sourced'}, 'tools': {'basis': 'vendor-docs', 'date': '2026-09-27', 'test': 'tests/test_adapter_claude_code.py::test_powershell_is_a_shell_tool_and_its_command_is_parsed'}, 'verdicts': {'basis': 'live-run', 'date': '2026-09-07', 'test': 'tests/test_golden_fixtures.py::test_wire_output_matches_the_frozen_fixture'}}, 'family': 'hook_json', 'fields': {'command': ('tool_input.command',), 'content': ('tool_input.content', 'tool_input.new_string', 'tool_input.new_source', 'content', 'tool_input.edits[].new_string'), 'cwd': ('cwd',), 'output': ('tool_output',), 'path': ('tool_input.file_path', 'tool_input.path', 'tool_input.notebook_path', 'file_path'), 'prompt': ('prompt',), 'session_id': ('session_id',), 'tool': ('tool_name',), 'tool_use_id': ('tool_use_id',)}, 'hook_entry': {'matcher': True, 'wrapper': 'hooks_map'}, 'needs_trust': False, 'repo_root_token': '${CLAUDE_PROJECT_DIR}', 'tools': {'shell': ('Bash', 'PowerShell'), 'write': ('Write', 'Edit', 'MultiEdit', 'NotebookEdit')}, 'verdicts': {'answer_events': ('PreToolUse', 'Stop', 'UserPromptSubmit'), 'bare_allow': 'silent', 'context_events': ('SessionStart', 'UserPromptSubmit'), 'context_source': 'context', 'degrade_notes': {'escalate': 'confirmation requested; this event cannot prompt, so it blocks', 'transform': 'input rewrite requested; this event cannot modify input, so it blocks'}, 'echo': 'reverse_map', 'gates': {'PreToolUse': {'grammar': 'G2', 'honours_escalate': True, 'honours_transform': True}, 'Stop': {'grammar': 'G1', 'honours_escalate': False, 'honours_transform': False}, 'UserPromptSubmit': {'grammar': 'G1', 'honours_escalate': False, 'honours_transform': False}}, 'note_style': 'suffix', 'reason_defaults': {'deny_gate': 'blocked', 'escalate_gate': 'confirmation required'}, 'transform_grammar': 'hook_specific_updated_input', 'vocabulary': ('allow', 'ask', 'block', 'deny'), 'vocabulary_basis': 'verified', 'words': {'block': 'block', 'deny': 'deny', 'escalate': 'ask', 'transform': 'allow', 'vouch': 'allow'}}} def claims(raw): @@ -697,19 +708,60 @@ def degrade(decision, event): # >>> agentseam handler >>> +ENGINE_BUDGET_SECONDS = 30 + +def engine_deadline(): + """The monotonic instant this invocation's budget runs out.""" + return _chock_time.monotonic() + ENGINE_BUDGET_SECONDS + +def engine_remaining(deadline): + """Seconds left before `deadline`, never negative.""" + return max(deadline - _chock_time.monotonic(), 0.0) + +def time_left(deadline, cmd): + """Seconds left for a subprocess; raises TimeoutExpired when none are, so the caller's timeout path runs.""" + left = deadline - _chock_time.monotonic() + if left <= 0: + raise _chock_subprocess.TimeoutExpired(cmd, 0) + return left + +def allowed_seconds(expired): + """The time a timed-out call was actually given, as text (what the deadline left it, not the full budget).""" + return f'{max(float(expired.timeout or 0), 0.0):.1f}s' + 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') +_CRASH_MARKERS = ('traceback (most recent call last)', 'syntax error', 'syntaxerror', 'unexpected eof') + +TOOL_UNKNOWN = 'unknown' + +ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' + +_POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') + +_GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') + +_GIT_BASH_DIRS = (('bin',), ('usr', 'bin')) + +_BASH_EXE = 'bash.exe' + +_WINDOWS_STUB_DIRS = ('/system32/', '/windowsapps/') + +_COREUTILS_MARKER = 'sed.exe' + +_WINDOWS = 'nt' + +_FOUND_BASH = {} GATE_LOG_ENV = 'CHOCK_GATE_LOG' _LOG_MAX_BYTES = 1048576 -_GUARD_TIMEOUT_SECONDS = 30 +_BASH_PROBE_SECONDS = 10 GUARD_BLOCKED = 'blocked' @@ -733,45 +785,109 @@ def guard_path_from_argv(argv: list[str]) -> _chock_Path | None: return _chock_Path(argv[i + 1]) return None -def find_bash(guard: _chock_Path) -> str | None: - """First interpreter that can actually see `guard`, or None.""" - for candidate in _BASH_CANDIDATES: +def _git_roots() -> list[_chock_Path]: + """Git for Windows install roots: from `git` on PATH (cmd/ or mingw64/bin/), then the defaults.""" + git = _chock_shutil.which('git') + near = [_chock_Path(git).parent.parent, _chock_Path(git).parent.parent.parent] if git else [] + return [*near, *(_chock_Path(root) for root in _GIT_FOR_WINDOWS)] + +def bash_candidates() -> list[str]: + """Bash interpreters to try, best first; on Windows Git's own, never the WSL launcher.""" + if _chock_os.name != _WINDOWS: + return list(_POSIX_BASH) + found = [str(root.joinpath(*sub, _BASH_EXE)) for root in _git_roots() for sub in _GIT_BASH_DIRS] + on_path = _chock_shutil.which('bash') + if on_path and (not any((stub in on_path.lower().replace('\\', '/') for stub in _WINDOWS_STUB_DIRS))): + found.append(on_path) + return [c for i, c in enumerate(found) if c not in found[:i] and _chock_Path(c).is_file()] + +def interpreter_env(interpreter: str) -> dict[str, str]: + """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" + env = dict(_chock_os.environ) + if _chock_os.name != _WINDOWS: + return env + home = _chock_Path(interpreter).parent + usr_bin = next((d for d in (home.parent / 'usr' / 'bin', home) if (d / _COREUTILS_MARKER).is_file()), None) + if usr_bin is not None: + env['PATH'] = str(usr_bin) + _chock_os.pathsep + env.get('PATH', '') + return env + +def find_bash(guard: _chock_Path, deadline: float | None=None) -> str | None: + """First bash that can actually see `guard`, probed once per process; the probes draw on `deadline`.""" + if 'bash' in _FOUND_BASH: + return _FOUND_BASH['bash'] + for candidate in bash_candidates(): + argv = [candidate, '-c', f'test -f "{guard.as_posix()}"'] try: - proc = _chock_subprocess.run([candidate, '-c', f'test -f "{guard.as_posix()}"'], capture_output=True, timeout=10, check=False) + limit = _BASH_PROBE_SECONDS if deadline is None else min(_BASH_PROBE_SECONDS, time_left(deadline, argv)) + proc = _chock_subprocess.run(argv, capture_output=True, timeout=limit, check=False) except (OSError, _chock_subprocess.SubprocessError): + if deadline is not None and engine_remaining(deadline) <= 0: + return None continue if proc.returncode == 0: + _FOUND_BASH['bash'] = candidate return candidate return None -def find_interpreter(guard: _chock_Path) -> str | None: +def find_interpreter(guard: _chock_Path, deadline: float | None=None) -> 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) + return find_bash(guard, deadline) + +def normalize_tool(tool: str | None) -> str: + """The hook payload's tool name as `bash` | `powershell` | `shell` | `unknown`.""" + name = (tool or '').lower() + if name in ('powershell', 'pwsh'): + return 'powershell' + if name == 'bash': + return 'bash' + if name == 'sh' or 'shell' in name: + return 'shell' + return TOOL_UNKNOWN + +def is_guard_crash(output: str) -> bool: + """True when a guard's exit-1 output is empty or shows an interpreter crash, not a refusal.""" + lowered = output.lower() + return not lowered or any((marker in lowered for marker in _CRASH_MARKERS)) + +def split_command(command: str) -> tuple[list[str], bool]: + """POSIX `shlex.split` argv, else a whitespace split; the flag is True on the fallback.""" + try: + return (_chock_shlex.split(command), False) + except ValueError: + return (command.split(), True) -def run_guard(guard: _chock_Path, command: str) -> str: +def run_guard(guard: _chock_Path, command: str, tool: str='') -> str: """`GUARD_BLOCKED` / `GUARD_ASKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" - return run_guard_detailed(guard, command)[0] + return run_guard_detailed(guard, command, tool)[0] + +def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: + """The ask a guard earns when no shell could run it: the budget ran out probing, or none can see it.""" + if engine_remaining(deadline) <= 0: + print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s finding a shell, not checked', file=sys.stderr) + return (GUARD_ERRORED, '') + reason = f'no usable bash was found to run {guard.name}; on Windows install Git for Windows (it ships bash), elsewhere put bash on PATH' + print(f'chock: {reason}', file=sys.stderr) + return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str) -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: 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, '') + args, fallback = split_command(command) if not args: return (GUARD_UNCHECKED, '') - interpreter = find_interpreter(guard) + deadline = engine_deadline() + interpreter = find_interpreter(guard, deadline) if interpreter is None: - print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) - return (GUARD_UNCHECKED, '') + return _no_interpreter(guard, deadline) try: - env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - 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) + env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} + if fallback: + env[ARGV_FALLBACK_ENV] = '1' + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: - print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) + print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) return (GUARD_ERRORED, '') except (OSError, UnicodeError) as exc: print(f'chock: guard could not run, not checked: {exc}', file=sys.stderr) @@ -779,10 +895,11 @@ def run_guard_detailed(guard: _chock_Path, command: str) -> tuple[str, str]: output = ((proc.stderr or '') + (proc.stdout or '')).strip() first_line = output.splitlines()[0].strip() if output else '' if proc.returncode == GUARD_VIOLATION: + if is_guard_crash(output): + print(f'chock: guard {_chock_Path(guard).name} exited 1 without a reason (crash?), not checked', file=sys.stderr) + return (GUARD_ERRORED, '') sys.stderr.write(proc.stdout or '') sys.stderr.write(proc.stderr or '') - if not output: - print(f'chock: blocked by {_chock_Path(guard).name} (guard gave no reason)', file=sys.stderr) return (GUARD_BLOCKED, first_line) if proc.returncode == GUARD_ASK_EXIT: sys.stderr.write(proc.stdout or '') @@ -793,6 +910,17 @@ def run_guard_detailed(guard: _chock_Path, command: str) -> tuple[str, str]: return (GUARD_ERRORED, '') return (GUARD_CLEAN, '') +def append_gate_log(chock_root: _chock_Path, record: dict) -> None: + """Append one record, stamped, to `/log/gate-events.jsonl`, rotating past the size cap.""" + log_dir = chock_root / 'log' + log_dir.mkdir(parents=True, exist_ok=True) + 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') + stamped = {'ts': _chock_datetime.now(_chock_timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'), **record} + with log_path.open('a', encoding='utf-8') as fh: + fh.write(json.dumps(stamped, ensure_ascii=False) + '\n') + def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: """Append one outcome record. Best effort: never raises, never changes the verdict.""" try: @@ -808,23 +936,19 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: break if artifact_root is None: return - log_dir = artifact_root / 'log' - log_dir.mkdir(parents=True, exist_ok=True) - 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': verdict} - with log_path.open('a', encoding='utf-8') as fh: - fh.write(json.dumps(record, ensure_ascii=False) + '\n') + record = {'policy_id': guard.parent.parent.name, 'surface': 'pre-tool-use', 'event': 'tool_use', 'kind': guard.stem, 'tool': tool, 'verdict': verdict} + append_gate_log(artifact_root, record) except Exception: return def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) - if guard is None or not guard.exists(): + if guard is None: return None - verdict, message = run_guard_detailed(guard, command) + if not guard.exists(): + return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") + verdict, message = run_guard_detailed(guard, command, tool) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -833,28 +957,766 @@ def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | N 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.") + why = message or f"the {guard.stem} guard did not complete (see this hook's stderr)" + return (VERDICT_ESCALATE, f'chock could not check this command against {guard.stem}: {why}. Approving runs it unchecked.') + return None + +_EDIT_KEYS = (('old_string', 'new_string'), ('oldString', 'newString'), ('old_str', 'new_str')) + +_EDIT_LIST = 'edits' + +_CRLF = '\r\n' + +def _edit_call_input(event): + raw = getattr(event, 'raw', None) + tool_input = raw.get('tool_input') if isinstance(raw, dict) else None + if isinstance(tool_input, str) and tool_input[:1] == '{': + try: + tool_input = json.loads(tool_input) + except ValueError: + return {} + return tool_input if isinstance(tool_input, dict) else {} + +def _pair(item): + if not isinstance(item, dict): + return None + for old, new in _EDIT_KEYS: + if isinstance(item.get(old), str) and isinstance(item.get(new), str): + return (item[old], item[new], item.get('replace_all') is True) + return None + +def edit_replacements(event): + """The (old, new, replace_all) replacements an edit call applies, in order; None if not an edit.""" + tool_input = _edit_call_input(event) + listed = tool_input.get(_EDIT_LIST) + found = [_pair(item) for item in (listed if isinstance(listed, list) else [tool_input])] + if not found or None in found: + return None + return found + +def _replace(text, old, new, every): + """`text` with the replacement applied, or None when `old` is not there to replace. + + The file is read with universal newlines, so a CRLF file (a Windows checkout) arrives as + LF; the call's strings are brought to LF too, whichever ending the client sent them with, + rather than calling a real edit a failure over line endings. + """ + old, new = (old.replace(_CRLF, '\n'), new.replace(_CRLF, '\n')) + if old == '': + return new if text == '' else None + if old not in text: + return None + return text.replace(old, new) if every else text.replace(old, new, 1) + +def edited_text(event, root=None): + """The file an edit call would leave behind, or None when it is not an edit or cannot be rebuilt.""" + replacements = edit_replacements(event) + path = getattr(event, 'path', None) + if replacements is None or not path: + return None + target = _chock_Path(path) + if not target.is_absolute(): + target = _chock_Path(root if root is not None else _chock_Path.cwd()) / target + try: + text = target.read_text(encoding='utf-8') if target.exists() else '' + except (OSError, UnicodeDecodeError): + return None + for old, new, every in replacements: + text = _replace(text, old, new, every) + if text is None: + return None + return text + +def added_from_event(event): + """What an edit call introduces -- what `added_lines` means for it -- or {} for any other call.""" + replacements = edit_replacements(event) + path = getattr(event, 'path', None) + if replacements is None or not path: + return {} + return {str(path): '\n'.join((new for _, new, _ in replacements))} + +_BEGIN = '*** Begin Patch' + +_END = '*** End Patch' + +_ADD = '*** Add File: ' + +_DELETE = '*** Delete File: ' + +_UPDATE = '*** Update File: ' + +_MOVE = '*** Move to: ' + +_EOF = '*** End of File' + +_HUNK = '@@' + +_OPS = (('add', _ADD), ('delete', _DELETE), ('update', _UPDATE)) + +def patch_text(event): + """The patch an apply_patch call carries, from its first to its last marker; None otherwise.""" + raw = getattr(event, 'raw', None) + tool_input = raw.get('tool_input') if isinstance(raw, dict) else None + command = tool_input.get('command') if isinstance(tool_input, dict) else None + if isinstance(command, list): + command = '\n'.join((part for part in command if isinstance(part, str))) + start = command.find(_BEGIN) if isinstance(command, str) else -1 + end = command.find(_END, start) if start >= 0 else -1 + return command[start:end + len(_END)] if end >= 0 else None + +def _body_line(current, line): + """Add one line under the file operation `current` is reading.""" + if line.startswith(_MOVE): + current[2] = line[len(_MOVE):].strip() + elif line.startswith(_HUNK): + current[3].append([line[len(_HUNK):].strip(), [], False]) + elif line.startswith(_EOF): + if current[3]: + current[3][-1][2] = True + elif current[0] == 'add': + current[3].append(line[1:] if line.startswith('+') else line) + else: + if not current[3]: + current[3].append(['', [], False]) + current[3][-1][1].append(line if line[:1] in (' ', '-', '+') else ' ' + line) + +def parse_patch(patch): + """[(op, path, moved_to, hunks)]: op is add/delete/update; a hunk is (anchor, lines, at_eof).""" + operations = [] + for line in patch.replace('\r\n', '\n').split('\n')[1:]: + if line.startswith(_END): + break + opened = next(((op, marker) for op, marker in _OPS if line.startswith(marker)), None) + if opened is not None: + operations.append([opened[0], line[len(opened[1]):].strip(), None, []]) + elif operations: + _body_line(operations[-1], line) + return [tuple(operation) for operation in operations] + +def _find(lines, wanted, start): + """Where `wanted` starts in `lines` at or after `start`, exactly or ignoring trailing space.""" + for loose in (False, True): + for index in range(start, len(lines) - len(wanted) + 1): + window = lines[index:index + len(wanted)] + if window == wanted or (loose and [w.rstrip() for w in window] == [w.rstrip() for w in wanted]): + return index + return None + +def apply_hunks(text, hunks): + """`text` with the hunks applied in order, or None when one cannot be placed.""" + lines = text.split('\n') + cursor = 0 + for anchor, body, at_eof in hunks: + if anchor: + found = _find(lines, [anchor], cursor) + if found is None: + found = next((i for i in range(cursor, len(lines)) if anchor.strip() in lines[i]), None) + if found is None: + return None + cursor = found + 1 + old = [line[1:] for line in body if line[:1] in (' ', '-')] + new = [line[1:] for line in body if line[:1] in (' ', '+')] + if not old: + at = len(lines) - (1 if lines and lines[-1] == '' else 0) if at_eof else cursor + else: + at = _find(lines, old, cursor) + if at is None: + return None + lines[at:at + len(old)] = new + cursor = at + len(new) + return '\n'.join(lines) + +def _read(root, path): + target = _chock_Path(path) + if not target.is_absolute(): + target = _chock_Path(root if root is not None else _chock_Path.cwd()) / target + try: + return target.read_text(encoding='utf-8') + except (OSError, UnicodeDecodeError): + return None + +def patched_files(event, root=None): + """{path: text} for every file an apply_patch call adds or updates and can be rebuilt.""" + patch = patch_text(event) + if patch is None: + return {} + files = {} + for op, path, moved_to, body in parse_patch(patch): + if op == 'add': + files[path] = '\n'.join(body) + '\n' + elif op == 'update': + before = _read(root, path) + after = apply_hunks(before, body) if before is not None else None + if after is not None: + files[moved_to or path] = after + return files + +def patch_added(event): + """{path: the lines the patch adds}, which is what `added_lines` means for it.""" + patch = patch_text(event) + if patch is None: + return {} + added = {} + for op, path, moved_to, body in parse_patch(patch): + if op == 'add': + added[path] = '\n'.join(body) + elif op == 'update': + added[moved_to or path] = '\n'.join((line[1:] for _, hunk, _ in body for line in hunk if line[:1] == '+')) + return added + +GATE_BLOCKED = 'blocked' + +GATE_CLEAN = 'clean' + +GATE_ERRORED = 'errored' + +GATE_ASKED = 'asked' + +GATE_WARNED = 'warned' + +VERDICT_WARN = 'warn' + +_EXIT_OUTCOME = {0: GATE_CLEAN, 1: GATE_BLOCKED, 3: GATE_ASKED, 4: GATE_WARNED} + +def runner_outcome(returncode, stderr): + """(outcome, message) for the runner's exit code and stderr: its words, only when it refused or spoke.""" + outcome = _EXIT_OUTCOME.get(returncode, GATE_ERRORED) + return (outcome, '' if outcome == GATE_CLEAN else (stderr or '').strip()) + +def gate_decision(outcome, message, gate): + """The decision an outcome earns, or None when the gate allowed: deny, escalate (ask) or warn.""" + policy = gate.parent.parent.name + spoken = {GATE_BLOCKED: (VERDICT_DENY, message or f'Blocked by chock policy: {policy}'), GATE_ASKED: (VERDICT_ESCALATE, message or f'Chock policy {policy} asks before this write.'), GATE_WARNED: (VERDICT_WARN, message or f'Chock policy {policy} warns about this write.'), GATE_ERRORED: (VERDICT_DENY, f'chock could not check this write: {message}. Refusing rather than reporting an allow it never established.')} + return spoken.get(outcome) + +OUTSIDE_KEY = 'outside_repo' + +_UP = '..' + +_DRIVE_AT = 1 + +_SEPARATORS = ('/', '\\') + +def _slashed(path): + return str(path).replace('\\', '/') + +def is_outside(path): + """Whether a path as the runtime names it lies outside the repository: absolute, or climbing out.""" + text = _slashed(path) + return text.startswith(('/', _UP + '/')) or text[_DRIVE_AT:_DRIVE_AT + 1] == ':' or text == _UP + +def expand_home(pattern): + """`~` (alone or before a separator) as this machine's home directory; anything else as given.""" + text = str(pattern) + if text[:1] == '~' and (len(text) == 1 or text[1] in _SEPARATORS): + return _chock_os.path.expanduser('~') + text[1:] + return text + +def _outside_fold(path): + """`path` with `.` and `..` folded lexically and doubled slashes dropped; a root's leading slash kept.""" + parts = [] + for part in _slashed(path).split('/'): + if part == _UP and parts and (parts[-1] not in ('', _UP)): + parts.pop() + elif part != '.' and (part or not parts): + parts.append(part) + return '/'.join(parts) + +def declared_outside(path, root, patterns, *, windows=None): + """The absolute, slash-separated path when it is outside `root` and matches a declared glob; else None.""" + windows = _chock_os.name == 'nt' if windows is None else windows + text = _slashed(path) + if not is_outside(text) or not patterns: + return None + absolute = text if not text.startswith(_UP) else _slashed(root) + '/' + text + absolute = _outside_fold(absolute) + subject = absolute.lower() if windows else absolute + for pattern in patterns: + wanted = _outside_fold(expand_home(pattern)) + if _chock_fnmatch.fnmatchcase(subject, wanted.lower() if windows else wanted): + return absolute + return None + +def outside_globs(gate): + """The `outside_repo` globs the compiled gate declares; none when it declares none or cannot be read.""" + try: + declared = json.loads(gate.read_text(encoding='utf-8')).get(OUTSIDE_KEY) + except (OSError, ValueError, AttributeError): + return [] + return [g for g in declared if isinstance(g, str)] if isinstance(declared, list) else [] + +def judged_files(files, root, patterns, repo_names): + """`files` ({path: text}) keyed the way a gate judges them. + + A path inside the repository is named by `repo_names(path)`; one outside is kept, under its + absolute path, only when a declared glob matches it. + """ + kept = {} + for path, text in files.items(): + for name in repo_names(path): + if not is_outside(name): + kept[name] = text + continue + found = declared_outside(name, root, patterns) + if found is not None: + kept[found] = text + return kept + +SESSION_MAX_ENTRIES = 500 + +SESSION_PRUNE_SECONDS = 7 * 24 * 3600 + +SESSION_FIELD_CAP = 300 + +SESSION_TRIM_BYTES = 131072 + +SESSION_DEDUPE_TAIL = 8 + +SESSION_STATE_PARTS = ('.chock', 'state') + +SESSION_INPUT_KEYS = ('tool_input', 'toolInput', 'toolArgs', 'input') + +SESSION_PATH_KEYS = ('file_path', 'filePath', 'path', 'notebook_path', 'target_file', 'TargetFile') + +SESSION_URL_KEYS = ('url', 'uri') + +SESSION_RESPONSE_KEYS = ('tool_response', 'toolResponse', 'tool_result', 'toolResult') + +SESSION_UNKNOWN_ID = 'unknown' + +SESSION_ID_MAX = 128 + +SESSION_MASK = '***' + +def tool_call_input(event): + """The tool's arguments as a dict, decoding the JSON-string form some vendors send; {} when absent.""" + raw = getattr(event, 'raw', None) + if not isinstance(raw, dict): + return {} + for key in SESSION_INPUT_KEYS: + value = raw.get(key) + if isinstance(value, str) and value[:1] == '{': + try: + value = json.loads(value) + except ValueError: + continue + if isinstance(value, dict): + return value + return {} + +def session_id_of(event): + """The session id as a safe file stem: separators and leading dots removed, never empty.""" + raw = getattr(event, 'raw', None) or {} + given = getattr(event, 'session_id', None) or (raw.get('session_id') if isinstance(raw, dict) else None) + stem = ''.join((c for c in str(given or '') if c.isalnum() or c in '._-')).lstrip('.') + return stem[:SESSION_ID_MAX] or SESSION_UNKNOWN_ID + +def session_log_path(root, session_id): + return _chock_Path(root).joinpath(*SESSION_STATE_PARTS, session_id + '.jsonl') + +def session_for(event, root): + """What a script gate receives as `session`: the id, the log's absolute path, this call's id.""" + session_id = session_id_of(event) + return {'id': session_id, 'log_path': session_log_path(root, session_id).as_posix(), 'tool_use_id': getattr(event, 'tool_use_id', None)} + +def _mask_assignments(command): + """`command` with the value of every `NAME=value` word masked: an env prefix is a secret's usual home.""" + words = [] + for word in command.split(' '): + name, eq, _ = word.partition('=') + words.append(name + eq + SESSION_MASK if eq and name.isidentifier() else word) + return ' '.join(words) + +def _bare_url(url): + """`url` without userinfo, query and fragment: where a token or a secret is usually carried.""" + url = url.split('#', 1)[0].split('?', 1)[0] + scheme, sep, rest = url.partition('://') + if not sep: + return url + authority, slash, tail = rest.partition('/') + return scheme + sep + authority.rsplit('@', 1)[-1] + slash + tail + +def _first_str(mapping, keys): + for key in keys: + value = mapping.get(key) + if isinstance(value, str) and value: + return value + return None + +def session_summary(event): + """The input of a tool call as {path?, command?, url?}: what a policy asks about, nothing else.""" + tool_input = tool_call_input(event) + path = getattr(event, 'path', None) or _first_str(tool_input, SESSION_PATH_KEYS) + command = getattr(event, 'command', None) or _first_str(tool_input, ('command',)) + url = _first_str(tool_input, SESSION_URL_KEYS) + summary = {} + if isinstance(path, str) and path: + summary['path'] = path[:SESSION_FIELD_CAP] + if isinstance(command, str) and command: + summary['command'] = _mask_assignments(command.splitlines()[0] if command.strip() else '')[:SESSION_FIELD_CAP] + if url: + summary['url'] = _bare_url(url)[:SESSION_FIELD_CAP] + return summary + +def _response_failed(response): + if not isinstance(response, dict): + return False + if response.get('is_error') is True or response.get('isError') is True or response.get('success') is False: + return True + code = response.get('exit_code', response.get('exitCode')) + return bool(response.get('error')) or (isinstance(code, int) and code != 0) + +def session_outcome(event): + """`ok` or `error` for a finished call, from what the payload says; None before it ran.""" + if event.event == 'tool_failure': + return 'error' + if event.event != 'post_tool': + return None + raw = getattr(event, 'raw', None) or {} + failed = any((_response_failed(raw.get(key)) for key in SESSION_RESPONSE_KEYS)) if isinstance(raw, dict) else False + return 'error' if failed else 'ok' + +def _same_call(line, record): + """Whether `line` is `record` already logged: same call id, or, with none, the same second and input.""" + try: + seen = json.loads(line) + except ValueError: + return False + if seen.get('phase') != record['phase']: + return False + if record['tool_use_id']: + return seen.get('tool_use_id') == record['tool_use_id'] + return not seen.get('tool_use_id') and all((seen.get(key) == record[key] for key in ('ts', 'tool', 'input', 'outcome'))) + +def _prune_old(directory): + cutoff = _chock_datetime.now(_chock_timezone.utc).timestamp() - SESSION_PRUNE_SECONDS + for old in directory.glob('*.jsonl'): + try: + if old.stat().st_mtime < cutoff: + old.unlink() + except OSError: + continue + +def _append_bounded(path, line, record): + lines = path.read_text(encoding='utf-8', errors='replace').splitlines() if path.exists() else [] + if any((_same_call(seen, record) for seen in lines[-SESSION_DEDUPE_TAIL:])): + return + if len(lines) < SESSION_MAX_ENTRIES and path.exists() and (path.stat().st_size < SESSION_TRIM_BYTES): + with path.open('a', encoding='utf-8') as fh: + fh.write(line + '\n') + return + kept = [*lines, line][-SESSION_MAX_ENTRIES:] + scratch = path.with_suffix('.tmp') + scratch.write_text('\n'.join(kept) + '\n', encoding='utf-8') + scratch.replace(path) + +def session_record(root, event, phase, outcome=None): + """Append one record for this call to its session's log. Best effort: never raises.""" + try: + path = session_log_path(root, session_id_of(event)) + path.parent.mkdir(parents=True, exist_ok=True) + record = {'ts': _chock_datetime.now(_chock_timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'), 'session_id': path.stem, 'phase': phase, 'tool': str(getattr(event, 'tool', None) or ''), 'input': session_summary(event), 'outcome': outcome, 'tool_use_id': getattr(event, 'tool_use_id', None)} + _append_bounded(path, json.dumps(record, ensure_ascii=False, sort_keys=True), record) + _prune_old(path.parent) + except Exception: + return + +REENTRY_CAP = 3 + +STOP_LEDGER_SUFFIX = '.stop.jsonl' + +GATE_LOG_PARTS = ('.chock', 'log', 'gate-events.jsonl') + +STOP_MODE_MASK = 61440 + +STOP_MODE_FILE = 32768 + +STOP_MODE_DIR = 16384 + +STOP_PHASE = 'stop' + +STOP_TURN_KEYS = ('turn_id', 'generation_id') + +STOP_COUNT_KEY = 'loop_count' + +STOP_DIGEST_CHARS = 16 + +REENTRY_CLEAN = 'clean' + +REENTRY_REFUSED = 'refused' + +REENTRY_CAPPED = 'cap-reached' + +REENTRY_UNTRACKED = 'untracked' + +STOP_LOG_BLOCK = 'block' + +STOP_LOG_ALLOW = 'allow' + +STOP_LOG_WARN = 'warn' + +_STOP_LAST = "\nThis is refusal {cap} of {cap} for this turn's end: the next stop ends the turn with these findings still on disk, and a commit will refuse them. A person must review them." + +_STOP_WARNING = 'chock {policy}: this turn ended with findings still on disk{why}. The turn was allowed to end so the client does not loop; a commit will refuse these findings, and a person must review them.\n' + +_STOP_WHY = {REENTRY_CAPPED: ' after {cap} refused stops', REENTRY_UNTRACKED: ' (the stop record under .chock/state could not be written, so re-entries cannot be counted)'} + +_STOP_SINCE = {True: ' (changed since the last refusal)', False: ' (unchanged since the last refusal)'} + +def reentries(event): + """How many times the client says this stop re-entered its hook: `loop_count`, else 1 for `stop_hook_active`.""" + raw = event.raw if isinstance(event.raw, dict) else {} + try: + counted = int(raw.get(STOP_COUNT_KEY) or 0) + except (TypeError, ValueError): + counted = 1 + return max(counted, 1 if raw.get('stop_hook_active') else 0) + +def _stop_turn(event): + """The vendor's turn id, when the payload carries one; None otherwise.""" + raw = event.raw if isinstance(event.raw, dict) else {} + for key in STOP_TURN_KEYS: + value = raw.get(key) + if isinstance(value, str) and value: + return value[:SESSION_ID_MAX] + return None + +def _stop_digest(*parts): + digest = _chock_hashlib.sha256() + for part in parts: + digest.update(str(part).encode('utf-8', 'replace') + b'\x00') + return digest.hexdigest()[:STOP_DIGEST_CHARS] + +def finding_digests(message, writes): + """One digest per flagged file, of its path and content: never the text itself, never the message's wording.""" + if not message: + return [] + lines = [line.strip().lstrip('-').strip() for line in message.splitlines()] + flagged = [p for p in writes if any((line.startswith(p + ':') for line in lines))] or list(writes) + if not flagged: + return [_stop_digest(message)] + return sorted({_stop_digest(p, writes[p]) for p in flagged}) + +def stop_ledger_path(root, session_id): + return _chock_Path(root).joinpath(*SESSION_STATE_PARTS, session_id + STOP_LEDGER_SUFFIX) + +def stop_ledger_safe(path): + """Whether `path`, its directory and `.chock` are what chock made: real directories and a regular file or none. + + lstat, so a symlink, FIFO, socket, device or directory in any place is refused before anything opens it: + opening a FIFO blocks the hook until its timeout, which a client may read as an allow. + """ + for target, kind in ((path, STOP_MODE_FILE), (path.parent, STOP_MODE_DIR), (path.parent.parent, STOP_MODE_DIR)): + try: + mode = _chock_os.lstat(target).st_mode + except FileNotFoundError: + continue + except OSError: + return False + if mode & STOP_MODE_MASK != kind: + return False + return True + +def _stop_lines(path): + """The ledger's lines; none when it is missing, unsafe or cannot be read.""" + if not stop_ledger_safe(path): + return [] + try: + fd = _chock_os.open(path, _chock_os.O_RDONLY | getattr(_chock_os, 'O_NOFOLLOW', 0) | getattr(_chock_os, 'O_NONBLOCK', 0)) + with _chock_os.fdopen(fd, encoding='utf-8', errors='replace') as fh: + return fh.read().splitlines() + except OSError: + return [] + +def _stop_parse(line): + try: + return json.loads(line) + except (ValueError, RecursionError): + return None + +def _stop_well_formed(seen, session_id): + stamp, index = (seen.get('ts'), seen.get('reentry')) + return isinstance(stamp, str) and bool(stamp) and (seen.get('session_id') == session_id) and (type(index) is int) and (index >= 0) + +def _stop_chain(lines, session_id, policy, turn): + """This turn's records, newest first, counting down without a gap to its first stop; None when broken.""" + chain = [] + for line in reversed(lines): + seen = _stop_parse(line) + if not isinstance(seen, dict) or seen.get('phase') != STOP_PHASE or seen.get('policy') != policy: + continue + if turn is not None and seen.get('turn') != turn: + continue + if not _stop_well_formed(seen, session_id): + return None + if chain and seen['reentry'] != chain[-1]['reentry'] - 1: + return None + chain.append(seen) + if seen['reentry'] == 0 or seen.get('anchor') is True: + return chain return None +def _stop_append(path, lines, record): + """Append one record, keeping the last `SESSION_MAX_ENTRIES`; whether it landed.""" + line = json.dumps(record, sort_keys=True) + if not stop_ledger_safe(path): + return False + try: + path.parent.mkdir(parents=True, exist_ok=True) + if len(lines) < SESSION_MAX_ENTRIES and (not path.exists() or path.stat().st_size < SESSION_TRIM_BYTES): + flags = _chock_os.O_WRONLY | _chock_os.O_APPEND | _chock_os.O_CREAT | getattr(_chock_os, 'O_NOFOLLOW', 0) | getattr(_chock_os, 'O_NONBLOCK', 0) + with _chock_os.fdopen(_chock_os.open(path, flags, 384), 'a', encoding='utf-8') as fh: + fh.write(line + '\n') + else: + scratch = path.with_name('%s.%d.tmp' % (path.name, _chock_os.getpid())) + scratch.write_text('\n'.join([*lines[-(SESSION_MAX_ENTRIES - 1):], line]) + '\n', encoding='utf-8') + scratch.replace(path) + _prune_old(path.parent) + except OSError: + return False + return True + +def _stop_gate_log(root, policy, verdict, reentry, findings): + """One gate-log record per re-entry; a warn is held (`would_block`), so it is kept even with the log off. + + `reentry` is (index, reentry_verdict, findings_changed or None). + """ + try: + if not stop_ledger_safe(_chock_Path(root).joinpath(*GATE_LOG_PARTS)): + return + held = verdict == STOP_LOG_WARN + if _chock_os.environ.get(GATE_LOG_ENV) == '0' and (not held): + return + index, reentry_verdict, changed = reentry + record = {'policy_id': policy, 'surface': 'stop-reentry', 'event': 'stop', 'kind': 'reentry', 'reentry': index, 'verdict': verdict, 'reentry_verdict': reentry_verdict, 'match_count': len(findings), 'matches': []} + if held: + record.update({'would_action': STOP_LOG_BLOCK, 'would_block': True}) + if changed is not None: + record['findings_changed'] = changed + append_gate_log(_chock_Path(root).joinpath(SESSION_STATE_PARTS[0]), record) + except Exception: + return + +def _stop_warn(policy, why, changed, message): + """The warning a turn ends with: said on stderr for every client, and returned for a `systemMessage`.""" + text = _STOP_WARNING.format(policy=policy, why=why.format(cap=REENTRY_CAP) + _STOP_SINCE.get(changed, '')) + text += message + with contextlib.suppress(Exception): + sys.stderr.write(text + '\n') + sys.stderr.flush() + return (VERDICT_WARN, text) + +def _stop_changed(chain, findings): + """Whether the findings differ from the last refusal's; None when there is no refusal to compare with.""" + refused = next((r for r in chain or () if r.get('verdict') == STOP_LOG_BLOCK), None) + if refused is None or not isinstance(refused.get('findings'), list): + return None + return sorted(refused['findings']) != findings + +def settle_stop(event, root, gate, decision, writes=None): + """The decision a Stop earns once this turn's earlier Stops are weighed against `decision`.""" + if decision is not None and decision[0] != VERDICT_DENY: + return decision + policy = gate.parent.parent.name + session_id = session_id_of(event) + path = stop_ledger_path(root, session_id) + lines = _stop_lines(path) + message = decision[1] if decision else '' + findings = finding_digests(message, writes or {}) + record = {'ts': _chock_datetime.now(_chock_timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'), 'session_id': session_id, 'phase': STOP_PHASE, 'policy': policy, 'turn': _stop_turn(event), 'findings': findings} + signalled = reentries(event) + if not signalled: + _stop_append(path, lines, {**record, 'reentry': 0, 'verdict': STOP_LOG_BLOCK if decision else STOP_LOG_ALLOW}) + return decision + chain = _stop_chain(lines, session_id, policy, record['turn']) + counted = isinstance(event.raw, dict) and STOP_COUNT_KEY in event.raw + index = signalled if counted else chain[0]['reentry'] + 1 if chain else 1 + capped = index > REENTRY_CAP + verdict = STOP_LOG_ALLOW if not decision else STOP_LOG_WARN if capped else STOP_LOG_BLOCK + landed = _stop_append(path, lines, {**record, 'reentry': index, 'verdict': verdict, 'anchor': chain is None}) + changed = _stop_changed(chain, findings) + if decision is None: + _stop_gate_log(root, policy, verdict, (index, REENTRY_CLEAN, None), findings) + return None + if capped: + _stop_gate_log(root, policy, verdict, (index, REENTRY_CAPPED, changed), findings) + return _stop_warn(policy, _STOP_WHY[REENTRY_CAPPED], changed, message) + if not landed and (not counted): + _stop_gate_log(root, policy, STOP_LOG_WARN, (index, REENTRY_UNTRACKED, None), findings) + return _stop_warn(policy, _STOP_WHY[REENTRY_UNTRACKED], None, message) + tracked = REENTRY_REFUSED if chain and landed else REENTRY_UNTRACKED + _stop_gate_log(root, policy, verdict, (index, tracked, changed), findings) + return (VERDICT_DENY, message + (_STOP_LAST.format(cap=REENTRY_CAP) if index == REENTRY_CAP else '')) + +UNREADABLE_STOP_LEDGER = 'unreadable-stop.jsonl' + +UNREADABLE_STOP_PHASE = 'unreadable-stop' + +UNREADABLE_STOP_WINDOW_SECONDS = 600 + +REENTRY_UNREADABLE = 'unreadable' + +_UNREADABLE_STOP_WARNING = 'chock {policy}: {cap} stops in a row had a payload chock could not read, so none of them was judged. The turn was allowed to end so the client does not loop; a commit will judge what is on disk, and a person must look at why the payload is unreadable.\n' + +_UNREADABLE_STOP_LAST = '\nThis is refusal {cap} of {cap} for stops chock cannot read: the next one ends the turn unchecked, and a commit will judge what is on disk. A person must look at why the payload is unreadable.' + +def _unreadable_stop_recent(lines, now): + """The refusals recorded inside the window; a record that does not parse, or has no time, counts as none.""" + kept = [] + for line in lines: + seen = _stop_parse(line) + stamp = seen.get('at') if isinstance(seen, dict) else None + refused = isinstance(seen, dict) and seen.get('phase') == UNREADABLE_STOP_PHASE + if refused and seen.get('verdict') == STOP_LOG_BLOCK and (type(stamp) in (int, float)) and (0 <= now - stamp < UNREADABLE_STOP_WINDOW_SECONDS): + kept.append(stamp) + return kept + +def settle_unreadable_stop(root, policy, refusal): + """(verdict, text) this unreadable Stop earns: a refusal with `refusal`, or a warning once the cap is spent. + + A ledger that is not a regular file in real directories, or cannot be written, keeps refusing: with nothing to count, only a refusal is safe. + """ + path = _chock_Path(root).joinpath(*SESSION_STATE_PARTS, UNREADABLE_STOP_LEDGER) + if not stop_ledger_safe(path): + return (VERDICT_DENY, refusal) + lines = _stop_lines(path) + now = _chock_datetime.now(_chock_timezone.utc).timestamp() + index = len(_unreadable_stop_recent(lines, now)) + 1 + capped = index > REENTRY_CAP + verdict = STOP_LOG_WARN if capped else STOP_LOG_BLOCK + _stop_append(path, lines, {'phase': UNREADABLE_STOP_PHASE, 'verdict': verdict, 'at': now, 'reentry': index}) + _stop_gate_log(root, policy, verdict, (index, REENTRY_UNREADABLE, None), []) + if capped: + text = _UNREADABLE_STOP_WARNING.format(policy=policy, cap=REENTRY_CAP) + with contextlib.suppress(Exception): + sys.stderr.write(text) + sys.stderr.flush() + return (VERDICT_WARN, text) + return (VERDICT_DENY, refusal + (_UNREADABLE_STOP_LAST.format(cap=REENTRY_CAP) if index == REENTRY_CAP else '')) + GATE_FLAG = '--gate' -_GATE_TIMEOUT_SECONDS = 30 +STOP_FLAG = '--stop' _GATE_DEPTH_TO_CHOCK = 3 _RUNNER_PARTS = ('bin', 'gate.py') +_PACKAGED_RUNNER = 'gate.py' + _GIT = 'git' -_DELETED = 'D' +_UTF8 = 'utf-8' -_RENAMED = 'R' +_PATH_ERRORS = 'surrogateescape' -GATE_BLOCKED = 'blocked' +_PARENT = '..' -GATE_CLEAN = 'clean' +_DRIVE_COLON = ':' -GATE_ERRORED = 'errored' +_DELETED = 'D' + +_RENAMED = 'R' VERDICT_DENY = 'deny' @@ -867,72 +1729,145 @@ def gate_path_from_argv(argv): return None def runner_for(gate): - """The vendored gate runner beside this compiled gate, or None when it is not there.""" + """The gate runner: beside the gate in a plugin, under .chock/bin in a repository, else None.""" + packaged = gate.resolve().parent / _PACKAGED_RUNNER + if packaged.exists(): + return packaged 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.""" +def writes_from_event(event, root=None): + """The one file this tool call would write, or none when it carries no file text. + + An edit call carries only the text it inserts; judged alone, that fragment has no imports, + no class and no neighbours, so a gate that reads a whole file (a script kind) finds nothing + in it. An edit is therefore judged as the file it would leave: the file on disk with the + call's replacements applied. When that cannot be rebuilt -- the file is unreadable, or the + text to replace is not in it, so the call itself will fail -- the fragment is judged as + before and the turn's end judges what actually landed. A Codex apply_patch call names its + files only inside the patch; each one it adds or updates is judged as the patch leaves it. + """ + patched = patched_files(event, root) + if patched: + return patched path = getattr(event, 'path', None) content = getattr(event, 'content', None) + edited = edited_text(event, root) + if path and edited is not None: + return {str(path): edited} 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.""" +def _folded(pure): + """`pure` with `..` folded lexically; an absolute path never climbs above its anchor.""" + parts = [] + for part in pure.parts: + if part != _PARENT: + parts.append(part) + elif len(parts) > 1: + parts.pop() + return type(pure)(*parts) + +def repo_relative(path, root): + """`path` relative to `root` in POSIX form, as scope globs are written; as given when outside `root`.""" + text = str(path) + if root is None: + return text + windows = _chock_os.name == 'nt' or str(root)[1:2] == _DRIVE_COLON + flavour = _chock_PureWindowsPath if windows else _chock_PurePosixPath 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] + return _folded(flavour(str(root)) / text).relative_to(flavour(str(root))).as_posix() + except ValueError: + pass + if windows != (_chock_os.name == 'nt'): + return text + try: + return _chock_Path(root, text).resolve().relative_to(_chock_Path(root).resolve()).as_posix() + except (OSError, ValueError, RuntimeError): + return text + +def repo_paths(path, root): + """Every repo-relative name a write is judged under: as written, and through symlinks and case.""" + lexical = repo_relative(path, root) + if root is None or (str(root)[1:2] == _DRIVE_COLON) != (_chock_os.name == 'nt'): + return (lexical,) + try: + resolved = _chock_Path(root, str(path)).resolve().relative_to(_chock_Path(root).resolve()).as_posix() + except (OSError, ValueError, RuntimeError): + return (lexical,) + return tuple(dict.fromkeys((lexical, resolved))) + +def changed_paths(repo_root, deadline=None, why=None): + """Every uncommitted path; None (reason appended to `why`) when git gave no answer, [] only outside a repository.""" + deadline, why = (engine_deadline() if deadline is None else deadline, [] if why is None else why) + argv = [_GIT, '-C', str(repo_root), 'status', '--porcelain=v1', '--untracked-files=all', '-z'] + try: + timeout = time_left(deadline, argv) + proc = _chock_subprocess.run(argv, capture_output=True, text=True, encoding=_UTF8, errors=_PATH_ERRORS, timeout=timeout, check=False) + except _chock_subprocess.TimeoutExpired as exc: + why.append(f'git status did not finish within {allowed_seconds(exc)}') + except (OSError, _chock_subprocess.SubprocessError) as exc: + why.append(f'git status could not run ({exc})') + else: + first = ((proc.stderr or '').strip().splitlines() or [''])[0] + if proc.returncode and 'not a git repository' in first: + return [] + if proc.returncode: + why.append(f'git status failed (exit {proc.returncode}): {first}') + if why: + sys.stderr.write(f'chock: {why[-1]}, worktree not checked\n') + return None + fields = iter([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) + if status.startswith(_RENAMED): + next(fields, None) + if _DELETED not in status and path: + paths.append(path) return paths -def writes_from_worktree(repo_root): - """What this turn actually left on disk, however it was written. +def writes_from_worktree(repo_root, deadline=None, why=None): + """What this turn actually left on disk, however it was written; None when it could not be listed. 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. """ + paths = changed_paths(repo_root, deadline, why) + if paths is None: + return None writes = {} - for path in changed_paths(repo_root): + for path in paths: 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.""" +def run_gate(gate, writes, event, run_in, deadline): + """Ask the vendored runner. Returns (outcome, message) and never decides for itself. + + `run_in` is (root, extra): the working directory, and the stdin keys `extra` adds: `added` (per + edited path, only the text the edit introduces; a runner that predates it judges the whole file, + which only ever refuses more) and the script `session`. The runner gets what is left of `deadline`. + """ + root, extra = run_in runner = runner_for(gate) if runner is None: return (GATE_ERRORED, 'the vendored gate runner is not installed beside this gate') + argv = [sys.executable, str(runner), 'run', '--gate', str(gate), '--event', event] 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) + proc = _chock_subprocess.run(argv, input=json.dumps({'writes': writes, **(extra or {})}), capture_output=True, text=True, encoding=_UTF8, errors='replace', timeout=time_left(deadline, argv), check=False, cwd=str(root) if root is not None else None) + except _chock_subprocess.TimeoutExpired as exc: + return (GATE_ERRORED, f'the gate runner gave no verdict within {allowed_seconds(exc)}') 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()) + return runner_outcome(proc.returncode, proc.stderr) _EVENT_ARG = {'pre_tool': 'pre-tool-use', 'stop': 'stop'} @@ -943,33 +1878,182 @@ def root_for(gate): parents = gate.resolve().parents if len(parents) <= _GATE_DEPTH_TO_CHOCK: return None + if parents[_GATE_DEPTH_TO_CHOCK - 1].name != 'compiled' or parents[_GATE_DEPTH_TO_CHOCK].name != '.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 {} +def repo_root_for(event, gate): + """The repository under judgement: the compiled layout's root, else where the agent works.""" root = root_for(gate) - return writes_from_worktree(root) if root is not None else {} + if root is not None: + return root + cwd = getattr(event, 'cwd', None) + return _chock_Path(cwd) if cwd else _chock_Path.cwd() + +def writes_for(event, gate, deadline=None, why=None): + """What this event puts under judgement: the call's own text, or what the turn left behind (None: unlisted).""" + if event.event == PRE_TOOL: + return writes_from_event(event, repo_root_for(event, gate)) + return writes_from_worktree(repo_root_for(event, gate), deadline, why) + +def _missing_gate(gate): + """A gate the hook names but that is not on disk: a broken install, so a refusal that says so.""" + return (VERDICT_DENY, f'chock gate {gate} is missing, so this write cannot be checked. Run `chock sync --repo .` to rebuild the compiled gates.') + +def _gate_says(gate, event, name, deadline): + """(decision, judged files): what the compiled gate says about this event, before a re-entered stop is weighed.""" + if not gate.exists(): + return (_missing_gate(gate), {}) + root = repo_root_for(event, gate) + outside = outside_globs(gate) + why = [] + listed = writes_for(event, gate, deadline, why) + if listed is None: + return (gate_decision(GATE_ERRORED, why[-1] if why else 'git status gave no answer', gate), {}) + writes = judged_files(listed, root, outside, lambda path: repo_paths(path, root)) + if not writes: + return (None, writes) + added = {**patch_added(event), **added_from_event(event)} if event.event == PRE_TOOL else {} + added = judged_files(added, root, outside, lambda path: repo_paths(path, root)) + added = {path: text for path, text in added.items() if path in writes} + extra = {**({'added': added} if added else {}), 'session': session_for(event, root)} + outcome, message = run_gate(gate, writes, name, (root, extra), deadline) + return (gate_decision(outcome, message, gate), writes) 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: + if gate is None or name is None: 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.') + decision, judged = _gate_says(gate, event, name, engine_deadline()) + if event.event == PRE_TOOL: + return decision + return settle_stop(event, repo_root_for(event, gate), gate, decision, judged) + +TOOL_CALL_FLAG = '--tool-call' + +RECORD_FLAG = '--record' + +TOOL_CALL_EVENT = 'tool_call' + +_TOOL_CALL_PRE = 'pre_tool' + +_TOOL_CALL_POST = ('post_tool', 'tool_failure') + +_TOOL_CALL_SCRIPT_KIND = 'script' + +_TOOL_CALL_REGEX_KIND = 'content_regex' + +_TOOL_CALL_BLOCKED = 'blocked' + +_TOOL_CALL_NEEDS_SESSION = frozenset({_TOOL_CALL_SCRIPT_KIND}) + +_TOOL_CALL_EXIT_VERDICTS = {0: None, 1: 'deny', 3: 'escalate', 4: 'warn'} + +_TOOL_CALL_CEILING = {'block': 'deny', 'ask': 'escalate', 'warn': 'warn'} + +_TOOL_CALL_RANK = {'warn': 0, 'escalate': 1, 'deny': 2} + +_TOOL_CALL_UNDECIDED = ' -- refusing rather than allowing what it never judged' + +def _flag_path(argv, flag): + if flag in argv: + index = argv.index(flag) + if index + 1 < len(argv): + return _chock_Path(argv[index + 1]) return None +def tool_call_matches(globs, tool): + """Whether `tool` (a tool name, exact case) matches any glob; no globs match nothing.""" + return any((_chock_fnmatch.fnmatchcase(tool, str(glob)) for glob in globs)) + +def _tool_call_refusal(spec, spoken): + return spoken or str(spec.get('message') or '').strip() or 'blocked by a chock tool_call gate' + +def _tool_call_regex(spec, tool_input): + """`deny` verdict when the input, as JSON text, matches the gate's `content_pattern`.""" + text = json.dumps(tool_input, sort_keys=True, ensure_ascii=False) + if _chock_re.search(str((spec.get('params') or {}).get('content_pattern', '')), text): + return ('deny', _tool_call_refusal(spec, '')) + return None + +def _tool_call_script(spec, root, payload, deadline=None): + """Run the policy's script on `payload`; a missing script, crash or timeout refuses.""" + named = str((spec.get('params') or {}).get('script', '')) + script = _chock_Path(root) / named + if not script.is_file(): + return ('deny', f'script gate: {named!r} is not installed{_TOOL_CALL_UNDECIDED}') + deadline = engine_deadline() if deadline is None else deadline + argv = [sys.executable, str(script)] + try: + proc = _chock_subprocess.run(argv, input=json.dumps(payload), capture_output=True, text=True, encoding='utf-8', errors='replace', cwd=str(root), timeout=time_left(deadline, argv), check=False) + except _chock_subprocess.TimeoutExpired as exc: + budget = f'gave no verdict within {allowed_seconds(exc)}' + return ('deny', f'script gate: {script.name} {budget}{_TOOL_CALL_UNDECIDED}') + except OSError as exc: + return ('deny', f'script gate: {script.name} could not run ({exc}){_TOOL_CALL_UNDECIDED}') + spoken = ((proc.stderr or '') + (proc.stdout or '')).strip() + if proc.returncode in _TOOL_CALL_EXIT_VERDICTS: + verdict = _TOOL_CALL_EXIT_VERDICTS[proc.returncode] + return None if verdict is None else (verdict, spoken or f'blocked by {script.name}') + first = spoken.splitlines()[0] if spoken else '' + detail = f': {first}' if first else '' + return ('deny', f'script gate: {script.name} exited {proc.returncode}{detail}{_TOOL_CALL_UNDECIDED}') + +def _tool_call_verdict(spec, root, event, deadline): + """None to allow, else (verdict, message) for this call; a kind a tool call cannot answer refuses.""" + tool, tool_input = (str(event.tool or ''), tool_call_input(event)) + kind = spec.get('kind') + if kind == _TOOL_CALL_REGEX_KIND: + return _tool_call_regex(spec, tool_input) + if kind == _TOOL_CALL_SCRIPT_KIND: + payload = {'event': TOOL_CALL_EVENT, 'repo_root': str(root), 'tool': tool, 'input': tool_input, 'session': session_for(event, root)} + return _tool_call_script(spec, root, payload, deadline) + return ('deny', f'chock tool_call gate: kind {kind!r} cannot judge a tool call{_TOOL_CALL_UNDECIDED}') + +def _tool_call_capped(spec, verdict): + """`verdict` held to the gate's declared action; an unknown action keeps the verdict as spoken.""" + ceiling = _TOOL_CALL_CEILING.get(spec.get('action', 'block')) + if verdict is None or ceiling is None: + return verdict + return (min(verdict[0], ceiling, key=_TOOL_CALL_RANK.__getitem__), verdict[1]) + +def _tool_call_spec(gate): + """The compiled gate as a dict, or (None, refusal) when the install is broken.""" + if not gate.exists(): + return (None, f'chock gate {gate} is missing, so this tool call cannot be checked. Run `chock sync --repo .` to rebuild the compiled gates.') + try: + return (json.loads(gate.read_text(encoding='utf-8')), None) + except (OSError, ValueError) as exc: + return (None, f'chock could not read {gate}: {exc}. Refusing rather than reporting an allow it never established.') + +def evaluate_tool_call(argv, event): + """The decision a tool_call gate earns this call, or None; also logs the call for the session. + + `--tool-call ` judges a PreToolUse call and, for a gate whose check reads the session, + logs it after the verdict, so the script reads prior calls only. `--record ` logs what + a call came to, after it ran (a vendor's PostToolUse and failure events). + """ + recording = _flag_path(argv, RECORD_FLAG) + if recording is not None: + if event.event in _TOOL_CALL_POST: + session_record(repo_root_for(event, recording), event, 'post', session_outcome(event)) + return None + gate = _flag_path(argv, TOOL_CALL_FLAG) + if gate is None or event.event != _TOOL_CALL_PRE: + return None + spec, broken = _tool_call_spec(gate) + if spec is None: + return ('deny', broken) + root = repo_root_for(event, gate) + verdict = None + if TOOL_CALL_EVENT in spec.get('on', []) and tool_call_matches((spec.get('params') or {}).get('tools', []), str(event.tool or '')): + verdict = _tool_call_capped(spec, _tool_call_verdict(spec, root, event, engine_deadline())) + if spec.get('kind') in _TOOL_CALL_NEEDS_SESSION: + session_record(root, event, 'pre', _TOOL_CALL_BLOCKED if verdict and verdict[0] == 'deny' else None) + return verdict + _GIT = _chock_shutil.which('git') or 'git' _INSTRUCTION = "Chock: this clone's git hooks are NOT installed -- git never clones hooks, so commit-time gates will not run locally until someone runs:\n pip install chock && chock sync --repo .\nRun that before the first commit. (The repo's CI gate, where wired, enforces regardless.)" @@ -981,7 +2065,7 @@ def _repo_root() -> _chock_Path: def _hooks_pre_commit(repo_root: _chock_Path) -> _chock_Path | None: """The active pre-commit hook path, honouring core.hooksPath. None when git is absent.""" try: - proc = _chock_subprocess.run([_GIT, 'rev-parse', '--git-path', 'hooks'], cwd=repo_root, capture_output=True, text=True, timeout=15, check=False) + proc = _chock_subprocess.run([_GIT, 'rev-parse', '--git-path', 'hooks'], cwd=repo_root, capture_output=True, text=True, encoding='utf-8', errors='surrogateescape', timeout=15, check=False) except (OSError, _chock_subprocess.TimeoutExpired): return None if proc.returncode != 0: @@ -1035,6 +2119,86 @@ def _chock_handle_session_start(event): ) return Decision.allow(context=_INSTRUCTION) +_CHOCK_UNREADABLE = ( + "chock could not read this hook's payload, so it could not check the call. " + "Refusing rather than reporting an allow it never established." +) + +#: A payload naming only this vendor's own wire event, so `respond` picks that event's grammar. +_CHOCK_WIRE_RAW = {PRE_TOOL: {'hook_event_name': 'PreToolUse'}, STOP: {'hook_event_name': 'Stop'}} + +#: The words a client reads as a refusal, in any of the answer fields a vendor grammar uses. +_CHOCK_REFUSING = ("deny", "block", "ask", "continue") +_CHOCK_ANSWER_KEYS = ("decision", "permission", "permissionDecision") + + +def _chock_spec_action(path): + try: + action = json.loads(path.read_text(encoding="utf-8")).get("action") + except Exception: + return "block" + return action if action in ("ask", "warn") else "block" + + +def _chock_declared(): + """(strictest action, event, gate) of the judging flags; (None, None, None) where this run judges nothing.""" + argv = sys.argv[1:] + actions = ["block"] if guard_path_from_argv(argv) is not None else [] + event = PRE_TOOL + gate = _flag_path(argv, GATE_FLAG) + if gate is not None: + actions.append(_chock_spec_action(gate)) + event = STOP if STOP_FLAG in argv or gate.parent.name == _EVENT_ARG[STOP] else PRE_TOOL + called = _flag_path(argv, TOOL_CALL_FLAG) + if called is not None: + actions.append(_chock_spec_action(called)) + for action in ("block", "ask", "warn"): + if action in actions: + return action, event, gate + return None, None, None + + +def _chock_refuses(text, code): + if code == 2: + return True + try: + body = json.loads(text) + except Exception: + return False + if not isinstance(body, dict): + return False + nested = body.get("hookSpecificOutput") + answers = [body] + ([nested] if isinstance(nested, dict) else []) + return any(part.get(key) in _CHOCK_REFUSING for part in answers for key in _CHOCK_ANSWER_KEYS) + + +def _chock_unreadable_stop(gate): + """The bounded verdict of an unreadable Stop: a refusal, or a warning once the cap is spent.""" + root = root_for(gate) or _chock_Path.cwd() + return settle_unreadable_stop(root, gate.parent.parent.name, _CHOCK_UNREADABLE) + + +def _chock_unreadable(): + action, kind, gate = _chock_declared() + if action in (None, "warn"): + return "", 0 + event = Event(AGENT, kind, raw=dict(_CHOCK_WIRE_RAW[kind])) + reason = _CHOCK_UNREADABLE + if kind == STOP and gate is not None: + verdict, reason = _chock_unreadable_stop(gate) + if verdict == VERDICT_WARN: + return _chock_warn_respond(degrade(Decision.warn(reason), event), event) + _report(reason + "\n") + decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) + text, code = respond(degrade(decision, event), event) + return (text, code) if _chock_refuses(text, code) else ("", 2) + + +def _spoken(verdict): + outcome, reason = verdict + if outcome == VERDICT_WARN: + return Decision.warn(reason) + return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) def _judge(event): @@ -1045,7 +2209,10 @@ def _judge(event): 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 _spoken(gated) + called = evaluate_tool_call(sys.argv[1:], event) + if called is not None: + return _spoken(called) if event.event == "session_start": return _chock_handle_session_start(event) return None @@ -1062,6 +2229,33 @@ def handle(event): "chock could not check this call (%s: %s). Refusing rather than reporting an " "allow it never established." % (type(exc).__name__, exc) ) +def _chock_warn_respond(decision, event): + # A warn never blocks. agentseam reduces WARN to a silent allow for every agent, which drops + # the reason, so the reason is added here in the one channel Claude Code documents for it: + # PreToolUse `additionalContext` (Claude reads it beside the tool result; no permissionDecision + # is sent, since `allow` would also skip the user's own permission prompt) and, at the turn's + # end, `systemMessage` (`_chock_stop_warning`). + text, code = respond(decision, event) + if text or degraded_from(decision) != WARN or not decision.reason: + return text, code + if event.event == PRE_TOOL: + body = {"hookSpecificOutput": {"hookEventName": "PreToolUse", "additionalContext": decision.reason}} + return json.dumps(body), code + return _chock_stop_warning(decision, event) or text, code + + +def _chock_stop_warning(decision, event): + # A warn at the turn's end never blocks: a Stop block would force the turn to continue. agentseam + # reduces WARN to a silent allow, which drops the reason, so it is added back as `systemMessage`, + # which Claude Code, Codex and VS Code Copilot each document as a warning shown to the user. + if event.event != STOP or degraded_from(decision) != WARN or not decision.reason: + return None + return json.dumps({"systemMessage": decision.reason}) + + +def _chock_stop_warn_respond(decision, event): + text, code = respond(decision, event) + return text or _chock_stop_warning(decision, event) or text, code # <<< agentseam handler <<< @@ -1107,6 +2301,55 @@ def _report(text): return +_STDERR_FD = 2 + + +def _divert_fd1(): + """Point fd 1 at stderr (devnull if there is none); the saved fd 1, or None if it could not.""" + try: + saved = os.dup(1) + except OSError: + return None + # With fd 2 closed, dup() hands back 2 itself, and "stderr" would be stdout again. + if saved > _STDERR_FD: + with contextlib.suppress(OSError): + os.dup2(_STDERR_FD, 1) + return saved + sink = os.open(os.devnull, os.O_WRONLY) + os.dup2(sink, 1) + os.close(sink) + return saved + + +def _flush(streams): + """Flush each stream that exists; a broken one is no reason to lose the verdict.""" + for stream in streams: + if stream is not None: + with contextlib.suppress(Exception): + stream.flush() + + +@contextlib.contextmanager +def _stdout_to_stderr(): + # Stdout is the verdict channel. A handler's stray print (or a child process it runs) + # ahead of the JSON makes the host fail to parse it, and Claude Code and Gemini CLI then + # treat the hook as a non-blocking error: a deny became an allow, witnessed live. + sink = sys.stderr if sys.stderr is not None else io.StringIO() + held = (sys.stdout, sys.__stdout__) + _flush(held) + saved = _divert_fd1() + try: + with contextlib.redirect_stdout(sink): + yield + finally: + if saved is not None: + # A stream the handler held on to (sys.__stdout__, a reference cached at import) + # buffers past redirect_stdout; flushed now it lands on stderr, not after the verdict. + _flush((sink, *held)) + os.dup2(saved, 1) + os.close(saved) + + def _decide(raw): """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" event = parse(raw) @@ -1116,7 +2359,9 @@ def _decide(raw): # vocabulary invites a decision made on a false premise. return "", 0 try: - decision = _coerce(handle(event)) + with _stdout_to_stderr(): + result = handle(event) + decision = _coerce(result) except Exception: # A door that cannot decide refuses. Escaping here would exit 1 with a traceback, # which every host reads as a non-blocking error and allows past; instead the @@ -1128,7 +2373,7 @@ def _decide(raw): "policy handler failed (%s); refusing rather than allowing what it could not judge" % sys.exc_info()[0].__name__ ) - return respond(degrade(decision, event), event) + return _chock_warn_respond(degrade(decision, event), event) def main(stdin=None, stdout=None, exit=True): @@ -1138,10 +2383,14 @@ def main(stdin=None, stdout=None, exit=True): try: raw = json.loads(_read_payload(stream)) except Exception: - # Malformed input is not the agent's fault to pay for: allow, stay silent. + raw = None + if not isinstance(raw, dict): + text, code = _chock_unreadable() + if text: + _emit(out, text) if exit: - sys.exit(0) - return 0 + sys.exit(code) + return code try: text, code = _decide(raw) except Exception: diff --git a/.chock/bin/launch.sh b/.chock/bin/launch.sh new file mode 100644 index 0000000..5db2694 --- /dev/null +++ b/.chock/bin/launch.sh @@ -0,0 +1,21 @@ +#!/bin/sh +# chock hook launcher. Every agent hook chock writes runs: +# git -c "alias.chock-hook=!test -f || { ...; exit 2; }; sh " chock-hook [args...] +# git runs the alias from the repository's top level under its own sh (bash, PowerShell +# and cmd.exe all pass that string through unchanged), so relative paths resolve and no +# absolute interpreter path is ever committed. This script picks the first Python that +# actually runs: `chock.python` (written into .git/config by `chock sync`), then PATH. +# A candidate must execute, not merely exist: Windows' python3.exe Store alias exists and +# exits 9009. With no working Python it refuses (exit 2) and says why -- never allows. +runtime="$1" +shift +configured="$(git config --get chock.python 2>/dev/null)" +# The recorded interpreter is probed like the rest: a venv whose base Python was removed +# still exists, and exec'ing it exits non-zero without a verdict, which agents let through. +for candidate in "$configured" python3 python py; do + if [ -n "$candidate" ] && "$candidate" -c 'import sys; sys.exit(sys.version_info < (3, 11,))' /dev/null 2>&1; then + exec "$candidate" -X utf8 "$runtime" "$@" + fi +done +echo "chock: no working Python 3.11+ found (tried chock.python, python3, python, py), so this hook cannot check anything. Install Python, or point chock at one: git config chock.python /path/to/python" >&2 +exit 2