Bump agentseam to 0.2.1, derive shell matcher and repo-root token

.chock/bin/antigravity.py

@@ -1,4 +1,4 @@
-# Generated by agentseam 0.2.0 -- bundle("antigravity"). Do not hand-edit, except the
+# Generated by agentseam 0.2.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.
 #
@@ -22,7 +22,7 @@
 import warnings as _warnings
 
 # ------------------------------------------------------------------------------
-# contract (agentseam 0.2.0)
+# contract (agentseam 0.2.1)
 
 """Canonical event vocabulary, normalized envelope, and decision type."""
 
@@ -64,17 +64,17 @@ class Event:
 
     __slots__ = (
         "agent",
-        "event",
-        "tool",
         "command",
-        "path",
         "content",
+        "cwd",
+        "event",
         "output",
+        "path",
         "prompt",
+        "raw",
         "session_id",
+        "tool",
         "tool_use_id",
-        "cwd",
-        "raw",
     )
 
     def __init__(
@@ -133,7 +133,7 @@ def __repr__(self):  # pragma: no cover - debugging aid
 class Decision:
     """What a handler wants to happen. Adapters translate this to vendor dialect."""
 
-    __slots__ = ("outcome", "reason", "updated_input", "evidence", "context")
+    __slots__ = ("context", "evidence", "outcome", "reason", "updated_input")
 
     #: Classmethods kept only so existing callers keep constructing; see .ask()/.rewrite().
     DEPRECATED_ALIASES = frozenset({"ask", "rewrite"})
@@ -203,7 +203,7 @@ def tool_input_of(raw):
     if isinstance(raw, str) and raw[:1] == "{":
         try:
             parsed = _json.loads(raw)
-        except Exception:
+        except _json.JSONDecodeError:
             return {}
         if isinstance(parsed, dict):
             return parsed
@@ -305,6 +305,10 @@ def hj_parse(cfg, raw, wire=None):
         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."""
     reverse = {}
@@ -327,14 +331,14 @@ def _context_body(name, value):
 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 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"]
+        if missing_input and _TRANSFORM_MISSING_INPUT in notes:
+            return notes[_TRANSFORM_MISSING_INPUT]
         return notes.get("transform")
     return None
 
@@ -361,46 +365,54 @@ def _refusal_text(v, decision, at_gate, wire=None):
     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):
-        value = _context_value(v, decision)
-        if at_context_event 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
+        return _g1_allow(v, decision, wire, name, words)
     if decision.outcome == TRANSFORM and gate["honours_transform"]:
-        if v.get("transform_grammar") == "hook_specific_tool_input":
-            return _json.dumps({"hookSpecificOutput": {"tool_input": decision.updated_input}}), 0
-        if decision.updated_input is not 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
+        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", {}))
+        and not (degraded_from(decision) == TRANSFORM and _ESCALATE_FROM_TRANSFORM in v.get("degrade_notes", {}))
     ):
-        reason = decision.reason or _default_for(v, decision, True, wire)
+        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, False, wire)}
+    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
@@ -442,49 +454,45 @@ def _hook_dict(cfg, command):
             entry[key] = value
     return entry
 
-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.
+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
 
-    `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.
-    """
+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"]
-    reverse = hj_reverse(cfg)
-    wrapper = hook_entry["wrapper"]
-    if wrapper == "flat_list":
-        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
-    if wrapper == "cursor":
-        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}
-    if wrapper == "flat_entries":
-        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}
     hooks = {}
     for ev in canonical_events:
         name = reverse.get(ev)
@@ -498,6 +506,23 @@ def hook_entry_config(cfg, canonical_events, command, matcher=None, fail_closed=
         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:
@@ -506,7 +531,7 @@ def antigravity_wire(raw):
         return "PostToolUse" if "error" in raw else "PreToolUse"
     return None
 
-def antigravity_claims(cfg, raw):
+def antigravity_claims(_cfg, raw):
     """Structural: `conversationId` with `workspacePaths` is Antigravity's own envelope."""
     if not isinstance(raw, dict):
         return False