From 09b28223f98ed6b9d58a44f4eb84ba51245cc67c Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 8 Sep 2026 10:20:41 -0700 Subject: [PATCH 1/6] Wake parents for parked director questions --- docs/ARCHITECTURE.md | 2 +- docs/TUI.md | 3 + src/subagent/fleet-report.ask-wake.test.ts | 118 ++++++++++++++ src/subagent/fleet-report.ts | 79 +++++++++ src/subagent/index.ts | 5 + src/tui/agent-ask-wake.test.ts | 177 +++++++++++++++++++++ src/tui/runner/wiring.ts | 14 +- src/tui/runtime-bridge.ts | 56 ++++++- src/tui/stream-event-map.ts | 6 + 9 files changed, 453 insertions(+), 7 deletions(-) create mode 100644 src/subagent/fleet-report.ask-wake.test.ts create mode 100644 src/tui/agent-ask-wake.test.ts diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index cc52f41cf..34a8b9a3d 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -209,7 +209,7 @@ Three distinct concepts (do not conflate them): The **`spawn_agent`** tool starts a fleet agent on a separate inference source (tier/profile resolved from settings) and returns immediately with an `agent_id`; **`wait_agents`** collects reports later. Declared fan-out is unlimited: excess dispatches enqueue rather than fail. `run()` is admitted by `src/subagent/admission.ts` (default burst window of 8 is race-avoidance so a 429 freeze can fire before a herd — not a declared-spawn cap). Occupancy is the whole first `run()`, including `wait_agents`. Nested children of an already-admitted parent bypass **capacity** so a nested orchestrator cannot deadlock while holding a slot; they still wait on a provider 429 pause. Drain is FIFO among currently admissible jobs (a paused provider is skipped, not head-of-line for every provider). Resume and followup inference re-enter the same queue. Queued workers report wait/list status `queued` (live, not failed). Lowering capacity never cancels in-flight work. Retryable provider 429s freeze new admits via the shared retry remapper in `createCorbitsRetryPolicy`; `quota_exhausted` does not freeze. `list_agents` remains mailbox-scoped. The dispatch brief separates durable `context`, actionable `prompt`, and optional `goals` (checklist seeds for the _child's_ own `manage_tasks` list). Implement/review dispatches (and their default directors) fail closed without non-empty `success_criteria`. The child returns a structured report (`Summary` / `Findings` / `Blockers` / `Paths`) plus a tools-used footer. Parent and child never share a `manage_tasks` list. -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. +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. When the parent TUI is not blocked in `wait_agents`, the runner watches the store for newly parked top-level questions and injects one coalesced wake turn at the next idle settle or gate close, so the parent still answers instead of hanging. Escalate to the human with **`ask_operator`** only when the parent cannot resolve it. 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. diff --git a/docs/TUI.md b/docs/TUI.md index 8b2e13aa6..cd87625b4 100644 --- a/docs/TUI.md +++ b/docs/TUI.md @@ -221,6 +221,9 @@ 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. +When a parked question lands while the parent is idle, the runner injects one +coalesced wake turn so the parent answers via `send_input` — the strip itself +never re-delivers it. `formatChromeZones` → `formatAgentsPanel` owns that paint. Geometry stays stack-only (`layoutMode: "stack"`, `railWidth: 0`); the zone max is diff --git a/src/subagent/fleet-report.ask-wake.test.ts b/src/subagent/fleet-report.ask-wake.test.ts new file mode 100644 index 000000000..13c118b5b --- /dev/null +++ b/src/subagent/fleet-report.ask-wake.test.ts @@ -0,0 +1,118 @@ +/** + * Parent-wake for workers parked in ask_director: the pure emitter-side + * diff. Bridge delivery (stash / coalesce / flush-once) lives in + * tui/agent-ask-wake.test.ts. + */ +import { describe, expect, test } from "bun:test"; +import { + createPendingAskWatch, + observePendingAsks, + pendingAskWakeText, + type FleetLane, +} from "./fleet-report.js"; + +function lane(overrides: Partial & { id: string }): FleetLane { + return { + description: overrides.id, + status: "running", + startedAt: 0, + lastActivityAt: 0, + currentToolName: null, + currentToolPreview: null, + currentToolStartedAt: null, + ...overrides, + }; +} + +interface FakeAsk { + readonly question: string; + readonly questionId: string; +} + +function peekAskFrom(asks: ReadonlyMap) { + return (id: string): FakeAsk | undefined => asks.get(id); +} + +describe("observePendingAsks", () => { + test("a parked top-level ask wakes once, naming the question", () => { + const asks = new Map([["a1", { question: "Which port?", questionId: "q1" }]] as const); + const first = observePendingAsks( + createPendingAskWatch(), + [lane({ id: "a1", agentId: "builder", description: "Build the thing" })], + peekAskFrom(asks), + ); + expect(first.wakes).toHaveLength(1); + expect(first.wakes[0]).toMatchObject({ + sessionId: "a1", + agentId: "builder", + description: "Build the thing", + question: "Which port?", + questionId: "q1", + }); + + // Store notifies again for the same question: no extra wake. + const repeat = observePendingAsks( + first.watch, + [lane({ id: "a1", agentId: "builder", description: "Build the thing" })], + peekAskFrom(asks), + ); + expect(repeat.wakes).toEqual([]); + }); + + test("a resolved ask drops from the watch; a re-ask with a new questionId wakes again", () => { + const asks = new Map([["a1", { question: "A?", questionId: "q1" }]] as const); + const first = observePendingAsks( + createPendingAskWatch(), + [lane({ id: "a1" })], + peekAskFrom(asks), + ); + expect(first.wakes).toHaveLength(1); + + const resolved = observePendingAsks(first.watch, [lane({ id: "a1" })], peekAskFrom(new Map())); + expect(resolved.wakes).toEqual([]); + + const reask = new Map([["a1", { question: "B?", questionId: "q2" }]] as const); + const again = observePendingAsks(resolved.watch, [lane({ id: "a1" })], peekAskFrom(reask)); + expect(again.wakes).toHaveLength(1); + expect(again.wakes[0]?.questionId).toBe("q2"); + }); + + test("nested-orchestrator asks never wake the root", () => { + const asks = new Map([["child", { question: "Q?", questionId: "q1" }]] as const); + const { wakes } = observePendingAsks( + createPendingAskWatch(), + [lane({ id: "child", parentSessionId: "orchestrator" })], + peekAskFrom(asks), + ); + expect(wakes).toEqual([]); + }); + + test("a lane that is not running does not wake", () => { + const asks = new Map([["a1", { question: "Q?", questionId: "q1" }]] as const); + const { wakes } = observePendingAsks( + createPendingAskWatch(), + [lane({ id: "a1", status: "done" })], + peekAskFrom(asks), + ); + expect(wakes).toEqual([]); + }); +}); + +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"); + // The worker raised it; the parent must not present it as operator-asked. + expect(text.toLowerCase()).toContain("worker"); + }); +}); diff --git a/src/subagent/fleet-report.ts b/src/subagent/fleet-report.ts index 50b60816c..cda62acd7 100644 --- a/src/subagent/fleet-report.ts +++ b/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,81 @@ export function liveFleetCount(lanes: readonly FleetLane[]): number { return lanes.filter((lane) => lane.status === "running").length; } +/** + * One parked ask_director question worth waking the parent for. `sessionId` + * and `agentId` differ only in namespace; the wake message and send_input + * both speak `agentId`. + */ +export interface PendingAskWake { + readonly sessionId: string; + readonly agentId: string; + readonly description: string; + readonly question: string; + readonly questionId: string; +} + +/** QuestionIds already woken, per lane. The caller keeps and hands it back. */ +export interface PendingAskWatch { + readonly questionIds: ReadonlyMap; +} + +export function createPendingAskWatch(): PendingAskWatch { + return { questionIds: new Map() }; +} + +/** + * Which parked asks are new since the last observation. Only top-level + * workers (no `parentSessionId`) wake the root — a nested orchestrator owns + * its own children's questions. A resolved ask drops from the watch, so a + * re-ask with a fresh questionId wakes again while repeat notifications for + * the same questionId stay silent. + */ +export function observePendingAsks( + previous: PendingAskWatch, + lanes: readonly FleetLane[], + peekAsk: (sessionId: string) => { question: string; questionId: string } | undefined, +): { watch: PendingAskWatch; wakes: readonly PendingAskWake[] } { + const questionIds = new Map(previous.questionIds); + const wakes: PendingAskWake[] = []; + for (const lane of lanes) { + if (lane.parentSessionId !== undefined) continue; + if (lane.status !== "running") { + questionIds.delete(lane.id); + continue; + } + const ask = peekAsk(lane.id); + if (ask === undefined) { + questionIds.delete(lane.id); + continue; + } + if (questionIds.get(lane.id) === ask.questionId) continue; + questionIds.set(lane.id, ask.questionId); + wakes.push({ + sessionId: lane.id, + agentId: lane.agentId ?? lane.id, + description: lane.description, + question: ask.question, + questionId: ask.questionId, + }); + } + return { watch: { questionIds }, wakes }; +} + +/** + * 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 — 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) targeting agent_id ${wake.agentId}; 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 } diff --git a/src/subagent/index.ts b/src/subagent/index.ts index bcdc1b861..cae686fc6 100644 --- a/src/subagent/index.ts +++ b/src/subagent/index.ts @@ -13,14 +13,19 @@ export type { export { createSubAgentSessionStore } from "./session-store.js"; export { createFleetWatch, + createPendingAskWatch, fleetDigest, FLEET_REPORT_SETTLE_MS, FLEET_STALL_POLL_MS, liveFleetCount, observeFleet, + observePendingAsks, + pendingAskWakeText, type FleetLane, type FleetObservation, type FleetWatch, + type PendingAskWake, + type PendingAskWatch, } from "./fleet-report.js"; export { EMPTY_THRASH_STATE, diff --git a/src/tui/agent-ask-wake.test.ts b/src/tui/agent-ask-wake.test.ts new file mode 100644 index 000000000..454f43077 --- /dev/null +++ b/src/tui/agent-ask-wake.test.ts @@ -0,0 +1,177 @@ +/** + * Bridge-side delivery of ask_director parent-wakes: stash while the parent + * turn is live or gated, coalesce, flush exactly once at settle or gate + * close, and drop the stash on session rotation. The pure emitter-side diff + * (which asks deserve a wake) lives in subagent/fleet-report.ask-wake.test.ts. + */ +import { describe, expect, test } from "bun:test"; +import { attachSessionBridge } from "./runtime-bridge"; +import { createLiveSessionPort } from "./live-session-port"; +import { createAppShell } from "./shell/index"; +import { withTestRenderer } from "./harness"; +import type { PendingAskWake } from "../subagent/fleet-report.js"; + +function wake(id: string, questionId: string): PendingAskWake { + return { + sessionId: id, + agentId: id, + description: `worker ${id}`, + question: "Which port?", + questionId, + }; +} + +function capturePort() { + const sends: string[] = []; + const port = createLiveSessionPort({ + send: (text) => { + sends.push(text); + }, + interrupt: () => {}, + deliver: () => {}, + }); + return { port, sends }; +} + +describe("agent ask wake delivery", () => { + test("a parked ask wakes an idle parent exactly once, repeats dedup", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }); + const { port, sends } = capturePort(); + const bridge = attachSessionBridge(shell, port); + try { + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toHaveLength(1); + expect(sends[0]).toContain("a1"); + expect(sends[0]).toContain("Which port?"); + expect(sends[0]).toContain("q1"); + expect(sends[0]).toContain("send_input"); + + // Repeat store notification for the same question: no second send. + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toHaveLength(1); + + // A resolved ask emits no wake event; an empty list must not send. + bridge.handle({ type: "agent-ask", asks: [] }); + expect(sends).toHaveLength(1); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("an ask parking mid-cycle defers to settle, then flushes once", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }); + const { port, sends } = capturePort(); + const bridge = attachSessionBridge(shell, port); + try { + bridge.handle({ type: "run", state: "busy" }); + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toEqual([]); + + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toHaveLength(1); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("multiple parked asks coalesce into one wake message", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }); + const { port, sends } = capturePort(); + const bridge = attachSessionBridge(shell, port); + try { + bridge.handle({ type: "run", state: "busy" }); + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + bridge.handle({ type: "agent-ask", asks: [wake("a2", "q2")] }); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toHaveLength(1); + expect(sends[0]).toContain("a1"); + expect(sends[0]).toContain("a2"); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("an open gate defers the wake; gate close flushes once", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }); + const { port, sends } = capturePort(); + const bridge = attachSessionBridge(shell, port); + try { + bridge.gateOpened(); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toEqual([]); + + bridge.gateClosed(); + expect(sends).toHaveLength(1); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("clearQueuedDelivery drops stashed wakes", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }); + const { port, sends } = capturePort(); + const bridge = attachSessionBridge(shell, port); + try { + bridge.handle({ type: "run", state: "busy" }); + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + bridge.clearQueuedDelivery(); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toEqual([]); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); +}); diff --git a/src/tui/runner/wiring.ts b/src/tui/runner/wiring.ts index 687b45cfc..391c62319 100644 --- a/src/tui/runner/wiring.ts +++ b/src/tui/runner/wiring.ts @@ -17,10 +17,12 @@ import { loadSentMessages } from "../../session/sent-messages.js"; import { setActiveDisposeHost } from "../../session/active-host.js"; import { createFleetWatch, + createPendingAskWatch, FLEET_REPORT_SETTLE_MS, FLEET_STALL_POLL_MS, liveFleetCount, observeFleet, + observePendingAsks, } from "../../subagent/index.js"; import { scheduleUpgradeNotice } from "../../upgrade/index.js"; import pkg from "../../../package.json" with { type: "json" }; @@ -130,12 +132,22 @@ export function wirePostStartup( // terminalizes. Store notifications fire per child event, not per status // flip, so emit only when the count itself moves. let lastLiveFleet = 0; + // Parked ask_director questions ride the same store subscription. The + // emitter-side watch dedups on questionId transitions, so only a newly + // parked question reaches the bridge; delivery timing is the bridge's. + let askWatch = createPendingAskWatch(); const unsubscribeFleetReport = services.subAgentSessions.subscribe(() => { - const fleet = liveFleetCount(services.subAgentSessions.list()); + const lanes = services.subAgentSessions.list(); + const fleet = liveFleetCount(lanes); if (fleet !== lastLiveFleet) { lastLiveFleet = fleet; services.emitter.emit("event", { type: "fleet", running: fleet }); } + const asks = observePendingAsks(askWatch, lanes, (id) => services.subAgentSessions.peekAsk(id)); + askWatch = asks.watch; + if (asks.wakes.length > 0) { + services.emitter.emit("event", { type: "agent-ask", asks: asks.wakes }); + } if (fleetSettle !== null) return; fleetSettle = setTimeout(() => { fleetSettle = null; diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index c105ca6dd..82352613b 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -67,6 +67,7 @@ import { fleetProgress, type AgentProgressSession, } from "./agent-progress.js"; +import { pendingAskWakeText, type PendingAskWake } from "../subagent/fleet-report.js"; /** Tool name a sub-agent dispatch call carries — its row gets live progress. */ const SPAWN_AGENT_TOOL_NAME = "spawn_agent"; @@ -268,6 +269,7 @@ function isBridgeInbound(event: { type: string }): event is BridgeInboundEvent { case "system": case "run": case "fleet": + case "agent-ask": case "tool.boundary": case "error": return true; @@ -353,6 +355,18 @@ export interface BridgeBag { * session-idle — and the hold releases when the count lands back at zero. */ liveFleet: number; + /** + * Worker asks parked in ask_director, keyed by session, waiting for a + * moment the parent can act on them (idle settle or last gate closing). + * Keying by session is what stops a repeat emitter notification from + * stashing the same question twice. + */ + pendingAskWake: Map; + /** + * Set once `submit` exists inside `attachSessionBridge`; the module-scope + * settle path (`settleRunToIdle`) and `gateClosed` re-enter through it. + */ + flushPendingAskWake: (() => void) | null; /** Last prompt actually sent — replay source for the quota auto-retry. */ lastSentMessage: string; /** One auto-retry per rate-limit window. */ @@ -900,11 +914,13 @@ function settleRunToIdle(shell: AppShell, bag: BridgeBag): void { // pending send now — the parent they were steering has stopped, so // each one starts its own turn — while follow-ups keep waiting. drainSteersAtBoundary(shell, bag); + bag.flushPendingAskWake?.(); return; } shell.session = setRunState(shell.session, "idle"); // Full drain: soft steers first, then follow-ups (drainOrder). drainAtBoundary(shell, bag); + bag.flushPendingAskWake?.(); } function applyInbound(shell: AppShell, bag: BridgeBag, event: BridgeInboundEvent): void { @@ -913,16 +929,24 @@ function applyInbound(shell: AppShell, bag: BridgeBag, event: BridgeInboundEvent // Fleet liveness owns no transcript row state, so it is handled before the // open-row machinery — a lane terminalizing mid-parent-stream must not // close the assistant row the parent's own deltas are growing. - if (event.type === "fleet") { + if (event.type === "fleet" || event.type === "agent-ask") { // Idle-with-fleet bookkeeping. A transition to zero while the parent is // already idle releases the hold: that moment is true session-idle, so // queued follow-ups drain now. While the parent is still working the // count just updates — the ordinary turn settle does the draining. - bag.liveFleet = event.running; - if (event.running === 0 && !bag.turn.isProcessing) { - settleRunToIdle(shell, bag); + if (event.type === "fleet") { + bag.liveFleet = event.running; + if (event.running === 0 && !bag.turn.isProcessing) { + settleRunToIdle(shell, bag); + } + paintChrome(shell); + return; } - paintChrome(shell); + // Parked ask_director questions: stash keyed by session (repeat store + // notifications overwrite rather than duplicate), then deliver now when + // the parent is free — otherwise they wait for settle or gate close. + for (const ask of event.asks) bag.pendingAskWake.set(ask.sessionId, ask); + bag.flushPendingAskWake?.(); return; } @@ -1011,6 +1035,8 @@ export function attachSessionBridge( disposed: false, turn: initialTurnState(now()), liveFleet: 0, + pendingAskWake: new Map(), + flushPendingAskWake: null, lastSentMessage: "", quotaFired: false, now, @@ -1306,6 +1332,21 @@ export function attachSessionBridge( paintChrome(shell); }; + /** + * Deliver stashed ask wakes as one turn, but only when the parent can act: + * never mid-cycle, and never while an operator gate holds the run. The + * gate-closed path flushes explicitly — a gate's `isProcessing` lingers by + * design, so the ordinary idle condition would never fire there. + */ + const flushPendingAskWake = (force = false): void => { + if (bag.pendingAskWake.size === 0) return; + if (!force && (bag.turn.isProcessing || bag.turn.blockedGateCount > 0)) return; + const asks = [...bag.pendingAskWake.values()]; + bag.pendingAskWake.clear(); + submit(asks.map((ask) => pendingAskWakeText(ask)).join("\n\n"), "immediate"); + }; + bag.flushPendingAskWake = () => flushPendingAskWake(); + const doInterrupt = (): void => { if (bag.disposed) return; closeOpenRow(shell, bag); @@ -1333,6 +1374,7 @@ export function attachSessionBridge( shell.session = createSessionQueue("idle"); bag.pendingEchoes.length = 0; bag.liveFleet = 0; + bag.pendingAskWake.clear(); bag.pendingRowUpdates.clear(); paintChrome(shell); }; @@ -1353,6 +1395,10 @@ export function attachSessionBridge( if (bag.disposed) return; bag.turn = turnStateGateClosed(bag.turn, now()); paintPhase(); + // Flush a wake stashed while the gate held the run — but only at true + // session-idle. A gate closing over a still-live parent cycle must not + // inject mid-cycle; that cycle's own settle flushes the stash instead. + flushPendingAskWake(bag.turn.blockedGateCount === 0 && shell.session.run === "idle"); }; const tick = (): void => { diff --git a/src/tui/stream-event-map.ts b/src/tui/stream-event-map.ts index 965cdaae5..e10b79ed3 100644 --- a/src/tui/stream-event-map.ts +++ b/src/tui/stream-event-map.ts @@ -16,6 +16,7 @@ import { } from "../inference-gateway-error.js"; import { isProviderFailurePresentationSuppressed } from "./provider/failure-attempt.js"; import type { RunState } from "./session-queue.js"; +import type { PendingAskWake } from "../subagent/fleet-report.js"; /** Canonical inbound events the bridge understands (fixtures + mapped reactor). */ export type BridgeInboundEvent = @@ -48,6 +49,11 @@ export type BridgeInboundEvent = * new turn rather than a queued steer. */ | { readonly type: "fleet"; readonly running: number } + /** + * Workers newly parked in ask_director (transition-only, emitter-side + * deduped). The bridge stashes and delivers them when the parent can act. + */ + | { readonly type: "agent-ask"; readonly asks: readonly PendingAskWake[] } | { readonly type: "tool.boundary" } | { readonly type: "error"; readonly message: string } /** From a2575cf9a4857cbfde2c5d056bd3e8141ed46c27 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 8 Sep 2026 14:32:26 -0700 Subject: [PATCH 2/6] Fix parent wake delivery across worker lifecycle transitions --- docs/ARCHITECTURE.md | 2 +- docs/TUI.md | 11 +- src/subagent/fleet-report.ask-wake.test.ts | 115 ++---- src/subagent/fleet-report.ts | 49 +-- src/subagent/index.ts | 4 +- src/tui/agent-ask-wake.test.ts | 426 +++++++++++++++------ src/tui/runner/exit.ts | 27 +- src/tui/runner/state.ts | 1 + src/tui/runner/wiring.ask-wake.test.ts | 323 ++++++++++++++++ src/tui/runner/wiring.ts | 59 +-- src/tui/runtime-bridge.ts | 79 ++-- src/tui/stream-event-map.ts | 4 +- src/tui/turn-state.test.ts | 10 + src/tui/turn-state.ts | 6 +- 14 files changed, 822 insertions(+), 294 deletions(-) create mode 100644 src/tui/runner/wiring.ask-wake.test.ts diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 34a8b9a3d..3a72f1a85 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -209,7 +209,7 @@ Three distinct concepts (do not conflate them): The **`spawn_agent`** tool starts a fleet agent on a separate inference source (tier/profile resolved from settings) and returns immediately with an `agent_id`; **`wait_agents`** collects reports later. Declared fan-out is unlimited: excess dispatches enqueue rather than fail. `run()` is admitted by `src/subagent/admission.ts` (default burst window of 8 is race-avoidance so a 429 freeze can fire before a herd — not a declared-spawn cap). Occupancy is the whole first `run()`, including `wait_agents`. Nested children of an already-admitted parent bypass **capacity** so a nested orchestrator cannot deadlock while holding a slot; they still wait on a provider 429 pause. Drain is FIFO among currently admissible jobs (a paused provider is skipped, not head-of-line for every provider). Resume and followup inference re-enter the same queue. Queued workers report wait/list status `queued` (live, not failed). Lowering capacity never cancels in-flight work. Retryable provider 429s freeze new admits via the shared retry remapper in `createCorbitsRetryPolicy`; `quota_exhausted` does not freeze. `list_agents` remains mailbox-scoped. The dispatch brief separates durable `context`, actionable `prompt`, and optional `goals` (checklist seeds for the _child's_ own `manage_tasks` list). Implement/review dispatches (and their default directors) fail closed without non-empty `success_criteria`. The child returns a structured report (`Summary` / `Findings` / `Blockers` / `Paths`) plus a tools-used footer. Parent and child never share a `manage_tasks` list. -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. When the parent TUI is not blocked in `wait_agents`, the runner watches the store for newly parked top-level questions and injects one coalesced wake turn at the next idle settle or gate close, so the parent still answers instead of hanging. Escalate to the human with **`ask_operator`** only when the parent cannot resolve it. +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. The TUI 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 target the worker session ID through `send_input`, 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. Escalate to the human with **`ask_operator`** only when the parent cannot resolve it. 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. diff --git a/docs/TUI.md b/docs/TUI.md index cd87625b4..78cb101b0 100644 --- a/docs/TUI.md +++ b/docs/TUI.md @@ -221,9 +221,14 @@ 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. -When a parked question lands while the parent is idle, the runner injects one -coalesced wake turn so the parent answers via `send_input` — the strip itself -never re-delivers it. +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` with the +worker's session ID, not its shared catalog ID. Each session/question identity is +delivered once while pending; the 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 diff --git a/src/subagent/fleet-report.ask-wake.test.ts b/src/subagent/fleet-report.ask-wake.test.ts index 13c118b5b..9a3ddfb8f 100644 --- a/src/subagent/fleet-report.ask-wake.test.ts +++ b/src/subagent/fleet-report.ask-wake.test.ts @@ -1,15 +1,5 @@ -/** - * Parent-wake for workers parked in ask_director: the pure emitter-side - * diff. Bridge delivery (stash / coalesce / flush-once) lives in - * tui/agent-ask-wake.test.ts. - */ import { describe, expect, test } from "bun:test"; -import { - createPendingAskWatch, - observePendingAsks, - pendingAskWakeText, - type FleetLane, -} from "./fleet-report.js"; +import { pendingAskSnapshot, pendingAskWakeText, type FleetLane } from "./fleet-report.js"; function lane(overrides: Partial & { id: string }): FleetLane { return { @@ -24,77 +14,52 @@ function lane(overrides: Partial & { id: string }): FleetLane { }; } -interface FakeAsk { - readonly question: string; - readonly questionId: string; -} - -function peekAskFrom(asks: ReadonlyMap) { - return (id: string): FakeAsk | undefined => asks.get(id); -} - -describe("observePendingAsks", () => { - test("a parked top-level ask wakes once, naming the question", () => { - const asks = new Map([["a1", { question: "Which port?", questionId: "q1" }]] as const); - const first = observePendingAsks( - createPendingAskWatch(), - [lane({ id: "a1", agentId: "builder", description: "Build the thing" })], - peekAskFrom(asks), - ); - expect(first.wakes).toHaveLength(1); - expect(first.wakes[0]).toMatchObject({ - sessionId: "a1", +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: "Build the thing", + description: worker.description, question: "Which port?", questionId: "q1", - }); - - // Store notifies again for the same question: no extra wake. - const repeat = observePendingAsks( - first.watch, - [lane({ id: "a1", agentId: "builder", description: "Build the thing" })], - peekAskFrom(asks), - ); - expect(repeat.wakes).toEqual([]); + })); + expect(pendingAskSnapshot(lanes, peek)).toEqual(expected); + expect(pendingAskSnapshot(lanes, peek)).toEqual(expected); }); - test("a resolved ask drops from the watch; a re-ask with a new questionId wakes again", () => { - const asks = new Map([["a1", { question: "A?", questionId: "q1" }]] as const); - const first = observePendingAsks( - createPendingAskWatch(), - [lane({ id: "a1" })], - peekAskFrom(asks), - ); - expect(first.wakes).toHaveLength(1); - - const resolved = observePendingAsks(first.watch, [lane({ id: "a1" })], peekAskFrom(new Map())); - expect(resolved.wakes).toEqual([]); - - const reask = new Map([["a1", { question: "B?", questionId: "q2" }]] as const); - const again = observePendingAsks(resolved.watch, [lane({ id: "a1" })], peekAskFrom(reask)); - expect(again.wakes).toHaveLength(1); - expect(again.wakes[0]?.questionId).toBe("q2"); - }); - - test("nested-orchestrator asks never wake the root", () => { - const asks = new Map([["child", { question: "Q?", questionId: "q1" }]] as const); - const { wakes } = observePendingAsks( - createPendingAskWatch(), - [lane({ id: "child", parentSessionId: "orchestrator" })], - peekAskFrom(asks), - ); - expect(wakes).toEqual([]); + 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("a lane that is not running does not wake", () => { - const asks = new Map([["a1", { question: "Q?", questionId: "q1" }]] as const); - const { wakes } = observePendingAsks( - createPendingAskWatch(), - [lane({ id: "a1", status: "done" })], - peekAskFrom(asks), + 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(wakes).toEqual([]); + expect(asks.map((ask) => ask.sessionId)).toEqual(["root"]); }); }); @@ -112,7 +77,7 @@ describe("pendingAskWakeText", () => { expect(text).toContain("Which port?"); expect(text).toContain("q1"); expect(text).toContain("send_input"); - // The worker raised it; the parent must not present it as operator-asked. + expect(text).toContain("targeting agent_id a1"); expect(text.toLowerCase()).toContain("worker"); }); }); diff --git a/src/subagent/fleet-report.ts b/src/subagent/fleet-report.ts index cda62acd7..234369b1e 100644 --- a/src/subagent/fleet-report.ts +++ b/src/subagent/fleet-report.ts @@ -122,9 +122,8 @@ export function liveFleetCount(lanes: readonly FleetLane[]): number { } /** - * One parked ask_director question worth waking the parent for. `sessionId` - * and `agentId` differ only in namespace; the wake message and send_input - * both speak `agentId`. + * 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; @@ -134,43 +133,17 @@ export interface PendingAskWake { readonly questionId: string; } -/** QuestionIds already woken, per lane. The caller keeps and hands it back. */ -export interface PendingAskWatch { - readonly questionIds: ReadonlyMap; -} - -export function createPendingAskWatch(): PendingAskWatch { - return { questionIds: new Map() }; -} - -/** - * Which parked asks are new since the last observation. Only top-level - * workers (no `parentSessionId`) wake the root — a nested orchestrator owns - * its own children's questions. A resolved ask drops from the watch, so a - * re-ask with a fresh questionId wakes again while repeat notifications for - * the same questionId stay silent. - */ -export function observePendingAsks( - previous: PendingAskWatch, +/** 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, -): { watch: PendingAskWatch; wakes: readonly PendingAskWake[] } { - const questionIds = new Map(previous.questionIds); - const wakes: PendingAskWake[] = []; +): readonly PendingAskWake[] { + const asks: PendingAskWake[] = []; for (const lane of lanes) { - if (lane.parentSessionId !== undefined) continue; - if (lane.status !== "running") { - questionIds.delete(lane.id); - continue; - } + if (lane.parentSessionId !== undefined || lane.status !== "running") continue; const ask = peekAsk(lane.id); - if (ask === undefined) { - questionIds.delete(lane.id); - continue; - } - if (questionIds.get(lane.id) === ask.questionId) continue; - questionIds.set(lane.id, ask.questionId); - wakes.push({ + if (ask === undefined) continue; + asks.push({ sessionId: lane.id, agentId: lane.agentId ?? lane.id, description: lane.description, @@ -178,7 +151,7 @@ export function observePendingAsks( questionId: ask.questionId, }); } - return { watch: { questionIds }, wakes }; + return asks; } /** @@ -192,7 +165,7 @@ export function pendingAskWakeText(wake: PendingAskWake): string { "", wake.question, "", - `The worker — not the operator — raised this. Answer it with send_input (soft) targeting agent_id ${wake.agentId}; do not relay to the operator unless it genuinely needs them.`, + `The worker — not the operator — raised this. Answer it with send_input (soft) targeting agent_id ${wake.sessionId}; do not relay to the operator unless it genuinely needs them.`, ].join("\n"); } diff --git a/src/subagent/index.ts b/src/subagent/index.ts index cae686fc6..22393f125 100644 --- a/src/subagent/index.ts +++ b/src/subagent/index.ts @@ -13,19 +13,17 @@ export type { export { createSubAgentSessionStore } from "./session-store.js"; export { createFleetWatch, - createPendingAskWatch, fleetDigest, FLEET_REPORT_SETTLE_MS, FLEET_STALL_POLL_MS, liveFleetCount, observeFleet, - observePendingAsks, + pendingAskSnapshot, pendingAskWakeText, type FleetLane, type FleetObservation, type FleetWatch, type PendingAskWake, - type PendingAskWatch, } from "./fleet-report.js"; export { EMPTY_THRASH_STATE, diff --git a/src/tui/agent-ask-wake.test.ts b/src/tui/agent-ask-wake.test.ts index 454f43077..a7fcac888 100644 --- a/src/tui/agent-ask-wake.test.ts +++ b/src/tui/agent-ask-wake.test.ts @@ -1,40 +1,59 @@ -/** - * Bridge-side delivery of ask_director parent-wakes: stash while the parent - * turn is live or gated, coalesce, flush exactly once at settle or gate - * close, and drop the stash on session rotation. The pure emitter-side diff - * (which asks deserve a wake) lives in subagent/fleet-report.ask-wake.test.ts. - */ import { describe, expect, test } from "bun:test"; import { attachSessionBridge } from "./runtime-bridge"; import { createLiveSessionPort } from "./live-session-port"; import { createAppShell } from "./shell/index"; import { withTestRenderer } from "./harness"; import type { PendingAskWake } from "../subagent/fleet-report.js"; +import { classifySubmission, createSubmitHandler } from "./runner/submit.js"; +import { routeQueuedDelivery } from "./queued-delivery.js"; +import { + armFeedbackCapture, + cancelFeedbackCapture, + isFeedbackCapturePending, + resetFeedbackStateForTests, +} from "../telemetry/feedback.js"; function wake(id: string, questionId: string): PendingAskWake { return { sessionId: id, - agentId: id, + agentId: "builder", description: `worker ${id}`, question: "Which port?", questionId, }; } -function capturePort() { - const sends: string[] = []; - const port = createLiveSessionPort({ - send: (text) => { - sends.push(text); +async function withWakeBridge( + run: (bridge: ReturnType, sends: string[]) => void, +) { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }); + const sends: string[] = []; + const send = (text: string) => { + sends.push(text); + }; + const bridge = attachSessionBridge( + shell, + createLiveSessionPort({ send, deliver: send, interrupt: () => {} }), + ); + try { + run(bridge, sends); + } finally { + bridge.dispose(); + shell.dispose(); + } }, - interrupt: () => {}, - deliver: () => {}, - }); - return { port, sends }; + { width: 80, height: 24 }, + ); } -describe("agent ask wake delivery", () => { - test("a parked ask wakes an idle parent exactly once, repeats dedup", async () => { +for (const action of ["retry", "interrupt", "reset", "dispose", "composer", "ordinary"] as const) { + test(`quota replay preserves submission origin (${action})`, async () => { await withTestRenderer( async (h) => { const shell = createAppShell(h.renderer, { @@ -42,24 +61,108 @@ describe("agent ask wake delivery", () => { wireKeys: false, run: "idle", }); - const { port, sends } = capturePort(); - const bridge = attachSessionBridge(shell, port); + const sends: string[] = []; + const composerSends: string[] = []; + const feedback: string[] = []; + let cancellations = 0; + let nowMs = 0; + let tick = () => {}; + const submit = createSubmitHandler({ + dispatchCommand: () => {}, + sendPrompt: (text) => { + composerSends.push(text); + sends.push(text); + }, + isFeedbackCapturePending, + onFeedbackText: (text) => { + feedback.push(text); + cancelFeedbackCapture(); + return "Thanks"; + }, + cancelFeedbackCapture: () => { + cancellations++; + cancelFeedbackCapture(); + }, + }); + const port = createLiveSessionPort({ + send: submit, + classifySubmit: (text) => + classifySubmission(text, { + feedbackPending: isFeedbackCapturePending(), + feedbackCaptureEnabled: true, + }), + interrupt: () => {}, + deliver: routeQueuedDelivery({ + send: (text) => { + sends.push(text); + }, + deliverSteer: () => { + throw new Error("quota replay must not live-inject"); + }, + parentCycleLive: () => bridge.parentCycleLive, + }), + }); + const bridge = attachSessionBridge(shell, port, { + now: () => nowMs, + schedule: (fn) => { + tick = fn; + // Retain the callback to exercise even a stale timer after disposal. + return () => {}; + }, + }); try { - bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); - expect(sends).toHaveLength(1); - expect(sends[0]).toContain("a1"); - expect(sends[0]).toContain("Which port?"); - expect(sends[0]).toContain("q1"); - expect(sends[0]).toContain("send_input"); - - // Repeat store notification for the same question: no second send. - bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); - expect(sends).toHaveLength(1); - - // A resolved ask emits no wake event; an empty list must not send. - bridge.handle({ type: "agent-ask", asks: [] }); - expect(sends).toHaveLength(1); + if (action === "ordinary") { + bridge.submit("operator prompt", "immediate"); + } else { + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "fleet", running: 1 }); + bridge.submit("held first", "queue"); + bridge.submit("held second", "queue"); + bridge.handle({ type: "inference.done", data: {} }); + if (action !== "composer") armFeedbackCapture(); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + if (action === "composer") bridge.submit("operator prompt", "immediate"); + } + const queued = shell.session.items; + const before = sends.length; + const replay = sends.at(-1); + bridge.handle({ + type: "inference.error", + data: { error: { category: "quota_exhausted", retryAfterMs: 1000 } }, + }); + if (action === "interrupt") bridge.interrupt(); + if (action === "reset") bridge.clearQueuedDelivery(); + if (action === "dispose") bridge.dispose(); + const afterCleanup = sends.length; + nowMs = 999; + tick(); + expect(sends).toHaveLength(afterCleanup); + nowMs = 1000; + tick(); + tick(); + if (action === "interrupt" || action === "reset" || action === "dispose") { + expect(sends).toHaveLength(afterCleanup); + expect(feedback).toEqual([]); + } else { + expect(sends).toHaveLength(before + 1); + expect(sends.at(-1)).toBe(replay); + expect(shell.session.items).toBe(queued); + if (action === "retry") { + expect(sends[0]).toBe(sends[1]); + expect(composerSends).toEqual([]); + expect(feedback).toEqual([]); + expect(cancellations).toBe(0); + expect(isFeedbackCapturePending()).toBe(true); + expect(queued.map((item) => item.text)).toEqual(["held first", "held second"]); + bridge.submit("actual feedback", "immediate"); + expect(feedback).toEqual(["actual feedback"]); + expect(sends).toHaveLength(2); + } else { + expect(composerSends).toEqual(["operator prompt", "operator prompt"]); + } + } } finally { + resetFeedbackStateForTests(); bridge.dispose(); shell.dispose(); } @@ -67,26 +170,78 @@ describe("agent ask wake delivery", () => { { width: 80, height: 24 }, ); }); +} - test("an ask parking mid-cycle defers to settle, then flushes once", async () => { +describe("agent ask wake delivery", () => { + test("synthetic wake bypasses armed feedback and leaves user followups held", async () => { await withTestRenderer( async (h) => { const shell = createAppShell(h.renderer, { terminal: { columns: 80, rows: 24 }, wireKeys: false, - run: "idle", }); - const { port, sends } = capturePort(); + const sends: string[] = []; + const feedback: string[] = []; + let cancellations = 0; + const submit = createSubmitHandler({ + dispatchCommand: () => {}, + sendPrompt: (text) => { + sends.push(text); + }, + isFeedbackCapturePending, + onFeedbackText: (text) => { + feedback.push(text); + cancelFeedbackCapture(); + return "Thanks"; + }, + cancelFeedbackCapture: () => { + cancellations++; + cancelFeedbackCapture(); + }, + }); + const port = createLiveSessionPort({ + send: submit, + classifySubmit: (text) => + classifySubmission(text, { + feedbackPending: isFeedbackCapturePending(), + feedbackCaptureEnabled: true, + }), + interrupt: () => {}, + deliver: routeQueuedDelivery({ + send: (text) => { + sends.push(text); + // Real delivery can synchronously settle and notify the bridge again. + bridge.handle({ type: "inference.done", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + }, + deliverSteer: () => { + throw new Error("wake must not live-inject"); + }, + parentCycleLive: () => bridge.parentCycleLive, + }), + }); const bridge = attachSessionBridge(shell, port); try { - bridge.handle({ type: "run", state: "busy" }); bridge.handle({ type: "inference.start", data: {} }); - bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); - expect(sends).toEqual([]); - + bridge.handle({ type: "fleet", running: 1 }); + bridge.submit("held followup", "queue"); bridge.handle({ type: "inference.done", data: {} }); + const held = shell.session; + armFeedbackCapture(); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toHaveLength(1); + expect(sends[0]).toContain("q1"); + expect(feedback).toEqual([]); + expect(cancellations).toBe(0); + expect(isFeedbackCapturePending()).toBe(true); + expect(shell.session.items).toEqual(held.items); + expect(held.items).toHaveLength(1); + expect(bridge.turn.isProcessing).toBe(false); + bridge.submit("actual feedback", "immediate"); + expect(feedback).toEqual(["actual feedback"]); expect(sends).toHaveLength(1); } finally { + resetFeedbackStateForTests(); bridge.dispose(); shell.dispose(); } @@ -95,83 +250,132 @@ describe("agent ask wake delivery", () => { ); }); - test("multiple parked asks coalesce into one wake message", async () => { - await withTestRenderer( - async (h) => { - const shell = createAppShell(h.renderer, { - terminal: { columns: 80, rows: 24 }, - wireKeys: false, - run: "idle", - }); - const { port, sends } = capturePort(); - const bridge = attachSessionBridge(shell, port); - try { - bridge.handle({ type: "run", state: "busy" }); - bridge.handle({ type: "inference.start", data: {} }); - bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); - bridge.handle({ type: "agent-ask", asks: [wake("a2", "q2")] }); + test("resolved snapshots remove deferred questions", async () => { + await withWakeBridge((bridge, sends) => { + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + bridge.handle({ type: "agent-ask", asks: [] }); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toEqual([]); + }); + }); + + test("final worker gate closes over an idle parent fleet hold", async () => { + await withWakeBridge((bridge, sends) => { + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "fleet", running: 2 }); + bridge.handle({ type: "inference.done", data: {} }); + bridge.gateOpened(); + bridge.gateOpened(); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + bridge.gateClosed(); + expect(sends).toEqual([]); + bridge.gateClosed(); + expect(sends).toHaveLength(1); + bridge.handle({ type: "inference.done", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toHaveLength(1); + }); + }); + + for (const settleFirst of [false, true]) { + test(`live parent and two gates wait for both conditions (settle first: ${settleFirst})`, async () => { + await withWakeBridge((bridge, sends) => { + bridge.gateOpened(); + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "fleet", running: 1 }); + bridge.gateOpened(); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + bridge.gateClosed(); + if (settleFirst) bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toEqual([]); + bridge.gateClosed(); + if (!settleFirst) { + expect(sends).toEqual([]); + expect(bridge.turn.isProcessing).toBe(true); bridge.handle({ type: "inference.done", data: {} }); - expect(sends).toHaveLength(1); - expect(sends[0]).toContain("a1"); - expect(sends[0]).toContain("a2"); - } finally { - bridge.dispose(); - shell.dispose(); } - }, - { width: 80, height: 24 }, - ); + expect(sends).toHaveLength(1); + }); + }); + } + + test("a parked ask wakes an idle parent exactly once, including notifications after settle", async () => { + await withWakeBridge((bridge, sends) => { + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toHaveLength(1); + expect(sends[0]).toContain("a1"); + expect(sends[0]).toContain("Which port?"); + expect(sends[0]).toContain("q1"); + expect(sends[0]).toContain("send_input"); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + bridge.handle({ type: "inference.done", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toHaveLength(1); + bridge.handle({ type: "agent-ask", asks: [] }); + expect(sends).toHaveLength(1); + }); + }); + + test("an ask parking mid-cycle defers to settle, then flushes once", async () => { + await withWakeBridge((bridge, sends) => { + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toEqual([]); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toHaveLength(1); + }); + }); + + test("multiple parked asks coalesce into one wake message", async () => { + await withWakeBridge((bridge, sends) => { + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1"), wake("a2", "q2")] }); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toHaveLength(1); + expect(sends[0]).toContain("a1"); + expect(sends[0]).toContain("a2"); + }); }); test("an open gate defers the wake; gate close flushes once", async () => { - await withTestRenderer( - async (h) => { - const shell = createAppShell(h.renderer, { - terminal: { columns: 80, rows: 24 }, - wireKeys: false, - run: "idle", - }); - const { port, sends } = capturePort(); - const bridge = attachSessionBridge(shell, port); - try { - bridge.gateOpened(); - bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); - expect(sends).toEqual([]); + await withWakeBridge((bridge, sends) => { + bridge.gateOpened(); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toEqual([]); + bridge.gateClosed(); + expect(sends).toHaveLength(1); + }); + }); - bridge.gateClosed(); - expect(sends).toHaveLength(1); - } finally { - bridge.dispose(); - shell.dispose(); - } - }, - { width: 80, height: 24 }, - ); + test("clearQueuedDelivery drops stashed wakes and resets delivered identities", async () => { + await withWakeBridge((bridge, sends) => { + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + bridge.clearQueuedDelivery(); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toEqual([]); + bridge.handle({ type: "agent-ask", asks: [wake("a2", "q2")] }); + expect(sends).toHaveLength(1); + bridge.clearQueuedDelivery(); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toHaveLength(1); + bridge.handle({ type: "agent-ask", asks: [wake("a2", "q2")] }); + expect(sends).toHaveLength(2); + }); }); - test("clearQueuedDelivery drops stashed wakes", async () => { - await withTestRenderer( - async (h) => { - const shell = createAppShell(h.renderer, { - terminal: { columns: 80, rows: 24 }, - wireKeys: false, - run: "idle", - }); - const { port, sends } = capturePort(); - const bridge = attachSessionBridge(shell, port); - try { - bridge.handle({ type: "run", state: "busy" }); - bridge.handle({ type: "inference.start", data: {} }); - bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); - bridge.clearQueuedDelivery(); - bridge.handle({ type: "inference.done", data: {} }); - expect(sends).toEqual([]); - } finally { - bridge.dispose(); - shell.dispose(); - } - }, - { width: 80, height: 24 }, - ); + test("disposed bridges cannot revive deferred wakes", async () => { + await withWakeBridge((bridge, sends) => { + bridge.gateOpened(); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + bridge.dispose(); + bridge.gateClosed(); + bridge.handle({ type: "inference.done", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toEqual([]); + }); }); }); diff --git a/src/tui/runner/exit.ts b/src/tui/runner/exit.ts index 36f76427d..f4786330c 100644 --- a/src/tui/runner/exit.ts +++ b/src/tui/runner/exit.ts @@ -39,6 +39,23 @@ import { const tuiLogger = getLogger([LOG_NAMESPACE_ROOT, "tui"]); +export function resetSessionForRotation( + state: Pick, + services: Pick, +): void { + const reset = (): void => { + services.deliveryGeneration.bump(); + cancelFeedbackCapture(); + services.emitter.emit("session.clear"); + services.subAgentSessions.cancelAll("Session cleared"); + }; + if (state.withFleetPublicationSuspended === undefined) { + reset(); + } else { + state.withFleetPublicationSuspended(reset); + } +} + export interface ResolveExitCodeArgs { runError: string | undefined; sinkError: string | undefined; @@ -427,15 +444,7 @@ export async function createRunLifecycle( // abort handles → child agent.close) before clearing the session store so // /clear does not leave orphaned child reactors burning tokens. const newSession = (): void => { - services.deliveryGeneration.bump(); - cancelFeedbackCapture(); - // Wipe the painted transcript immediately. The product host listens for - // session.clear; the Ink App used to clear its own stream unconditionally - // and that path never moved to OpenTUI. - services.emitter.emit("session.clear"); - // Cancel live workers before rotation so /clear does not leave orphaned - // child reactors burning tokens under the old session id. - services.subAgentSessions.cancelAll("Session cleared"); + resetSessionForRotation(state, services); // Backend rotation is always enqueued regardless of contention; the queue // serialises it behind any in-progress op. Sub-agents nest under the new // session automatically because getWorkdirBase reads the live sessionId. diff --git a/src/tui/runner/state.ts b/src/tui/runner/state.ts index 0db63a254..523680f0b 100644 --- a/src/tui/runner/state.ts +++ b/src/tui/runner/state.ts @@ -238,6 +238,7 @@ export interface RunnerState { ) => Promise; shutdownRuntime?: () => Promise; stopFleetReporting?: () => void; + withFleetPublicationSuspended?: (reset: () => void) => void; } export function recordRunError(state: RunnerState, err: unknown): void { diff --git a/src/tui/runner/wiring.ask-wake.test.ts b/src/tui/runner/wiring.ask-wake.test.ts new file mode 100644 index 000000000..5354a8f06 --- /dev/null +++ b/src/tui/runner/wiring.ask-wake.test.ts @@ -0,0 +1,323 @@ +import { expect, test } from "bun:test"; +import { EventEmitter } from "node:events"; +import { createSubAgentSessionStore } from "../../subagent/session-store.js"; +import { createFleetMailbox, createWaitAgentsTool } from "../../subagent/agent-fleet.js"; +import { createFleetWakePublisher } from "./wiring.js"; +import { attachSessionBridge, type BridgeInboundEvent } from "../runtime-bridge.js"; +import { createLiveSessionPort } from "../live-session-port.js"; +import { createAppShell } from "../shell/index.js"; +import { withTestRenderer } from "../harness.js"; +import { resetSessionForRotation } from "./exit.js"; +import { clearTranscript } from "../shell/chrome.js"; +import { createDeliveryGeneration, createLeftoverSend } from "../queued-delivery.js"; +import { createSessionOperationQueue } from "../session-operation-queue.js"; + +test("failed reset releases publication without flushing partially cancelled workers", () => { + const store = createSubAgentSessionStore(); + const emitter = new EventEmitter(); + const publisher = createFleetWakePublisher(store, emitter); + const events: BridgeInboundEvent[] = []; + emitter.on("event", (event: BridgeInboundEvent) => events.push(event)); + const unsubscribe = store.subscribe(publisher.publish); + try { + store.start({ id: "old", agentId: "builder", description: "old", brief: "build" }); + store.markRunning("old"); + store.registerAsk("old", { + question: "Old question?", + questionId: "old-question", + resolve: () => {}, + reject: () => {}, + }); + events.length = 0; + const error = new Error("reset failed"); + expect(() => + publisher.withSuspended(() => { + store.wake(); + throw error; + }), + ).toThrow(error); + expect(events).toEqual([]); + publisher.withSuspended(() => store.cancelAll("Session cleared")); + expect(events).toEqual([ + { type: "agent-ask", asks: [] }, + { type: "fleet", running: 0 }, + ]); + events.length = 0; + publisher.withSuspended(() => store.cancelAll("Session cleared")); + expect(events).toEqual([ + { type: "agent-ask", asks: [] }, + { type: "fleet", running: 0 }, + ]); + } finally { + unsubscribe(); + } +}); + +for (const phase of ["settled", "prequeued", "deferred"] as const) { + test(`rotation suppresses old worker snapshots (${phase}), including repeated resets`, async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + }); + const store = createSubAgentSessionStore(); + const emitter = new EventEmitter(); + const deliveryGeneration = createDeliveryGeneration(); + const queue = createSessionOperationQueue(); + const sends: string[] = []; + const scheduled: string[] = []; + const deliver = createLeftoverSend({ + enqueue: queue.enqueue, + ingest: async (text, attachments) => ({ text, attachments }), + send: (text) => { + sends.push(text); + }, + captureGeneration: deliveryGeneration.capture, + onFailure: (error) => { + throw error; + }, + }); + const bridge = attachSessionBridge( + shell, + createLiveSessionPort({ + send: (text) => { + sends.push(text); + }, + deliver: (text) => { + scheduled.push(text); + deliver(text); + }, + interrupt: () => {}, + }), + ); + let resetting = false; + const resetEvents: BridgeInboundEvent[] = []; + const repainted: string[] = []; + emitter.on("event", (event: BridgeInboundEvent) => { + bridge.handle(event); + if (resetting) { + resetEvents.push(event); + repainted.push(...shell.streamLog.map((row) => row.text)); + } + }); + emitter.on("session.clear", () => { + clearTranscript(shell); + bridge.clearQueuedDelivery(); + }); + const publisher = createFleetWakePublisher(store, emitter); + const unsubscribe = store.subscribe(publisher.publish); + const start = (id: string) => { + store.start({ id, agentId: "builder", description: id, brief: "build" }); + store.markRunning(id); + store.registerAsk(id, { + question: `question ${id}`, + questionId: `question-${id}`, + resolve: () => {}, + reject: () => {}, + }); + }; + try { + for (let round = 0; round < 2; round++) { + bridge.handle({ type: "inference.start", data: {} }); + start(`old-${round}-one`); + start(`old-${round}-two`); + if (phase !== "deferred") bridge.handle({ type: "inference.done", data: {} }); + if (phase === "settled") { + await queue.awaitTail(); + bridge.handle({ type: "inference.done", data: {} }); + } + const beforeScheduled = scheduled.length; + const beforeSent = sends.length; + resetting = true; + resetSessionForRotation( + { withFleetPublicationSuspended: publisher.withSuspended }, + { deliveryGeneration, emitter, subAgentSessions: store }, + ); + resetting = false; + expect(scheduled).toHaveLength(beforeScheduled); + expect(repainted).toEqual([]); + expect(resetEvents).toEqual([ + { type: "agent-ask", asks: [] }, + { type: "fleet", running: 0 }, + ]); + expect(store.list().every((worker) => worker.status === "cancelled")).toBe(true); + await queue.awaitTail(); + expect(sends).toHaveLength(beforeSent); + bridge.handle({ type: "inference.done", data: {} }); + resetEvents.length = 0; + } + const before = sends.length; + start("new-worker"); + await queue.awaitTail(); + expect(sends).toHaveLength(before + 1); + expect(sends.at(-1)).toContain("question new-worker"); + bridge.handle({ type: "inference.done", data: {} }); + store.wake(); + await queue.awaitTail(); + expect(sends).toHaveLength(before + 1); + } finally { + unsubscribe(); + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); +} + +for (const removal of ["answer", "cancel", "terminal", "remove", "replace"] as const) { + test(`production pending snapshot drops a deferred ask on ${removal}`, async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + }); + const sends: string[] = []; + const send = (text: string) => { + sends.push(text); + }; + const bridge = attachSessionBridge( + shell, + createLiveSessionPort({ send, deliver: send, interrupt: () => {} }), + ); + const store = createSubAgentSessionStore(); + const emitter = new EventEmitter(); + const events: BridgeInboundEvent[] = []; + emitter.on("event", (event: BridgeInboundEvent) => { + events.push(event); + bridge.handle(event); + }); + const publisher = createFleetWakePublisher(store, emitter); + const unsubscribe = store.subscribe(publisher.publish); + try { + bridge.handle({ type: "inference.start", data: {} }); + const worker = store.start({ + id: "worker-session", + agentId: "builder", + description: "work", + brief: "build", + }); + store.markRunning(worker.id); + store.registerAsk(worker.id, { + question: "Which port?", + questionId: "q1", + resolve: () => {}, + reject: () => {}, + }); + if (removal === "answer") { + const mailbox = createFleetMailbox(store); + mailbox.register(worker.id); + const wait = createWaitAgentsTool({ sessions: store, fleetRecords: mailbox }); + if (wait.kind !== "full") throw new Error("expected full wait tool"); + const result = await wait.handler( + { + id: "wait-call", + name: "wait_agents", + arguments: { targets: [worker.id], timeout_ms: 1000 }, + }, + new AbortController().signal, + ); + expect(result.content).toContain("awaiting_director"); + store.sendInputOne(worker.id, "8080"); + } + if (removal === "cancel") store.cancelAsk(worker.id); + if (removal === "terminal") store.complete(worker.id, "done"); + if (removal === "remove") store.clear(); + if (removal === "replace") { + store.start({ + id: worker.id, + agentId: "builder", + description: "replacement", + brief: "build", + }); + } + expect(events.at(-1)?.type === "fleet" ? events.at(-2) : events.at(-1)).toEqual({ + type: "agent-ask", + asks: [], + }); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toEqual([]); + } finally { + unsubscribe(); + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); +} + +test("same catalog workers answer by session, reconcile one resolution and replacement, exclude nested asks", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + }); + const sends: string[] = []; + const send = (text: string) => { + sends.push(text); + }; + const bridge = attachSessionBridge( + shell, + createLiveSessionPort({ send, deliver: send, interrupt: () => {} }), + ); + const store = createSubAgentSessionStore(); + const emitter = new EventEmitter(); + emitter.on("event", (event: BridgeInboundEvent) => bridge.handle(event)); + const publisher = createFleetWakePublisher(store, emitter); + const unsubscribe = store.subscribe(publisher.publish); + const answers: string[] = []; + const ask = (id: string, questionId: string) => + store.registerAsk(id, { + question: questionId, + questionId, + resolve: (answer) => { + answers.push(`${id}:${answer}`); + }, + reject: () => {}, + }); + try { + bridge.handle({ type: "inference.start", data: {} }); + for (const id of ["session-one", "session-two", "nested"]) { + store.start({ + id, + agentId: "builder", + description: id, + brief: "build", + ...(id === "nested" ? { parentSessionId: "session-one" } : {}), + }); + store.markRunning(id); + ask(id, `question-${id}`); + } + expect(store.sendInputOne("session-one", "8080").ok).toBe(true); + expect(answers).toEqual(["session-one:8080"]); + expect(store.hasPendingAsk("session-two")).toBe(true); + store.cancelAsk("session-two"); + ask("session-two", "replacement-question"); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toHaveLength(1); + expect(sends[0]).toContain("targeting agent_id session-two"); + expect(sends[0]).toContain("replacement-question"); + expect(sends[0]).not.toContain("question-session-two"); + expect(sends[0]).not.toContain("question-session-one"); + expect(sends[0]).not.toContain("question-nested"); + bridge.handle({ type: "inference.done", data: {} }); + store.wake(); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toHaveLength(1); + expect(store.sendInputOne("session-two", "9090").ok).toBe(true); + expect(answers).toEqual(["session-one:8080", "session-two:9090"]); + } finally { + unsubscribe(); + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); +}); diff --git a/src/tui/runner/wiring.ts b/src/tui/runner/wiring.ts index 391c62319..0984cdaac 100644 --- a/src/tui/runner/wiring.ts +++ b/src/tui/runner/wiring.ts @@ -17,12 +17,11 @@ import { loadSentMessages } from "../../session/sent-messages.js"; import { setActiveDisposeHost } from "../../session/active-host.js"; import { createFleetWatch, - createPendingAskWatch, FLEET_REPORT_SETTLE_MS, FLEET_STALL_POLL_MS, liveFleetCount, observeFleet, - observePendingAsks, + pendingAskSnapshot, } from "../../subagent/index.js"; import { scheduleUpgradeNotice } from "../../upgrade/index.js"; import pkg from "../../../package.json" with { type: "json" }; @@ -53,6 +52,39 @@ import { LOG_NAMESPACE_ROOT } from "../../branding.js"; const tuiLogger = getLogger([LOG_NAMESPACE_ROOT, "tui"]); +export function createFleetWakePublisher( + sessions: RunnerServices["subAgentSessions"], + emitter: RunnerServices["emitter"], +) { + let lastLiveFleet = 0; + let suspended = false; + const publish = (): void => { + if (suspended) return; + const lanes = sessions.list(); + // Reconcile even an empty snapshot before a fleet drop can settle the parent. + const asks = pendingAskSnapshot(lanes, (id) => sessions.peekAsk(id)); + emitter.emit("event", { type: "agent-ask", asks }); + const fleet = liveFleetCount(sessions.list()); + if (fleet !== lastLiveFleet) { + lastLiveFleet = fleet; + emitter.emit("event", { type: "fleet", running: fleet }); + } + }; + const withSuspended = (reset: () => void): void => { + suspended = true; + // The bridge clears its fleet count even when the store's count is unchanged. + lastLiveFleet = -1; + try { + reset(); + } finally { + suspended = false; + } + // A failed cancellation must not publish its partially reset snapshot. + publish(); + }; + return { publish, withSuspended }; +} + export function wirePostStartup( state: RunnerState, services: RunnerServices, @@ -127,27 +159,10 @@ export function wirePostStartup( for (const update of observation.updates) surfaceSystemNotice(hostOf(state).shell, update); }; let fleetSettle: ReturnType | null = null; - // Live-lane count feeds the bridge's idle-with-fleet hold (CL-7057): the - // run stays busy after the parent turn settles until the last lane - // terminalizes. Store notifications fire per child event, not per status - // flip, so emit only when the count itself moves. - let lastLiveFleet = 0; - // Parked ask_director questions ride the same store subscription. The - // emitter-side watch dedups on questionId transitions, so only a newly - // parked question reaches the bridge; delivery timing is the bridge's. - let askWatch = createPendingAskWatch(); + const fleetWakePublisher = createFleetWakePublisher(services.subAgentSessions, services.emitter); + state.withFleetPublicationSuspended = fleetWakePublisher.withSuspended; const unsubscribeFleetReport = services.subAgentSessions.subscribe(() => { - const lanes = services.subAgentSessions.list(); - const fleet = liveFleetCount(lanes); - if (fleet !== lastLiveFleet) { - lastLiveFleet = fleet; - services.emitter.emit("event", { type: "fleet", running: fleet }); - } - const asks = observePendingAsks(askWatch, lanes, (id) => services.subAgentSessions.peekAsk(id)); - askWatch = asks.watch; - if (asks.wakes.length > 0) { - services.emitter.emit("event", { type: "agent-ask", asks: asks.wakes }); - } + fleetWakePublisher.publish(); if (fleetSettle !== null) return; fleetSettle = setTimeout(() => { fleetSettle = null; diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index 82352613b..f8bb3fa50 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -362,13 +362,15 @@ export interface BridgeBag { * stashing the same question twice. */ pendingAskWake: Map; + deliveredAskWake: Map; /** - * Set once `submit` exists inside `attachSessionBridge`; the module-scope + * Set inside `attachSessionBridge`; the module-scope * settle path (`settleRunToIdle`) and `gateClosed` re-enter through it. */ flushPendingAskWake: (() => void) | null; /** Last prompt actually sent — replay source for the quota auto-retry. */ lastSentMessage: string; + lastSentOrigin: "composer" | "internal" | null; /** One auto-retry per rate-limit window. */ quotaFired: boolean; now: () => number; @@ -942,10 +944,12 @@ function applyInbound(shell: AppShell, bag: BridgeBag, event: BridgeInboundEvent paintChrome(shell); return; } - // Parked ask_director questions: stash keyed by session (repeat store - // notifications overwrite rather than duplicate), then deliver now when - // the parent is free — otherwise they wait for settle or gate close. - for (const ask of event.asks) bag.pendingAskWake.set(ask.sessionId, ask); + bag.pendingAskWake = new Map(event.asks.map((ask) => [ask.sessionId, ask])); + for (const [sessionId, questionId] of bag.deliveredAskWake) { + if (bag.pendingAskWake.get(sessionId)?.questionId !== questionId) { + bag.deliveredAskWake.delete(sessionId); + } + } bag.flushPendingAskWake?.(); return; } @@ -1036,8 +1040,10 @@ export function attachSessionBridge( turn: initialTurnState(now()), liveFleet: 0, pendingAskWake: new Map(), + deliveredAskWake: new Map(), flushPendingAskWake: null, lastSentMessage: "", + lastSentOrigin: null, quotaFired: false, now, toolRows: new Map(), @@ -1235,6 +1241,13 @@ export function attachSessionBridge( if (settled) settleRun(); }; + const recordLastSent = ( + replay: { text: string; origin: "composer" | "internal" } | null, + ): void => { + bag.lastSentMessage = replay === null ? "" : replay.text; + bag.lastSentOrigin = replay === null ? null : replay.origin; + }; + const submit = ( text: string, kind: "queue" | "steer" | "immediate" | "reinject", @@ -1286,7 +1299,7 @@ export function attachSessionBridge( meta: "stop", }); bag.port.interrupt(); - bag.lastSentMessage = ""; + recordLastSent(null); bag.turn = turnStateOnInterrupt(bag.turn, now()); } @@ -1307,12 +1320,12 @@ export function attachSessionBridge( ...(kind === "reinject" ? { meta: "reinject" } : {}), }); bag.pendingEchoes.push(t); - bag.port.sendImmediate(t, attachments); shell.session = setRunState(shell.session, "busy"); - bag.lastSentMessage = t; + recordLastSent({ text: t, origin: "composer" }); bag.turn = turnStateOnSubmit(bag.turn, now()); paintChrome(shell); paintPhase(); + bag.port.sendImmediate(t, attachments); return; } @@ -1332,18 +1345,26 @@ export function attachSessionBridge( paintChrome(shell); }; - /** - * Deliver stashed ask wakes as one turn, but only when the parent can act: - * never mid-cycle, and never while an operator gate holds the run. The - * gate-closed path flushes explicitly — a gate's `isProcessing` lingers by - * design, so the ordinary idle condition would never fire there. - */ - const flushPendingAskWake = (force = false): void => { - if (bag.pendingAskWake.size === 0) return; - if (!force && (bag.turn.isProcessing || bag.turn.blockedGateCount > 0)) return; - const asks = [...bag.pendingAskWake.values()]; - bag.pendingAskWake.clear(); - submit(asks.map((ask) => pendingAskWakeText(ask)).join("\n\n"), "immediate"); + const sendInternalText = (text: string): void => { + appendStreamRow(shell, { role: "user", text }); + bag.pendingEchoes.push(text); + shell.session = setRunState(shell.session, "busy"); + recordLastSent({ text, origin: "internal" }); + bag.turn = turnStateOnSubmit(bag.turn, now()); + paintChrome(shell); + paintPhase(); + // Harness turns are not composer input: /feedback must never consume them. + bag.port.deliver({ id: crypto.randomUUID(), text, kind: "queue", enqueuedAt: now() }); + }; + const flushPendingAskWake = (): void => { + if (bag.disposed || bag.turn.isProcessing || bag.turn.blockedGateCount > 0) return; + const asks = [...bag.pendingAskWake.values()].filter( + (ask) => bag.deliveredAskWake.get(ask.sessionId) !== ask.questionId, + ); + if (asks.length === 0) return; + // Outbound delivery can synchronously re-enter through store/stream events. + for (const ask of asks) bag.deliveredAskWake.set(ask.sessionId, ask.questionId); + sendInternalText(asks.map((ask) => pendingAskWakeText(ask)).join("\n\n")); }; bag.flushPendingAskWake = () => flushPendingAskWake(); @@ -1365,16 +1386,18 @@ export function attachSessionBridge( drainAtBoundary(shell, bag); // Clearing the last prompt is what stops the quota loop from replaying a // turn the operator (or the watchdog) deliberately stopped. - bag.lastSentMessage = ""; + recordLastSent(null); bag.turn = turnStateOnInterrupt(bag.turn, now()); paintPhase(); }; const clearQueuedDelivery = (): void => { if (bag.disposed) return; shell.session = createSessionQueue("idle"); + recordLastSent(null); bag.pendingEchoes.length = 0; bag.liveFleet = 0; bag.pendingAskWake.clear(); + bag.deliveredAskWake.clear(); bag.pendingRowUpdates.clear(); paintChrome(shell); }; @@ -1395,10 +1418,7 @@ export function attachSessionBridge( if (bag.disposed) return; bag.turn = turnStateGateClosed(bag.turn, now()); paintPhase(); - // Flush a wake stashed while the gate held the run — but only at true - // session-idle. A gate closing over a still-live parent cycle must not - // inject mid-cycle; that cycle's own settle flushes the stash instead. - flushPendingAskWake(bag.turn.blockedGateCount === 0 && shell.session.run === "idle"); + flushPendingAskWake(); }; const tick = (): void => { @@ -1415,12 +1435,13 @@ export function attachSessionBridge( }) ) { bag.quotaFired = true; - const replay = bag.lastSentMessage; + const replay = { text: bag.lastSentMessage, origin: bag.lastSentOrigin }; bag.turn = clearQuotaWait(bag.turn); setStatusFlash(shell, "rate limit cleared — resubmitting", { ttlMs: RUNTIME_FLASH_MS, }); - submit(replay, "immediate"); + if (replay.origin === "internal") sendInternalText(replay.text); + else submit(replay.text, "immediate"); return; } @@ -1521,6 +1542,10 @@ export function attachSessionBridge( dispose: () => { flushOpenRow(shell, bag); bag.disposed = true; + recordLastSent(null); + bag.pendingAskWake.clear(); + bag.deliveredAskWake.clear(); + bag.flushPendingAskWake = null; applyCadence(null); clearShellBridgeHooks(shell); bridges.delete(shell); diff --git a/src/tui/stream-event-map.ts b/src/tui/stream-event-map.ts index e10b79ed3..b7f3d3b5b 100644 --- a/src/tui/stream-event-map.ts +++ b/src/tui/stream-event-map.ts @@ -50,8 +50,8 @@ export type BridgeInboundEvent = */ | { readonly type: "fleet"; readonly running: number } /** - * Workers newly parked in ask_director (transition-only, emitter-side - * deduped). The bridge stashes and delivers them when the parent can act. + * Authoritative snapshot of currently pending top-level ask_director + * questions, including empty. The bridge reconciles and dedups delivery. */ | { readonly type: "agent-ask"; readonly asks: readonly PendingAskWake[] } | { readonly type: "tool.boundary" } diff --git a/src/tui/turn-state.test.ts b/src/tui/turn-state.test.ts index cfc095abf..f6d60da99 100644 --- a/src/tui/turn-state.test.ts +++ b/src/tui/turn-state.test.ts @@ -23,6 +23,16 @@ const fold = ( initialTurnState(startMs), ); +test("an idle worker gate does not manufacture parent processing", () => { + const opened = turnStateGateOpened(initialTurnState(0)); + expect(opened.isProcessing).toBe(false); + expect(opened.status).toBe("blocked"); + expect(turnStateGateClosed(opened, 1).status).toBe("idle"); + const started = turnStateFromEvent(opened, { type: "inference.start" }, 2); + expect(started.isProcessing).toBe(true); + expect(turnStateGateClosed(started, 3).status).toBe("running"); +}); + describe("turnStateFromEvent", () => { test("start awaits the first token", () => { const s = fold([{ type: "inference.start" }]); diff --git a/src/tui/turn-state.ts b/src/tui/turn-state.ts index ba6e0b339..33038b7ac 100644 --- a/src/tui/turn-state.ts +++ b/src/tui/turn-state.ts @@ -214,14 +214,14 @@ export function turnStateGateOpened(state: TurnState): TurnState { return { ...state, status: "blocked", - isProcessing: true, blockedGateCount, }; } /** * A gate resolved. Only the last outstanding gate clearing returns the turn - * to "running" — earlier ones just decrement the count. `lastActivityAt` + * to "running" if the parent is processing, otherwise "idle" — earlier ones + * just decrement the count. `lastActivityAt` * moves to `nowMs` so the stall clock restarts from the moment the operator * actually answered, rather than crediting silence spent reading the prompt. */ @@ -230,7 +230,7 @@ export function turnStateGateClosed(state: TurnState, nowMs: number): TurnState if (blockedGateCount > 0) return { ...state, blockedGateCount }; return { ...state, - status: state.status === "blocked" ? "running" : state.status, + status: state.status === "blocked" ? (state.isProcessing ? "running" : "idle") : state.status, lastActivityAt: nowMs, blockedGateCount, }; From 07dc44e4c9c8be42ad4bbcb90c9512686de30776 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 8 Sep 2026 20:53:41 -0700 Subject: [PATCH 3/6] Point parent wakes at the send_input target field --- docs/ARCHITECTURE.md | 4 +++- docs/TUI.md | 4 ++-- src/subagent/fleet-report.ask-wake.test.ts | 2 +- src/subagent/fleet-report.ts | 2 +- src/tui/runner/wiring.ask-wake.test.ts | 2 +- src/tui/runner/wiring.ts | 2 +- 6 files changed, 9 insertions(+), 7 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 3a72f1a85..5dd0fdb9a 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -209,7 +209,9 @@ Three distinct concepts (do not conflate them): The **`spawn_agent`** tool starts a fleet agent on a separate inference source (tier/profile resolved from settings) and returns immediately with an `agent_id`; **`wait_agents`** collects reports later. Declared fan-out is unlimited: excess dispatches enqueue rather than fail. `run()` is admitted by `src/subagent/admission.ts` (default burst window of 8 is race-avoidance so a 429 freeze can fire before a herd — not a declared-spawn cap). Occupancy is the whole first `run()`, including `wait_agents`. Nested children of an already-admitted parent bypass **capacity** so a nested orchestrator cannot deadlock while holding a slot; they still wait on a provider 429 pause. Drain is FIFO among currently admissible jobs (a paused provider is skipped, not head-of-line for every provider). Resume and followup inference re-enter the same queue. Queued workers report wait/list status `queued` (live, not failed). Lowering capacity never cancels in-flight work. Retryable provider 429s freeze new admits via the shared retry remapper in `createCorbitsRetryPolicy`; `quota_exhausted` does not freeze. `list_agents` remains mailbox-scoped. The dispatch brief separates durable `context`, actionable `prompt`, and optional `goals` (checklist seeds for the _child's_ own `manage_tasks` list). Implement/review dispatches (and their default directors) fail closed without non-empty `success_criteria`. The child returns a structured report (`Summary` / `Findings` / `Blockers` / `Paths`) plus a tools-used footer. Parent and child never share a `manage_tasks` list. -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. The TUI 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 target the worker session ID through `send_input`, 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. Escalate to the human with **`ask_operator`** only when the parent cannot resolve it. +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. diff --git a/docs/TUI.md b/docs/TUI.md index 78cb101b0..4e3d121dc 100644 --- a/docs/TUI.md +++ b/docs/TUI.md @@ -224,8 +224,8 @@ An `ask_director` lane stays live and reads as waiting on the director, not stal 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` with the -worker's session ID, not its shared catalog ID. Each session/question identity is +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. Each session/question identity is delivered once while pending; the strip never re-delivers it. Synthetic wakes use the idle delivery path, bypassing composer `/feedback` capture and leaving queued user follow-ups untouched. diff --git a/src/subagent/fleet-report.ask-wake.test.ts b/src/subagent/fleet-report.ask-wake.test.ts index 9a3ddfb8f..a08ba5354 100644 --- a/src/subagent/fleet-report.ask-wake.test.ts +++ b/src/subagent/fleet-report.ask-wake.test.ts @@ -77,7 +77,7 @@ describe("pendingAskWakeText", () => { expect(text).toContain("Which port?"); expect(text).toContain("q1"); expect(text).toContain("send_input"); - expect(text).toContain("targeting agent_id a1"); + expect(text).toContain("using target a1"); expect(text.toLowerCase()).toContain("worker"); }); }); diff --git a/src/subagent/fleet-report.ts b/src/subagent/fleet-report.ts index 234369b1e..c02d4c4a6 100644 --- a/src/subagent/fleet-report.ts +++ b/src/subagent/fleet-report.ts @@ -165,7 +165,7 @@ export function pendingAskWakeText(wake: PendingAskWake): string { "", wake.question, "", - `The worker — not the operator — raised this. Answer it with send_input (soft) targeting agent_id ${wake.sessionId}; do not relay to the operator unless it genuinely needs them.`, + `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"); } diff --git a/src/tui/runner/wiring.ask-wake.test.ts b/src/tui/runner/wiring.ask-wake.test.ts index 5354a8f06..c7b3c7318 100644 --- a/src/tui/runner/wiring.ask-wake.test.ts +++ b/src/tui/runner/wiring.ask-wake.test.ts @@ -301,7 +301,7 @@ test("same catalog workers answer by session, reconcile one resolution and repla ask("session-two", "replacement-question"); bridge.handle({ type: "inference.done", data: {} }); expect(sends).toHaveLength(1); - expect(sends[0]).toContain("targeting agent_id session-two"); + expect(sends[0]).toContain("using target session-two"); expect(sends[0]).toContain("replacement-question"); expect(sends[0]).not.toContain("question-session-two"); expect(sends[0]).not.toContain("question-session-one"); diff --git a/src/tui/runner/wiring.ts b/src/tui/runner/wiring.ts index 0984cdaac..46cf0b827 100644 --- a/src/tui/runner/wiring.ts +++ b/src/tui/runner/wiring.ts @@ -64,7 +64,7 @@ export function createFleetWakePublisher( // Reconcile even an empty snapshot before a fleet drop can settle the parent. const asks = pendingAskSnapshot(lanes, (id) => sessions.peekAsk(id)); emitter.emit("event", { type: "agent-ask", asks }); - const fleet = liveFleetCount(sessions.list()); + const fleet = liveFleetCount(lanes); if (fleet !== lastLiveFleet) { lastLiveFleet = fleet; emitter.emit("event", { type: "fleet", running: fleet }); From b2c13b9fe4caee76e16a5961c8c2bb7c76b5bba8 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Wed, 9 Sep 2026 10:55:59 -0700 Subject: [PATCH 4/6] Teach the parent the idle-send director wake contract Wait JSON running-plus-question was the wrong pull contract, and an idle parent still has to see parked questions after a stop. --- docs/PRODUCT.md | 2 +- docs/TUI.md | 5 +- src/agent/directors/skywalker/package.test.ts | 8 +- src/agent/directors/skywalker/package.ts | 2 +- src/tui/agent-ask-wake.test.ts | 104 ++++++++++++++++++ src/tui/runner/submit.ts | 1 + src/tui/runner/wiring.ts | 3 +- src/tui/runtime-bridge.ts | 5 +- 8 files changed, 121 insertions(+), 9 deletions(-) diff --git a/docs/PRODUCT.md b/docs/PRODUCT.md index 5373d29c3..d9121a1a7 100644 --- a/docs/PRODUCT.md +++ b/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. diff --git a/docs/TUI.md b/docs/TUI.md index 4e3d121dc..1b711b866 100644 --- a/docs/TUI.md +++ b/docs/TUI.md @@ -225,8 +225,9 @@ 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. Each session/question identity is -delivered once while pending; the strip never re-delivers it. Synthetic wakes use +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. diff --git a/src/agent/directors/skywalker/package.test.ts b/src/agent/directors/skywalker/package.test.ts index 378dac0c5..73cbe39d9 100644 --- a/src/agent/directors/skywalker/package.test.ts +++ b/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/); }); diff --git a/src/agent/directors/skywalker/package.ts b/src/agent/directors/skywalker/package.ts index 85759fab1..6ed508804 100644 --- a/src/agent/directors/skywalker/package.ts +++ b/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. diff --git a/src/tui/agent-ask-wake.test.ts b/src/tui/agent-ask-wake.test.ts index a7fcac888..ccac1e4e6 100644 --- a/src/tui/agent-ask-wake.test.ts +++ b/src/tui/agent-ask-wake.test.ts @@ -378,4 +378,108 @@ describe("agent ask wake delivery", () => { expect(sends).toEqual([]); }); }); + + for (const stop of ["interrupt", "stall abort"] as const) { + test(`${stop} flushes a stashed ask once the parent is idle`, async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }); + const sends: string[] = []; + let nowMs = 0; + let tick = () => {}; + const bridge = attachSessionBridge( + shell, + createLiveSessionPort({ + send: (text) => { + sends.push(text); + }, + deliver: (text) => { + sends.push(text); + }, + interrupt: () => {}, + }), + { + now: () => nowMs, + stallTimeoutMs: 1_000, + stallNoticeMs: 400, + schedule: (fn) => { + tick = fn; + return () => {}; + }, + }, + ); + try { + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "inference.text.delta", data: { token: "ok" } }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toEqual([]); + if (stop === "interrupt") { + bridge.interrupt(); + } else { + nowMs = 1_000; + tick(); + } + expect(sends).toHaveLength(1); + expect(sends[0]).toContain("q1"); + expect(bridge.turn.isProcessing).toBe(true); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + } + + test("a wake question with bracket lines does not spoof attachment-echo matching", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }); + const sends: string[] = []; + const send = (text: string) => { + sends.push(text); + }; + const bridge = attachSessionBridge( + shell, + createLiveSessionPort({ send, deliver: send, interrupt: () => {} }), + ); + try { + const ask = { + ...wake("a1", "q1"), + question: "Choose:\n[1] 8080\n[2] 9090", + }; + bridge.handle({ type: "agent-ask", asks: [ask] }); + expect(sends).toHaveLength(1); + const wakeText = sends[0]; + if (wakeText === undefined) throw new Error("expected wake text"); + bridge.handle({ + type: "message.received", + data: { message: { content: wakeText } }, + }); + expect(shell.streamLog.filter((row) => row.role === "user")).toHaveLength(1); + bridge.submit("hello", "immediate"); + bridge.handle({ + type: "message.received", + data: { message: { content: "hello\n[1 image attached: shot.png]" } }, + }); + expect( + shell.streamLog.filter((row) => row.role === "user").map((row) => row.text), + ).toEqual([wakeText, "hello"]); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); }); diff --git a/src/tui/runner/submit.ts b/src/tui/runner/submit.ts index 773732905..0ee7145c6 100644 --- a/src/tui/runner/submit.ts +++ b/src/tui/runner/submit.ts @@ -351,6 +351,7 @@ export function createDeliverRouting( }, recordSent: (text) => { if (text.trim().length === 0) return; + if (text.startsWith("ask_director wake")) return; void appendSentMessage(state.config.cwd, state.sessionId, text).catch((err: unknown) => { tuiLogger.debug("sent-message append failed: {error}", { error: err instanceof Error ? err.message : String(err), diff --git a/src/tui/runner/wiring.ts b/src/tui/runner/wiring.ts index 46cf0b827..567bbb5bb 100644 --- a/src/tui/runner/wiring.ts +++ b/src/tui/runner/wiring.ts @@ -79,7 +79,8 @@ export function createFleetWakePublisher( } finally { suspended = false; } - // A failed cancellation must not publish its partially reset snapshot. + // Reached only after reset() returns. A throw leaves publication suppressed + // so a failed cancellation cannot publish its partially reset snapshot. publish(); }; return { publish, withSuspended }; diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index f8bb3fa50..16a3274f1 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -449,7 +449,7 @@ function resolvePort(handlers?: SessionPortHandlers): SessionPort { * `message.received` word that note differently, so echoes match on content. */ function promptContent(text: string): string { - const note = text.indexOf("\n["); + const note = text.search(/\n\[\d+ images? attached:/); return (note === -1 ? text : text.slice(0, note)).trim(); } @@ -1366,7 +1366,7 @@ export function attachSessionBridge( for (const ask of asks) bag.deliveredAskWake.set(ask.sessionId, ask.questionId); sendInternalText(asks.map((ask) => pendingAskWakeText(ask)).join("\n\n")); }; - bag.flushPendingAskWake = () => flushPendingAskWake(); + bag.flushPendingAskWake = flushPendingAskWake; const doInterrupt = (): void => { if (bag.disposed) return; @@ -1389,6 +1389,7 @@ export function attachSessionBridge( recordLastSent(null); bag.turn = turnStateOnInterrupt(bag.turn, now()); paintPhase(); + flushPendingAskWake(); }; const clearQueuedDelivery = (): void => { if (bag.disposed) return; From d6c619e4e5c0744e870961627b133d750e2d8606 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Wed, 9 Sep 2026 12:46:28 -0700 Subject: [PATCH 5/6] Skip leftover ingest for director wake text Leftover hops used operator ingest, so a parked @path was rewritten before echo matching could consume the raw wake. --- src/tui/agent-ask-wake.test.ts | 88 ++++++++++++++++++++++++++++++++- src/tui/queued-delivery.test.ts | 29 +++++++++++ src/tui/queued-delivery.ts | 11 ++++- 3 files changed, 126 insertions(+), 2 deletions(-) diff --git a/src/tui/agent-ask-wake.test.ts b/src/tui/agent-ask-wake.test.ts index ccac1e4e6..de7b5ea81 100644 --- a/src/tui/agent-ask-wake.test.ts +++ b/src/tui/agent-ask-wake.test.ts @@ -5,7 +5,13 @@ import { createAppShell } from "./shell/index"; import { withTestRenderer } from "./harness"; import type { PendingAskWake } from "../subagent/fleet-report.js"; import { classifySubmission, createSubmitHandler } from "./runner/submit.js"; -import { routeQueuedDelivery } from "./queued-delivery.js"; +import { + createDeliveryGeneration, + createLeftoverSend, + routeQueuedDelivery, +} from "./queued-delivery.js"; +import { createSessionOperationQueue } from "./session-operation-queue.js"; +import { ingestOperatorPrompt } from "./prompt-attachments.js"; import { armFeedbackCapture, cancelFeedbackCapture, @@ -482,4 +488,84 @@ describe("agent ask wake delivery", () => { { width: 80, height: 24 }, ); }); + + test("idle leftover wake keeps an @path in the question raw and consumes the echo", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }); + const sent: string[] = []; + const attachments: number[] = []; + const ingested: string[] = []; + const queue = createSessionOperationQueue(); + const leftoverSend = createLeftoverSend({ + enqueue: queue.enqueue, + ingest: async (text, pending) => { + ingested.push(text); + return ingestOperatorPrompt( + text, + "/repo", + async () => { + throw new Error("wake leftover must not load image paths"); + }, + pending, + ); + }, + send: (text, pending) => { + sent.push(text); + attachments.push(pending.length); + }, + captureGeneration: createDeliveryGeneration().capture, + onFailure: (error) => { + throw error; + }, + }); + const send = (text: string) => { + leftoverSend(text); + }; + const bridge = attachSessionBridge( + shell, + createLiveSessionPort({ + send, + deliver: routeQueuedDelivery({ + send, + deliverSteer: () => { + throw new Error("wake must not live-inject"); + }, + parentCycleLive: () => bridge.parentCycleLive, + }), + interrupt: () => {}, + }), + ); + try { + const ask = { + ...wake("a1", "q1"), + question: "Should I edit @src/foo.ts?", + }; + bridge.handle({ type: "agent-ask", asks: [ask] }); + await queue.awaitTail(); + expect(sent).toHaveLength(1); + const wakeText = sent[0]; + if (wakeText === undefined) throw new Error("expected wake text"); + expect(wakeText).toContain("@src/foo.ts"); + expect(wakeText).not.toContain("(not found)"); + expect(attachments).toEqual([0]); + expect(ingested).toEqual([]); + expect(shell.streamLog.filter((row) => row.role === "user")).toHaveLength(1); + bridge.handle({ + type: "message.received", + data: { message: { content: wakeText } }, + }); + expect(shell.streamLog.filter((row) => row.role === "user")).toHaveLength(1); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); }); diff --git a/src/tui/queued-delivery.test.ts b/src/tui/queued-delivery.test.ts index 8e4a63ec9..c5d54c7e4 100644 --- a/src/tui/queued-delivery.test.ts +++ b/src/tui/queued-delivery.test.ts @@ -240,6 +240,35 @@ describe("createLeftoverSend", () => { expect(recorded).toEqual(["follow-up"]); }); + test("leftover send skips ingest for ask_director wake and still ingests operator prompts", async () => { + const sent: string[] = []; + const ingested: string[] = []; + const { enqueue, awaitTail } = createSessionOperationQueue(); + const leftoverSend = createLeftoverSend({ + enqueue, + ingest: async (text, pending) => { + ingested.push(text); + return { text: `${text} ingested`, attachments: pending }; + }, + send: (text) => { + sent.push(text); + }, + captureGeneration: () => () => true, + onFailure: (err) => { + throw err; + }, + }); + + leftoverSend("ask_director wake — see @src/foo.ts"); + leftoverSend("please read @src/foo.ts"); + await awaitTail(); + expect(ingested).toEqual(["please read @src/foo.ts"]); + expect(sent).toEqual([ + "ask_director wake — see @src/foo.ts", + "please read @src/foo.ts ingested", + ]); + }); + test("generation bump drops leftover send but not a sibling Enter send", async () => { const leftoverSent: string[] = []; const enterSent: string[] = []; diff --git a/src/tui/queued-delivery.ts b/src/tui/queued-delivery.ts index b307cb461..e5cb55e0e 100644 --- a/src/tui/queued-delivery.ts +++ b/src/tui/queued-delivery.ts @@ -121,9 +121,18 @@ export function createLiveSteerDeliver( /** * Leftover / queue drain hop: capture generation at hop time, ingest, then * send only if /clear|/new has not bumped. Operator Enter must not use this. + * `ask_director wake` leftover is passed through raw so worker @paths and + * image mentions are not rewritten as operator attachments. */ export function createLeftoverSend( args: CreateLeftoverSendArgs, ): (text: string, attachments?: readonly PendingImageAttachment[]) => void { - return createGenerationGatedHop({ ...args, hop: args.send }); + return createGenerationGatedHop({ + ...args, + hop: args.send, + ingest: async (text, pending) => + text.startsWith("ask_director wake") + ? { text, attachments: pending } + : args.ingest(text, pending), + }); } From 7034fe0c418f85857aabcd2e102c031bf5136c2b Mon Sep 17 00:00:00 2001 From: Sawyer Date: Wed, 9 Sep 2026 15:20:02 -0700 Subject: [PATCH 6/6] Share the ask_director wake prefix across leftover ingest and sent-message skip Leftover ingest skip and sent-message skip both matched the wake prefix as a string literal. Export ASK_DIRECTOR_WAKE_PREFIX from fleet-report (the wake text source) and use it in queued-delivery and submit. --- src/subagent/fleet-report.ts | 4 +++- src/tui/queued-delivery.ts | 3 ++- src/tui/runner/submit.ts | 3 ++- 3 files changed, 7 insertions(+), 3 deletions(-) diff --git a/src/subagent/fleet-report.ts b/src/subagent/fleet-report.ts index c02d4c4a6..74e41b504 100644 --- a/src/subagent/fleet-report.ts +++ b/src/subagent/fleet-report.ts @@ -154,6 +154,8 @@ export function pendingAskSnapshot( 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 @@ -161,7 +163,7 @@ export function pendingAskSnapshot( */ export function pendingAskWakeText(wake: PendingAskWake): string { return [ - `ask_director wake — worker ${wake.agentId} (${wake.description}) parked question ${wake.questionId} while this session was not collecting:`, + `${ASK_DIRECTOR_WAKE_PREFIX} — worker ${wake.agentId} (${wake.description}) parked question ${wake.questionId} while this session was not collecting:`, "", wake.question, "", diff --git a/src/tui/queued-delivery.ts b/src/tui/queued-delivery.ts index e5cb55e0e..88a12854d 100644 --- a/src/tui/queued-delivery.ts +++ b/src/tui/queued-delivery.ts @@ -10,6 +10,7 @@ import type { PendingImageAttachment } from "./image-attachments.js"; import type { ProductHostDeliver } from "./product-host.js"; +import { ASK_DIRECTOR_WAKE_PREFIX } from "../subagent/fleet-report.js"; export interface RouteQueuedDeliveryArgs { send: (text: string, attachments?: readonly PendingImageAttachment[]) => void; @@ -131,7 +132,7 @@ export function createLeftoverSend( ...args, hop: args.send, ingest: async (text, pending) => - text.startsWith("ask_director wake") + text.startsWith(ASK_DIRECTOR_WAKE_PREFIX) ? { text, attachments: pending } : args.ingest(text, pending), }); diff --git a/src/tui/runner/submit.ts b/src/tui/runner/submit.ts index 0ee7145c6..edea2a307 100644 --- a/src/tui/runner/submit.ts +++ b/src/tui/runner/submit.ts @@ -37,6 +37,7 @@ import type { InferenceAttemptIdentity } from "./state.js"; import { tuiSendFailureMessage } from "./send-failure-message.js"; import type { ProviderFailureAttempt } from "../provider/failure-attempt.js"; import type { Agent } from "@intx/agent"; +import { ASK_DIRECTOR_WAKE_PREFIX } from "../../subagent/fleet-report.js"; import { hostOf, type RunnerServices, type RunnerState } from "./state.js"; import { LOG_NAMESPACE_ROOT } from "../../branding.js"; @@ -351,7 +352,7 @@ export function createDeliverRouting( }, recordSent: (text) => { if (text.trim().length === 0) return; - if (text.startsWith("ask_director wake")) return; + if (text.startsWith(ASK_DIRECTOR_WAKE_PREFIX)) return; void appendSentMessage(state.config.cwd, state.sessionId, text).catch((err: unknown) => { tuiLogger.debug("sent-message append failed: {error}", { error: err instanceof Error ? err.message : String(err),