sessions: ✨ Lift settled-dispatch closes and own-teammate TaskStops

CHANGELOG.md

@@ -8,12 +8,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-- **The general pack denies the harness's `TaskStop` tool.** The mandatory guard matches
-  the exact tool name on `PreToolUse` and `PermissionRequest`. A bare task id does not tell a
-  disposable shell task from a workflow, agent, or teammate session, so the caller's own
-  child workflows and subagents stay protected too. The denial names the target and asks
-  the caller to let it finish, or ask the owner to end it or name it in a cc-notes answer
-  for a later session.
+- **The general pack denies `TaskStop` unless a grant permits it.** The
+  mandatory guard matches the exact tool name on `PreToolUse` and `PermissionRequest`. A
+  bare task id does not tell a disposable shell task from a workflow, agent, or teammate
+  session, so the caller's own child workflows and subagents need a grant too. The denial
+  names the target and asks the caller to let it finish, or ask the owner to end it or name
+  it in a cc-notes answer for a later session.
 - **Session guards spend grants for an owner-named terminal close, launchd service stop,
   or `TaskStop`.** The `sessions.close`, `sessions.launchctl`, and `sessions.task-stop`
   kinds scope permission to a terminal handle, service label, or task id. The budget is
@@ -42,6 +42,46 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   grant refusal reasons reach the user as `sessions: ...`. This replaces the unreleased
   `# ccx:owner-authorized=<answer id>` annotation, which allowed unlimited reuse and
   answers written during the acting session.
+- **A standing owner ruling can lift a close of a settled dispatch's idle terminal.** The
+  general pack's `sessions.close-settled` kind runs only after the per-terminal
+  `sessions.close` lift denies. Only the root session coordinating the dispatch's Run
+  qualifies. The hook event has no `agent_id`, and the request's `ORCA_TERMINAL_HANDLE` is set
+  and matches that Run's `coordinator_handle` from `orca orchestration run-show`.
+  It requires a literal
+  `orca terminal close --terminal <handle>` whose dispatch Orca's worker list records with
+  `dispatchStatus` of `completed` or `failed`, a successful `tui-idle` check, an idle
+  prompt on the screen, and a readable terminal process tree. New `StandingRulings` evidence in
+  `grants/evidence.py` reads cc-notes answers by id with `ccn answer show`: `c9b27c1` for
+  settled-dispatch `orca-gc` closes and `6190a4a` for failed-launch orphans. Only rulings
+  last written before the acting session started count; each is pinned live to its
+  revision, and an id absent from the repository (exit 3) supplies no evidence. The grants
+  `Judge`, capt-hook's small model, decides whether a ruling covers the terminal's Orca
+  record; the proposal names the Run and describes the caller as its coordinator.
+  The minted grant has empty scope, unlimited uses, and no expiry; each close
+  spends it again and is judged again. After the judge allows, the guard re-reads the
+  worker row and idle prompt. The same dispatch must still hold the terminal as
+  `completed` or `failed`, and the agent must still be idle. A failed recheck blocks the
+  close, releases the reserved spend, and reports the change in `systemMessage`.
+  Other terminal lanes, in-process teammates, callers outside Orca, and another Run's
+  coordinator get no lift. Live or unsettled dispatches, busy agents,
+  terminals without dispatches, rulings written during the session, unreadable process
+  trees, loops, and batches never receive this lift.
+- **An agent can stop a teammate its own transcript proves it spawned.**
+  `sessions.task-stop` now accepts `OwnTeammate` evidence for a `<name>@session-<8 hex>`
+  task id. The acting agent's transcript must record an `Agent` or `Task` call whose
+  `toolUseResult` has `status: teammate_spawned` and that exact `teammate_id`; the stop is
+  logged as a spend of a one-use deterministic grant. The team id is the Claude process's
+  session id at spawn and can differ from the hook payload's session id after a resume.
+  This fixes the refusal of `aig-no-delete-plan@session-67c0e5da` from session `900424b6`
+  by proving the spawn instead of matching session ids. Bare ids, teammates this agent
+  never spawned, a sibling's teammates, the root's teammates when a lane asks, and other
+  sessions' teammates receive no own-teammate lift.
+- **Tests pin the sessions guard's acceptance of `orca-gc` invocations.**
+  The guard accepts `orca-gc --run <run> [--dispatch <ctx>]`; no guard code change was
+  needed for it.
+- **Grants support unlimited uses and absent cc-notes answers.**
+  `Grants.mint` accepts `None` for unlimited uses, and `ccn_answer` returns `None` when
+  `ccn answer show` reports not-found with exit 3.
 
 ### Fixed
 

captain_hook/builtin_packs/general/hooks/_sessions.py

@@ -23,7 +23,7 @@
 from captain_hook.cmd import Cmd
 from captain_hook.command_schemas import ORCA, OSASCRIPT
 from captain_hook.dispatch import SYNC_DEADLINE_MARGIN_SECONDS, collect_budget
-from captain_hook.grants import Allowed, Evidence, Grants, Proposal, Rulings
+from captain_hook.grants import Allowed, Evidence, Grants, Judge, Proposal, Rulings, StandingRulings
 from captain_hook.guard_literal import FOLD_TABLE, GUARDED_WORD, QUOTING_CHARS, names_guarded
 from captain_hook.util import proc, reqenv
 from captain_hook.util.payload import command_texts
@@ -142,6 +142,20 @@
 RECEIPT_SLACK = timedelta(seconds=1)
 RETRY_WINDOW = timedelta(minutes=2)
 LATER_SESSION = "name it in a cc-notes answer for a later session"
+SETTLED = frozenset({"completed", "failed"})
+CLASS_RULINGS = ("c9b27c1", "6190a4a")
+SETTLED_CLOSE_RULES = (
+    "The owner's standing rulings in the evidence cover a class of Orca terminal closes rather than one named "
+    "terminal: c9b27c1 lets the root close a settled dispatch's idle terminal, as orca-gc does, and 6190a4a lets it "
+    "close an orphan terminal its own failed launch created. The proposed action's payload is Orca's record for the "
+    "terminal: the dispatch holding it, its Run, and that dispatch's status. The guard has already proven the status "
+    "is completed or failed, the terminal's agent idles at a prompt, and the caller is the root session coordinating "
+    "that Run. Allow only when a ruling's words cover closing "
+    "a terminal with this record, and cite that ruling. Deny when the rulings exclude this record, or limit the class "
+    "to terminals it is not."
+)
+TEAMMATE_TASK = re.compile(r"[\w.-]+@session-[0-9a-f]{8}")
+SPAWN_TOOLS = frozenset({"Agent", "Task"})
 INLINE_LOGIN = "/usr/bin/login -flpq dev /bin/bash --noprofile --norc -p -c orca-tcc-login"
 INLINE_TABLE = (
     "    1     0     1    0 Thu Jan  1 00:00:00 2026 /sbin/launchd\n"
@@ -204,11 +218,63 @@ def inline_create(
     ]
 
 
+def inline_spawn(task: str, *, tool: str = "Agent", status: str = "teammate_spawned") -> list[dict[str, Any]]:
+    name, _, team = task.partition("@")
+    return [
+        *INLINE_TRANSCRIPT,
+        {
+            "type": "assistant",
+            "message": {
+                "role": "assistant",
+                "content": [{"type": "tool_use", "id": "toolu_spawn", "name": tool, "input": {"name": name}}],
+            },
+        },
+        {
+            "type": "user",
+            "message": {
+                "role": "user",
+                "content": [{"type": "tool_result", "tool_use_id": "toolu_spawn", "content": f"agent_id: {task}"}],
+            },
+            "toolUseResult": {"status": status, "teammate_id": task, "agent_id": task, "name": name, "team_name": team},
+        },
+    ]
+
+
 def inline_workers(*handles: str) -> str:
     rows = [{"dispatchId": f"ctx_{index}", "agentTerminalHandle": handle} for index, handle in enumerate(handles)]
     return json.dumps({"ok": True, "result": {"workers": rows, "page": {}, "scope": {"source": "all"}}})
 
 
+def inline_worker(handle: str, status: str, stage: str = "settled") -> str:
+    row = {
+        "dispatchId": "ctx_settled",
+        "runId": "run_inline",
+        "dispatchStatus": status,
+        "agentTerminalHandle": handle,
+        "projection": {"stage": {"detail": stage}},
+    }
+    return json.dumps({"ok": True, "result": {"workers": [row], "page": {}, "scope": {"source": "all"}}})
+
+
+def inline_class_rulings(*, written: datetime = INLINE_STARTED - timedelta(days=1)) -> dict[str, str]:
+    bodies = {
+        "c9b27c1": "Owner: after a dispatch settles the root runs orca-gc to close that dispatch's idle terminal; "
+        "never a live or unsettled dispatch.",
+        "6190a4a": "Owner: the root may close an orphan Orca terminal its own failed launch created.",
+    }
+    return {
+        f"ccn answer show {ident}": json.dumps(
+            {"id": f"{ident}aaaa", "title": "Close settled terminals", "body": body, "updated_at": written.isoformat()}
+        )
+        for ident, body in bodies.items()
+    }
+
+
+def inline_run(coordinator: str) -> dict[str, str]:
+    shown = {"ok": True, "result": {"run": {"id": "run_inline", "coordinator_handle": coordinator}}}
+    return {"orca orchestration run-show --id run_inline": json.dumps(shown)}
+
+
 def inline_screen(*tail: str) -> str:
     return json.dumps({"ok": True, "result": {"terminal": {"source": "screen", "tail": list(tail)}}})
 
@@ -783,7 +849,7 @@ def runs_once(call: Call, scan: Scan) -> bool:
     )
 
 
-def dispatch_of(handle: str) -> str | Unreadable | None:
+def worker_of(handle: str) -> dict[str, Any] | Unreadable | None:
     cursor: tuple[str, ...] = ()
     while True:
         argv = ("orca", "orchestration", "worker-list", "--limit", str(WORKER_PAGE), *cursor, "--json")
@@ -794,7 +860,7 @@ def dispatch_of(handle: str) -> str | Unreadable | None:
                 return Unreadable("Orca scoped its worker list to one Run")
             holder = next(
                 (
-                    row["dispatchId"]
+                    row
                     for row in page["workers"]
                     if handle in (row.get("agentTerminalHandle"), (row.get("resource") or {}).get("terminalHandle"))
                 ),
@@ -881,10 +947,11 @@ def collect(self, evt: BaseHookEvent, action: Proposal) -> list[Evidence]:
         if found is None:
             return []
         use, bash = found
-        holder = dispatch_of(handle)
+        holder = worker_of(handle)
         ready = idle(handle) if holder is None else False
         if holder is not None or ready is not True:
-            logger.bind(terminal=handle, dispatch=holder, idle=ready).info("a terminal this session created is busy")
+            dispatch = holder.get("dispatchId") if isinstance(holder, dict) else holder
+            logger.bind(terminal=handle, dispatch=dispatch, idle=ready).info("a terminal this session created is busy")
             return []
         return [
             Evidence(
@@ -899,6 +966,68 @@ def collect(self, evt: BaseHookEvent, action: Proposal) -> list[Evidence]:
         ]
 
 
+def settled_worker(handle: str) -> dict[str, Any] | None:
+    worker = worker_of(handle)
+    if not isinstance(worker, dict) or worker.get("dispatchStatus") not in SETTLED or idle(handle) is not True:
+        return None
+    return worker
+
+
+def coordinates(evt: BaseHookEvent, run: str | None) -> bool:
+    caller = reqenv.getenv("ORCA_TERMINAL_HANDLE")
+    if evt.agent_id is not None or not caller or not run:
+        return False
+    shown = ("orca", "orchestration", "run-show", "--id", run, "--json")
+    return orca_json(shown, "result", "run", "coordinator_handle") == caller
+
+
+def settled_close(evt: BaseHookEvent, handle: str, tab: bool) -> Proposal | None:
+    if (worker := settled_worker(handle)) is None or not coordinates(evt, run := worker.get("runId")):
+        return None
+    dispatch, status = worker["dispatchId"], worker["dispatchStatus"]
+    stage = ((worker.get("projection") or {}).get("stage") or {}).get("detail")
+    return Proposal(
+        scope={},
+        payload={"terminal": handle, "tab": tab, "dispatch": dispatch, "run": run, "status": status},
+        summary=f"the coordinator of run {run} closes terminal {handle}, whose dispatch {dispatch} is {status} "
+        f"({stage}) and whose agent idles at a prompt",
+    )
+
+
+def still_settled(action: Proposal) -> bool:
+    worker = settled_worker(action.payload["terminal"])
+    return worker is not None and worker["dispatchId"] == action.payload["dispatch"]
+
+
+def spawned(use: Any, task: str) -> bool:
+    if use.call.name not in SPAWN_TOOLS or use.result is None:
+        return False
+    result = use.result.tool_use_result
+    return isinstance(result, dict) and result.get("status") == "teammate_spawned" and result.get("teammate_id") == task
+
+
+@dataclass(frozen=True, slots=True)
+class OwnTeammate:
+    def collect(self, evt: BaseHookEvent, action: Proposal) -> list[Evidence]:
+        task = action.scope["task"]
+        if TEAMMATE_TASK.fullmatch(task) is None:
+            return []
+        found = next((use for turn in evt.ctx.t.turns for use in turn.tool_uses if spawned(use, task)), None)
+        if found is None:
+            return []
+        return [
+            Evidence(
+                id=f"teammate:{task}",
+                source="teammate",
+                quote=task,
+                said_at=found.result_ts or found.ts,
+                detail=f"this agent's own transcript records spawning {task} as its teammate",
+                key=f"teammate:{evt.session_id}/{evt.agent_id or 'main'}/{task}",
+                live=True,
+            )
+        ]
+
+
 def rulings_naming(key: str) -> Rulings:
     return Rulings(search=lambda action: action.scope[key])
 
@@ -923,9 +1052,21 @@ def rulings_naming(key: str) -> Rulings:
 TASK_STOP = Grants(
     "sessions.task-stop",
     ("task",),
-    evidence=(rulings_naming("task"),),
+    evidence=(OwnTeammate(), rulings_naming("task")),
     replay=RETRY_WINDOW,
-    would_allow="Have the owner name the task id in a cc-notes answer before the stopping session starts.",
+    would_allow="Stop only a teammate this agent spawned, by its `<name>@session-<id>` task id, or have the owner "
+    "name the task id in a cc-notes answer before the stopping session starts.",
+    hook="sessions",
+)
+SETTLED_CLOSE = Grants(
+    "sessions.close-settled",
+    (),
+    judge=Judge(rules=SETTLED_CLOSE_RULES),
+    evidence=(StandingRulings(CLASS_RULINGS),),
+    mint=None,
+    ttl=None,
+    would_allow="Close only an idle terminal whose dispatch Orca records as settled, under a standing owner ruling "
+    "that predates the closing session.",
     hook="sessions",
 )