diff --git a/.chock/bin/antigravity.py b/.chock/bin/antigravity.py deleted file mode 100755 index 647b4245..00000000 --- a/.chock/bin/antigravity.py +++ /dev/null @@ -1,973 +0,0 @@ -# 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. -# -# Self-contained: stdlib only, no "import agentseam" anywhere in this file. Regenerate -# with `agentseam.bundler.bundle("antigravity")` (same agentseam version -> identical bytes, -# except your own edits inside the handler block) rather than patching this by hand. -# https://github.com/open-coder-ai/agentseam - -from __future__ import annotations - -import json -_json = json -import sys - -import os as _chock_os -import shlex as _chock_shlex -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.3.0) - -"""Canonical event vocabulary, normalized envelope, and decision type.""" - - - -SESSION_START = "session_start" -SESSION_END = "session_end" -PROMPT_SUBMIT = "prompt_submit" -PRE_TOOL = "pre_tool" -POST_TOOL = "post_tool" -TOOL_FAILURE = "tool_failure" -PRE_COMPACT = "pre_compact" -STOP = "stop" -SUBAGENT_START = "subagent_start" -SUBAGENT_STOP = "subagent_stop" -INSTRUCTIONS_LOADED = "instructions_loaded" -FILE_CHANGED = "file_changed" - -UNKNOWN = "unknown" - -EVENTS = ( - SESSION_START, - SESSION_END, - PROMPT_SUBMIT, - PRE_TOOL, - POST_TOOL, - TOOL_FAILURE, - PRE_COMPACT, - STOP, - SUBAGENT_START, - SUBAGENT_STOP, - INSTRUCTIONS_LOADED, - FILE_CHANGED, -) - - -class Event: - """One agent lifecycle event, normalized.""" - - __slots__ = ( - "agent", - "command", - "content", - "cwd", - "event", - "output", - "path", - "prompt", - "raw", - "session_id", - "tool", - "tool_use_id", - ) - - def __init__( - self, - agent, - event, - *, - tool=None, - command=None, - path=None, - content=None, - output=None, - prompt=None, - session_id=None, - tool_use_id=None, - cwd=None, - raw=None, - ): - self.agent = agent - self.event = event - self.tool = tool - self.command = command - self.path = path - self.content = content - self.output = output - self.prompt = prompt - self.session_id = session_id - self.tool_use_id = tool_use_id - self.cwd = cwd - self.raw = raw if raw is not None else {} - - def __repr__(self): # pragma: no cover - debugging aid - return "Event(%s/%s tool=%r path=%r)" % (self.agent, self.event, self.tool, self.path) - - -ALLOW = "allow" -DENY = "deny" -ESCALATE = "escalate" -TRANSFORM = "transform" -WARN = "warn" -VOUCH = "vouch" - -# Pre-ACS-alignment names. Same strings as their ACS-named counterparts, so every existing -# `is`/`==` comparison against the old constant keeps working untouched. -ASK = ESCALATE -REWRITE = TRANSFORM - -_CANONICAL_OUTCOMES = (ALLOW, DENY, ESCALATE, TRANSFORM, WARN, VOUCH) - -# The literal spellings a caller might still pass to Decision(outcome, ...) directly, mapped -# to the value that now backs them. Only needed for the raw-string constructor path -- -# Decision.ask()/.rewrite() below build the canonical outcome themselves. -_LEGACY_SPELLING = {"ask": ESCALATE, "rewrite": TRANSFORM} - - -class Decision: - """What a handler wants to happen. Adapters translate this to vendor dialect.""" - - __slots__ = ("context", "evidence", "outcome", "reason", "updated_input") - - #: Classmethods kept only so existing callers keep constructing; see .ask()/.rewrite(). - DEPRECATED_ALIASES = frozenset({"ask", "rewrite"}) - - def __init__(self, outcome, reason=None, updated_input=None, evidence=None, context=None): - outcome = _LEGACY_SPELLING.get(outcome, outcome) - if outcome not in _CANONICAL_OUTCOMES: - raise ValueError("unknown outcome: %r" % (outcome,)) - self.outcome = outcome - self.reason = reason - self.updated_input = updated_input - self.evidence = evidence or {} - self.context = context - - @classmethod - def allow(cls, reason=None, evidence=None, context=None): - return cls(ALLOW, reason, evidence=evidence, context=context) - - @classmethod - def deny(cls, reason, evidence=None, context=None): - return cls(DENY, reason, evidence=evidence, context=context) - - @classmethod - def escalate(cls, reason, evidence=None, context=None): - """Defer the action to the host's own approval path (ACS `escalate`).""" - return cls(ESCALATE, reason, evidence=evidence, context=context) - - @classmethod - def ask(cls, reason, evidence=None, context=None): - """Deprecated alias of escalate() -- kept so existing callers keep constructing.""" - _warnings.warn("Decision.ask() is deprecated; use Decision.escalate()", DeprecationWarning, stacklevel=2) - return cls.escalate(reason, evidence=evidence, context=context) - - @classmethod - def transform(cls, updated_input, reason=None, evidence=None, context=None): - """Replace the tool input wholesale (ACS `transform`, at whole-value granularity).""" - return cls(TRANSFORM, reason, updated_input=updated_input, evidence=evidence, context=context) - - @classmethod - def rewrite(cls, updated_input, reason=None, evidence=None, context=None): - """Deprecated alias of transform() -- kept so existing callers keep constructing.""" - _warnings.warn("Decision.rewrite() is deprecated; use Decision.transform()", DeprecationWarning, stacklevel=2) - return cls.transform(updated_input, reason, evidence=evidence, context=context) - - @classmethod - def warn(cls, reason=None, evidence=None, context=None): - """Permit the action with no change, recording a warning (ACS `warn`).""" - return cls(WARN, reason, evidence=evidence, context=context) - - @classmethod - def vouch(cls, reason=None, evidence=None, context=None): - return cls(VOUCH, reason, evidence=evidence, context=context) - - def __repr__(self): # pragma: no cover - debugging aid - return "Decision(%s, %r)" % (self.outcome, self.reason) - - -def degraded_from(decision): - """What this decision was before the dispatcher reduced it, or None.""" - return (decision.evidence or {}).get("degraded_from") - - -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] == "{": - try: - parsed = _json.loads(raw) - except _json.JSONDecodeError: - return {} - if isinstance(parsed, dict): - return parsed - return {} - - -# ------------------------------------------------------------------------------ -# antigravity family engine (trimmed to what this entry uses) - -def _wire_name(cfg, raw): - for key in cfg["claims"].get("event_key", ()): - name = raw.get(key) - if name is not None: - return name - return None - -def _segment(node, part): - """One path segment: a dict key, or `key[N]` indexing the list under it.""" - if part.endswith("]") and "[" in part: - key, _, index = part[:-1].partition("[") - items = node.get(key) if isinstance(node, dict) else None - i = int(index) - return items[i] if isinstance(items, (list, tuple)) and len(items) > i else None - return node.get(part) if isinstance(node, dict) else None - -def _walk(node, path): - """A dotted path off `node`; `a[].b` joins `b` over `a`'s dict items, `a[0]` indexes.""" - if "[]." in path: - head, sub = path.split("[].", 1) - items = _walk(node, head) - if not isinstance(items, (list, tuple)): - return None - joined = "\n".join(str(item.get(sub, "")) for item in items if isinstance(item, dict)) - return joined or None - for part in path.split("."): - node = _segment(node, part) - if node is None: - return None - return node - -def _lookup(raw, ti, key): - """One config key: a `tool_input.` path walks the decoded tool input, else the payload.""" - if key.startswith("tool_input."): - return _walk(ti, key[len("tool_input.") :]) - return _walk(raw, key) - -def _field(raw, ti, chain): - value = None - for key in chain: - if value: - break - value = value or _lookup(raw, ti, key) - return value - -_FIELD_META = ("tool_input", "content_only_for_write_tools", "stringify") - -def _tool_input_raw(cfg, raw): - for key in cfg["fields"].get("tool_input", ("tool_input",)): - value = raw.get(key) - if value is not None: - return value - return None - -def _canonical_of(cfg, name): - """The canonical event for one wire name; an entry with no `events` at all (antigravity, - whose payloads never carry one) maps the shape-inferred name back through `wire_events`.""" - events = cfg["events"] - if not events: - return {wire: canonical for canonical, wire in cfg.get("wire_events", {}).items()}.get(name, UNKNOWN) - return events.get(name, UNKNOWN) - -def hj_parse(cfg, raw, wire=None): - """Normalise one payload along the entry's ordered field-fallback chains. - - `wire` is the pre-resolved wire event name for the shape-inferred families; the - marker families resolve it from the payload's own event key. - """ - ti = tool_input_of(_tool_input_raw(cfg, raw)) - fields = {name: _field(raw, ti, chain) for name, chain in cfg["fields"].items() if name not in _FIELD_META} - if cfg["fields"].get("content_only_for_write_tools") and fields.get("tool") not in cfg["tools"].get("write", ()): - fields["content"] = None - if isinstance(fields.get("output"), (dict, list)): - fields["output"] = _json.dumps(fields["output"]) - for name in cfg["fields"].get("stringify", ()): - if fields.get(name) is not None: - fields[name] = str(fields[name]) - return Event( - cfg["agent"], - _canonical_of(cfg, wire if wire is not None else _wire_name(cfg, raw)), - tool=fields.get("tool"), - command=fields.get("command"), - path=fields.get("path"), - content=fields.get("content"), - output=fields.get("output"), - prompt=fields.get("prompt"), - session_id=fields.get("session_id"), - tool_use_id=fields.get("tool_use_id"), - cwd=fields.get("cwd"), - raw=raw, - ) - -_ESCALATE_FROM_TRANSFORM = "escalate_from_transform" - -_TRANSFORM_MISSING_INPUT = "transform_missing_input" - -def hj_reverse(cfg): - """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: - reverse[canonical] = name - reverse.update(cfg.get("wire_events", {})) - return reverse - -def _context_value(v, decision): - if v.get("context_source") == "context": - return decision.context - if v.get("context_source") == "reason": - return decision.reason - return None - -def _context_body(name, value): - return _json.dumps({"hookSpecificOutput": {"hookEventName": name, "additionalContext": value}}), 0 - -def _note_for(v, decision, at_gate, missing_input): - notes = v.get("degrade_notes", {}) - if decision.outcome == ESCALATE: - if degraded_from(decision) == TRANSFORM and _ESCALATE_FROM_TRANSFORM in notes: - return notes[_ESCALATE_FROM_TRANSFORM] - if at_gate and "escalate_gate" in notes: - return notes["escalate_gate"] - return notes.get("escalate") - if decision.outcome == TRANSFORM: - if missing_input and _TRANSFORM_MISSING_INPUT in notes: - return notes[_TRANSFORM_MISSING_INPUT] - return notes.get("transform") - return None - -def _default_for(v, decision, at_gate, wire=None): - gate_defaults = v.get("gate_reason_defaults", {}) - if wire in gate_defaults: - return gate_defaults[wire] - defaults = v.get("reason_defaults", {}) - key = {DENY: "deny", ESCALATE: "escalate", TRANSFORM: "transform"}.get(decision.outcome, "deny") - if at_gate and key + "_gate" in defaults: - return defaults[key + "_gate"] - return defaults.get(key, "blocked by policy") - -def _refusal_text(v, decision, at_gate, wire=None): - note = _note_for(v, decision, at_gate, decision.updated_input is None) - default = _default_for(v, decision, at_gate, wire) - if note and "%s" in note: - # A template note fills (the reason or its default, the wire event name) itself. - return note % (decision.reason or default, wire) - reason = decision.reason - if v.get("note_style") == "suffix": - reason = reason or default - return "%s (%s)" % (reason, note) if note else reason - text = "%s (%s)" % (reason, note) if reason and note else (note or reason) - return text or default - -def _g1_allow(v, decision, wire, name, words): - value = _context_value(v, decision) - if wire in v.get("context_events", ()) and value: - return _context_body(name, value) - if wire in v.get("allow_silent_events", ()): - return "", 0 - if "allow" in words: - out = {"decision": words["allow"]} - if v.get("allow_context_key") and decision.outcome == ALLOW and value: - out[v["allow_context_key"]] = value - return _json.dumps(out), 0 - return "", 0 - -def _g1_transform(v, decision, name, words): - """None when the transform isn't representable here; `_g1` falls through to the block path.""" - if v.get("transform_grammar") == "hook_specific_tool_input": - return _json.dumps({"hookSpecificOutput": {"tool_input": decision.updated_input}}), 0 - if decision.updated_input is None: - return None - if v.get("transform_grammar") == "top_level_updated_input": - out = {"decision": words.get("transform", "allow"), "updatedInput": decision.updated_input} - if decision.reason: - out["reason"] = decision.reason - return _json.dumps(out), 0 - return _json.dumps({"hookSpecificOutput": {"hookEventName": name, "updatedInput": decision.updated_input}}), 0 - -def _g1(v, gate, decision, wire, name): - """Block dialect: a top-level decision word, or silence/context where nothing is read.""" - words = dict(v.get("words", {})) - words.update(v.get("words_at", {}).get(wire, {})) - at_context_event = wire in v.get("context_events", ()) - if decision.outcome in (ALLOW, VOUCH, WARN): - return _g1_allow(v, decision, wire, name, words) - if decision.outcome == TRANSFORM and gate["honours_transform"]: - transformed = _g1_transform(v, decision, name, words) - if transformed is not None: - return transformed - if ( - decision.outcome == ESCALATE - and gate["honours_escalate"] - and "escalate" in words - # An escalate the dispatcher degraded a transform into is a block where the entry - # names that degradation (antigravity): prompting would offer the unmodified call. - and not (degraded_from(decision) == TRANSFORM and _ESCALATE_FROM_TRANSFORM in v.get("degrade_notes", {})) - ): - reason = decision.reason or _default_for(v, decision, at_gate=True, wire=wire) - return _json.dumps({"decision": words["escalate"], "reason": reason}), 0 - out = {"decision": words.get("block", "block"), "reason": _refusal_text(v, decision, at_gate=False, wire=wire)} - if at_context_event and v.get("context_source") == "context" and decision.context: - out["hookSpecificOutput"] = {"hookEventName": name, "additionalContext": decision.context} - return _json.dumps(out), 0 - -def hj_respond(cfg, decision, event, wire=None): - """(stdout_text, exit_code) in this entry's dialect for the gate the payload names. - - `wire` is the pre-resolved wire event name for the shape-inferred families; the - marker families resolve it from the payload's own event key. - """ - v = cfg["verdicts"] - if wire is None: - wire = _wire_name(cfg, event.raw or {}) - if wire in v.get("empty_object_events", ()): - return _json.dumps({}), 0 - if wire is None: - wire = v.get("default_wire_event") - if wire is None and v.get("missing_wire") == "reverse_map": - wire = hj_reverse(cfg).get(event.event) - name = wire if v.get("echo") == "payload" else hj_reverse(cfg).get(event.event, "PreToolUse") - gate = v["gates"].get(wire) - if gate is None: - value = _context_value(v, decision) - if wire in v.get("context_events", ()) and value: - return _context_body(name, value) - return "", 0 - if gate["grammar"] == "G2": - return _g2(v, gate, decision, name) - return _g1(v, gate, decision, wire, name) - -_WINDOWS_KEYS = ("commandWindows", "windows") - -def _hook_dict(cfg, command): - entry = {"type": "command", "command": command} - for key, value in cfg["hook_entry"].get("entry_extra", {}).items(): - if key in _WINDOWS_KEYS: - entry[key] = powershell_command(command) - else: - entry[key] = value - return entry - -def _flat_list_wrapper(hook_entry, reverse, canonical_events, command, matcher): - rules = [] - for ev in canonical_events: - name = reverse.get(ev) - if not name: - continue - rule = {"event": name, "command": command} - if matcher and hook_entry["matcher"]: - rule["matcher"] = matcher - rules.append(rule) - return rules - -def _cursor_wrapper(cfg, reverse, canonical_events, command, *, fail_closed): - gates = cfg["verdicts"]["answer_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: - entry["failClosed"] = True - hooks.setdefault(name, []).append(entry) - return {"version": 1, "hooks": hooks} - -def _flat_entries_wrapper(hook_entry, reverse, canonical_events, command): - hooks = {} - for ev in canonical_events: - name = reverse.get(ev) - if not name: - continue - hooks.setdefault(name, []).append({"command": command}) - extra = hook_entry.get("also_wires", {}).get(ev) - if extra: - hooks.setdefault(extra, []).append({"command": command}) - return {"hooks": hooks} - -def _default_wrapper(cfg, reverse, canonical_events, command, matcher): - hook_entry = cfg["hook_entry"] - hooks = {} - for ev in canonical_events: - name = reverse.get(ev) - if not name: - continue - entry = {"hooks": [_hook_dict(cfg, command)]} - if matcher and hook_entry["matcher"]: - entry["matcher"] = matcher - hooks.setdefault(name, []).append(entry) - if hook_entry.get("group"): - return {hook_entry["group"]: hooks} - return hooks if hook_entry.get("bare") else {"hooks": hooks} - -def hook_entry_config(cfg, canonical_events, command, matcher=None, *, fail_closed=True): - """The vendor's hooks-config fragment wiring `command` for these canonical events. - - `fail_closed` is read only by the `cursor` wrapper, whose gates fail open unless the - entry says otherwise; a False installs an observer, not a gate. - """ - hook_entry = cfg["hook_entry"] - reverse = hj_reverse(cfg) - wrapper = hook_entry["wrapper"] - if wrapper == "flat_list": - return _flat_list_wrapper(hook_entry, reverse, canonical_events, command, matcher) - if wrapper == "cursor": - return _cursor_wrapper(cfg, reverse, canonical_events, command, fail_closed=fail_closed) - if wrapper == "flat_entries": - return _flat_entries_wrapper(hook_entry, reverse, canonical_events, command) - return _default_wrapper(cfg, reverse, canonical_events, command, matcher) - -def antigravity_wire(raw): - """Name the event from shape; ties go to PreToolUse so the gate stays a gate.""" - if "terminationReason" in raw or "fullyIdle" in raw: - return "Stop" - if isinstance(raw.get("toolCall"), dict): - return "PostToolUse" if "error" in raw else "PreToolUse" - return None - -def antigravity_claims(_cfg, raw): - """Structural: `conversationId` with `workspacePaths` is Antigravity's own envelope.""" - if not isinstance(raw, dict): - return False - return "conversationId" in raw and isinstance(raw.get("workspacePaths"), list) - -def antigravity_parse(cfg, raw): - return hj_parse(cfg, raw, wire=antigravity_wire(raw)) - -def antigravity_respond(cfg, decision, event): - return hj_respond(cfg, decision, event, wire=antigravity_wire(event.raw or {})) - - -# ------------------------------------------------------------------------------ -# antigravity vendor config + engine binding - -AGENT = "antigravity" - -VENDOR = {'agent': 'antigravity', 'claims': {'mode': 'shape_inferred'}, 'config_format': 'json', 'config_path': '.agents/hooks.json', 'display': 'Antigravity', 'events': {}, 'evidence': {'claims': {'basis': 'vendor-docs', 'date': '2026-08-26', 'test': 'tests/test_examples.py::test_each_payload_is_claimed_by_its_own_adapter'}, 'config_path': {'basis': 'vendor-docs', 'date': '2026-08-26', 'test': 'tests/test_vendor_config.py::test_config_path_agrees_with_matrix'}, 'events': {'basis': 'vendor-docs', 'date': '2026-08-26', 'test': 'tests/test_examples.py::test_each_payload_parses_to_the_event_it_is_filed_under'}, 'family': {'basis': 'vendor-docs', 'date': '2026-08-26', 'test': 'tests/test_golden_fixtures.py::test_wire_output_matches_the_frozen_fixture'}, 'fields': {'basis': 'vendor-docs', 'date': '2026-08-26', 'test': 'tests/test_vendor_config.py::test_entries_match_recount'}, 'hook_entry': {'basis': 'vendor-docs', 'date': '2026-08-26', 'test': 'tests/test_golden_fixtures.py::test_hook_config_matches_the_frozen_fixture_on_both_matcher_paths'}, 'tools': {'basis': 'vendor-docs', 'date': '2026-08-26', 'test': 'tests/test_vendor_config.py::test_entries_match_recount'}, 'verdicts': {'basis': 'vendor-docs', 'date': '2026-08-26', 'test': 'tests/test_golden_fixtures.py::test_wire_output_matches_the_frozen_fixture'}}, 'family': 'antigravity', 'fields': {'command': ('toolCall.args.CommandLine',), 'content': ('toolCall.args.CodeContent', 'toolCall.args.ReplacementContent', 'toolCall.args.ReplacementChunks[].ReplacementContent'), 'cwd': ('toolCall.args.Cwd', 'workspacePaths[0]'), 'output': ('error',), 'path': ('toolCall.args.TargetFile', 'toolCall.args.AbsolutePath'), 'session_id': ('conversationId',), 'stringify': ('tool_use_id',), 'tool': ('toolCall.name',), 'tool_use_id': ('stepIdx',)}, 'hook_entry': {'group': 'agentseam', 'matcher': True, 'wrapper': 'hooks_map'}, 'needs_trust': False, 'tools': {}, 'verdicts': {'answer_events': ('PreToolUse', 'Stop'), 'bare_allow': 'unverified', 'default_wire_event': 'PreToolUse', 'degrade_notes': {'escalate': 'Antigravity cannot prompt at Stop', 'escalate_from_transform': 'Antigravity cannot modify a tool call', 'transform': 'Antigravity cannot modify a tool call'}, 'empty_object_events': ('PostToolUse',), 'gate_reason_defaults': {'Stop': 'policy requires more work'}, 'gates': {'PreToolUse': {'grammar': 'G1', 'honours_escalate': True, 'honours_transform': False}, 'Stop': {'grammar': 'G1', 'honours_escalate': False, 'honours_transform': False}}, 'note_style': 'because', 'reason_defaults': {'escalate': 'confirmation required'}, 'vocabulary': ('allow', 'ask', 'continue', 'deny', 'deny_unless_prior_grant', 'force_ask', 'stop'), 'vocabulary_basis': 'verified', 'words': {'allow': 'allow', 'block': 'deny', 'escalate': 'ask'}, 'words_at': {'Stop': {'allow': 'stop', 'block': 'continue'}}}, 'wire_events': {'post_tool': 'PostToolUse', 'pre_tool': 'PreToolUse', 'stop': 'Stop'}} - - -def claims(raw): - return antigravity_claims(VENDOR, raw) - - -def parse(raw): - return antigravity_parse(VENDOR, raw) - - -def respond(decision, event): - return antigravity_respond(VENDOR, decision, event) - - -def hook_config(canonical_events, command, matcher=None): - return hook_entry_config(VENDOR, canonical_events, command, matcher) - - -# ------------------------------------------------------------------------------ -# runtime (agentseam dispatch, specialized to antigravity) - -_VOUCH_SPEAKS = False -_WARN_SPEAKS = False -_TRANSFORM_EVENTS = frozenset(()) - - -def degrade(decision, event): - """Reduce a decision to what 'antigravity' can actually honor, honestly. - - This is agentseam.dispatch.degrade(), specialized to one fixed agent so this file needs - no "import agentseam" -- no evidence establishes that an explicit approval word means anything beyond a plain allow here, so vouch degrades to one; see agentseam.allow_semantics. - """ - if decision.outcome == TRANSFORM and event.event not in _TRANSFORM_EVENTS: - evidence = dict(decision.evidence) - evidence["degraded_from"] = TRANSFORM - return Decision.escalate( - decision.reason or "input requires modification before it can run", evidence=evidence - ) - if decision.outcome == VOUCH and not _VOUCH_SPEAKS: - evidence = dict(decision.evidence) - evidence["degraded_from"] = VOUCH - return Decision.allow(decision.reason, evidence=evidence, context=decision.context) - if decision.outcome == WARN and not _WARN_SPEAKS: - evidence = dict(decision.evidence) - evidence["degraded_from"] = WARN - return Decision.allow(decision.reason, evidence=evidence, context=decision.context) - return decision - - -# >>> 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' - -_LOG_MAX_BYTES = 1048576 - -_GUARD_TIMEOUT_SECONDS = 30 - -GUARD_BLOCKED = 'blocked' - -GUARD_CLEAN = 'clean' - -GUARD_ASKED = 'asked' - -GUARD_UNCHECKED = 'unchecked' - -GUARD_ERRORED = 'errored' - -VERDICT_DENY = 'deny' - -VERDICT_ESCALATE = 'escalate' - -def guard_path_from_argv(argv: list[str]) -> _chock_Path | None: - """The `--guard ` argument a vendored runtime was invoked with, or None.""" - if '--guard' in argv: - i = argv.index('--guard') - if i + 1 < len(argv): - 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: - try: - proc = _chock_subprocess.run([candidate, '-c', f'test -f "{guard.as_posix()}"'], capture_output=True, timeout=10, check=False) - except (OSError, _chock_subprocess.SubprocessError): - continue - if proc.returncode == 0: - 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_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, '') - if not args: - 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([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, '') - except (OSError, UnicodeError) as exc: - print(f'chock: guard could not run, not checked: {exc}', file=sys.stderr) - 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 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 '') - sys.stderr.write(proc.stderr or '') - return (GUARD_ASKED, first_line) - if proc.returncode != 0: - 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, *, 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': - return - guard = guard.resolve() - if guard.parent.name != 'implementations': - return - artifact_root = None - for parent in guard.parents: - if (parent / '.chock').is_dir(): - artifact_root = parent / '.chock' - 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') - 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(): - return None - 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' - -_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 ` 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 <<< - - -def _coerce(result): - if result is None: - return Decision.allow() - if isinstance(result, Decision): - return result - raise TypeError("handle() must return a Decision or None, got %r" % (type(result),)) - - -def _read_payload(stream): - # Decode bytes ourselves rather than trust the platform locale -- a BOM'd or non-UTF-8 - # stdin must not silently disable the gate. See agentseam.dispatch for the incident - # this guards: a Windows console's cp1252 layer turning a UTF-8 BOM into three bytes - # json cannot parse, with the hook allowing everything while claiming enforcement. - buffer = getattr(stream, "buffer", None) - if buffer is not None: - return buffer.read().decode("utf-8-sig", errors="replace") - return stream.read().lstrip("\ufeff") - - -def _emit(out, text): - buffer = getattr(out, "buffer", None) - if buffer is not None: - buffer.write(text.encode("utf-8")) - buffer.flush() - else: - out.write(text) - out.flush() - - -def main(stdin=None, stdout=None, exit=True): - """Read one payload from stdin, dispatch, emit the 'antigravity' response, exit.""" - stream = stdin if stdin is not None else sys.stdin - out = stdout if stdout is not None else sys.stdout - try: - raw = json.loads(_read_payload(stream)) - except Exception: - # Malformed input is not the agent's fault to pay for: allow, stay silent. - if exit: - sys.exit(0) - return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) - if text: - _emit(out, text) - if exit: - sys.exit(code) - return code - - -if __name__ == "__main__": # pragma: no cover - main() - diff --git a/.chock/bin/claude_code.py b/.chock/bin/claude_code.py index 6185c6f1..e4c6207b 100755 --- a/.chock/bin/claude_code.py +++ b/.chock/bin/claude_code.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.0 -- bundle("claude_code"). Do not hand-edit, except the +# Generated by agentseam 0.3.1 -- 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. # @@ -19,10 +19,11 @@ import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.0) +# contract (agentseam 0.3.1) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -203,7 +204,9 @@ def tool_input_of(raw): if isinstance(raw, str) and raw[:1] == "{": try: parsed = _json.loads(raw) - except _json.JSONDecodeError: + except (ValueError, RecursionError): + # JSONDecodeError is a ValueError; a string nested past the interpreter's + # limit raises RecursionError instead, and neither is an input to crash on. return {} if isinstance(parsed, dict): return parsed @@ -230,9 +233,22 @@ def looks_like_claude_code(raw): PROBES = {"looks_like_claude_code": looks_like_claude_code} +UNREADABLE_NAME = "" + +def wire_name_of(raw, key): + """The event name under `key`: text, None when absent, UNREADABLE_NAME when not text. + + Total over anything `json.loads` can return: a payload that is not an object names no + event, and a name that is not a string is not one any adapter can map. + """ + name = raw.get(key) if isinstance(raw, dict) else None + if name is None or isinstance(name, str): + return name + return UNREADABLE_NAME + def _wire_name(cfg, raw): for key in cfg["claims"].get("event_key", ()): - name = raw.get(key) + name = wire_name_of(raw, key) if name is not None: return name return None @@ -317,6 +333,8 @@ def _field(raw, ti, chain): _FIELD_META = ("tool_input", "content_only_for_write_tools", "stringify") def _tool_input_raw(cfg, raw): + if not isinstance(raw, dict): + return None for key in cfg["fields"].get("tool_input", ("tool_input",)): value = raw.get(key) if value is not None: @@ -1076,6 +1094,43 @@ def _emit(out, text): out.flush() +def _report(text): + # Best effort, never raising: a diagnostic must not pre-empt the refusal it accompanies. + # A hook's stderr is whatever the host gave it -- closed (then sys.stderr is None, and + # the traceback module's default print would land on STDOUT, inside the verdict), a + # console code page that cannot hold the payload text quoted in the exception, a pipe + # nobody reads. + try: + sys.stderr.write(text) + sys.stderr.flush() + except Exception: + return + + +def _decide(raw): + """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" + event = parse(raw) + if event.event == UNKNOWN: + # A vendor event this adapter has no mapping for. handle() is not called: it + # reasons about the canonical vocabulary, and handing it something outside that + # vocabulary invites a decision made on a false premise. + return "", 0 + try: + decision = _coerce(handle(event)) + 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 + # failure is printed for the operator and the gate answers deny in 'claude_code''s + # own dialect. Only the failure's class is named to the host: its message may quote + # the very payload content the policy was inspecting. + _report(traceback.format_exc()) + decision = Decision.deny( + "policy handler failed (%s); refusing rather than allowing what it could not judge" + % sys.exc_info()[0].__name__ + ) + return respond(degrade(decision, event), event) + + def main(stdin=None, stdout=None, exit=True): """Read one payload from stdin, dispatch, emit the 'claude_code' response, exit.""" stream = stdin if stdin is not None else sys.stdin @@ -1087,16 +1142,14 @@ def main(stdin=None, stdout=None, exit=True): if exit: sys.exit(0) return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) + try: + text, code = _decide(raw) + except Exception: + # Past the handler, which _decide() already answers for: a fault in this file on a + # payload it did decode. With no Event to answer in dialect, the one refusal left is + # the host's blocking exit code, and nothing on stdout to be misread as a verdict. + _report(traceback.format_exc()) + text, code = "", 2 if text: _emit(out, text) if exit: diff --git a/.chock/bin/codex_cli.py b/.chock/bin/codex_cli.py index 7a723000..ebad2939 100755 --- a/.chock/bin/codex_cli.py +++ b/.chock/bin/codex_cli.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.0 -- bundle("codex_cli"). Do not hand-edit, except the +# Generated by agentseam 0.3.1 -- bundle("codex_cli"). 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. # @@ -18,10 +18,11 @@ import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.0) +# contract (agentseam 0.3.1) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -202,7 +203,9 @@ def tool_input_of(raw): if isinstance(raw, str) and raw[:1] == "{": try: parsed = _json.loads(raw) - except _json.JSONDecodeError: + except (ValueError, RecursionError): + # JSONDecodeError is a ValueError; a string nested past the interpreter's + # limit raises RecursionError instead, and neither is an input to crash on. return {} if isinstance(parsed, dict): return parsed @@ -216,9 +219,22 @@ def powershell_command(command): """`command` rewritten so PowerShell will actually run it.""" return command if command.lstrip().startswith("&") else "& " + command +UNREADABLE_NAME = "" + +def wire_name_of(raw, key): + """The event name under `key`: text, None when absent, UNREADABLE_NAME when not text. + + Total over anything `json.loads` can return: a payload that is not an object names no + event, and a name that is not a string is not one any adapter can map. + """ + name = raw.get(key) if isinstance(raw, dict) else None + if name is None or isinstance(name, str): + return name + return UNREADABLE_NAME + def _wire_name(cfg, raw): for key in cfg["claims"].get("event_key", ()): - name = raw.get(key) + name = wire_name_of(raw, key) if name is not None: return name return None @@ -303,6 +319,8 @@ def _field(raw, ti, chain): _FIELD_META = ("tool_input", "content_only_for_write_tools", "stringify") def _tool_input_raw(cfg, raw): + if not isinstance(raw, dict): + return None for key in cfg["fields"].get("tool_input", ("tool_input",)): value = raw.get(key) if value is not None: @@ -994,6 +1012,43 @@ def _emit(out, text): out.flush() +def _report(text): + # Best effort, never raising: a diagnostic must not pre-empt the refusal it accompanies. + # A hook's stderr is whatever the host gave it -- closed (then sys.stderr is None, and + # the traceback module's default print would land on STDOUT, inside the verdict), a + # console code page that cannot hold the payload text quoted in the exception, a pipe + # nobody reads. + try: + sys.stderr.write(text) + sys.stderr.flush() + except Exception: + return + + +def _decide(raw): + """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" + event = parse(raw) + if event.event == UNKNOWN: + # A vendor event this adapter has no mapping for. handle() is not called: it + # reasons about the canonical vocabulary, and handing it something outside that + # vocabulary invites a decision made on a false premise. + return "", 0 + try: + decision = _coerce(handle(event)) + 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 + # failure is printed for the operator and the gate answers deny in 'codex_cli''s + # own dialect. Only the failure's class is named to the host: its message may quote + # the very payload content the policy was inspecting. + _report(traceback.format_exc()) + decision = Decision.deny( + "policy handler failed (%s); refusing rather than allowing what it could not judge" + % sys.exc_info()[0].__name__ + ) + return respond(degrade(decision, event), event) + + def main(stdin=None, stdout=None, exit=True): """Read one payload from stdin, dispatch, emit the 'codex_cli' response, exit.""" stream = stdin if stdin is not None else sys.stdin @@ -1005,16 +1060,14 @@ def main(stdin=None, stdout=None, exit=True): if exit: sys.exit(0) return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) + try: + text, code = _decide(raw) + except Exception: + # Past the handler, which _decide() already answers for: a fault in this file on a + # payload it did decode. With no Event to answer in dialect, the one refusal left is + # the host's blocking exit code, and nothing on stdout to be misread as a verdict. + _report(traceback.format_exc()) + text, code = "", 2 if text: _emit(out, text) if exit: diff --git a/.chock/bin/cursor.py b/.chock/bin/cursor.py index 579e8959..498bab48 100755 --- a/.chock/bin/cursor.py +++ b/.chock/bin/cursor.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.0 -- bundle("cursor"). Do not hand-edit, except the +# Generated by agentseam 0.3.1 -- bundle("cursor"). 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. # @@ -18,10 +18,11 @@ import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.0) +# contract (agentseam 0.3.1) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -202,7 +203,9 @@ def tool_input_of(raw): if isinstance(raw, str) and raw[:1] == "{": try: parsed = _json.loads(raw) - except _json.JSONDecodeError: + except (ValueError, RecursionError): + # JSONDecodeError is a ValueError; a string nested past the interpreter's + # limit raises RecursionError instead, and neither is an input to crash on. return {} if isinstance(parsed, dict): return parsed @@ -212,9 +215,22 @@ def tool_input_of(raw): # ------------------------------------------------------------------------------ # cursor family engine (trimmed to what this entry uses) +UNREADABLE_NAME = "" + +def wire_name_of(raw, key): + """The event name under `key`: text, None when absent, UNREADABLE_NAME when not text. + + Total over anything `json.loads` can return: a payload that is not an object names no + event, and a name that is not a string is not one any adapter can map. + """ + name = raw.get(key) if isinstance(raw, dict) else None + if name is None or isinstance(name, str): + return name + return UNREADABLE_NAME + def _wire_name(cfg, raw): for key in cfg["claims"].get("event_key", ()): - name = raw.get(key) + name = wire_name_of(raw, key) if name is not None: return name return None @@ -260,6 +276,8 @@ def _field(raw, ti, chain): _FIELD_META = ("tool_input", "content_only_for_write_tools", "stringify") def _tool_input_raw(cfg, raw): + if not isinstance(raw, dict): + return None for key in cfg["fields"].get("tool_input", ("tool_input",)): value = raw.get(key) if value is not None: @@ -424,8 +442,8 @@ def hook_entry_config(cfg, canonical_events, command, matcher=None, *, fail_clos def cursor_wire(raw): """The wire event name, inferred from shape when the payload names none.""" - name = raw.get("hook_event_name") - if name is None: + name = wire_name_of(raw, "hook_event_name") + if name is None and isinstance(raw, dict): return "afterFileEdit" if isinstance(raw.get("edits"), list) else "beforeShellExecution" return name @@ -433,7 +451,7 @@ def cursor_claims(cfg, raw): """True when this payload looks like Cursor's shape.""" if not isinstance(raw, dict): return False - name = raw.get("hook_event_name") + name = wire_name_of(raw, "hook_event_name") if name in cfg["events"]: if name in _AMBIGUOUS_NAMES: return any(k in raw for k in _MARKERS) @@ -458,7 +476,7 @@ def _wire_of(cfg, event): `tool` is read only for an Event carrying no payload, where `parse` left the inferred name there; without a payload there is nothing to re-infer from. """ - name = (event.raw or {}).get("hook_event_name") + name = wire_name_of(event.raw, "hook_event_name") if name in cfg["events"]: return name if event.raw: @@ -913,6 +931,43 @@ def _emit(out, text): out.flush() +def _report(text): + # Best effort, never raising: a diagnostic must not pre-empt the refusal it accompanies. + # A hook's stderr is whatever the host gave it -- closed (then sys.stderr is None, and + # the traceback module's default print would land on STDOUT, inside the verdict), a + # console code page that cannot hold the payload text quoted in the exception, a pipe + # nobody reads. + try: + sys.stderr.write(text) + sys.stderr.flush() + except Exception: + return + + +def _decide(raw): + """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" + event = parse(raw) + if event.event == UNKNOWN: + # A vendor event this adapter has no mapping for. handle() is not called: it + # reasons about the canonical vocabulary, and handing it something outside that + # vocabulary invites a decision made on a false premise. + return "", 0 + try: + decision = _coerce(handle(event)) + 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 + # failure is printed for the operator and the gate answers deny in 'cursor''s + # own dialect. Only the failure's class is named to the host: its message may quote + # the very payload content the policy was inspecting. + _report(traceback.format_exc()) + decision = Decision.deny( + "policy handler failed (%s); refusing rather than allowing what it could not judge" + % sys.exc_info()[0].__name__ + ) + return respond(degrade(decision, event), event) + + def main(stdin=None, stdout=None, exit=True): """Read one payload from stdin, dispatch, emit the 'cursor' response, exit.""" stream = stdin if stdin is not None else sys.stdin @@ -924,16 +979,14 @@ def main(stdin=None, stdout=None, exit=True): if exit: sys.exit(0) return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) + try: + text, code = _decide(raw) + except Exception: + # Past the handler, which _decide() already answers for: a fault in this file on a + # payload it did decode. With no Event to answer in dialect, the one refusal left is + # the host's blocking exit code, and nothing on stdout to be misread as a verdict. + _report(traceback.format_exc()) + text, code = "", 2 if text: _emit(out, text) if exit: diff --git a/.chock/bin/devin.py b/.chock/bin/devin.py index 860ee582..4c468978 100755 --- a/.chock/bin/devin.py +++ b/.chock/bin/devin.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.0 -- bundle("devin"). Do not hand-edit, except the +# Generated by agentseam 0.3.1 -- bundle("devin"). 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. # @@ -18,10 +18,11 @@ import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.0) +# contract (agentseam 0.3.1) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -202,7 +203,9 @@ def tool_input_of(raw): if isinstance(raw, str) and raw[:1] == "{": try: parsed = _json.loads(raw) - except _json.JSONDecodeError: + except (ValueError, RecursionError): + # JSONDecodeError is a ValueError; a string nested past the interpreter's + # limit raises RecursionError instead, and neither is an input to crash on. return {} if isinstance(parsed, dict): return parsed @@ -229,9 +232,22 @@ def looks_like_claude_code(raw): PROBES = {"looks_like_claude_code": looks_like_claude_code} +UNREADABLE_NAME = "" + +def wire_name_of(raw, key): + """The event name under `key`: text, None when absent, UNREADABLE_NAME when not text. + + Total over anything `json.loads` can return: a payload that is not an object names no + event, and a name that is not a string is not one any adapter can map. + """ + name = raw.get(key) if isinstance(raw, dict) else None + if name is None or isinstance(name, str): + return name + return UNREADABLE_NAME + def _wire_name(cfg, raw): for key in cfg["claims"].get("event_key", ()): - name = raw.get(key) + name = wire_name_of(raw, key) if name is not None: return name return None @@ -316,6 +332,8 @@ def _field(raw, ti, chain): _FIELD_META = ("tool_input", "content_only_for_write_tools", "stringify") def _tool_input_raw(cfg, raw): + if not isinstance(raw, dict): + return None for key in cfg["fields"].get("tool_input", ("tool_input",)): value = raw.get(key) if value is not None: @@ -976,6 +994,43 @@ def _emit(out, text): out.flush() +def _report(text): + # Best effort, never raising: a diagnostic must not pre-empt the refusal it accompanies. + # A hook's stderr is whatever the host gave it -- closed (then sys.stderr is None, and + # the traceback module's default print would land on STDOUT, inside the verdict), a + # console code page that cannot hold the payload text quoted in the exception, a pipe + # nobody reads. + try: + sys.stderr.write(text) + sys.stderr.flush() + except Exception: + return + + +def _decide(raw): + """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" + event = parse(raw) + if event.event == UNKNOWN: + # A vendor event this adapter has no mapping for. handle() is not called: it + # reasons about the canonical vocabulary, and handing it something outside that + # vocabulary invites a decision made on a false premise. + return "", 0 + try: + decision = _coerce(handle(event)) + 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 + # failure is printed for the operator and the gate answers deny in 'devin''s + # own dialect. Only the failure's class is named to the host: its message may quote + # the very payload content the policy was inspecting. + _report(traceback.format_exc()) + decision = Decision.deny( + "policy handler failed (%s); refusing rather than allowing what it could not judge" + % sys.exc_info()[0].__name__ + ) + return respond(degrade(decision, event), event) + + def main(stdin=None, stdout=None, exit=True): """Read one payload from stdin, dispatch, emit the 'devin' response, exit.""" stream = stdin if stdin is not None else sys.stdin @@ -987,16 +1042,14 @@ def main(stdin=None, stdout=None, exit=True): if exit: sys.exit(0) return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) + try: + text, code = _decide(raw) + except Exception: + # Past the handler, which _decide() already answers for: a fault in this file on a + # payload it did decode. With no Event to answer in dialect, the one refusal left is + # the host's blocking exit code, and nothing on stdout to be misread as a verdict. + _report(traceback.format_exc()) + text, code = "", 2 if text: _emit(out, text) if exit: diff --git a/.chock/bin/gemini_cli.py b/.chock/bin/gemini_cli.py index 5cdf2cc2..db789c29 100755 --- a/.chock/bin/gemini_cli.py +++ b/.chock/bin/gemini_cli.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.0 -- bundle("gemini_cli"). Do not hand-edit, except the +# Generated by agentseam 0.3.1 -- bundle("gemini_cli"). 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. # @@ -18,10 +18,11 @@ import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.0) +# contract (agentseam 0.3.1) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -202,7 +203,9 @@ def tool_input_of(raw): if isinstance(raw, str) and raw[:1] == "{": try: parsed = _json.loads(raw) - except _json.JSONDecodeError: + except (ValueError, RecursionError): + # JSONDecodeError is a ValueError; a string nested past the interpreter's + # limit raises RecursionError instead, and neither is an input to crash on. return {} if isinstance(parsed, dict): return parsed @@ -229,9 +232,22 @@ def looks_like_claude_code(raw): PROBES = {"looks_like_claude_code": looks_like_claude_code} +UNREADABLE_NAME = "" + +def wire_name_of(raw, key): + """The event name under `key`: text, None when absent, UNREADABLE_NAME when not text. + + Total over anything `json.loads` can return: a payload that is not an object names no + event, and a name that is not a string is not one any adapter can map. + """ + name = raw.get(key) if isinstance(raw, dict) else None + if name is None or isinstance(name, str): + return name + return UNREADABLE_NAME + def _wire_name(cfg, raw): for key in cfg["claims"].get("event_key", ()): - name = raw.get(key) + name = wire_name_of(raw, key) if name is not None: return name return None @@ -316,6 +332,8 @@ def _field(raw, ti, chain): _FIELD_META = ("tool_input", "content_only_for_write_tools", "stringify") def _tool_input_raw(cfg, raw): + if not isinstance(raw, dict): + return None for key in cfg["fields"].get("tool_input", ("tool_input",)): value = raw.get(key) if value is not None: @@ -976,6 +994,43 @@ def _emit(out, text): out.flush() +def _report(text): + # Best effort, never raising: a diagnostic must not pre-empt the refusal it accompanies. + # A hook's stderr is whatever the host gave it -- closed (then sys.stderr is None, and + # the traceback module's default print would land on STDOUT, inside the verdict), a + # console code page that cannot hold the payload text quoted in the exception, a pipe + # nobody reads. + try: + sys.stderr.write(text) + sys.stderr.flush() + except Exception: + return + + +def _decide(raw): + """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" + event = parse(raw) + if event.event == UNKNOWN: + # A vendor event this adapter has no mapping for. handle() is not called: it + # reasons about the canonical vocabulary, and handing it something outside that + # vocabulary invites a decision made on a false premise. + return "", 0 + try: + decision = _coerce(handle(event)) + 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 + # failure is printed for the operator and the gate answers deny in 'gemini_cli''s + # own dialect. Only the failure's class is named to the host: its message may quote + # the very payload content the policy was inspecting. + _report(traceback.format_exc()) + decision = Decision.deny( + "policy handler failed (%s); refusing rather than allowing what it could not judge" + % sys.exc_info()[0].__name__ + ) + return respond(degrade(decision, event), event) + + def main(stdin=None, stdout=None, exit=True): """Read one payload from stdin, dispatch, emit the 'gemini_cli' response, exit.""" stream = stdin if stdin is not None else sys.stdin @@ -987,16 +1042,14 @@ def main(stdin=None, stdout=None, exit=True): if exit: sys.exit(0) return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) + try: + text, code = _decide(raw) + except Exception: + # Past the handler, which _decide() already answers for: a fault in this file on a + # payload it did decode. With no Event to answer in dialect, the one refusal left is + # the host's blocking exit code, and nothing on stdout to be misread as a verdict. + _report(traceback.format_exc()) + text, code = "", 2 if text: _emit(out, text) if exit: diff --git a/.chock/bin/grok.py b/.chock/bin/grok.py index 0f4fceb0..8f09b67c 100755 --- a/.chock/bin/grok.py +++ b/.chock/bin/grok.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.0 -- bundle("grok"). Do not hand-edit, except the +# Generated by agentseam 0.3.1 -- bundle("grok"). 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. # @@ -18,10 +18,11 @@ import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.0) +# contract (agentseam 0.3.1) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -202,7 +203,9 @@ def tool_input_of(raw): if isinstance(raw, str) and raw[:1] == "{": try: parsed = _json.loads(raw) - except _json.JSONDecodeError: + except (ValueError, RecursionError): + # JSONDecodeError is a ValueError; a string nested past the interpreter's + # limit raises RecursionError instead, and neither is an input to crash on. return {} if isinstance(parsed, dict): return parsed @@ -212,9 +215,22 @@ def tool_input_of(raw): # ------------------------------------------------------------------------------ # flat_decision family engine (trimmed to what this entry uses) +UNREADABLE_NAME = "" + +def wire_name_of(raw, key): + """The event name under `key`: text, None when absent, UNREADABLE_NAME when not text. + + Total over anything `json.loads` can return: a payload that is not an object names no + event, and a name that is not a string is not one any adapter can map. + """ + name = raw.get(key) if isinstance(raw, dict) else None + if name is None or isinstance(name, str): + return name + return UNREADABLE_NAME + def _wire_name(cfg, raw): for key in cfg["claims"].get("event_key", ()): - name = raw.get(key) + name = wire_name_of(raw, key) if name is not None: return name return None @@ -299,6 +315,8 @@ def _field(raw, ti, chain): _FIELD_META = ("tool_input", "content_only_for_write_tools", "stringify") def _tool_input_raw(cfg, raw): + if not isinstance(raw, dict): + return None for key in cfg["fields"].get("tool_input", ("tool_input",)): value = raw.get(key) if value is not None: @@ -959,6 +977,43 @@ def _emit(out, text): out.flush() +def _report(text): + # Best effort, never raising: a diagnostic must not pre-empt the refusal it accompanies. + # A hook's stderr is whatever the host gave it -- closed (then sys.stderr is None, and + # the traceback module's default print would land on STDOUT, inside the verdict), a + # console code page that cannot hold the payload text quoted in the exception, a pipe + # nobody reads. + try: + sys.stderr.write(text) + sys.stderr.flush() + except Exception: + return + + +def _decide(raw): + """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" + event = parse(raw) + if event.event == UNKNOWN: + # A vendor event this adapter has no mapping for. handle() is not called: it + # reasons about the canonical vocabulary, and handing it something outside that + # vocabulary invites a decision made on a false premise. + return "", 0 + try: + decision = _coerce(handle(event)) + 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 + # failure is printed for the operator and the gate answers deny in 'grok''s + # own dialect. Only the failure's class is named to the host: its message may quote + # the very payload content the policy was inspecting. + _report(traceback.format_exc()) + decision = Decision.deny( + "policy handler failed (%s); refusing rather than allowing what it could not judge" + % sys.exc_info()[0].__name__ + ) + return respond(degrade(decision, event), event) + + def main(stdin=None, stdout=None, exit=True): """Read one payload from stdin, dispatch, emit the 'grok' response, exit.""" stream = stdin if stdin is not None else sys.stdin @@ -970,16 +1025,14 @@ def main(stdin=None, stdout=None, exit=True): if exit: sys.exit(0) return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) + try: + text, code = _decide(raw) + except Exception: + # Past the handler, which _decide() already answers for: a fault in this file on a + # payload it did decode. With no Event to answer in dialect, the one refusal left is + # the host's blocking exit code, and nothing on stdout to be misread as a verdict. + _report(traceback.format_exc()) + text, code = "", 2 if text: _emit(out, text) if exit: diff --git a/.chock/bin/tabnine.py b/.chock/bin/tabnine.py index 88c2225d..bcb4148f 100755 --- a/.chock/bin/tabnine.py +++ b/.chock/bin/tabnine.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.0 -- bundle("tabnine"). Do not hand-edit, except the +# Generated by agentseam 0.3.1 -- bundle("tabnine"). 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. # @@ -18,10 +18,11 @@ import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.0) +# contract (agentseam 0.3.1) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -202,7 +203,9 @@ def tool_input_of(raw): if isinstance(raw, str) and raw[:1] == "{": try: parsed = _json.loads(raw) - except _json.JSONDecodeError: + except (ValueError, RecursionError): + # JSONDecodeError is a ValueError; a string nested past the interpreter's + # limit raises RecursionError instead, and neither is an input to crash on. return {} if isinstance(parsed, dict): return parsed @@ -212,9 +215,22 @@ def tool_input_of(raw): # ------------------------------------------------------------------------------ # flat_decision family engine (trimmed to what this entry uses) +UNREADABLE_NAME = "" + +def wire_name_of(raw, key): + """The event name under `key`: text, None when absent, UNREADABLE_NAME when not text. + + Total over anything `json.loads` can return: a payload that is not an object names no + event, and a name that is not a string is not one any adapter can map. + """ + name = raw.get(key) if isinstance(raw, dict) else None + if name is None or isinstance(name, str): + return name + return UNREADABLE_NAME + def _wire_name(cfg, raw): for key in cfg["claims"].get("event_key", ()): - name = raw.get(key) + name = wire_name_of(raw, key) if name is not None: return name return None @@ -299,6 +315,8 @@ def _field(raw, ti, chain): _FIELD_META = ("tool_input", "content_only_for_write_tools", "stringify") def _tool_input_raw(cfg, raw): + if not isinstance(raw, dict): + return None for key in cfg["fields"].get("tool_input", ("tool_input",)): value = raw.get(key) if value is not None: @@ -959,6 +977,43 @@ def _emit(out, text): out.flush() +def _report(text): + # Best effort, never raising: a diagnostic must not pre-empt the refusal it accompanies. + # A hook's stderr is whatever the host gave it -- closed (then sys.stderr is None, and + # the traceback module's default print would land on STDOUT, inside the verdict), a + # console code page that cannot hold the payload text quoted in the exception, a pipe + # nobody reads. + try: + sys.stderr.write(text) + sys.stderr.flush() + except Exception: + return + + +def _decide(raw): + """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" + event = parse(raw) + if event.event == UNKNOWN: + # A vendor event this adapter has no mapping for. handle() is not called: it + # reasons about the canonical vocabulary, and handing it something outside that + # vocabulary invites a decision made on a false premise. + return "", 0 + try: + decision = _coerce(handle(event)) + 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 + # failure is printed for the operator and the gate answers deny in 'tabnine''s + # own dialect. Only the failure's class is named to the host: its message may quote + # the very payload content the policy was inspecting. + _report(traceback.format_exc()) + decision = Decision.deny( + "policy handler failed (%s); refusing rather than allowing what it could not judge" + % sys.exc_info()[0].__name__ + ) + return respond(degrade(decision, event), event) + + def main(stdin=None, stdout=None, exit=True): """Read one payload from stdin, dispatch, emit the 'tabnine' response, exit.""" stream = stdin if stdin is not None else sys.stdin @@ -970,16 +1025,14 @@ def main(stdin=None, stdout=None, exit=True): if exit: sys.exit(0) return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) + try: + text, code = _decide(raw) + except Exception: + # Past the handler, which _decide() already answers for: a fault in this file on a + # payload it did decode. With no Event to answer in dialect, the one refusal left is + # the host's blocking exit code, and nothing on stdout to be misread as a verdict. + _report(traceback.format_exc()) + text, code = "", 2 if text: _emit(out, text) if exit: diff --git a/.chock/bin/vscode_copilot.py b/.chock/bin/vscode_copilot.py index 2cc45d96..6218e705 100755 --- a/.chock/bin/vscode_copilot.py +++ b/.chock/bin/vscode_copilot.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.0 -- bundle("vscode_copilot"). Do not hand-edit, except the +# Generated by agentseam 0.3.1 -- bundle("vscode_copilot"). 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. # @@ -18,10 +18,11 @@ import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.0) +# contract (agentseam 0.3.1) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -202,7 +203,9 @@ def tool_input_of(raw): if isinstance(raw, str) and raw[:1] == "{": try: parsed = _json.loads(raw) - except _json.JSONDecodeError: + except (ValueError, RecursionError): + # JSONDecodeError is a ValueError; a string nested past the interpreter's + # limit raises RecursionError instead, and neither is an input to crash on. return {} if isinstance(parsed, dict): return parsed @@ -264,11 +267,22 @@ def powershell_command(command): _VSCODE_ENVELOPE = "timestamp" +def _name(raw): + """The payload's own event name as text; None when it names none, UNKNOWN when the + value is not text (a list or object cannot be looked up, and would raise if tried).""" + if not isinstance(raw, dict): + return None + name = raw.get("hook_event_name") or raw.get("hookEventName") + if name is None or isinstance(name, str): + return name + return UNKNOWN + + def claims(raw): """True for a payload from either product.""" if not isinstance(raw, dict): return False - name = raw.get("hook_event_name") or raw.get("hookEventName") + name = _name(raw) if name in EVENT_MAP and _VSCODE_ENVELOPE in raw and "turn_id" not in raw: return True if any(k in raw for k in _CODEX_MARKERS + _CURSOR_MARKERS): @@ -280,6 +294,8 @@ def claims(raw): def parse(raw): + if not isinstance(raw, dict): + return Event(AGENT, UNKNOWN, raw=raw) ti = raw.get("tool_input") ti = tool_input_of(ti) tool = raw.get("tool_name") or raw.get("toolName") @@ -293,7 +309,7 @@ def parse(raw): else: path = ti.get("filePath") or ti.get("file_path") or ti.get("path") content = ti.get("content") or ti.get("newText") or ti.get("new_str") - name = raw.get("hook_event_name") or raw.get("hookEventName") or "preToolUse" + name = _name(raw) or "preToolUse" return Event( AGENT, EVENT_MAP.get(name, UNKNOWN), @@ -324,8 +340,7 @@ def is_memory_write(event): def _echoed_name(event): """This event's own vendor spelling, out of the payload; VS Code's name if there is none.""" - raw = event.raw or {} - return raw.get("hook_event_name") or raw.get("hookEventName") or REVERSE_EVENT_MAP.get(event.event, "PreToolUse") + return _name(event.raw) or REVERSE_EVENT_MAP.get(event.event, "PreToolUse") def _refusal_reason(decision): @@ -787,6 +802,43 @@ def _emit(out, text): out.flush() +def _report(text): + # Best effort, never raising: a diagnostic must not pre-empt the refusal it accompanies. + # A hook's stderr is whatever the host gave it -- closed (then sys.stderr is None, and + # the traceback module's default print would land on STDOUT, inside the verdict), a + # console code page that cannot hold the payload text quoted in the exception, a pipe + # nobody reads. + try: + sys.stderr.write(text) + sys.stderr.flush() + except Exception: + return + + +def _decide(raw): + """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" + event = parse(raw) + if event.event == UNKNOWN: + # A vendor event this adapter has no mapping for. handle() is not called: it + # reasons about the canonical vocabulary, and handing it something outside that + # vocabulary invites a decision made on a false premise. + return "", 0 + try: + decision = _coerce(handle(event)) + 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 + # failure is printed for the operator and the gate answers deny in 'vscode_copilot''s + # own dialect. Only the failure's class is named to the host: its message may quote + # the very payload content the policy was inspecting. + _report(traceback.format_exc()) + decision = Decision.deny( + "policy handler failed (%s); refusing rather than allowing what it could not judge" + % sys.exc_info()[0].__name__ + ) + return respond(degrade(decision, event), event) + + def main(stdin=None, stdout=None, exit=True): """Read one payload from stdin, dispatch, emit the 'vscode_copilot' response, exit.""" stream = stdin if stdin is not None else sys.stdin @@ -798,16 +850,14 @@ def main(stdin=None, stdout=None, exit=True): if exit: sys.exit(0) return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) + try: + text, code = _decide(raw) + except Exception: + # Past the handler, which _decide() already answers for: a fault in this file on a + # payload it did decode. With no Event to answer in dialect, the one refusal left is + # the host's blocking exit code, and nothing on stdout to be misread as a verdict. + _report(traceback.format_exc()) + text, code = "", 2 if text: _emit(out, text) if exit: diff --git a/.chock/bin/windsurf.py b/.chock/bin/windsurf.py index 538b028e..90ae1820 100755 --- a/.chock/bin/windsurf.py +++ b/.chock/bin/windsurf.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.0 -- bundle("windsurf"). Do not hand-edit, except the +# Generated by agentseam 0.3.1 -- bundle("windsurf"). 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. # @@ -18,10 +18,11 @@ import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.0) +# contract (agentseam 0.3.1) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -202,7 +203,9 @@ def tool_input_of(raw): if isinstance(raw, str) and raw[:1] == "{": try: parsed = _json.loads(raw) - except _json.JSONDecodeError: + except (ValueError, RecursionError): + # JSONDecodeError is a ValueError; a string nested past the interpreter's + # limit raises RecursionError instead, and neither is an input to crash on. return {} if isinstance(parsed, dict): return parsed @@ -212,9 +215,22 @@ def tool_input_of(raw): # ------------------------------------------------------------------------------ # windsurf family engine (trimmed to what this entry uses) +UNREADABLE_NAME = "" + +def wire_name_of(raw, key): + """The event name under `key`: text, None when absent, UNREADABLE_NAME when not text. + + Total over anything `json.loads` can return: a payload that is not an object names no + event, and a name that is not a string is not one any adapter can map. + """ + name = raw.get(key) if isinstance(raw, dict) else None + if name is None or isinstance(name, str): + return name + return UNREADABLE_NAME + def _wire_name(cfg, raw): for key in cfg["claims"].get("event_key", ()): - name = raw.get(key) + name = wire_name_of(raw, key) if name is not None: return name return None @@ -260,6 +276,8 @@ def _field(raw, ti, chain): _FIELD_META = ("tool_input", "content_only_for_write_tools", "stringify") def _tool_input_raw(cfg, raw): + if not isinstance(raw, dict): + return None for key in cfg["fields"].get("tool_input", ("tool_input",)): value = raw.get(key) if value is not None: @@ -448,25 +466,28 @@ def hook_entry_config(cfg, canonical_events, command, matcher=None, *, fail_clos _MCP_EVENTS = ("pre_mcp_tool_use", "post_mcp_tool_use") +def _tool_info(raw): + info = raw.get("tool_info") if isinstance(raw, dict) else None + return info if isinstance(info, dict) else {} + def windsurf_wire(raw): """The wire event name, inferred from `tool_info` when the payload names none.""" - name = raw.get("hook_event_name") - if name is not None: + name = wire_name_of(raw, "hook_event_name") + if name is not None or not isinstance(raw, dict): return name - info = raw.get("tool_info") or {} - return "pre_run_command" if info.get("command_line") else "pre_user_prompt" + return "pre_run_command" if _tool_info(raw).get("command_line") else "pre_user_prompt" def windsurf_claims(cfg, raw): if not isinstance(raw, dict): return False - if raw.get("hook_event_name") in cfg["events"]: + if wire_name_of(raw, "hook_event_name") in cfg["events"]: return True return "trajectory_id" in raw and isinstance(raw.get("tool_info"), dict) def windsurf_parse(cfg, raw): name = windsurf_wire(raw) event = hj_parse(cfg, raw, wire=name) - info = raw.get("tool_info") or {} + info = _tool_info(raw) if name in _MCP_EVENTS: joined = "%s/%s" % (info["server"], info["tool"]) if info.get("server") and info.get("tool") else None event.tool = joined or info.get("tool") @@ -870,6 +891,43 @@ def _emit(out, text): out.flush() +def _report(text): + # Best effort, never raising: a diagnostic must not pre-empt the refusal it accompanies. + # A hook's stderr is whatever the host gave it -- closed (then sys.stderr is None, and + # the traceback module's default print would land on STDOUT, inside the verdict), a + # console code page that cannot hold the payload text quoted in the exception, a pipe + # nobody reads. + try: + sys.stderr.write(text) + sys.stderr.flush() + except Exception: + return + + +def _decide(raw): + """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" + event = parse(raw) + if event.event == UNKNOWN: + # A vendor event this adapter has no mapping for. handle() is not called: it + # reasons about the canonical vocabulary, and handing it something outside that + # vocabulary invites a decision made on a false premise. + return "", 0 + try: + decision = _coerce(handle(event)) + 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 + # failure is printed for the operator and the gate answers deny in 'windsurf''s + # own dialect. Only the failure's class is named to the host: its message may quote + # the very payload content the policy was inspecting. + _report(traceback.format_exc()) + decision = Decision.deny( + "policy handler failed (%s); refusing rather than allowing what it could not judge" + % sys.exc_info()[0].__name__ + ) + return respond(degrade(decision, event), event) + + def main(stdin=None, stdout=None, exit=True): """Read one payload from stdin, dispatch, emit the 'windsurf' response, exit.""" stream = stdin if stdin is not None else sys.stdin @@ -881,16 +939,14 @@ def main(stdin=None, stdout=None, exit=True): if exit: sys.exit(0) return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) + try: + text, code = _decide(raw) + except Exception: + # Past the handler, which _decide() already answers for: a fault in this file on a + # payload it did decode. With no Event to answer in dialect, the one refusal left is + # the host's blocking exit code, and nothing on stdout to be misread as a verdict. + _report(traceback.format_exc()) + text, code = "", 2 if text: _emit(out, text) if exit: diff --git a/CHANGELOG.md b/CHANGELOG.md index d80e00c9..13e95aa8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -101,6 +101,16 @@ `.devin/`, ...) and vendored ten runtimes; the SessionStart arm hook was installed whether or not claude was named (`recompile.wired_vendors()`). +- **`agentseam` bumped to 0.3.1.** Its own `dispatch.handle()` now refuses a handler that + raises in the vendor's dialect, instead of letting the exception escape; every vendored + runtime's `main()` gains the matching `_decide()`/`_report()` wrapping, so a policy handler + that raises is answered with a deny instead of an unhandled traceback. `dispatch.py.tmpl`'s + own try/except in `gate/runtime_bundle.py` already refused on failure from the outside and + is now redundant with agentseam's inner one for the vendored runtimes this repo emits; it + stays for this release as defense in depth rather than being pulled the same cycle as the + bump. 0.3.1 also writes `dump()` output as LF on every platform, which chock's own code + paths never call. Every adopter's next `chock sync` rewrites `.chock/bin/*.py`. + ## 0.9.0 — In-session enforcement for content policies: the write path, the `stop` backstop, and the waiver that could not be self-served - **The vendored runtime refuses when it cannot decide.** `handle()` in every diff --git a/pyproject.toml b/pyproject.toml index 9becc334..8c44ca96 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -60,7 +60,7 @@ dependencies = [ # It never ships to an adopter repo -- the vendored PreToolUse/SessionStart runners # stay self-contained stdlib-only files, unaffected by this dependency. Pinned exact # per plan/spine-a/contract.md: the wave boundary is the published PyPI artifact. - "agentseam==0.3.0", + "agentseam==0.3.1", ] [project.optional-dependencies] diff --git a/requirements/brand-assets.in b/requirements/brand-assets.in index 4092b19c..488253e6 100644 --- a/requirements/brand-assets.in +++ b/requirements/brand-assets.in @@ -1,5 +1,5 @@ pyyaml>=6.0 jsonschema>=4.18,<5 referencing>=0.35,<0.38 -agentseam==0.3.0 +agentseam==0.3.1 cairosvg==2.9.0 diff --git a/requirements/brand-assets.txt b/requirements/brand-assets.txt index bfdb9a6b..6c258c6a 100644 --- a/requirements/brand-assets.txt +++ b/requirements/brand-assets.txt @@ -4,9 +4,9 @@ # # pip-compile --generate-hashes --output-file=requirements/brand-assets.txt --strip-extras requirements/brand-assets.in # -agentseam==0.3.0 \ - --hash=sha256:04ceaca8671b7720bb174106232bf3d2a6a2eb08fcae2c02cc27274333f275da \ - --hash=sha256:6efa82835bfb8365ad1687dd00fa7e18326b8d9153764444ccb697d807e77d3b +agentseam==0.3.1 \ + --hash=sha256:4e5b249162204f0bd90c60e2c67a8417c999881a786d4d18a4e85dca570c6564 \ + --hash=sha256:db7099395724122989ef8fbf1115d09da2cb0194fff9ce2a3ebe979793fc5271 # via -r requirements/brand-assets.in attrs==26.1.0 \ --hash=sha256:c647aa4a12dfbad9333ca4e71fe62ddc36f4e63b2d260a37a8b83d2f043ac309 \ diff --git a/src/chock/scaffold/recompile.py b/src/chock/scaffold/recompile.py index d6927123..c8ccee97 100644 --- a/src/chock/scaffold/recompile.py +++ b/src/chock/scaffold/recompile.py @@ -14,6 +14,7 @@ from chock.emit import write_generated_json from chock.hooks.in_agent_install import WIRED_VENDORS, install_hooks, install_label, installed_policy_ids from chock.hooks.installers import get_hooks_dir, install_policy_hooks +from chock.hooks.runtime_vendor import runtime_rel from chock.hooks.sessionstart_install import install_sessionstart_hook from chock.index.cli import cmd_refresh from chock.output import warn @@ -60,6 +61,23 @@ def wired_vendors(agents: list[str]) -> tuple[str, ...]: return tuple(v for v in WIRED_VENDORS if v in chosen) +def _prune_unwired_runtimes(repo_root: Path, wired: tuple[str, ...]) -> None: + """Delete a vendored runtime for a vendor `wired` no longer names. + + `install_hooks`/`vendor_runtime` only ever write a runtime for a wired vendor, so one a + vendor lost (dropped from `supported_agents`, or never in `CHOCK_AGENT`) is never + rewritten and never removed either -- it sits stale, drifting against every future + render, and `vendored_differences` still checks any file it finds on disk. + """ + for vendor in WIRED_VENDORS: + if vendor in wired: + continue + stale = repo_root / runtime_rel(vendor) + if stale.exists(): + stale.unlink() + print(f"Removed {runtime_rel(vendor).as_posix()} ({vendor} not in supported_agents)") + + def compiled_differences(repo_root: Path | str, agents: list[str]) -> list[str]: """Every way the committed compiled tree differs from what the manifests produce now.""" repo_root = Path(repo_root) @@ -161,6 +179,7 @@ def recompile(repo_root: Path | str, agents: list[str], *, skip_hooks: bool = Fa install_policy_hooks(repo_root, get_hooks_dir(repo_root)) wired = wired_vendors(agents) + _prune_unwired_runtimes(repo_root, wired) try: if CHOCK_AGENT["claude"] in wired and install_sessionstart_hook(repo_root): print("Registered SessionStart arm hook in .claude/settings.json") diff --git a/tests/fixtures/runtime_goldens/antigravity.py b/tests/fixtures/runtime_goldens/antigravity.py index 647b4245..bfaa5c80 100644 --- a/tests/fixtures/runtime_goldens/antigravity.py +++ b/tests/fixtures/runtime_goldens/antigravity.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.0 -- bundle("antigravity"). Do not hand-edit, except the +# Generated by agentseam 0.3.1 -- 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. # @@ -18,10 +18,11 @@ import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.0) +# contract (agentseam 0.3.1) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -202,7 +203,9 @@ def tool_input_of(raw): if isinstance(raw, str) and raw[:1] == "{": try: parsed = _json.loads(raw) - except _json.JSONDecodeError: + except (ValueError, RecursionError): + # JSONDecodeError is a ValueError; a string nested past the interpreter's + # limit raises RecursionError instead, and neither is an input to crash on. return {} if isinstance(parsed, dict): return parsed @@ -212,9 +215,22 @@ def tool_input_of(raw): # ------------------------------------------------------------------------------ # antigravity family engine (trimmed to what this entry uses) +UNREADABLE_NAME = "" + +def wire_name_of(raw, key): + """The event name under `key`: text, None when absent, UNREADABLE_NAME when not text. + + Total over anything `json.loads` can return: a payload that is not an object names no + event, and a name that is not a string is not one any adapter can map. + """ + name = raw.get(key) if isinstance(raw, dict) else None + if name is None or isinstance(name, str): + return name + return UNREADABLE_NAME + def _wire_name(cfg, raw): for key in cfg["claims"].get("event_key", ()): - name = raw.get(key) + name = wire_name_of(raw, key) if name is not None: return name return None @@ -260,6 +276,8 @@ def _field(raw, ti, chain): _FIELD_META = ("tool_input", "content_only_for_write_tools", "stringify") def _tool_input_raw(cfg, raw): + if not isinstance(raw, dict): + return None for key in cfg["fields"].get("tool_input", ("tool_input",)): value = raw.get(key) if value is not None: @@ -536,6 +554,8 @@ def hook_entry_config(cfg, canonical_events, command, matcher=None, *, fail_clos def antigravity_wire(raw): """Name the event from shape; ties go to PreToolUse so the gate stays a gate.""" + if not isinstance(raw, dict): + return None if "terminationReason" in raw or "fullyIdle" in raw: return "Stop" if isinstance(raw.get("toolCall"), dict): @@ -940,6 +960,43 @@ def _emit(out, text): out.flush() +def _report(text): + # Best effort, never raising: a diagnostic must not pre-empt the refusal it accompanies. + # A hook's stderr is whatever the host gave it -- closed (then sys.stderr is None, and + # the traceback module's default print would land on STDOUT, inside the verdict), a + # console code page that cannot hold the payload text quoted in the exception, a pipe + # nobody reads. + try: + sys.stderr.write(text) + sys.stderr.flush() + except Exception: + return + + +def _decide(raw): + """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" + event = parse(raw) + if event.event == UNKNOWN: + # A vendor event this adapter has no mapping for. handle() is not called: it + # reasons about the canonical vocabulary, and handing it something outside that + # vocabulary invites a decision made on a false premise. + return "", 0 + try: + decision = _coerce(handle(event)) + 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 + # failure is printed for the operator and the gate answers deny in 'antigravity''s + # own dialect. Only the failure's class is named to the host: its message may quote + # the very payload content the policy was inspecting. + _report(traceback.format_exc()) + decision = Decision.deny( + "policy handler failed (%s); refusing rather than allowing what it could not judge" + % sys.exc_info()[0].__name__ + ) + return respond(degrade(decision, event), event) + + def main(stdin=None, stdout=None, exit=True): """Read one payload from stdin, dispatch, emit the 'antigravity' response, exit.""" stream = stdin if stdin is not None else sys.stdin @@ -951,16 +1008,14 @@ def main(stdin=None, stdout=None, exit=True): if exit: sys.exit(0) return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) + try: + text, code = _decide(raw) + except Exception: + # Past the handler, which _decide() already answers for: a fault in this file on a + # payload it did decode. With no Event to answer in dialect, the one refusal left is + # the host's blocking exit code, and nothing on stdout to be misread as a verdict. + _report(traceback.format_exc()) + text, code = "", 2 if text: _emit(out, text) if exit: diff --git a/tests/fixtures/runtime_goldens/claude_code.py b/tests/fixtures/runtime_goldens/claude_code.py index 6185c6f1..e4c6207b 100644 --- a/tests/fixtures/runtime_goldens/claude_code.py +++ b/tests/fixtures/runtime_goldens/claude_code.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.0 -- bundle("claude_code"). Do not hand-edit, except the +# Generated by agentseam 0.3.1 -- 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. # @@ -19,10 +19,11 @@ import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.0) +# contract (agentseam 0.3.1) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -203,7 +204,9 @@ def tool_input_of(raw): if isinstance(raw, str) and raw[:1] == "{": try: parsed = _json.loads(raw) - except _json.JSONDecodeError: + except (ValueError, RecursionError): + # JSONDecodeError is a ValueError; a string nested past the interpreter's + # limit raises RecursionError instead, and neither is an input to crash on. return {} if isinstance(parsed, dict): return parsed @@ -230,9 +233,22 @@ def looks_like_claude_code(raw): PROBES = {"looks_like_claude_code": looks_like_claude_code} +UNREADABLE_NAME = "" + +def wire_name_of(raw, key): + """The event name under `key`: text, None when absent, UNREADABLE_NAME when not text. + + Total over anything `json.loads` can return: a payload that is not an object names no + event, and a name that is not a string is not one any adapter can map. + """ + name = raw.get(key) if isinstance(raw, dict) else None + if name is None or isinstance(name, str): + return name + return UNREADABLE_NAME + def _wire_name(cfg, raw): for key in cfg["claims"].get("event_key", ()): - name = raw.get(key) + name = wire_name_of(raw, key) if name is not None: return name return None @@ -317,6 +333,8 @@ def _field(raw, ti, chain): _FIELD_META = ("tool_input", "content_only_for_write_tools", "stringify") def _tool_input_raw(cfg, raw): + if not isinstance(raw, dict): + return None for key in cfg["fields"].get("tool_input", ("tool_input",)): value = raw.get(key) if value is not None: @@ -1076,6 +1094,43 @@ def _emit(out, text): out.flush() +def _report(text): + # Best effort, never raising: a diagnostic must not pre-empt the refusal it accompanies. + # A hook's stderr is whatever the host gave it -- closed (then sys.stderr is None, and + # the traceback module's default print would land on STDOUT, inside the verdict), a + # console code page that cannot hold the payload text quoted in the exception, a pipe + # nobody reads. + try: + sys.stderr.write(text) + sys.stderr.flush() + except Exception: + return + + +def _decide(raw): + """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" + event = parse(raw) + if event.event == UNKNOWN: + # A vendor event this adapter has no mapping for. handle() is not called: it + # reasons about the canonical vocabulary, and handing it something outside that + # vocabulary invites a decision made on a false premise. + return "", 0 + try: + decision = _coerce(handle(event)) + 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 + # failure is printed for the operator and the gate answers deny in 'claude_code''s + # own dialect. Only the failure's class is named to the host: its message may quote + # the very payload content the policy was inspecting. + _report(traceback.format_exc()) + decision = Decision.deny( + "policy handler failed (%s); refusing rather than allowing what it could not judge" + % sys.exc_info()[0].__name__ + ) + return respond(degrade(decision, event), event) + + def main(stdin=None, stdout=None, exit=True): """Read one payload from stdin, dispatch, emit the 'claude_code' response, exit.""" stream = stdin if stdin is not None else sys.stdin @@ -1087,16 +1142,14 @@ def main(stdin=None, stdout=None, exit=True): if exit: sys.exit(0) return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) + try: + text, code = _decide(raw) + except Exception: + # Past the handler, which _decide() already answers for: a fault in this file on a + # payload it did decode. With no Event to answer in dialect, the one refusal left is + # the host's blocking exit code, and nothing on stdout to be misread as a verdict. + _report(traceback.format_exc()) + text, code = "", 2 if text: _emit(out, text) if exit: diff --git a/tests/fixtures/runtime_goldens/codex_cli.py b/tests/fixtures/runtime_goldens/codex_cli.py index 7a723000..ebad2939 100644 --- a/tests/fixtures/runtime_goldens/codex_cli.py +++ b/tests/fixtures/runtime_goldens/codex_cli.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.0 -- bundle("codex_cli"). Do not hand-edit, except the +# Generated by agentseam 0.3.1 -- bundle("codex_cli"). 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. # @@ -18,10 +18,11 @@ import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.0) +# contract (agentseam 0.3.1) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -202,7 +203,9 @@ def tool_input_of(raw): if isinstance(raw, str) and raw[:1] == "{": try: parsed = _json.loads(raw) - except _json.JSONDecodeError: + except (ValueError, RecursionError): + # JSONDecodeError is a ValueError; a string nested past the interpreter's + # limit raises RecursionError instead, and neither is an input to crash on. return {} if isinstance(parsed, dict): return parsed @@ -216,9 +219,22 @@ def powershell_command(command): """`command` rewritten so PowerShell will actually run it.""" return command if command.lstrip().startswith("&") else "& " + command +UNREADABLE_NAME = "" + +def wire_name_of(raw, key): + """The event name under `key`: text, None when absent, UNREADABLE_NAME when not text. + + Total over anything `json.loads` can return: a payload that is not an object names no + event, and a name that is not a string is not one any adapter can map. + """ + name = raw.get(key) if isinstance(raw, dict) else None + if name is None or isinstance(name, str): + return name + return UNREADABLE_NAME + def _wire_name(cfg, raw): for key in cfg["claims"].get("event_key", ()): - name = raw.get(key) + name = wire_name_of(raw, key) if name is not None: return name return None @@ -303,6 +319,8 @@ def _field(raw, ti, chain): _FIELD_META = ("tool_input", "content_only_for_write_tools", "stringify") def _tool_input_raw(cfg, raw): + if not isinstance(raw, dict): + return None for key in cfg["fields"].get("tool_input", ("tool_input",)): value = raw.get(key) if value is not None: @@ -994,6 +1012,43 @@ def _emit(out, text): out.flush() +def _report(text): + # Best effort, never raising: a diagnostic must not pre-empt the refusal it accompanies. + # A hook's stderr is whatever the host gave it -- closed (then sys.stderr is None, and + # the traceback module's default print would land on STDOUT, inside the verdict), a + # console code page that cannot hold the payload text quoted in the exception, a pipe + # nobody reads. + try: + sys.stderr.write(text) + sys.stderr.flush() + except Exception: + return + + +def _decide(raw): + """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" + event = parse(raw) + if event.event == UNKNOWN: + # A vendor event this adapter has no mapping for. handle() is not called: it + # reasons about the canonical vocabulary, and handing it something outside that + # vocabulary invites a decision made on a false premise. + return "", 0 + try: + decision = _coerce(handle(event)) + 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 + # failure is printed for the operator and the gate answers deny in 'codex_cli''s + # own dialect. Only the failure's class is named to the host: its message may quote + # the very payload content the policy was inspecting. + _report(traceback.format_exc()) + decision = Decision.deny( + "policy handler failed (%s); refusing rather than allowing what it could not judge" + % sys.exc_info()[0].__name__ + ) + return respond(degrade(decision, event), event) + + def main(stdin=None, stdout=None, exit=True): """Read one payload from stdin, dispatch, emit the 'codex_cli' response, exit.""" stream = stdin if stdin is not None else sys.stdin @@ -1005,16 +1060,14 @@ def main(stdin=None, stdout=None, exit=True): if exit: sys.exit(0) return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) + try: + text, code = _decide(raw) + except Exception: + # Past the handler, which _decide() already answers for: a fault in this file on a + # payload it did decode. With no Event to answer in dialect, the one refusal left is + # the host's blocking exit code, and nothing on stdout to be misread as a verdict. + _report(traceback.format_exc()) + text, code = "", 2 if text: _emit(out, text) if exit: diff --git a/tests/fixtures/runtime_goldens/cursor.py b/tests/fixtures/runtime_goldens/cursor.py index 579e8959..498bab48 100644 --- a/tests/fixtures/runtime_goldens/cursor.py +++ b/tests/fixtures/runtime_goldens/cursor.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.0 -- bundle("cursor"). Do not hand-edit, except the +# Generated by agentseam 0.3.1 -- bundle("cursor"). 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. # @@ -18,10 +18,11 @@ import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.0) +# contract (agentseam 0.3.1) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -202,7 +203,9 @@ def tool_input_of(raw): if isinstance(raw, str) and raw[:1] == "{": try: parsed = _json.loads(raw) - except _json.JSONDecodeError: + except (ValueError, RecursionError): + # JSONDecodeError is a ValueError; a string nested past the interpreter's + # limit raises RecursionError instead, and neither is an input to crash on. return {} if isinstance(parsed, dict): return parsed @@ -212,9 +215,22 @@ def tool_input_of(raw): # ------------------------------------------------------------------------------ # cursor family engine (trimmed to what this entry uses) +UNREADABLE_NAME = "" + +def wire_name_of(raw, key): + """The event name under `key`: text, None when absent, UNREADABLE_NAME when not text. + + Total over anything `json.loads` can return: a payload that is not an object names no + event, and a name that is not a string is not one any adapter can map. + """ + name = raw.get(key) if isinstance(raw, dict) else None + if name is None or isinstance(name, str): + return name + return UNREADABLE_NAME + def _wire_name(cfg, raw): for key in cfg["claims"].get("event_key", ()): - name = raw.get(key) + name = wire_name_of(raw, key) if name is not None: return name return None @@ -260,6 +276,8 @@ def _field(raw, ti, chain): _FIELD_META = ("tool_input", "content_only_for_write_tools", "stringify") def _tool_input_raw(cfg, raw): + if not isinstance(raw, dict): + return None for key in cfg["fields"].get("tool_input", ("tool_input",)): value = raw.get(key) if value is not None: @@ -424,8 +442,8 @@ def hook_entry_config(cfg, canonical_events, command, matcher=None, *, fail_clos def cursor_wire(raw): """The wire event name, inferred from shape when the payload names none.""" - name = raw.get("hook_event_name") - if name is None: + name = wire_name_of(raw, "hook_event_name") + if name is None and isinstance(raw, dict): return "afterFileEdit" if isinstance(raw.get("edits"), list) else "beforeShellExecution" return name @@ -433,7 +451,7 @@ def cursor_claims(cfg, raw): """True when this payload looks like Cursor's shape.""" if not isinstance(raw, dict): return False - name = raw.get("hook_event_name") + name = wire_name_of(raw, "hook_event_name") if name in cfg["events"]: if name in _AMBIGUOUS_NAMES: return any(k in raw for k in _MARKERS) @@ -458,7 +476,7 @@ def _wire_of(cfg, event): `tool` is read only for an Event carrying no payload, where `parse` left the inferred name there; without a payload there is nothing to re-infer from. """ - name = (event.raw or {}).get("hook_event_name") + name = wire_name_of(event.raw, "hook_event_name") if name in cfg["events"]: return name if event.raw: @@ -913,6 +931,43 @@ def _emit(out, text): out.flush() +def _report(text): + # Best effort, never raising: a diagnostic must not pre-empt the refusal it accompanies. + # A hook's stderr is whatever the host gave it -- closed (then sys.stderr is None, and + # the traceback module's default print would land on STDOUT, inside the verdict), a + # console code page that cannot hold the payload text quoted in the exception, a pipe + # nobody reads. + try: + sys.stderr.write(text) + sys.stderr.flush() + except Exception: + return + + +def _decide(raw): + """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" + event = parse(raw) + if event.event == UNKNOWN: + # A vendor event this adapter has no mapping for. handle() is not called: it + # reasons about the canonical vocabulary, and handing it something outside that + # vocabulary invites a decision made on a false premise. + return "", 0 + try: + decision = _coerce(handle(event)) + 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 + # failure is printed for the operator and the gate answers deny in 'cursor''s + # own dialect. Only the failure's class is named to the host: its message may quote + # the very payload content the policy was inspecting. + _report(traceback.format_exc()) + decision = Decision.deny( + "policy handler failed (%s); refusing rather than allowing what it could not judge" + % sys.exc_info()[0].__name__ + ) + return respond(degrade(decision, event), event) + + def main(stdin=None, stdout=None, exit=True): """Read one payload from stdin, dispatch, emit the 'cursor' response, exit.""" stream = stdin if stdin is not None else sys.stdin @@ -924,16 +979,14 @@ def main(stdin=None, stdout=None, exit=True): if exit: sys.exit(0) return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) + try: + text, code = _decide(raw) + except Exception: + # Past the handler, which _decide() already answers for: a fault in this file on a + # payload it did decode. With no Event to answer in dialect, the one refusal left is + # the host's blocking exit code, and nothing on stdout to be misread as a verdict. + _report(traceback.format_exc()) + text, code = "", 2 if text: _emit(out, text) if exit: diff --git a/tests/fixtures/runtime_goldens/devin.py b/tests/fixtures/runtime_goldens/devin.py index 860ee582..4c468978 100644 --- a/tests/fixtures/runtime_goldens/devin.py +++ b/tests/fixtures/runtime_goldens/devin.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.0 -- bundle("devin"). Do not hand-edit, except the +# Generated by agentseam 0.3.1 -- bundle("devin"). 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. # @@ -18,10 +18,11 @@ import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.0) +# contract (agentseam 0.3.1) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -202,7 +203,9 @@ def tool_input_of(raw): if isinstance(raw, str) and raw[:1] == "{": try: parsed = _json.loads(raw) - except _json.JSONDecodeError: + except (ValueError, RecursionError): + # JSONDecodeError is a ValueError; a string nested past the interpreter's + # limit raises RecursionError instead, and neither is an input to crash on. return {} if isinstance(parsed, dict): return parsed @@ -229,9 +232,22 @@ def looks_like_claude_code(raw): PROBES = {"looks_like_claude_code": looks_like_claude_code} +UNREADABLE_NAME = "" + +def wire_name_of(raw, key): + """The event name under `key`: text, None when absent, UNREADABLE_NAME when not text. + + Total over anything `json.loads` can return: a payload that is not an object names no + event, and a name that is not a string is not one any adapter can map. + """ + name = raw.get(key) if isinstance(raw, dict) else None + if name is None or isinstance(name, str): + return name + return UNREADABLE_NAME + def _wire_name(cfg, raw): for key in cfg["claims"].get("event_key", ()): - name = raw.get(key) + name = wire_name_of(raw, key) if name is not None: return name return None @@ -316,6 +332,8 @@ def _field(raw, ti, chain): _FIELD_META = ("tool_input", "content_only_for_write_tools", "stringify") def _tool_input_raw(cfg, raw): + if not isinstance(raw, dict): + return None for key in cfg["fields"].get("tool_input", ("tool_input",)): value = raw.get(key) if value is not None: @@ -976,6 +994,43 @@ def _emit(out, text): out.flush() +def _report(text): + # Best effort, never raising: a diagnostic must not pre-empt the refusal it accompanies. + # A hook's stderr is whatever the host gave it -- closed (then sys.stderr is None, and + # the traceback module's default print would land on STDOUT, inside the verdict), a + # console code page that cannot hold the payload text quoted in the exception, a pipe + # nobody reads. + try: + sys.stderr.write(text) + sys.stderr.flush() + except Exception: + return + + +def _decide(raw): + """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" + event = parse(raw) + if event.event == UNKNOWN: + # A vendor event this adapter has no mapping for. handle() is not called: it + # reasons about the canonical vocabulary, and handing it something outside that + # vocabulary invites a decision made on a false premise. + return "", 0 + try: + decision = _coerce(handle(event)) + 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 + # failure is printed for the operator and the gate answers deny in 'devin''s + # own dialect. Only the failure's class is named to the host: its message may quote + # the very payload content the policy was inspecting. + _report(traceback.format_exc()) + decision = Decision.deny( + "policy handler failed (%s); refusing rather than allowing what it could not judge" + % sys.exc_info()[0].__name__ + ) + return respond(degrade(decision, event), event) + + def main(stdin=None, stdout=None, exit=True): """Read one payload from stdin, dispatch, emit the 'devin' response, exit.""" stream = stdin if stdin is not None else sys.stdin @@ -987,16 +1042,14 @@ def main(stdin=None, stdout=None, exit=True): if exit: sys.exit(0) return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) + try: + text, code = _decide(raw) + except Exception: + # Past the handler, which _decide() already answers for: a fault in this file on a + # payload it did decode. With no Event to answer in dialect, the one refusal left is + # the host's blocking exit code, and nothing on stdout to be misread as a verdict. + _report(traceback.format_exc()) + text, code = "", 2 if text: _emit(out, text) if exit: diff --git a/tests/fixtures/runtime_goldens/gemini_cli.py b/tests/fixtures/runtime_goldens/gemini_cli.py index 5cdf2cc2..db789c29 100644 --- a/tests/fixtures/runtime_goldens/gemini_cli.py +++ b/tests/fixtures/runtime_goldens/gemini_cli.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.0 -- bundle("gemini_cli"). Do not hand-edit, except the +# Generated by agentseam 0.3.1 -- bundle("gemini_cli"). 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. # @@ -18,10 +18,11 @@ import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.0) +# contract (agentseam 0.3.1) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -202,7 +203,9 @@ def tool_input_of(raw): if isinstance(raw, str) and raw[:1] == "{": try: parsed = _json.loads(raw) - except _json.JSONDecodeError: + except (ValueError, RecursionError): + # JSONDecodeError is a ValueError; a string nested past the interpreter's + # limit raises RecursionError instead, and neither is an input to crash on. return {} if isinstance(parsed, dict): return parsed @@ -229,9 +232,22 @@ def looks_like_claude_code(raw): PROBES = {"looks_like_claude_code": looks_like_claude_code} +UNREADABLE_NAME = "" + +def wire_name_of(raw, key): + """The event name under `key`: text, None when absent, UNREADABLE_NAME when not text. + + Total over anything `json.loads` can return: a payload that is not an object names no + event, and a name that is not a string is not one any adapter can map. + """ + name = raw.get(key) if isinstance(raw, dict) else None + if name is None or isinstance(name, str): + return name + return UNREADABLE_NAME + def _wire_name(cfg, raw): for key in cfg["claims"].get("event_key", ()): - name = raw.get(key) + name = wire_name_of(raw, key) if name is not None: return name return None @@ -316,6 +332,8 @@ def _field(raw, ti, chain): _FIELD_META = ("tool_input", "content_only_for_write_tools", "stringify") def _tool_input_raw(cfg, raw): + if not isinstance(raw, dict): + return None for key in cfg["fields"].get("tool_input", ("tool_input",)): value = raw.get(key) if value is not None: @@ -976,6 +994,43 @@ def _emit(out, text): out.flush() +def _report(text): + # Best effort, never raising: a diagnostic must not pre-empt the refusal it accompanies. + # A hook's stderr is whatever the host gave it -- closed (then sys.stderr is None, and + # the traceback module's default print would land on STDOUT, inside the verdict), a + # console code page that cannot hold the payload text quoted in the exception, a pipe + # nobody reads. + try: + sys.stderr.write(text) + sys.stderr.flush() + except Exception: + return + + +def _decide(raw): + """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" + event = parse(raw) + if event.event == UNKNOWN: + # A vendor event this adapter has no mapping for. handle() is not called: it + # reasons about the canonical vocabulary, and handing it something outside that + # vocabulary invites a decision made on a false premise. + return "", 0 + try: + decision = _coerce(handle(event)) + 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 + # failure is printed for the operator and the gate answers deny in 'gemini_cli''s + # own dialect. Only the failure's class is named to the host: its message may quote + # the very payload content the policy was inspecting. + _report(traceback.format_exc()) + decision = Decision.deny( + "policy handler failed (%s); refusing rather than allowing what it could not judge" + % sys.exc_info()[0].__name__ + ) + return respond(degrade(decision, event), event) + + def main(stdin=None, stdout=None, exit=True): """Read one payload from stdin, dispatch, emit the 'gemini_cli' response, exit.""" stream = stdin if stdin is not None else sys.stdin @@ -987,16 +1042,14 @@ def main(stdin=None, stdout=None, exit=True): if exit: sys.exit(0) return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) + try: + text, code = _decide(raw) + except Exception: + # Past the handler, which _decide() already answers for: a fault in this file on a + # payload it did decode. With no Event to answer in dialect, the one refusal left is + # the host's blocking exit code, and nothing on stdout to be misread as a verdict. + _report(traceback.format_exc()) + text, code = "", 2 if text: _emit(out, text) if exit: diff --git a/tests/fixtures/runtime_goldens/grok.py b/tests/fixtures/runtime_goldens/grok.py index 0f4fceb0..8f09b67c 100644 --- a/tests/fixtures/runtime_goldens/grok.py +++ b/tests/fixtures/runtime_goldens/grok.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.0 -- bundle("grok"). Do not hand-edit, except the +# Generated by agentseam 0.3.1 -- bundle("grok"). 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. # @@ -18,10 +18,11 @@ import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.0) +# contract (agentseam 0.3.1) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -202,7 +203,9 @@ def tool_input_of(raw): if isinstance(raw, str) and raw[:1] == "{": try: parsed = _json.loads(raw) - except _json.JSONDecodeError: + except (ValueError, RecursionError): + # JSONDecodeError is a ValueError; a string nested past the interpreter's + # limit raises RecursionError instead, and neither is an input to crash on. return {} if isinstance(parsed, dict): return parsed @@ -212,9 +215,22 @@ def tool_input_of(raw): # ------------------------------------------------------------------------------ # flat_decision family engine (trimmed to what this entry uses) +UNREADABLE_NAME = "" + +def wire_name_of(raw, key): + """The event name under `key`: text, None when absent, UNREADABLE_NAME when not text. + + Total over anything `json.loads` can return: a payload that is not an object names no + event, and a name that is not a string is not one any adapter can map. + """ + name = raw.get(key) if isinstance(raw, dict) else None + if name is None or isinstance(name, str): + return name + return UNREADABLE_NAME + def _wire_name(cfg, raw): for key in cfg["claims"].get("event_key", ()): - name = raw.get(key) + name = wire_name_of(raw, key) if name is not None: return name return None @@ -299,6 +315,8 @@ def _field(raw, ti, chain): _FIELD_META = ("tool_input", "content_only_for_write_tools", "stringify") def _tool_input_raw(cfg, raw): + if not isinstance(raw, dict): + return None for key in cfg["fields"].get("tool_input", ("tool_input",)): value = raw.get(key) if value is not None: @@ -959,6 +977,43 @@ def _emit(out, text): out.flush() +def _report(text): + # Best effort, never raising: a diagnostic must not pre-empt the refusal it accompanies. + # A hook's stderr is whatever the host gave it -- closed (then sys.stderr is None, and + # the traceback module's default print would land on STDOUT, inside the verdict), a + # console code page that cannot hold the payload text quoted in the exception, a pipe + # nobody reads. + try: + sys.stderr.write(text) + sys.stderr.flush() + except Exception: + return + + +def _decide(raw): + """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" + event = parse(raw) + if event.event == UNKNOWN: + # A vendor event this adapter has no mapping for. handle() is not called: it + # reasons about the canonical vocabulary, and handing it something outside that + # vocabulary invites a decision made on a false premise. + return "", 0 + try: + decision = _coerce(handle(event)) + 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 + # failure is printed for the operator and the gate answers deny in 'grok''s + # own dialect. Only the failure's class is named to the host: its message may quote + # the very payload content the policy was inspecting. + _report(traceback.format_exc()) + decision = Decision.deny( + "policy handler failed (%s); refusing rather than allowing what it could not judge" + % sys.exc_info()[0].__name__ + ) + return respond(degrade(decision, event), event) + + def main(stdin=None, stdout=None, exit=True): """Read one payload from stdin, dispatch, emit the 'grok' response, exit.""" stream = stdin if stdin is not None else sys.stdin @@ -970,16 +1025,14 @@ def main(stdin=None, stdout=None, exit=True): if exit: sys.exit(0) return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) + try: + text, code = _decide(raw) + except Exception: + # Past the handler, which _decide() already answers for: a fault in this file on a + # payload it did decode. With no Event to answer in dialect, the one refusal left is + # the host's blocking exit code, and nothing on stdout to be misread as a verdict. + _report(traceback.format_exc()) + text, code = "", 2 if text: _emit(out, text) if exit: diff --git a/tests/fixtures/runtime_goldens/tabnine.py b/tests/fixtures/runtime_goldens/tabnine.py index 88c2225d..bcb4148f 100644 --- a/tests/fixtures/runtime_goldens/tabnine.py +++ b/tests/fixtures/runtime_goldens/tabnine.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.0 -- bundle("tabnine"). Do not hand-edit, except the +# Generated by agentseam 0.3.1 -- bundle("tabnine"). 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. # @@ -18,10 +18,11 @@ import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.0) +# contract (agentseam 0.3.1) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -202,7 +203,9 @@ def tool_input_of(raw): if isinstance(raw, str) and raw[:1] == "{": try: parsed = _json.loads(raw) - except _json.JSONDecodeError: + except (ValueError, RecursionError): + # JSONDecodeError is a ValueError; a string nested past the interpreter's + # limit raises RecursionError instead, and neither is an input to crash on. return {} if isinstance(parsed, dict): return parsed @@ -212,9 +215,22 @@ def tool_input_of(raw): # ------------------------------------------------------------------------------ # flat_decision family engine (trimmed to what this entry uses) +UNREADABLE_NAME = "" + +def wire_name_of(raw, key): + """The event name under `key`: text, None when absent, UNREADABLE_NAME when not text. + + Total over anything `json.loads` can return: a payload that is not an object names no + event, and a name that is not a string is not one any adapter can map. + """ + name = raw.get(key) if isinstance(raw, dict) else None + if name is None or isinstance(name, str): + return name + return UNREADABLE_NAME + def _wire_name(cfg, raw): for key in cfg["claims"].get("event_key", ()): - name = raw.get(key) + name = wire_name_of(raw, key) if name is not None: return name return None @@ -299,6 +315,8 @@ def _field(raw, ti, chain): _FIELD_META = ("tool_input", "content_only_for_write_tools", "stringify") def _tool_input_raw(cfg, raw): + if not isinstance(raw, dict): + return None for key in cfg["fields"].get("tool_input", ("tool_input",)): value = raw.get(key) if value is not None: @@ -959,6 +977,43 @@ def _emit(out, text): out.flush() +def _report(text): + # Best effort, never raising: a diagnostic must not pre-empt the refusal it accompanies. + # A hook's stderr is whatever the host gave it -- closed (then sys.stderr is None, and + # the traceback module's default print would land on STDOUT, inside the verdict), a + # console code page that cannot hold the payload text quoted in the exception, a pipe + # nobody reads. + try: + sys.stderr.write(text) + sys.stderr.flush() + except Exception: + return + + +def _decide(raw): + """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" + event = parse(raw) + if event.event == UNKNOWN: + # A vendor event this adapter has no mapping for. handle() is not called: it + # reasons about the canonical vocabulary, and handing it something outside that + # vocabulary invites a decision made on a false premise. + return "", 0 + try: + decision = _coerce(handle(event)) + 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 + # failure is printed for the operator and the gate answers deny in 'tabnine''s + # own dialect. Only the failure's class is named to the host: its message may quote + # the very payload content the policy was inspecting. + _report(traceback.format_exc()) + decision = Decision.deny( + "policy handler failed (%s); refusing rather than allowing what it could not judge" + % sys.exc_info()[0].__name__ + ) + return respond(degrade(decision, event), event) + + def main(stdin=None, stdout=None, exit=True): """Read one payload from stdin, dispatch, emit the 'tabnine' response, exit.""" stream = stdin if stdin is not None else sys.stdin @@ -970,16 +1025,14 @@ def main(stdin=None, stdout=None, exit=True): if exit: sys.exit(0) return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) + try: + text, code = _decide(raw) + except Exception: + # Past the handler, which _decide() already answers for: a fault in this file on a + # payload it did decode. With no Event to answer in dialect, the one refusal left is + # the host's blocking exit code, and nothing on stdout to be misread as a verdict. + _report(traceback.format_exc()) + text, code = "", 2 if text: _emit(out, text) if exit: diff --git a/tests/fixtures/runtime_goldens/vscode_copilot.py b/tests/fixtures/runtime_goldens/vscode_copilot.py index 2cc45d96..6218e705 100644 --- a/tests/fixtures/runtime_goldens/vscode_copilot.py +++ b/tests/fixtures/runtime_goldens/vscode_copilot.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.0 -- bundle("vscode_copilot"). Do not hand-edit, except the +# Generated by agentseam 0.3.1 -- bundle("vscode_copilot"). 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. # @@ -18,10 +18,11 @@ import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.0) +# contract (agentseam 0.3.1) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -202,7 +203,9 @@ def tool_input_of(raw): if isinstance(raw, str) and raw[:1] == "{": try: parsed = _json.loads(raw) - except _json.JSONDecodeError: + except (ValueError, RecursionError): + # JSONDecodeError is a ValueError; a string nested past the interpreter's + # limit raises RecursionError instead, and neither is an input to crash on. return {} if isinstance(parsed, dict): return parsed @@ -264,11 +267,22 @@ def powershell_command(command): _VSCODE_ENVELOPE = "timestamp" +def _name(raw): + """The payload's own event name as text; None when it names none, UNKNOWN when the + value is not text (a list or object cannot be looked up, and would raise if tried).""" + if not isinstance(raw, dict): + return None + name = raw.get("hook_event_name") or raw.get("hookEventName") + if name is None or isinstance(name, str): + return name + return UNKNOWN + + def claims(raw): """True for a payload from either product.""" if not isinstance(raw, dict): return False - name = raw.get("hook_event_name") or raw.get("hookEventName") + name = _name(raw) if name in EVENT_MAP and _VSCODE_ENVELOPE in raw and "turn_id" not in raw: return True if any(k in raw for k in _CODEX_MARKERS + _CURSOR_MARKERS): @@ -280,6 +294,8 @@ def claims(raw): def parse(raw): + if not isinstance(raw, dict): + return Event(AGENT, UNKNOWN, raw=raw) ti = raw.get("tool_input") ti = tool_input_of(ti) tool = raw.get("tool_name") or raw.get("toolName") @@ -293,7 +309,7 @@ def parse(raw): else: path = ti.get("filePath") or ti.get("file_path") or ti.get("path") content = ti.get("content") or ti.get("newText") or ti.get("new_str") - name = raw.get("hook_event_name") or raw.get("hookEventName") or "preToolUse" + name = _name(raw) or "preToolUse" return Event( AGENT, EVENT_MAP.get(name, UNKNOWN), @@ -324,8 +340,7 @@ def is_memory_write(event): def _echoed_name(event): """This event's own vendor spelling, out of the payload; VS Code's name if there is none.""" - raw = event.raw or {} - return raw.get("hook_event_name") or raw.get("hookEventName") or REVERSE_EVENT_MAP.get(event.event, "PreToolUse") + return _name(event.raw) or REVERSE_EVENT_MAP.get(event.event, "PreToolUse") def _refusal_reason(decision): @@ -787,6 +802,43 @@ def _emit(out, text): out.flush() +def _report(text): + # Best effort, never raising: a diagnostic must not pre-empt the refusal it accompanies. + # A hook's stderr is whatever the host gave it -- closed (then sys.stderr is None, and + # the traceback module's default print would land on STDOUT, inside the verdict), a + # console code page that cannot hold the payload text quoted in the exception, a pipe + # nobody reads. + try: + sys.stderr.write(text) + sys.stderr.flush() + except Exception: + return + + +def _decide(raw): + """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" + event = parse(raw) + if event.event == UNKNOWN: + # A vendor event this adapter has no mapping for. handle() is not called: it + # reasons about the canonical vocabulary, and handing it something outside that + # vocabulary invites a decision made on a false premise. + return "", 0 + try: + decision = _coerce(handle(event)) + 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 + # failure is printed for the operator and the gate answers deny in 'vscode_copilot''s + # own dialect. Only the failure's class is named to the host: its message may quote + # the very payload content the policy was inspecting. + _report(traceback.format_exc()) + decision = Decision.deny( + "policy handler failed (%s); refusing rather than allowing what it could not judge" + % sys.exc_info()[0].__name__ + ) + return respond(degrade(decision, event), event) + + def main(stdin=None, stdout=None, exit=True): """Read one payload from stdin, dispatch, emit the 'vscode_copilot' response, exit.""" stream = stdin if stdin is not None else sys.stdin @@ -798,16 +850,14 @@ def main(stdin=None, stdout=None, exit=True): if exit: sys.exit(0) return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) + try: + text, code = _decide(raw) + except Exception: + # Past the handler, which _decide() already answers for: a fault in this file on a + # payload it did decode. With no Event to answer in dialect, the one refusal left is + # the host's blocking exit code, and nothing on stdout to be misread as a verdict. + _report(traceback.format_exc()) + text, code = "", 2 if text: _emit(out, text) if exit: diff --git a/tests/fixtures/runtime_goldens/windsurf.py b/tests/fixtures/runtime_goldens/windsurf.py index 538b028e..90ae1820 100644 --- a/tests/fixtures/runtime_goldens/windsurf.py +++ b/tests/fixtures/runtime_goldens/windsurf.py @@ -1,4 +1,4 @@ -# Generated by agentseam 0.3.0 -- bundle("windsurf"). Do not hand-edit, except the +# Generated by agentseam 0.3.1 -- bundle("windsurf"). 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. # @@ -18,10 +18,11 @@ import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path +import traceback import warnings as _warnings # ------------------------------------------------------------------------------ -# contract (agentseam 0.3.0) +# contract (agentseam 0.3.1) """Canonical event vocabulary, normalized envelope, and decision type.""" @@ -202,7 +203,9 @@ def tool_input_of(raw): if isinstance(raw, str) and raw[:1] == "{": try: parsed = _json.loads(raw) - except _json.JSONDecodeError: + except (ValueError, RecursionError): + # JSONDecodeError is a ValueError; a string nested past the interpreter's + # limit raises RecursionError instead, and neither is an input to crash on. return {} if isinstance(parsed, dict): return parsed @@ -212,9 +215,22 @@ def tool_input_of(raw): # ------------------------------------------------------------------------------ # windsurf family engine (trimmed to what this entry uses) +UNREADABLE_NAME = "" + +def wire_name_of(raw, key): + """The event name under `key`: text, None when absent, UNREADABLE_NAME when not text. + + Total over anything `json.loads` can return: a payload that is not an object names no + event, and a name that is not a string is not one any adapter can map. + """ + name = raw.get(key) if isinstance(raw, dict) else None + if name is None or isinstance(name, str): + return name + return UNREADABLE_NAME + def _wire_name(cfg, raw): for key in cfg["claims"].get("event_key", ()): - name = raw.get(key) + name = wire_name_of(raw, key) if name is not None: return name return None @@ -260,6 +276,8 @@ def _field(raw, ti, chain): _FIELD_META = ("tool_input", "content_only_for_write_tools", "stringify") def _tool_input_raw(cfg, raw): + if not isinstance(raw, dict): + return None for key in cfg["fields"].get("tool_input", ("tool_input",)): value = raw.get(key) if value is not None: @@ -448,25 +466,28 @@ def hook_entry_config(cfg, canonical_events, command, matcher=None, *, fail_clos _MCP_EVENTS = ("pre_mcp_tool_use", "post_mcp_tool_use") +def _tool_info(raw): + info = raw.get("tool_info") if isinstance(raw, dict) else None + return info if isinstance(info, dict) else {} + def windsurf_wire(raw): """The wire event name, inferred from `tool_info` when the payload names none.""" - name = raw.get("hook_event_name") - if name is not None: + name = wire_name_of(raw, "hook_event_name") + if name is not None or not isinstance(raw, dict): return name - info = raw.get("tool_info") or {} - return "pre_run_command" if info.get("command_line") else "pre_user_prompt" + return "pre_run_command" if _tool_info(raw).get("command_line") else "pre_user_prompt" def windsurf_claims(cfg, raw): if not isinstance(raw, dict): return False - if raw.get("hook_event_name") in cfg["events"]: + if wire_name_of(raw, "hook_event_name") in cfg["events"]: return True return "trajectory_id" in raw and isinstance(raw.get("tool_info"), dict) def windsurf_parse(cfg, raw): name = windsurf_wire(raw) event = hj_parse(cfg, raw, wire=name) - info = raw.get("tool_info") or {} + info = _tool_info(raw) if name in _MCP_EVENTS: joined = "%s/%s" % (info["server"], info["tool"]) if info.get("server") and info.get("tool") else None event.tool = joined or info.get("tool") @@ -870,6 +891,43 @@ def _emit(out, text): out.flush() +def _report(text): + # Best effort, never raising: a diagnostic must not pre-empt the refusal it accompanies. + # A hook's stderr is whatever the host gave it -- closed (then sys.stderr is None, and + # the traceback module's default print would land on STDOUT, inside the verdict), a + # console code page that cannot hold the payload text quoted in the exception, a pipe + # nobody reads. + try: + sys.stderr.write(text) + sys.stderr.flush() + except Exception: + return + + +def _decide(raw): + """(stdout_text, exit_code) for one decoded payload: agentseam.dispatch.handle(), inlined.""" + event = parse(raw) + if event.event == UNKNOWN: + # A vendor event this adapter has no mapping for. handle() is not called: it + # reasons about the canonical vocabulary, and handing it something outside that + # vocabulary invites a decision made on a false premise. + return "", 0 + try: + decision = _coerce(handle(event)) + 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 + # failure is printed for the operator and the gate answers deny in 'windsurf''s + # own dialect. Only the failure's class is named to the host: its message may quote + # the very payload content the policy was inspecting. + _report(traceback.format_exc()) + decision = Decision.deny( + "policy handler failed (%s); refusing rather than allowing what it could not judge" + % sys.exc_info()[0].__name__ + ) + return respond(degrade(decision, event), event) + + def main(stdin=None, stdout=None, exit=True): """Read one payload from stdin, dispatch, emit the 'windsurf' response, exit.""" stream = stdin if stdin is not None else sys.stdin @@ -881,16 +939,14 @@ def main(stdin=None, stdout=None, exit=True): if exit: sys.exit(0) return 0 - event = parse(raw) - if event.event == UNKNOWN: - # A vendor event this adapter has no mapping for. handle() is not called: it - # reasons about the canonical vocabulary, and handing it something outside that - # vocabulary invites a decision made on a false premise. - if exit: - sys.exit(0) - return 0 - decision = degrade(_coerce(handle(event)), event) - text, code = respond(decision, event) + try: + text, code = _decide(raw) + except Exception: + # Past the handler, which _decide() already answers for: a fault in this file on a + # payload it did decode. With no Event to answer in dialect, the one refusal left is + # the host's blocking exit code, and nothing on stdout to be misread as a verdict. + _report(traceback.format_exc()) + text, code = "", 2 if text: _emit(out, text) if exit: diff --git a/tests/test_sync_wires_only_supported_agents.py b/tests/test_sync_wires_only_supported_agents.py index c2090ee0..446de8a0 100644 --- a/tests/test_sync_wires_only_supported_agents.py +++ b/tests/test_sync_wires_only_supported_agents.py @@ -49,3 +49,18 @@ def test_a_cursor_only_repo_gets_no_claude_settings(tmp_path: Path) -> None: assert (repo / ".cursor" / "hooks.json").exists() assert not (repo / ".claude" / "settings.json").exists() + + +def test_a_vendor_dropped_from_supported_agents_has_its_runtime_pruned(tmp_path: Path) -> None: + """A vendor's `.chock/bin/.py` outlives it leaving `supported_agents` otherwise: + + nothing rewrites it once it is unwired, so it drifts against every future render forever. + """ + repo = _repo(tmp_path) + recompile(repo, ["claude", "cursor"], skip_hooks=False) + assert (repo / ".chock" / "bin" / "cursor.py").exists() + + recompile(repo, ["claude"], skip_hooks=False) + + assert not (repo / ".chock" / "bin" / "cursor.py").exists() + assert (repo / ".chock" / "bin" / "claude_code.py").exists()