diff --git a/captain_hook/builtin_packs/general/hooks/grants.py b/captain_hook/builtin_packs/general/hooks/grants.py
index afc6b71f..8d27d30a 100644
--- a/captain_hook/builtin_packs/general/hooks/grants.py
+++ b/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]
diff --git a/captain_hook/context.py b/captain_hook/context.py
index d700181c..52aa2901 100644
--- a/captain_hook/context.py
+++ b/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
 
diff --git a/captain_hook/grants/declare.py b/captain_hook/grants/declare.py
index 8143a020..a8dec151 100644
--- a/captain_hook/grants/declare.py
+++ b/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:
diff --git a/captain_hook/grants/evidence.py b/captain_hook/grants/evidence.py
index 56283535..658ed17a 100644
--- a/captain_hook/grants/evidence.py
+++ b/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:
diff --git a/captain_hook/grants/orca.py b/captain_hook/grants/orca.py
new file mode 100644
index 00000000..25688e02
--- /dev/null
+++ b/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"
+        )
diff --git a/captain_hook/grants/store.py b/captain_hook/grants/store.py
index a3391f8b..1e84f29d 100644
--- a/captain_hook/grants/store.py
+++ b/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(
diff --git a/captain_hook/primitives/llm.py b/captain_hook/primitives/llm.py
index e97fd311..a54de951 100644
--- a/captain_hook/primitives/llm.py
+++ b/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
 
diff --git a/captain_hook/snapshots/client.py b/captain_hook/snapshots/client.py
index 0cd493f0..b4f72ff4 100644
--- a/captain_hook/snapshots/client.py
+++ b/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
diff --git a/captain_hook/transcripts.py b/captain_hook/transcripts.py
index e4be3a8a..5116f313 100644
--- a/captain_hook/transcripts.py
+++ b/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]
diff --git a/docs/guide/grants.qmd b/docs/guide/grants.qmd
index 4ba115d9..528563a6 100644
--- a/docs/guide/grants.qmd
+++ b/docs/guide/grants.qmd
@@ -199,7 +199,7 @@ The defaults are `model="small"`, `specialty="review"`, and `deadline=20` second
 `contexts` accepts the same prompt contexts as other LLM hooks.
 `transcript` and `root_transcript` default to `False`; set either to a message
 count, `"recent"`, or `"full"` when the judge needs more context.
-`root_excerpt` maps an event to search needles in the root transcript.
+`root_excerpt` maps an event to search needles in the last 16 MiB of the root transcript.
 `tool_results=False` keeps tool output out of those windows by default.
 
 The evidence providers determine which words can authorize an action:
@@ -220,11 +220,14 @@ judge raises `ValueError`: those sources need a judge to interpret the owner's w
 selection, typed answer, and per-question notes. It distinguishes what was shown
 from what was chosen. The builtin `general` pack records answers on
 `PostToolUse(AskUserQuestion)` as tree-bound `ask` grants with one use and a
-24-hour expiry. These are evidence records; the declaration mints its own kind
-after the judge allows.
+7-day expiry, and a root session's own prompts on `UserPromptSubmit` as `words`
+grants with the same expiry. These are evidence records; the declaration mints its
+own kind after the judge allows.
 
 `Asked` and `OwnerWords` default to a window of 60 and accept a `needles` callback
-to include root-session excerpts farther back. Set
+to include matching events from the last 16 MiB of the root transcript. Evidence
+reads never scan a whole transcript: owner words older than these windows reach the
+judge only through the recorded `ask` and `words` records. Set
 `needles=lambda evt: ("C_EXAMPLE",)` to search for that channel id.
 `OwnerWords(machine=("<machine-message>",))` excludes prompts containing that marker.
 Tool output, teammate messages, and notifications are not owner evidence.
@@ -257,8 +260,10 @@ one answer naming two terminals mints two grants. Rulings are live evidence, so
 stored grants recheck the answer's revision before spending; editing the answer
 stops its existing grant from covering the action.
 
-Stored `ask` evidence is collected only while its record is unexpired and
-unrevoked. `Asked` also reads recent transcript answers. To stop an already minted
+Stored `ask` and `words` evidence is collected only while its record is unexpired
+and unrevoked. The judge of a stored grant also reads expired records said after the
+grant was minted, so a withdrawal never lapses while the grant is usable. `Asked`
+also reads recent transcript answers. To stop an already minted
 permission, revoke the grant of the declaration's kind.
 
 ## Follow the check order
@@ -489,6 +494,25 @@ The CLI reads the adopting session from the session environment variables even
 when `--tree` is explicit. `grant show` logs each adoption with its destination
 tree, timestamp, session, and agent.
 
+An Orca lane adopts its Run coordinator's grants without a command. The builtin
+`general` pack records the Orca terminal of each attended Claude session from
+`ORCA_TERMINAL_HANDLE`, once per session. A `claude -p` session is unattended and
+records nothing.
+
+On a grants check in an Orca terminal, the framework finds the
+dispatched worker for that terminal with `orca orchestration worker-list`, and reads
+the Run's coordinator terminal with `orca orchestration run-show`. It then adopts
+every spendable grant minted in the session tree recorded for that terminal. The
+coordinator's recorded `words` and `ask` records are never adopted, so one approval
+still spends once. Each adoption is logged with agent `orca:<run>`, so both trees
+share the budget, and `grant show` lists the lane.
+
+A lane pins the coordinator session it first binds for each Run. When the coordinator
+terminal's record moves to another session, the lane adopts nothing more from it.
+The session re-reads the Run from Orca after one minute, or when its terminal
+changes, so a finished worker stops adopting. Outside Orca, or with no recorded
+coordinator session, nothing is adopted.
+
 Inspect and revoke a grant with its printed id in place of `GRANT_ID`:
 
 ```bash
diff --git a/tests/conftest.py b/tests/conftest.py
index 4163bdb1..269da668 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -89,6 +89,8 @@ def clean_state(tmp_path_factory: pytest.TempPathFactory, monkeypatch: pytest.Mo
     # The SessionEnd reviewer skips headless entrypoints (sdk-*); scrub it so tests don't
     # inherit the ambient CLAUDE_CODE_ENTRYPOINT of a pytest run launched inside claude.
     monkeypatch.delenv("CLAUDE_CODE_ENTRYPOINT", raising=False)
+    monkeypatch.delenv("ORCA_TERMINAL_HANDLE", raising=False)
+    monkeypatch.delenv("CLAUDE_CODE_SESSION_ATTENDED", raising=False)
     config_dir = tmp_path_factory.mktemp("claude-config")
     (config_dir / "plugins").mkdir()
     monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(config_dir))
diff --git a/tests/test_context.py b/tests/test_context.py
index c813648e..7d1f191c 100644
--- a/tests/test_context.py
+++ b/tests/test_context.py
@@ -639,7 +639,7 @@ def test_root_transcript_block_renders_the_lane_root_and_is_empty_outside_a_lane
         )
         assert HookContext(session=SessionStore(None), transcript=lane, settings=None).root_transcript_block() == ""
 
-    def test_root_excerpt_keeps_only_matching_events_from_the_whole_root(self) -> None:
+    def test_root_excerpt_keeps_only_matching_events_from_the_root(self) -> None:
         from captain_hook.context import render_window
         from captain_hook.prompt import Prompt
         from captain_hook.testing.helpers import fixture_file, fixture_session
@@ -684,6 +684,19 @@ def test_root_excerpt_keeps_only_matching_events_from_the_whole_root(self) -> No
         outside = HookContext(session=SessionStore(None), transcript=ctx.transcript, settings=None)
         assert outside.root_excerpt(["x"]) is None
 
+    def test_root_excerpt_reads_only_the_roots_trailing_window(self) -> None:
+        from captain_hook.context import render_window
+        from captain_hook.testing.helpers import fixture_file
+        from captain_hook.transcripts import root_excerpt
+
+        root = fixture_file(
+            [T.user("approve the old draft"), *(T.user(f"status {i}") for i in range(50)), T.user("approve the new")]
+        )
+        last = len(root.read_bytes().splitlines(keepends=True)[-1])
+        for tail_bytes in (last, last + 10):
+            excerpt = root_excerpt(root, ["approve"], around=0, tail_bytes=tail_bytes)
+            assert render_window(excerpt, window=None, tool_results=False, budget=None) == "user: approve the new"
+
     def test_root_excerpt_keeps_an_answer_that_later_agent_echoes_outnumber(self) -> None:
         from captain_hook.testing.helpers import fixture_file
         from captain_hook.transcripts import root_excerpt
diff --git a/tests/test_grants.py b/tests/test_grants.py
index 429de8f6..e4d08751 100644
--- a/tests/test_grants.py
+++ b/tests/test_grants.py
@@ -35,6 +35,7 @@
 )
 from captain_hook.grants import cli as grant_cli_module
 from captain_hook.grants import evidence as evidence_module
+from captain_hook.grants import orca as orca_module
 from captain_hook.grants.cli import grant as grant_cli
 from captain_hook.hook_lint import result_violations
 from captain_hook.types import Action, HookResult, HookSpec, RegisteredHook
@@ -714,3 +715,161 @@ def test_spending_an_unknown_grant_names_it(tmp_path: Path) -> None:
         grant_cli, [*argv, "--session", TREE, "--call", "p", "--fingerprint", "f", "--summary", "s"]
     )
     assert refused.exit_code == 1 and "no grant 000000000000" in refused.output
+
+
+def recorded_words(quote: str, *, expires: datetime, ago: timedelta = timedelta(days=2)) -> Grant:
+    said = evidence_module.words_evidence(quote, store.now() - ago)
+    return store.mint(
+        Grant(
+            id=store.new_id(),
+            kind="words",
+            tree=TREE,
+            scope={},
+            evidence=[said],
+            source_key=said.key,
+            expires=expires,
+            author="words@test",
+            created=store.now() - ago,
+        )
+    )
+
+
+def test_a_lane_reads_its_roots_older_words_from_the_recorded_index(tmp_path: Path) -> None:
+    recorded_words("post the AIG summary in that thread", expires=store.now() + timedelta(days=5))
+    recorded_words("post anything anywhere", expires=store.now() - timedelta(minutes=1))
+    lane = event(tmp_path, session="lane-session")
+    lane.ctx.root_path = tmp_path / f"{TREE}.jsonl"
+    quotes = [item.quote for item in OwnerWords().collect(lane, proposal(lane))]
+    assert quotes == ["post the AIG summary in that thread"]
+
+
+def in_orca(monkeypatch: pytest.MonkeyPatch, handle: str) -> None:
+    monkeypatch.setenv("ORCA_TERMINAL_HANDLE", handle)
+    monkeypatch.setenv("CLAUDE_CODE_SESSION_ATTENDED", "1")
+
+
+def fake_orca(monkeypatch: pytest.MonkeyPatch, status: list[str] | None = None) -> list[str]:
+    calls: list[str] = []
+    lane = status or ["dispatched"]
+
+    def resolve(handle: str) -> tuple[str | None, str | None]:
+        calls.append(handle)
+        return ("run_1", "term_root") if handle == "term_lane" and lane[0] == "dispatched" else (None, None)
+
+    monkeypatch.setattr(orca_module, "resolve", resolve)
+    return calls
+
+
+def test_an_attended_orca_session_records_its_terminal_once(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    in_orca(monkeypatch, "term_root")
+    evt = event(tmp_path)
+    orca_module.record_terminal(evt)
+    orca_module.record_terminal(evt)
+    assert store.terminal_tree("term_root") == TREE
+    assert not store.record_terminal("term_root", TREE)
+    monkeypatch.setenv("CLAUDE_CODE_SESSION_ATTENDED", "0")
+    monkeypatch.setenv("ORCA_TERMINAL_HANDLE", "term_nested")
+    orca_module.record_terminal(event(tmp_path / "nested", session="nested-print"))
+    assert store.terminal_tree("term_nested") is None
+
+
+def test_an_orca_lane_spends_its_coordinators_grant(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    store.record_terminal("term_root", TREE)
+    grant = minted(uses=2)
+    calls = fake_orca(monkeypatch)
+    in_orca(monkeypatch, "term_lane")
+    lane = event(tmp_path / "lane", "lane", session="lane-root")
+    assert declared().check(lane)
+    assert [(found.tree, found.agent) for found in store.adoptions(grant.id)] == [("lane-root", "orca:run_1")]
+    assert declared().check(event(tmp_path / "lane", "again", session="lane-root", call="c2"))
+    assert calls == ["term_lane"]
+    assert not declared().check(event(tmp_path, "root", call="c3"))
+
+
+def test_no_orca_binds_nothing(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    store.record_terminal("term_root", TREE)
+    minted(uses=None)
+    calls = fake_orca(monkeypatch)
+    monkeypatch.delenv("ORCA_TERMINAL_HANDLE", raising=False)
+    assert not declared().check(event(tmp_path, session="lane-root"))
+    in_orca(monkeypatch, "term_unknown")
+    assert not declared().check(event(tmp_path / "other", session="other-root", call="c2"))
+    assert calls == ["term_unknown"]
+
+
+def test_an_orca_lane_never_adopts_the_coordinators_recorded_words(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    store.record_terminal("term_root", TREE)
+    words = recorded_words("send one reply in that thread", expires=store.now() + timedelta(days=5))
+    grant = minted(uses=1)
+    fake_orca(monkeypatch)
+    in_orca(monkeypatch, "term_lane")
+    assert declared().check(event(tmp_path / "a", "a", session="lane-a"))
+    assert store.adoptions(words.id) == [] and [found.tree for found in store.adoptions(grant.id)] == ["lane-a"]
+    judged = declared(judge=Judge("rules"), evidence=(OwnerWords(),))
+    lane_b = event(tmp_path / "b", "b", session="lane-b", call="c2", allow=True, reason="ok", relied_on=[words.id])
+    assert evidence_module.recorded(lane_b, "words") == []
+    assert not judged.check(lane_b)
+    assert not declared().check(event(tmp_path, "root", call="c3"))
+
+
+def test_a_reused_coordinator_terminal_never_lends_the_new_sessions_grants(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    store.record_terminal("term_root", TREE)
+    minted(uses=None)
+    fake_orca(monkeypatch)
+    in_orca(monkeypatch, "term_lane")
+    assert declared().check(event(tmp_path / "lane", "lane", session="lane-root"))
+    store.record_terminal("term_root", "unrelated-root")
+    unrelated = store.mint(
+        Grant(id=store.new_id(), kind="test.write", tree="unrelated-root", scope=SCOPE, author="t", created=store.now())
+    )
+    monkeypatch.setattr(orca_module, "BINDING_TTL", timedelta(0))
+    declared().check(event(tmp_path / "lane", "again", session="lane-root", call="c2"))
+    assert store.adoptions(unrelated.id) == []
+
+
+def test_an_orca_binding_is_revalidated_after_its_ttl(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    store.record_terminal("term_root", TREE)
+    first = minted(uses=None)
+    status = ["dispatched"]
+    calls = fake_orca(monkeypatch, status)
+    in_orca(monkeypatch, "term_lane")
+    assert declared().check(event(tmp_path / "lane", "lane", session="lane-root"))
+    assert [found.tree for found in store.adoptions(first.id)] == ["lane-root"]
+    monkeypatch.setenv("ORCA_TERMINAL_HANDLE", "term_moved")
+    declared().check(event(tmp_path / "lane", "moved", session="lane-root", call="c2"))
+    assert calls == ["term_lane", "term_moved"]
+    monkeypatch.setenv("ORCA_TERMINAL_HANDLE", "term_lane")
+    monkeypatch.setattr(orca_module, "BINDING_TTL", timedelta(0))
+    status[0] = "completed"
+    later = minted(uses=None)
+    declared().check(event(tmp_path / "lane", "done", session="lane-root", call="c3"))
+    assert store.adoptions(later.id) == []
+
+
+def test_a_withdrawal_whose_record_expired_still_reaches_the_judge(tmp_path: Path) -> None:
+    grant = minted(approved={"text": "ok"}, uses=None)
+    recorded_words("stop posting there", expires=store.now() - timedelta(seconds=1), ago=timedelta(seconds=30))
+    evt = event(tmp_path, "ok", allow=False, reason="the owner withdrew it", withdrawn=True)
+    denied = declared(rules=(ContentMatches(),), judge=Judge("rules"), evidence=(OwnerWords(),)).check(evt)
+    assert isinstance(denied, Denied) and "withdrew" in denied.reason
+    assert "stop posting there" in evt.ctx.call_llm.call_args_list[0].args[0].system_text
+    assert store.load(grant.id).revoked is not None
+
+
+@pytest.mark.parametrize(
+    ("status", "expected"),
+    [("dispatched", ("run_inline", "term_root")), ("completed", (None, None))],
+    ids=["dispatched", "settled"],
+)
+def test_a_lane_resolves_its_run_through_orcas_worker_list(status: str, expected: tuple[str | None, ...]) -> None:
+    from captain_hook.builtin_packs.general.hooks._sessions import inline_run, inline_worker
+    from captain_hook.testing.helpers import stubbed_commands
+
+    commands = {"orca orchestration worker-list": inline_worker("term_lane", status), **inline_run("term_root")}
+    with stubbed_commands(commands):
+        assert orca_module.resolve("term_lane") == expected
+        assert orca_module.resolve("term_other") == (None, None)
diff --git a/tests/test_pack_sessions.py b/tests/test_pack_sessions.py
index 749667ac..6c1dd798 100644
--- a/tests/test_pack_sessions.py
+++ b/tests/test_pack_sessions.py
@@ -34,6 +34,7 @@
 from captain_hook.dispatch import SYNC_DEADLINE_MARGIN_SECONDS, dispatch
 from captain_hook.events import PreToolUseEvent
 from captain_hook.grants import evidence as evidence_module
+from captain_hook.grants import orca as orca_grants
 from captain_hook.grants import store
 from captain_hook.loader import discover_pack
 from captain_hook.session import SessionStore
@@ -754,6 +755,7 @@ def test_a_class_lift_rechecks_the_orca_record_after_the_judge_allows(
     ) -> None:
         monkeypatch.setattr(_sessions, "idle", lambda handle, seen=iter(idles): next(seen))
         monkeypatch.setattr(_sessions, "worker_of", lambda handle, seen=iter(workers): next(seen))
+        monkeypatch.setattr(orca_grants, "resolve", lambda handle: (None, None))
         with stubbed_commands(SETTLED):
             envelope = envelope_of(bash(AGENT_CLOSE, llm=CLASS_ALLOW), tmp_path, env=ROOT)
         assert envelope is not None
diff --git a/tests/test_snapshot_owner.py b/tests/test_snapshot_owner.py
index 421dcb38..8fd0f5f5 100644
--- a/tests/test_snapshot_owner.py
+++ b/tests/test_snapshot_owner.py
@@ -1,6 +1,7 @@
 import io
 import json
 import struct
+import sys
 import threading
 import time
 from types import SimpleNamespace
@@ -8,7 +9,14 @@
 import pytest
 from jsonschema import ValidationError
 
-from captain_hook.snapshots.client import CORE_SCHEMA, HOST_SCHEMA, MAX_FRAME_BYTES, SnapshotProtocolError, encode_frame
+from captain_hook.snapshots.client import (
+    CORE_SCHEMA,
+    HOST_SCHEMA,
+    MAX_FRAME_BYTES,
+    Bridge,
+    SnapshotProtocolError,
+    encode_frame,
+)
 from captain_hook.snapshots.validation import checked, validator
 from captain_hook.snapshots.worker import OWNER_ADMISSION, OwnerService, empty_usage, failure, handshake, read_frame
 
@@ -509,3 +517,25 @@ def native_call(body, *, context, cancellation):
     assert captured[0]["registry_generation"] == "content-fingerprint"
     assert captured[0]["authority"] == authority
     assert captured[0]["claimant"] == "fixture"
+
+
+REFUSING_HOST = """
+import json, struct, sys
+def read():
+    (size,) = struct.unpack(">I", sys.stdin.buffer.read(4))
+    return json.loads(sys.stdin.buffer.read(size))
+def write(value):
+    body = json.dumps(value).encode()
+    sys.stdout.buffer.write(struct.pack(">I", len(body)) + body)
+    sys.stdout.buffer.flush()
+write(read())
+frame = read()
+write({"protocol": 1, "op": "error", "id": frame["id"], "error": "captain: snapshot admission queue exhausted"})
+"""
+
+
+def test_the_bridge_names_the_error_the_host_returns():
+    bridge = Bridge((sys.executable, "-c", REFUSING_HOST))
+    with pytest.raises(SnapshotProtocolError, match="snapshot host failed the request: captain: snapshot admission"):
+        bridge({"schema": HOST_SCHEMA})
+    bridge.close()