Merge pull request #841 from corbitsdev/cl-7524-wake-parents-for-parked-director-questions

Wake parents for parked director questions

Author: TheGreatAxios

docs/ARCHITECTURE.md

@@ -211,6 +211,8 @@ The **`spawn_agent`** tool starts a fleet agent on a separate inference source (
 
 Workers ask the spawning parent with **`ask_director`** (not the human). That parks a question while the worker stays `running`. **`wait_agents`** returns `awaiting_director` with a question payload — that is not terminal. The parent answers with **`send_input`**, then **`wait_agents`** again. Escalate to the human with **`ask_operator`** only when the parent cannot resolve it.
 
+When the parent TUI is not blocked in `wait_agents`, the runner publishes an authoritative snapshot of currently pending top-level questions on each store notification, including empty snapshots before fleet-count updates. During synchronous session rotation, a runner-owned barrier suppresses both publications before delivery-generation invalidation, transcript clearing, and worker cancellation; successful reset reconciles a fresh snapshot before resuming asynchronous backend rebuild. The bridge drops resolved, cancelled, replaced, terminal, and removed asks and delivers each session/question identity once while pending. A coalesced wake starts only when the parent is not processing and every operator gate is closed, including parent-idle fleet holds where the shell stays busy. Worker gates do not manufacture parent processing. Replies use `send_input`'s `target` field with the worker session ID, never its shared catalog ID. Synthetic wakes use `SessionPort.deliver` through queued-delivery's idle-send path without entering the user follow-up queue or composer `/feedback` capture.
+
 When profiles exist (local `.agents/agents/` and/or enabled **`kind: "agent"`** plugins, including **data-only** markdown plugins with no `index.ts`), the chat model also receives **`search_agents`** — a lexical index over profile id, description, and role text so the model can discover ids before calling `spawn_agent(agent=...)`. Results include each match's full loaded system prompt / body so the parent can inspect plugin or Claude marketplace agents without `read_file` on paths outside the session cwd (path-escape blocks those roots by design; writes remain blocked). `spawn_agent` and `search_agents` are core tools on the primary session.
 
 Built-in directors with `spawn.maySpawn` may themselves call `spawn_agent` (one hop only): nested dispatch installs the mailbox-scoped fleet verbs (`spawn_agent`, `wait_agents`, `list_agents`, …) with `allowOrchestrator: false` so the tree bottoms out. Profile-sourced `orchestrator: true` is rejected before a session starts because it has no trusted tier/authority semantics today. Fleet discovery (`search_agents`) stays Tier 1 only. Unknown `agent` ids fail closed.

docs/PRODUCT.md

@@ -168,7 +168,7 @@ Corbits Code fans work out to short-lived **fleet agents** — workers with thei
 
 - **Agents** are runtime entities (primary session or child).
 - **Tasks** are checklist items owned by one agent via `manage_tasks`.
-- **Fleet agents** are spawned with `spawn_agent` / `wait_agents`. Workers ask the parent with `ask_director`. When `wait_agents` returns status `running` plus a question payload, the parent answers with `send_input`, then `wait_agents` again. Escalate to the human only with `ask_operator`.
+- **Fleet agents** are spawned with `spawn_agent` / `wait_agents`. Workers ask the parent with `ask_director`. That parks a question while the worker stays `running`. `wait_agents` returns `awaiting_director` with a question payload — that is not terminal. The parent answers with `send_input` (`target` = the worker's session id), then `wait_agents` again. When the parent TUI is not blocked in `wait_agents`, a parked question arrives as a synthetic idle-send wake. Escalate to the human only with `ask_operator`.
 
 Dispatch uses a structured brief (context / goal / optional goals seed) and returns a structured report. The TUI Agents strip and fleet board show who is running; live tool progress updates the status bar without dumping the child transcript into the parent chat. There is no turn budget. A tool-less final turn completes only with the four-heading report envelope; without it, one nudge is given and a second tool-less turn without the envelope salvages as `incomplete-report-stop`. A silent worker (no activity for `stallTimeoutMs`, opt-in) gets one continuation nudge, then salvages as `stalled` if a second consecutive check finds no activity. An opt-in `deadlineMs`, or an operator cancel, can also end a run early. Each of these returns a salvage report so a runaway or idle child cannot quietly burn a large token budget or look done after prose alone.
 

docs/TUI.md

@@ -221,6 +221,15 @@ status / current tool) — Amp/Codex-style lanes without a FLEET header board:
 ```
 
 An `ask_director` lane stays live and reads as waiting on the director, not stalled.
+The runner snapshots currently pending root-worker questions, dropping resolved,
+cancelled, replaced, terminal, or removed asks before delivery. It sends one
+coalesced wake when the parent is not processing and all operator gates are closed,
+even while live workers hold the shell busy. Replies use `send_input`'s `target`
+field with the worker's session ID, not its shared catalog ID. The runner
+publishes snapshots and the bridge delivers each session/question identity
+once while pending; the agents strip never re-delivers it. Synthetic wakes use
+the idle delivery path, bypassing composer `/feedback` capture and leaving queued
+user follow-ups untouched.
 
 `formatChromeZones` → `formatAgentsPanel` owns that paint. Geometry stays
 stack-only (`layoutMode: "stack"`, `railWidth: 0`); the zone max is

src/agent/directors/skywalker/package.test.ts

@@ -205,11 +205,15 @@ describe("skywalkerPackage", () => {
     expect(skywalkerPackage.systemPrompt).not.toMatch(/\bleaves\b/i);
   });
 
-  test("systemPrompt answers wait_agents questions via send_input", () => {
+  test("systemPrompt answers parked director questions via send_input", () => {
     const p = skywalkerPackage.systemPrompt;
     expect(p).toContain("ask_director");
     expect(p).toContain("send_input");
-    expect(p).toMatch(/wait_agents returns status running plus a question/i);
+    expect(p).toContain("awaiting_director");
+    expect(p).toContain("idle-send");
+    expect(p).toContain("target = that worker's session id");
+    expect(p).toContain("target = worker session id");
+    expect(p).not.toMatch(/wait_agents returns status running plus a question/i);
     expect(p).toMatch(/Escalate with ask_operator only when you cannot resolve it/);
   });
 

src/agent/directors/skywalker/package.ts

@@ -20,7 +20,7 @@ Idle-orchestrator: fire one or more spawn_agent calls in a turn — each returns
 
 # Operator updates (mandatory while fleet is live)
 
-You are the chat surface. Workers cannot ask_operator; they ask_director. When wait_agents returns status running plus a question, answer with send_input, then wait_agents again. Escalate with ask_operator only when you cannot resolve it. While any specialist is running:
+You are the chat surface. Workers cannot ask_operator; they ask_director. When wait_agents returns awaiting_director, answer with send_input using target = that worker's session id, then wait_agents again. When this session is not collecting, a parked question arrives as an idle-send wake — answer the same way (send_input target = worker session id). Escalate with ask_operator only when you cannot resolve it. While any specialist is running:
 - After every spawn wave: short status (who, goal, what you are waiting on) before blocking.
 - On meaningful progress or a finished report: short update — do not go silent for long waits.
 - When the operator messages mid-run: answer them first (COMMUNICATION). Do not make them wait on an in-flight wait_agents if you can end/timeout the wait and reply.

src/subagent/fleet-report.ask-wake.test.ts

@@ -0,0 +1,83 @@
+import { describe, expect, test } from "bun:test";
+import { pendingAskSnapshot, pendingAskWakeText, type FleetLane } from "./fleet-report.js";
+
+function lane(overrides: Partial<FleetLane> & { id: string }): FleetLane {
+  return {
+    description: overrides.id,
+    status: "running",
+    startedAt: 0,
+    lastActivityAt: 0,
+    currentToolName: null,
+    currentToolPreview: null,
+    currentToolStartedAt: null,
+    ...overrides,
+  };
+}
+
+describe("pendingAskSnapshot", () => {
+  test("repeated calls return complete identical snapshots for distinct sessions sharing a catalog", () => {
+    const lanes = [
+      lane({ id: "a1", agentId: "builder", description: "Build the thing" }),
+      lane({ id: "a2", agentId: "builder" }),
+    ];
+    const peek = () => ({ question: "Which port?", questionId: "q1" });
+    const expected = lanes.map((worker) => ({
+      sessionId: worker.id,
+      agentId: "builder",
+      description: worker.description,
+      question: "Which port?",
+      questionId: "q1",
+    }));
+    expect(pendingAskSnapshot(lanes, peek)).toEqual(expected);
+    expect(pendingAskSnapshot(lanes, peek)).toEqual(expected);
+  });
+
+  test("resolution, removal and replacement are reflected without prior watch state", () => {
+    const lanes = [lane({ id: "a1" })];
+    const asks = new Map([["a1", { question: "A?", questionId: "q1" }]]);
+    const peek = (id: string) => asks.get(id);
+    expect(pendingAskSnapshot(lanes, peek)[0]?.questionId).toBe("q1");
+    expect(pendingAskSnapshot([], peek)).toEqual([]);
+    asks.clear();
+    expect(pendingAskSnapshot(lanes, peek)).toEqual([]);
+    asks.set("a1", { question: "B?", questionId: "q2" });
+    expect(pendingAskSnapshot(lanes, peek)[0]).toMatchObject({
+      sessionId: "a1",
+      question: "B?",
+      questionId: "q2",
+    });
+  });
+
+  test("only running root workers with a pending question are included", () => {
+    const lanes = [
+      lane({ id: "root" }),
+      lane({ id: "child", parentSessionId: "orchestrator" }),
+      lane({ id: "done", status: "done" }),
+      lane({ id: "cancelled", status: "cancelled" }),
+      lane({ id: "no-ask" }),
+    ];
+    const asks = pendingAskSnapshot(lanes, (id) =>
+      id === "no-ask" ? undefined : { question: "Q?", questionId: "q1" },
+    );
+    expect(asks.map((ask) => ask.sessionId)).toEqual(["root"]);
+  });
+});
+
+describe("pendingAskWakeText", () => {
+  test("names agent, description, question and question id, and routes to send_input", () => {
+    const text = pendingAskWakeText({
+      sessionId: "a1",
+      agentId: "builder",
+      description: "Build the thing",
+      question: "Which port?",
+      questionId: "q1",
+    });
+    expect(text).toContain("builder");
+    expect(text).toContain("Build the thing");
+    expect(text).toContain("Which port?");
+    expect(text).toContain("q1");
+    expect(text).toContain("send_input");
+    expect(text).toContain("using target a1");
+    expect(text.toLowerCase()).toContain("worker");
+  });
+});

src/subagent/fleet-report.ts

@@ -29,6 +29,10 @@ export interface FleetLane {
   readonly error?: string;
   /** Machine-readable forced-stop reason (see SubAgentSession.stopReason). */
   readonly stopReason?: string;
+  /** Catalog agent id (SubAgentSession.agentId); the wake message names it. */
+  readonly agentId?: string;
+  /** Set on nested (one-hop) dispatches; such asks never wake the root. */
+  readonly parentSessionId?: string;
 }
 
 interface LaneMark {
@@ -117,6 +121,56 @@ export function liveFleetCount(lanes: readonly FleetLane[]): number {
   return lanes.filter((lane) => lane.status === "running").length;
 }
 
+/**
+ * One parked ask_director question. Replies target the unique `sessionId`;
+ * `agentId` is only the descriptive catalog identity shared by workers.
+ */
+export interface PendingAskWake {
+  readonly sessionId: string;
+  readonly agentId: string;
+  readonly description: string;
+  readonly question: string;
+  readonly questionId: string;
+}
+
+/** Nested orchestrators own their children's questions; only root workers wake the TUI. */
+export function pendingAskSnapshot(
+  lanes: readonly FleetLane[],
+  peekAsk: (sessionId: string) => { question: string; questionId: string } | undefined,
+): readonly PendingAskWake[] {
+  const asks: PendingAskWake[] = [];
+  for (const lane of lanes) {
+    if (lane.parentSessionId !== undefined || lane.status !== "running") continue;
+    const ask = peekAsk(lane.id);
+    if (ask === undefined) continue;
+    asks.push({
+      sessionId: lane.id,
+      agentId: lane.agentId ?? lane.id,
+      description: lane.description,
+      question: ask.question,
+      questionId: ask.questionId,
+    });
+  }
+  return asks;
+}
+
+export const ASK_DIRECTOR_WAKE_PREFIX = "ask_director wake";
+
+/**
+ * The wake turn text. It must read as the worker's question reaching the
+ * parent, not as the operator being asked — the parent answers via
+ * send_input itself and only escalates when it genuinely cannot.
+ */
+export function pendingAskWakeText(wake: PendingAskWake): string {
+  return [
+    `${ASK_DIRECTOR_WAKE_PREFIX} — worker ${wake.agentId} (${wake.description}) parked question ${wake.questionId} while this session was not collecting:`,
+    "",
+    wake.question,
+    "",
+    `The worker — not the operator — raised this. Answer it with send_input (soft) using target ${wake.sessionId}; do not relay to the operator unless it genuinely needs them.`,
+  ].join("\n");
+}
+
 type Change =
   | { readonly kind: "dispatched"; readonly line: string }
   | { readonly kind: "done"; readonly line: string }

src/subagent/index.ts

@@ -18,9 +18,12 @@ export {
   FLEET_STALL_POLL_MS,
   liveFleetCount,
   observeFleet,
+  pendingAskSnapshot,
+  pendingAskWakeText,
   type FleetLane,
   type FleetObservation,
   type FleetWatch,
+  type PendingAskWake,
 } from "./fleet-report.js";
 export {
   EMPTY_THRASH_STATE,