From dab784c0178942970cfe9add8a4cad35582d181c Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 8 Sep 2026 10:20:41 -0700 Subject: [PATCH 1/3] 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 8b476ef4d..8b691033e 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 d44e42db9..395fd876c 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 7205e47e9798919d0d25be0da76b5465c0c9c740 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 8 Sep 2026 14:32:26 -0700 Subject: [PATCH 2/3] 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 8b691033e..4b39f6b29 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 395fd876c..12755a39f 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 482db17d9e309114a0efde847a9c1d72e8b7a143 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 8 Sep 2026 20:53:41 -0700 Subject: [PATCH 3/3] 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 4b39f6b29..603a1afe8 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 12755a39f..bfcbb451f 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 });