grants: 🐛 Bound evidence reads and bind Orca lanes to their Run coordinator's grants

captain_hook/builtin_packs/general/hooks/grants.py

@@ -6,6 +6,7 @@
 from captain_hook import Allow, Block, Event, Input, LambdaCondition, Tool, on
 from captain_hook.grants import store
 from captain_hook.grants.evidence import answer_evidence, machine_written, parse_answer, tree_of, words_evidence
+from captain_hook.grants.orca import record_terminal
 from captain_hook.grants.records import Evidence, Grant
 
 if TYPE_CHECKING:
@@ -54,6 +55,16 @@ def record_owner_words(evt: UserPromptSubmitEvent) -> HookResult | None:
     return None
 
 
+@on(
+    Event.SessionStart | Event.UserPromptSubmit | Event.PreToolUse,
+    respect_gitignore=False,
+    skip_planning_agents=False,
+)
+def record_orca_terminal(evt: BaseHookEvent) -> HookResult | None:
+    record_terminal(evt)
+    return None
+
+
 def mints(evt: BaseHookEvent) -> bool:
     for call in evt.cmd.calls():
         words = [word.value or "" for word in call.command.words]

captain_hook/context.py

@@ -284,7 +284,7 @@ def root_transcript_block(
 
     def root_excerpt(self, needles: Sequence[str], *, around: int = 2) -> Session | None:
         """The root session's events that mention any of ``needles``, with ``around`` events either side,
-        from the whole transcript rather than its tail; ``None`` outside a lane.
+        from the last 16 MiB of its transcript rather than its event tail; ``None`` outside a lane.
         """
         from captain_hook.transcripts import root_excerpt
 

captain_hook/grants/declare.py

@@ -15,8 +15,9 @@
 from loguru import logger
 
 from captain_hook.grants import store
-from captain_hook.grants.evidence import Asked, OwnerWords, tree_of, verbatim
+from captain_hook.grants.evidence import Asked, OwnerWords, lapsed, tree_of, verbatim
 from captain_hook.grants.judge import GrantVerdict, Judge, JudgeFailed
+from captain_hook.grants.orca import adopt_coordinator
 from captain_hook.grants.records import Allowed, Denied, Evidence, Grant, Proposal
 
 if TYPE_CHECKING:
@@ -180,6 +181,7 @@ def check(self, evt: BaseHookEvent, action: Proposal | None = None) -> Allowed |
             action = self.action(evt)
         scope = self.canonical(action)
         tree = tree_of(evt)
+        adopt_coordinator(evt)
         collected: list[Evidence] | None = None
 
         def session() -> list[Evidence]:
@@ -200,11 +202,15 @@ def session() -> list[Evidence]:
                 refusals.append(f"grant {grant.id} rests on {stale}, which changed after the grant was minted.")
                 continue
             reason, relied = f"covered by grant {grant.id}", [item.id for item in grant.evidence]
-            since = [
-                item
-                for item in (session() if self.judge is not None else ())
-                if item.source in OWNER_SOURCES and item.said_at is not None and item.said_at > grant.created
-            ]
+            later = [*session(), *lapsed(evt, grant.created)] if self.judge is not None else []
+            since = sorted(
+                {
+                    item.id: item
+                    for item in later
+                    if item.source in OWNER_SOURCES and item.said_at is not None and item.said_at > grant.created
+                }.values(),
+                key=lambda item: item.said_at.timestamp() if item.said_at else 0.0,
+            )
             allowed_by_rule = any(ruling.verdict == "allow" for ruling in rulings)
             if self.judge is not None and (since or not allowed_by_rule):
                 try:

captain_hook/grants/evidence.py

@@ -165,18 +165,37 @@ def ask_evidence(use: Any) -> list[Evidence]:
     return answer_evidence(use.ref.tool_use_id or f"{use.ts:%s}", *result, use.result_ts or use.ts)
 
 
-def recorded_asks(evt: BaseHookEvent) -> list[Evidence]:
+def recorded(evt: BaseHookEvent, kind: str) -> list[Evidence]:
     from captain_hook.grants import store
 
     at = store.now()
     return [
         item
-        for grant in store.grants("ask", tree_of(evt))
+        for grant in store.grants(kind, tree_of(evt))
         if grant.revoked is None and (grant.expires is None or grant.expires > at)
         for item in grant.evidence
     ]
 
 
+def lapsed(evt: BaseHookEvent, since: datetime) -> list[Evidence]:
+    """The owner's recorded words and answers in *evt*'s tree said after *since* whose records have expired.
+
+    A grant minted before them still reaches the judge with them, so an expired record never ends a
+    withdrawal while the grant it withdrew stays usable.
+    """
+    from captain_hook.grants import store
+
+    at = store.now()
+    return [
+        item
+        for kind in store.EVIDENCE_KINDS
+        for grant in store.grants(kind, tree_of(evt))
+        if grant.revoked is None and grant.expires is not None and grant.expires <= at
+        for item in grant.evidence
+        if item.said_at is not None and item.said_at > since
+    ]
+
+
 @dataclass(frozen=True, slots=True)
 class Asked:
     """The owner's AskUserQuestion answers in the session tree, with every option and preview they saw."""
@@ -187,7 +206,7 @@ class Asked:
     def collect(self, evt: BaseHookEvent, action: Proposal) -> list[Evidence]:
         needles = self.needles(evt) if self.needles else ()
         read = {item.id: item for use in owner_uses(evt, needles, self.window) for item in ask_evidence(use)}
-        merged = {item.id: item for item in recorded_asks(evt)} | read
+        merged = {item.id: item for item in recorded(evt, "ask")} | read
         return sorted(merged.values(), key=lambda item: item.said_at.timestamp() if item.said_at else 0.0)
 
 
@@ -201,12 +220,6 @@ def words_evidence(text: str, at: datetime | None) -> Evidence:
     return Evidence(id=key, source="words", quote=text, said_at=at, key=key)
 
 
-def recorded_words(evt: BaseHookEvent) -> list[Evidence]:
-    from captain_hook.grants import store
-
-    return [item for grant in store.grants("words", tree_of(evt)) for item in grant.evidence]
-
-
 def queued_words(turn: Any) -> list[tuple[str, Any]]:
     return [
         (event.detail.prompt or "", event)
@@ -232,7 +245,7 @@ class OwnerWords:
     window: int = OWNER_WINDOW
 
     def collect(self, evt: BaseHookEvent, action: Proposal) -> list[Evidence]:
-        items = {item.id: item for item in recorded_words(evt) if not machine_written(item.quote, self.machine)}
+        items = {item.id: item for item in recorded(evt, "words") if not machine_written(item.quote, self.machine)}
         needles = self.needles(evt) if self.needles else ()
         for session, prompts_are_owner in owner_sessions(evt, needles):
             for turn in session.recent_messages(self.window).turns:

captain_hook/grants/orca.py

@@ -0,0 +1,92 @@
+"""An Orca lane reads its Run coordinator's grants, bound through Orca's own orchestration record."""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+from typing import TYPE_CHECKING
+
+from loguru import logger
+from pydantic import BaseModel, Field
+
+from captain_hook.grants import store
+from captain_hook.grants.evidence import tree_of
+from captain_hook.util import reqenv
+
+if TYPE_CHECKING:
+    from captain_hook.events import BaseHookEvent
+
+BINDING_TTL = timedelta(minutes=1)
+
+
+class OrcaRun(BaseModel):
+    terminal: str | None = None
+    run: str | None = None
+    coordinator: str | None = None
+    checked: datetime | None = None
+    pinned: dict[str, str] = Field(default_factory=dict[str, str])
+
+
+def terminal() -> str | None:
+    return reqenv.getenv("ORCA_TERMINAL_HANDLE") or None
+
+
+def attended() -> bool:
+    return reqenv.getenv("CLAUDE_CODE_SESSION_ATTENDED") == "1"
+
+
+def resolve(handle: str) -> tuple[str | None, str | None]:
+    from captain_hook.builtin_packs.general.hooks._sessions import orca_json, worker_of
+
+    worker = worker_of(handle)
+    if not isinstance(worker, dict) or worker.get("dispatchStatus") != "dispatched" or not (run := worker.get("runId")):
+        return None, None
+    shown = ("orca", "orchestration", "run-show", "--id", run, "--json")
+    coordinator = orca_json(shown, "result", "run", "coordinator_handle")
+    return (run, coordinator) if isinstance(coordinator, str) else (None, None)
+
+
+def pin(run: str | None, coordinator: str | None) -> str:
+    return f"{run} {coordinator}"
+
+
+def bound_run(evt: BaseHookEvent, handle: str) -> OrcaRun:
+    slot = evt.ctx.session[OrcaRun]
+    at = store.now()
+    cached = slot.get(OrcaRun())
+    if cached.terminal == handle and cached.checked is not None and at - cached.checked < BINDING_TTL:
+        return cached
+    run, coordinator = resolve(handle)
+    pinned = cached.pinned
+    if run is not None and coordinator is not None and pin(run, coordinator) not in pinned:
+        if (tree := store.terminal_tree(coordinator)) is not None:
+            pinned = pinned | {pin(run, coordinator): tree}
+    found = OrcaRun(terminal=handle, run=run, coordinator=coordinator, checked=at, pinned=pinned)
+    slot.set(found)
+    return found
+
+
+def coordinator_tree(binding: OrcaRun) -> str | None:
+    if binding.coordinator is None or (pinned := binding.pinned.get(pin(binding.run, binding.coordinator))) is None:
+        return None
+    return pinned if store.terminal_tree(binding.coordinator) == pinned else None
+
+
+def record_terminal(evt: BaseHookEvent) -> None:
+    if (handle := terminal()) is None or not attended() or not evt.ctx.session.once(handle, scope="orca-terminal"):
+        return
+    if store.record_terminal(handle, tree_of(evt)):
+        logger.bind(terminal=handle, tree=tree_of(evt)).info("recorded the orca terminal's session tree")
+
+
+def adopt_coordinator(evt: BaseHookEvent) -> None:
+    if (handle := terminal()) is None:
+        return
+    binding = bound_run(evt, handle)
+    tree = tree_of(evt)
+    if (coordinator := coordinator_tree(binding)) is None or coordinator == tree:
+        return
+    agent = f"orca:{binding.run}"
+    if adopted := store.adopt_tree(coordinator, tree=tree, session=evt.session_id, agent=agent):
+        logger.bind(run=binding.run, coordinator=coordinator, tree=tree, adopted=adopted).info(
+            "adopted the orca coordinator's grants"
+        )

captain_hook/grants/store.py

@@ -15,6 +15,7 @@
 from captain_hook.util.paths import resolve_state_dir
 
 RESERVATION_TTL = timedelta(minutes=2)
+EVIDENCE_KINDS = ("ask", "words")
 SCHEMA = """
 CREATE TABLE IF NOT EXISTS grants (
     id TEXT PRIMARY KEY,
@@ -45,6 +46,12 @@
     agent TEXT NOT NULL,
     PRIMARY KEY (grant_id, tree)
 );
+CREATE TABLE IF NOT EXISTS orca_terminals (
+    handle TEXT NOT NULL,
+    tree TEXT NOT NULL,
+    at TEXT NOT NULL,
+    PRIMARY KEY (handle, tree)
+);
 CREATE INDEX IF NOT EXISTS grants_by_kind ON grants (kind, tree);
 CREATE INDEX IF NOT EXISTS spends_by_grant ON spends (grant_id);
 CREATE INDEX IF NOT EXISTS spends_by_call ON spends (tool_use_id, state);
@@ -303,6 +310,41 @@ def adopt(grant_id: str, *, tree: str, session: str, agent: str) -> Adoption:
     return adoption
 
 
+def adopt_tree(source: str, *, tree: str, session: str, agent: str) -> int:
+    """Adopt every spendable grant minted in *source* into *tree*, logged like :func:`adopt`.
+
+    The owner's recorded words and answers stay in their own tree, so a second tree never mints a fresh
+    budget from an approval *source* already spent. Returns how many adoptions were new.
+    """
+    with connect() as db:
+        return db.execute(
+            "INSERT OR IGNORE INTO adoptions (grant_id, tree, at, session, agent)"
+            " SELECT id, ?, ?, ?, ? FROM grants WHERE tree = ? AND kind NOT IN (SELECT value FROM json_each(?))",
+            (tree, now().isoformat(), session, agent, source, json.dumps(EVIDENCE_KINDS)),
+        ).rowcount
+
+
+def record_terminal(handle: str, tree: str) -> bool:
+    """Record that session tree *tree* runs in the Orca terminal *handle*; ``False`` when already recorded."""
+    with connect() as db:
+        return (
+            db.execute(
+                "INSERT OR IGNORE INTO orca_terminals (handle, tree, at) VALUES (?, ?, ?)",
+                (handle, tree, now().isoformat()),
+            ).rowcount
+            == 1
+        )
+
+
+def terminal_tree(handle: str) -> str | None:
+    """The session tree most recently recorded in the Orca terminal *handle*."""
+    with connect() as db:
+        row = db.execute(
+            "SELECT tree FROM orca_terminals WHERE handle = ? ORDER BY at DESC LIMIT 1", (handle,)
+        ).fetchone()
+    return None if row is None else row[0]
+
+
 def adoptions(grant_id: str) -> list[Adoption]:
     with connect() as db:
         rows = db.execute(

captain_hook/primitives/llm.py

@@ -106,7 +106,7 @@ def llm_evaluate[M: BaseModel](
     the event fires inside a subagent or teammate lane, adds that window of the root session that
     spawned the lane as ``<root_transcript>``, so a judge can read the user's words a lane never saw.
     ``root_excerpt`` maps the event to needles (a quote, a thread, the text being judged) and adds every
-    root event that mentions one as ``<root_excerpt>``, however far back the root transcript it sits.
+    event in the last 16 MiB of the root transcript that mentions one as ``<root_excerpt>``.
     """
     from cc_transcript.render import clip
 

captain_hook/snapshots/client.py

@@ -187,6 +187,8 @@ def __call__(self, request: dict[str, object]) -> dict[str, Any]:
             process.stdin.write(encode_frame(frame))
             process.stdin.flush()
             response = read_frame(process.stdout)
+            if response.get("op") == "error" and response.get("id") == self._id:
+                raise SnapshotProtocolError(f"snapshot host failed the request: {response.get('error')}")
             if (
                 type(response.get("protocol")) is not int
                 or type(response.get("id")) is not int

captain_hook/transcripts.py

@@ -2,6 +2,7 @@
 
 import hashlib
 import json
+import os
 import re
 import threading
 from collections import deque
@@ -30,6 +31,7 @@
 # must not smuggle path separators or traversal past that trust boundary.
 INVALID_SESSION_ID = re.compile(r"[/\\]|\x00|^\.\.?$")
 ROOT_TAIL_EVENTS = 256
+ROOT_EXCERPT_BYTES = 16 * 1024 * 1024
 
 
 def user_classifier(events: Sequence[TranscriptEvent], *, path: Path | None = None) -> UserClassifier:
@@ -326,11 +328,18 @@ def root_transcript(path: str | Path, events: int) -> LazyTranscript:
     )
 
 
-def root_excerpt(path: str | Path, needles: Sequence[str], *, around: int = 2, limit: int = 20) -> Session:
-    """The earliest and the newest ``limit`` events of a lane's root session transcript whose line contains
-    any of ``needles``, each with ``around`` events either side, streamed from the whole file so an answer
-    far older than the tail still reaches the judge. The earliest are kept because a quote is first said by
-    whoever it came from and echoed by agents after, so the newest alone drop the words and keep the echoes.
+def root_excerpt(
+    path: str | Path,
+    needles: Sequence[str],
+    *,
+    around: int = 2,
+    limit: int = 20,
+    tail_bytes: int = ROOT_EXCERPT_BYTES,
+) -> Session:
+    """The earliest and the newest ``limit`` events in the last ``tail_bytes`` of a lane's root session
+    transcript whose line contains any of ``needles``, each with ``around`` events either side. Owner words
+    older than that tail reach a grant judge through the store's recorded ``words`` and ``ask`` records.
+    The earliest are kept because a quote is first said by whoever it came from and echoed by agents after.
     A needle matches as typed or JSON-escaped, with or without its non-ASCII escaped.
     """
     from cc_transcript.parser import parse_events_from_bytes
@@ -352,6 +361,11 @@ def root_excerpt(path: str | Path, needles: Sequence[str], *, around: int = 2, l
     after = 0
     reqenv.checkpoint()
     with Path(path).open("rb") as transcript:
+        if (start := transcript.seek(0, os.SEEK_END) - tail_bytes) > 0:
+            transcript.seek(start - 1)
+            transcript.readline()
+        else:
+            transcript.seek(0)
         for line in transcript:
             if any(form in line for form in forms):
                 window = [*before, line]