From af1eaf8595ee5de8df551c29ce6ccf19255c44c9 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Fri, 11 Sep 2026 12:57:12 -0700 Subject: [PATCH] Refuse list_agents after a parked director question is surfaced --- CHANGELOG.md | 5 + docs/ARCHITECTURE.md | 2 +- docs/PRODUCT.md | 2 +- src/agent/directors/skywalker/package.test.ts | 3 +- src/agent/directors/skywalker/package.ts | 4 +- src/agent/prompts.ts | 4 +- src/subagent/agent-fleet.test.ts | 384 +++++++++++++++++- src/subagent/agent-fleet.ts | 101 ++++- src/subagent/poll-exempt.test.ts | 13 + src/tui/agent-ask-wake.test.ts | 19 + src/tui/runner/wiring.ask-wake.test.ts | 104 +++++ src/tui/runner/wiring.ts | 9 + src/tui/runtime-bridge.ts | 14 + 13 files changed, 641 insertions(+), 23 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c6303c3bb..226227cf7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,11 @@ parallel copies under `docs/` or `scripts/notes/`. At cut time: rename - System-originated inbound (`message.received` without the operator flag — background shell exits, mailbox mail, fleet-dry and compact continuations) paints as a visible system row instead of a user prompt. +- After a parked `ask_director` is surfaced (TUI idle-send wake, nested + `wait_agents` with a question payload, or a successful `list_agents`), + further `list_agents` calls fail closed until `send_input` answers or the + ask is dropped. Do not poll `list_agents`. + ## [0.3.21] - 2026-09-11 diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index c79cf62a9..5027269a8 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -216,7 +216,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`. On the TUI primary, Skywalker idles after spawn and occupancy delivers mailbox mail as system inbound when a worker finishes or fails (including while siblings still run). Nested orchestrators still collect with **`wait_agents`**. 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`. On the TUI primary that arrives as an idle-send wake (and yields an in-flight `wait_agents` as a timeout without the question payload). Nested **`wait_agents`** still returns `awaiting_director` with a question payload — that is not terminal. The parent answers with **`send_input`**, then continues. 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`. On the TUI primary that arrives as an idle-send wake (and yields an in-flight `wait_agents` as a timeout without the question payload). Nested **`wait_agents`** still returns `awaiting_director` with a question payload — that is not terminal. Once a parked ask is surfaced (TUI wake, nested `wait_agents` with a question payload, or a successful `list_agents`), further `list_agents` calls fail closed until **`send_input`** answers or the ask is dropped — `list_agents` is not a poll. The parent answers with **`send_input`**, then continues. 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. diff --git a/docs/PRODUCT.md b/docs/PRODUCT.md index 276a0b9b4..e382b08f7 100644 --- a/docs/PRODUCT.md +++ b/docs/PRODUCT.md @@ -170,7 +170,7 @@ Corbits Code fans work out to short-lived **fleet agents** — workers with thei - **Agents** are runtime entities (primary session or child). - **Tasks** are checklist items owned by one agent via `manage_tasks`. -- **Fleet agents** are spawned with `spawn_agent`. On the TUI primary, mailbox mail arrives as inbound when a worker finishes or fails — do not poll `wait_agents`. Nested orchestrators still collect with `wait_agents`. Workers ask the parent with `ask_director`. That parks a question while the worker stays `running`. Nested `wait_agents` returns `awaiting_director` with a question payload — that is not terminal. The parent answers with `send_input` (`target` = the worker's session id). When the parent TUI is not blocked in `wait_agents`, a parked question arrives as a synthetic idle-send wake. Escalate to the human only with `ask_operator`. +- **Fleet agents** are spawned with `spawn_agent`. On the TUI primary, mailbox mail arrives as inbound when a worker finishes or fails — do not poll `wait_agents`. Nested orchestrators still collect with `wait_agents`. Workers ask the parent with `ask_director`. That parks a question while the worker stays `running`. Nested `wait_agents` returns `awaiting_director` with a question payload — that is not terminal. Once a parked ask is surfaced, `list_agents` fails closed until `send_input` answers or the ask is dropped — do not poll `list_agents`. The parent answers with `send_input` (`target` = the worker's session id). When the parent TUI is not blocked in `wait_agents`, a parked question arrives as a synthetic idle-send wake. Escalate to the human only with `ask_operator`. Dispatch uses a structured brief (context / goal / optional goals seed) and returns a structured report. The TUI Agents strip and fleet board show who is running; live tool progress updates the status bar without dumping the child transcript into the parent chat. There is no turn budget. A tool-less final turn completes only with the four-heading report envelope. Printed `` markup in assistant text gets one corrective nudge to issue a real tool call and does not count as the wrap-up; without the envelope, one incomplete-report nudge is given and a second tool-less turn without the envelope salvages as `incomplete-report-stop`. A silent worker (no activity for `stallTimeoutMs`, opt-in) gets one continuation nudge, then salvages as `stalled` only after a full `stallTimeoutMs` grace with still no activity — queued checks inside that window wait, they do not salvage. An opt-in `deadlineMs`, or an operator cancel, can also end a run early. Each of these returns a salvage report so a runaway or idle child cannot quietly burn a large token budget or look done after prose alone. diff --git a/src/agent/directors/skywalker/package.test.ts b/src/agent/directors/skywalker/package.test.ts index 4f80b5e01..9fbcc9aa2 100644 --- a/src/agent/directors/skywalker/package.test.ts +++ b/src/agent/directors/skywalker/package.test.ts @@ -227,7 +227,8 @@ describe("skywalkerPackage", () => { const p = skywalkerPackage.systemPrompt; expect(p).toContain("ask_director"); expect(p).toContain("send_input"); - expect(p).toContain("awaiting_director"); + expect(p).toContain("Do not poll list_agents"); + expect(p).not.toContain("list_agents shows awaiting_director"); expect(p).toContain("idle-send"); expect(p).toMatch(/target = (that worker's |worker )session id/); expect(p).not.toMatch( diff --git a/src/agent/directors/skywalker/package.ts b/src/agent/directors/skywalker/package.ts index 3d06c85bf..f96c14094 100644 --- a/src/agent/directors/skywalker/package.ts +++ b/src/agent/directors/skywalker/package.ts @@ -16,11 +16,11 @@ You do not do the specialists' jobs by default. For tiny bounded product edits, Do not run long-blocking jobs on the parent (evals, full test suites, long installs, long-running implementation). Dispatch intern (mechanical shell), tester (suite / repro), or builder (substantial code). Path tools (write_file/edit_file/delete_file) are the DIY surface; shell file-writes stay denied. -Idle-orchestrator: fire one or more spawn_agent calls in a turn — each returns immediately with an agent_id and does not hold the parent. Then **reply to the operator** with who is running and **end the turn**. Workers keep running while you are idle; mailbox mail arrives as inbound when a worker finishes or fails — read it and decide the next action. Do not poll wait_agents. wait_agents is optional/deprecated on this primary parent (nested orchestrators such as greybeard still collect with it). list_agents shows the fleet without blocking. interrupt_agent unblocks an in-flight wait immediately. Enter mid-run delivers at the next parent tool.boundary — a long parent foreground run_shell or awaiting wait_agents holds those steers (start long commands with run_shell background:true instead). A bare spawn_agent does not. When the fleet goes dry the runtime re-enters with collected reports. +Idle-orchestrator: fire one or more spawn_agent calls in a turn — each returns immediately with an agent_id and does not hold the parent. Then **reply to the operator** with who is running and **end the turn**. Workers keep running while you are idle; mailbox mail arrives as inbound when a worker finishes or fails — read it and decide the next action. Do not poll wait_agents. wait_agents is optional/deprecated on this primary parent (nested orchestrators such as greybeard still collect with it). list_agents shows the fleet without blocking; do not poll list_agents. interrupt_agent unblocks an in-flight wait immediately. Enter mid-run delivers at the next parent tool.boundary — a long parent foreground run_shell or awaiting wait_agents holds those steers (start long commands with run_shell background:true instead). A bare spawn_agent does not. When the fleet goes dry the runtime re-enters with collected reports. # Operator updates (mandatory while fleet is live) -You are the chat surface. Workers cannot ask_operator; they ask_director. A parked question arrives as an idle-send wake (list_agents shows awaiting_director) — answer with send_input using target = that worker's session id. Escalate with ask_operator only when you cannot resolve it. While any specialist is running: +You are the chat surface. Workers cannot ask_operator; they ask_director. A parked question arrives as an idle-send wake — answer with send_input using target = that worker's session id. Do not poll list_agents. Escalate with ask_operator only when you cannot resolve it. While any specialist is running: - After every spawn wave: short status (who, goal, what you are waiting on) then end the turn. - On mailbox mail or a finished report: short update — do not go silent. - When the operator messages mid-run: answer them first (COMMUNICATION). Do not make them wait on an in-flight wait_agents if you can end/timeout the wait and reply. diff --git a/src/agent/prompts.ts b/src/agent/prompts.ts index cbf4bc8e4..8ac05ce48 100644 --- a/src/agent/prompts.ts +++ b/src/agent/prompts.ts @@ -174,7 +174,7 @@ export function buildGuidelines( : [ "", "Orchestration:", - "- Break multi-step or parallel work into focused worker dispatches with distinct lenses; prefer `spawn_agent` (fire several in one turn when jobs are independent), then reply with who is running and end the turn — workers keep running while you are idle. Mailbox mail arrives as inbound when a worker finishes; read it and do not poll `wait_agents`. Nested orchestrators still collect with `wait_agents`. `list_agents` shows the fleet without blocking.", + "- Break multi-step or parallel work into focused worker dispatches with distinct lenses; prefer `spawn_agent` (fire several in one turn when jobs are independent), then reply with who is running and end the turn — workers keep running while you are idle. Mailbox mail arrives as inbound when a worker finishes; read it and do not poll `wait_agents`. Nested orchestrators still collect with `wait_agents`. `list_agents` shows the fleet without blocking; after a parked ask is surfaced, answer with `send_input` and do not poll `list_agents`.", "- Pass the typed spawn contract: `intent`, `success_criteria` (done-when; required for implement/review and their default directors), `do_not` (scope fence), and `report_focus`. Free-form `prompt` without `success_criteria` fail-closes for implement/review and their default directors.", "- After workers return, classify fail / incomplete-report vs parent-initiated interrupt vs operator-cancel vs clean complete. Fail-path (`status: failed` or salvage `incomplete-report`): diagnose from the report or error and MAY spawn one successor with a changed brief. Parent-initiated interrupt (`interrupt_agent` / `send_input` with `interrupt:true` unblocks wait with `stop_reason: interrupted`): the worker is often still running and often has no report — `resume_agent` or re-wait; do not `spawn_agent` a successor against a still-live worker. Successor only if that session is no longer resumable. Operator-cancel (`stop_reason` cancelled): wait for the operator; do not auto-retry. Identical brief: refuse. Merge Summary/Findings into a coherent answer for the operator; do not paste raw fleet-agent dumps.", "- Use manage_tasks for your own coordination checklist; spawning workers is `spawn_agent`, not manage_tasks.", @@ -244,6 +244,8 @@ const TOOL_SUMMARIES: Record = { "start a worker agent and return immediately with agent_id; pass returned ids from search_agents as agent=...", wait_agents: "optional/deprecated on the primary parent — mailbox mail arrives as inbound; nested orchestrators still wait for spawned workers by agent_id; returns awaiting_director when a worker asks, without collecting that session", + list_agents: + "list this session's spawn_agent workers without blocking; after a parked ask_director is surfaced, returns an error until send_input answers or the ask is dropped — do not poll", search_agents: "find agent profiles by role or team before spawning with spawn_agent(agent=...); default results are id, description, and spawn metadata — pass include_body=true for the loaded system prompt / body", manage_tasks: diff --git a/src/subagent/agent-fleet.test.ts b/src/subagent/agent-fleet.test.ts index e440ba8ca..b786a4068 100644 --- a/src/subagent/agent-fleet.test.ts +++ b/src/subagent/agent-fleet.test.ts @@ -110,6 +110,53 @@ function waitUntilMailboxTerminal( }); } +function waitUntilAwaitingDirector( + mailbox: ReturnType, + sessions: ReturnType, + id: string, +): Promise { + return new Promise((resolve) => { + const done = (): boolean => + mailbox.peek(id)?.status === "awaiting_director"; + if (done()) { + resolve(); + return; + } + const unsub = sessions.subscribe(() => { + if (done()) { + unsub(); + resolve(); + } + }); + if (done()) { + unsub(); + resolve(); + } + }); +} + +async function callListAgents( + list: ReturnType, +): Promise<{ content: string; isError?: boolean }> { + if (list.kind !== "full") throw new Error("expected full tool"); + const result = await list.handler( + { + id: `list-${Math.random()}`, + name: "list_agents", + arguments: {}, + }, + new AbortController().signal, + ); + const content = + typeof result.content === "string" + ? result.content + : JSON.stringify(result.content); + return { + content, + ...(result.isError !== undefined ? { isError: result.isError } : {}), + }; +} + async function callToolRaw( tool: | ReturnType @@ -2318,10 +2365,6 @@ describe("list_agents", () => { return gate.promise; }); const spawn = createSpawnAgentTool(deps); - const wait = createWaitAgentsTool({ - sessions: deps.sessions, - fleetRecords: deps.fleetRecords, - }); const list = createListAgentsTool({ sessions: deps.sessions, fleetRecords: deps.fleetRecords, @@ -2331,17 +2374,11 @@ describe("list_agents", () => { prompt: "do it", intent: "explore", }); - await callTool(wait, { targets: [spawned.agent_id], timeout_ms: 5000 }); - if (list.kind !== "full") throw new Error("expected full tool"); - const raw = await list.handler( - { id: "list-ask-1", name: "list_agents", arguments: {} }, - new AbortController().signal, - ); - const content = - typeof raw.content === "string" - ? raw.content - : JSON.stringify(raw.content); - const parsed = JSON.parse(content) as { + const id = spawned.agent_id as string; + await waitUntilAwaitingDirector(deps.fleetRecords, deps.sessions, id); + const listed = await callListAgents(list); + expect(listed.isError).not.toBe(true); + const parsed = parseFleetJson(listed.content) as { agents: { agent_id: string; status: string; @@ -2352,7 +2389,7 @@ describe("list_agents", () => { }[]; }; expect(parsed.agents).toHaveLength(1); - expect(defined(parsed.agents[0]).agent_id).toBe(spawned.agent_id as string); + expect(defined(parsed.agents[0]).agent_id).toBe(id); expect(defined(parsed.agents[0]).status).toBe("awaiting_director"); expect(defined(parsed.agents[0]).collected).toBe(false); expect(defined(parsed.agents[0]).description).toBe("need a path"); @@ -2361,9 +2398,324 @@ describe("list_agents", () => { ); expect(defined(parsed.agents[0]).question_id).toBe("ask-1"); expect(list.definition.description).toContain("question_id"); + expect(list.definition.description).toContain("send_input"); gate.resolve({ report: "done" }); }); + test("errors after wait_agents surfaces awaiting_director with a question", async () => { + const gate = deferred(); + const deps = makeDeps(async (params) => { + params.onAgentReady?.({ + close: async () => undefined, + interrupt: () => undefined, + followup: async () => "", + deliver: () => undefined, + }); + void params.askDirectorPort + ?.register({ + question: "which file should I edit?", + questionId: "ask-1", + }) + .catch(() => undefined); + return gate.promise; + }); + const spawn = createSpawnAgentTool(deps); + const wait = createWaitAgentsTool({ + sessions: deps.sessions, + fleetRecords: deps.fleetRecords, + }); + const list = createListAgentsTool({ + sessions: deps.sessions, + fleetRecords: deps.fleetRecords, + }); + const spawned = await callTool(spawn, { + description: "need a path", + prompt: "do it", + intent: "explore", + }); + const id = spawned.agent_id as string; + const waited = await callTool(wait, { targets: [id], timeout_ms: 5000 }); + expect(waited.timed_out).toBe(false); + const first = defined((waited.results as Record[])[0]); + expect(first.status).toBe("awaiting_director"); + expect(first.question).toBe("which file should I edit?"); + expect(first.question_id).toBe("ask-1"); + const listed = await callListAgents(list); + expect(listed.isError).toBe(true); + expect(listed.content.startsWith("Error:")).toBe(true); + expect(listed.content).toContain("send_input"); + expect(listed.content).toContain(id); + expect(listed.content).toContain("ask-1"); + expect(listed.content).toContain("which file should I edit?"); + expect(listed.content).not.toContain('"agents"'); + gate.resolve({ report: "done" }); + }); + + test("second list with the same parked snapshot errors", async () => { + const gate = deferred(); + const deps = makeDeps(async (params) => { + params.onAgentReady?.({ + close: async () => undefined, + interrupt: () => undefined, + followup: async () => "", + deliver: () => undefined, + }); + void params.askDirectorPort + ?.register({ + question: "which file should I edit?", + questionId: "ask-1", + }) + .catch(() => undefined); + return gate.promise; + }); + const spawn = createSpawnAgentTool(deps); + const list = createListAgentsTool({ + sessions: deps.sessions, + fleetRecords: deps.fleetRecords, + }); + const spawned = await callTool(spawn, { + description: "need a path", + prompt: "do it", + intent: "explore", + }); + const id = spawned.agent_id as string; + await waitUntilAwaitingDirector(deps.fleetRecords, deps.sessions, id); + const first = await callListAgents(list); + expect(first.isError).not.toBe(true); + const parsed = parseFleetJson(first.content) as { + agents: { status: string; question_id?: string }[]; + }; + expect(defined(parsed.agents[0]).status).toBe("awaiting_director"); + const second = await callListAgents(list); + expect(second.isError).toBe(true); + expect(second.content.startsWith("Error:")).toBe(true); + expect(second.content).toContain("send_input"); + expect(second.content).toContain(id); + expect(second.content).toContain("ask-1"); + expect(second.content).toContain("which file should I edit?"); + expect(second.content).not.toContain('"agents"'); + gate.resolve({ report: "done" }); + }); + + test("send_input clears the gate so list_agents works again", async () => { + const gate = deferred(); + let answerP: Promise | undefined; + const deps = makeDeps(async (params) => { + params.onAgentReady?.({ + close: async () => undefined, + interrupt: () => undefined, + followup: async () => "", + deliver: () => { + throw new Error( + "soft send_input must not deliver while an ask is pending", + ); + }, + }); + answerP = params.askDirectorPort?.register({ + question: "which file should I edit?", + questionId: "ask-1", + }); + return gate.promise; + }); + const spawn = createSpawnAgentTool(deps); + const list = createListAgentsTool({ + sessions: deps.sessions, + fleetRecords: deps.fleetRecords, + }); + const sendInput = createSendInputTool({ + sessions: deps.sessions, + fleetRecords: deps.fleetRecords, + }); + const spawned = await callTool(spawn, { + description: "need a path", + prompt: "do it", + intent: "explore", + }); + const id = spawned.agent_id as string; + await waitUntilAwaitingDirector(deps.fleetRecords, deps.sessions, id); + expect((await callListAgents(list)).isError).not.toBe(true); + expect((await callListAgents(list)).isError).toBe(true); + await callTool(sendInput, { target: id, message: "edit src/foo.ts" }); + expect(await answerP).toBe("edit src/foo.ts"); + const after = await callListAgents(list); + expect(after.isError).not.toBe(true); + const parsed = parseFleetJson(after.content) as { + agents: { agent_id: string; status: string }[]; + }; + expect(defined(parsed.agents[0]).agent_id).toBe(id); + expect(defined(parsed.agents[0]).status).not.toBe("awaiting_director"); + gate.resolve({ report: "done" }); + }); + + test("interrupt_agent drops the ask so list_agents works again", async () => { + const gate = deferred(); + const deps = makeDeps(async (params) => { + params.onAgentReady?.({ + close: async () => undefined, + interrupt: () => undefined, + followup: async () => "", + deliver: () => undefined, + }); + void params.askDirectorPort + ?.register({ + question: "which file should I edit?", + questionId: "ask-1", + }) + .catch(() => undefined); + return gate.promise; + }); + const spawn = createSpawnAgentTool(deps); + const list = createListAgentsTool({ + sessions: deps.sessions, + fleetRecords: deps.fleetRecords, + }); + const interrupt = createInterruptAgentTool({ + sessions: deps.sessions, + fleetRecords: deps.fleetRecords, + }); + const spawned = await callTool(spawn, { + description: "need a path", + prompt: "do it", + intent: "explore", + }); + const id = spawned.agent_id as string; + await waitUntilAwaitingDirector(deps.fleetRecords, deps.sessions, id); + expect((await callListAgents(list)).isError).not.toBe(true); + expect((await callListAgents(list)).isError).toBe(true); + if (interrupt.kind !== "full") throw new Error("expected full tool"); + await interrupt.handler( + { id: "int-ask", name: "interrupt_agent", arguments: { target: id } }, + new AbortController().signal, + ); + const after = await callListAgents(list); + expect(after.isError).not.toBe(true); + const parsed = parseFleetJson(after.content) as { + agents: { agent_id: string; status: string }[]; + }; + expect(defined(parsed.agents[0]).agent_id).toBe(id); + expect(defined(parsed.agents[0]).status).not.toBe("awaiting_director"); + gate.resolve({ report: "done" }); + }); + + test("a new question_id is listable once", async () => { + const gate = deferred(); + let port: RunSubAgentParams["askDirectorPort"]; + const deps = makeDeps(async (params) => { + params.onAgentReady?.({ + close: async () => undefined, + interrupt: () => undefined, + followup: async () => "", + deliver: () => { + throw new Error( + "soft send_input must not deliver while an ask is pending", + ); + }, + }); + port = params.askDirectorPort; + void params.askDirectorPort + ?.register({ + question: "which file should I edit?", + questionId: "ask-1", + }) + .catch(() => undefined); + return gate.promise; + }); + const spawn = createSpawnAgentTool(deps); + const list = createListAgentsTool({ + sessions: deps.sessions, + fleetRecords: deps.fleetRecords, + }); + const sendInput = createSendInputTool({ + sessions: deps.sessions, + fleetRecords: deps.fleetRecords, + }); + const spawned = await callTool(spawn, { + description: "need a path", + prompt: "do it", + intent: "explore", + }); + const id = spawned.agent_id as string; + await waitUntilAwaitingDirector(deps.fleetRecords, deps.sessions, id); + expect((await callListAgents(list)).isError).not.toBe(true); + expect((await callListAgents(list)).isError).toBe(true); + await callTool(sendInput, { target: id, message: "edit src/foo.ts" }); + expect(port).toBeDefined(); + void defined(port) + .register({ + question: "which test should I add?", + questionId: "ask-2", + }) + .catch(() => undefined); + await waitUntilAwaitingDirector(deps.fleetRecords, deps.sessions, id); + const next = await callListAgents(list); + expect(next.isError).not.toBe(true); + const parsed = parseFleetJson(next.content) as { + agents: { question_id?: string; question?: string }[]; + }; + expect(defined(parsed.agents[0]).question_id).toBe("ask-2"); + expect(defined(parsed.agents[0]).question).toBe("which test should I add?"); + const blocked = await callListAgents(list); + expect(blocked.isError).toBe(true); + expect(blocked.content).toContain("ask-2"); + expect(blocked.content).not.toContain("ask-1"); + gate.resolve({ report: "done" }); + }); + + test("yield wait does not stamp; first list still surfaces once", async () => { + const gate = deferred(); + const deps = makeDeps(async (params) => { + params.onAgentReady?.({ + close: async () => undefined, + interrupt: () => undefined, + followup: async () => "", + deliver: () => undefined, + }); + void params.askDirectorPort + ?.register({ + question: "which file should I edit?", + questionId: "ask-1", + }) + .then( + () => undefined, + () => undefined, + ); + return gate.promise; + }); + const spawn = createSpawnAgentTool(deps); + let id = ""; + const wait = createWaitAgentsTool({ + sessions: deps.sessions, + fleetRecords: deps.fleetRecords, + shouldYieldWait: () => + deps.fleetRecords.peek(id)?.status === "awaiting_director", + }); + const list = createListAgentsTool({ + sessions: deps.sessions, + fleetRecords: deps.fleetRecords, + }); + const spawned = await callTool(spawn, { + description: "need a path", + prompt: "do it", + intent: "explore", + }); + id = spawned.agent_id as string; + const waited = await callTool(wait, { targets: [id], timeout_ms: 5_000 }); + expect(waited.timed_out).toBe(true); + const row = defined((waited.results as Record[])[0]); + expect(row.status).toBe("awaiting_director"); + expect(row.question).toBeUndefined(); + expect(row.question_id).toBeUndefined(); + const first = await callListAgents(list); + expect(first.isError).not.toBe(true); + const parsed = parseFleetJson(first.content) as { + agents: { question_id?: string }[]; + }; + expect(defined(parsed.agents[0]).question_id).toBe("ask-1"); + const second = await callListAgents(list); + expect(second.isError).toBe(true); + gate.resolve({ report: "ok" }); + }); + test("interrupt_agent leaves the strip after the linger window", async () => { const gate = deferred(); const deps = makeDeps(async (params) => { diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index 537e08b00..a490a0701 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -168,6 +168,18 @@ function waitStatusFromVerbLifecycle( /** Payload cap: uncollected pinned terminal records still holding a report. */ export const MAX_FLEET_RECORDS = 200; +export interface ParkedAskSurface { + readonly id: string; + readonly questionId: string; +} + +function parkedAskFingerprint(rows: readonly ParkedAskSurface[]): string { + return [...rows] + .map((row) => `${row.id}\0${row.questionId}`) + .sort() + .join("\n"); +} + /** * Per-install wait mailbox over the session store. Session lifecycle is the * source of wait status unless this overlay forces interrupted or has frozen @@ -176,6 +188,7 @@ export const MAX_FLEET_RECORDS = 200; class FleetMailbox { private readonly records = new Map(); private readonly sessions: SubAgentSessionStore; + private lastSurfacedParkedAsks: string | undefined; constructor(sessions: SubAgentSessionStore) { this.sessions = sessions; @@ -185,6 +198,24 @@ class FleetMailbox { }); } + /** + * Snapshot the parked ask set the parent already saw (TUI wake, wait_agents + * question payload, or a successful list_agents). Empty clears the stamp. + * Independent of TUI deliveredAskWake. + */ + noteParkedAsksSurfaced(rows: readonly ParkedAskSurface[]): void { + this.lastSurfacedParkedAsks = + rows.length === 0 ? undefined : parkedAskFingerprint(rows); + } + + parkedAsksAlreadySurfaced(rows: readonly ParkedAskSurface[]): boolean { + return ( + rows.length > 0 && + this.lastSurfacedParkedAsks !== undefined && + this.lastSurfacedParkedAsks === parkedAskFingerprint(rows) + ); + } + register(id: string): void { const existing = this.records.get(id); // start() drops pinCounts on call-id reuse. Re-pin whenever the overlay @@ -1730,6 +1761,24 @@ export function createWaitAgentsTool(deps: WaitAgentsDeps): AgentTool { }; }); + if (!yielded) { + const surfaced: ParkedAskSurface[] = []; + for (const row of results) { + if ( + row.status === "awaiting_director" && + "question" in row && + typeof row.question === "string" && + "question_id" in row && + typeof row.question_id === "string" + ) { + surfaced.push({ id: row.agent_id, questionId: row.question_id }); + } + } + if (surfaced.length > 0) { + deps.fleetRecords.noteParkedAsksSurfaced(surfaced); + } + } + return fleetResult(call.id, fleetJson({ results, timed_out: timedOut })); }, }); @@ -1741,17 +1790,64 @@ export const listAgentsToolDefinition: ToolDefinition = { "List the workers this session started with spawn_agent — the same fleet wait_agents " + "collects. Does not list siblings or another orchestrator's workers. Each entry is id, " + "director, description, wait status, lifecycle, stop_reason when recorded, and whether wait_agents already collected it. " + - "When status is awaiting_director, the entry also includes question and question_id.", + "When status is awaiting_director, the entry also includes question and question_id. " + + "After parked ask_director questions are already surfaced (idle-send wake, wait_agents with a question payload, or a prior list), " + + "list_agents returns an error until you answer with send_input (soft) or the ask is dropped. Do not poll list_agents.", inputSchema: { type: "object", properties: {}, }, }; +function collectParkedAsks( + fleetRecords: FleetMailboxHandle, +): { id: string; questionId: string; question: string }[] { + const parked: { id: string; questionId: string; question: string }[] = []; + for (const id of fleetRecords.ids()) { + const record = fleetRecords.peek(id); + if ( + record?.status !== "awaiting_director" || + record.questionId === undefined || + record.question === undefined + ) { + continue; + } + parked.push({ + id, + questionId: record.questionId, + question: record.question, + }); + } + return parked; +} + +function refuseSurfacedParkedAsks( + callId: string, + parked: readonly { id: string; questionId: string; question: string }[], +): ToolResult { + const targets = parked.map((row) => row.id).join(", "); + const lines = parked.map( + (row) => `target ${row.id} question_id ${row.questionId}: ${row.question}`, + ); + return fleetResult( + callId, + [ + `Error: list_agents is not progress while parked ask_director questions are already surfaced. Answer with send_input (soft) using target ${targets}, or escalate with ask_operator.`, + ...lines, + ].join("\n"), + ); +} + export function createListAgentsTool(deps: WaitAgentsDeps): AgentTool { return tool({ definition: listAgentsToolDefinition, handler: async (call, _signal): Promise => { + const parked = collectParkedAsks(deps.fleetRecords); + if (parked.length === 0) { + deps.fleetRecords.noteParkedAsksSurfaced([]); + } else if (deps.fleetRecords.parkedAsksAlreadySurfaced(parked)) { + return refuseSurfacedParkedAsks(call.id, parked); + } const agents = deps.fleetRecords.ids().map((id) => { const record = deps.fleetRecords.peek(id); const session = deps.sessions.get(id); @@ -1778,6 +1874,9 @@ export function createListAgentsTool(deps: WaitAgentsDeps): AgentTool { : {}), }; }); + if (parked.length > 0) { + deps.fleetRecords.noteParkedAsksSurfaced(parked); + } return fleetResult(call.id, fleetJson({ agents })); }, }); diff --git a/src/subagent/poll-exempt.test.ts b/src/subagent/poll-exempt.test.ts index 11033cbf1..b73d9c849 100644 --- a/src/subagent/poll-exempt.test.ts +++ b/src/subagent/poll-exempt.test.ts @@ -140,6 +140,19 @@ describe("isPollOnlyPendingBatch", () => { ).toBe(false); }); + test("list_agents is never poll-exempt", () => { + expect( + isPollOnlyPendingBatch( + [call("list_agents")], + [ + result({ + agents: [{ agent_id: "w0", status: "awaiting_director" }], + }), + ], + ), + ).toBe(false); + }); + test("empty batches and misaligned results are never exempt", () => { expect(isPollOnlyPendingBatch([], [])).toBe(false); expect( diff --git a/src/tui/agent-ask-wake.test.ts b/src/tui/agent-ask-wake.test.ts index d9d685c94..aff44359d 100644 --- a/src/tui/agent-ask-wake.test.ts +++ b/src/tui/agent-ask-wake.test.ts @@ -206,6 +206,25 @@ for (const action of [ } describe("agent ask wake delivery", () => { + test("onAskWakeSent fires only after a wake is actually sent", async () => { + await withWakeBridge((bridge, sends) => { + const sent: string[] = []; + bridge.setOnAskWakeSent((asks) => { + sent.push(...asks.map((ask) => `${ask.sessionId}:${ask.questionId}`)); + }); + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toEqual([]); + expect(sent).toEqual([]); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toHaveLength(1); + expect(sent).toEqual(["a1:q1"]); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toHaveLength(1); + expect(sent).toEqual(["a1:q1"]); + }); + }); + test("synthetic wake bypasses armed feedback and leaves user followups held", async () => { await withTestRenderer( async (h) => { diff --git a/src/tui/runner/wiring.ask-wake.test.ts b/src/tui/runner/wiring.ask-wake.test.ts index 0c3814b68..1b321b420 100644 --- a/src/tui/runner/wiring.ask-wake.test.ts +++ b/src/tui/runner/wiring.ask-wake.test.ts @@ -3,6 +3,7 @@ import { EventEmitter } from "node:events"; import { createSubAgentSessionStore } from "../../subagent/session-store.js"; import { createFleetMailbox, + createListAgentsTool, createWaitAgentsTool, } from "../../subagent/agent-fleet.js"; import { createFleetWakePublisher } from "./wiring.js"; @@ -363,3 +364,106 @@ test("same catalog workers answer by session, reconcile one resolution and repla { width: 80, height: 24 }, ); }); + +test("yield wait does not stamp; sending pendingAskWakeText gates list_agents", 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 store = createSubAgentSessionStore(); + const mailbox = createFleetMailbox(store); + const bridge = attachSessionBridge( + shell, + createLiveSessionPort({ + send, + deliver: send, + interrupt: () => undefined, + }), + ); + bridge.setOnAskWakeSent((asks) => { + mailbox.noteParkedAsksSurfaced( + asks.map((ask) => ({ + id: ask.sessionId, + questionId: ask.questionId, + })), + ); + }); + const emitter = new EventEmitter(); + emitter.on("event", (event: BridgeInboundEvent) => 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", + }); + mailbox.register(worker.id); + store.markRunning(worker.id); + store.registerAsk(worker.id, { + question: "Which port?", + questionId: "q1", + resolve: () => undefined, + reject: () => undefined, + }); + const wait = createWaitAgentsTool({ + sessions: store, + fleetRecords: mailbox, + shouldYieldWait: () => + mailbox.peek(worker.id)?.status === "awaiting_director", + }); + if (wait.kind !== "full") throw new Error("expected full wait tool"); + const waited = await wait.handler( + { + id: "wait-call", + name: "wait_agents", + arguments: { targets: [worker.id], timeout_ms: 1000 }, + }, + new AbortController().signal, + ); + const waitContent = + typeof waited.content === "string" + ? waited.content + : JSON.stringify(waited.content); + expect(waitContent).toContain("awaiting_director"); + expect(waitContent).not.toContain("Which port?"); + expect(sends).toEqual([]); + const list = createListAgentsTool({ + sessions: store, + fleetRecords: mailbox, + }); + if (list.kind !== "full") throw new Error("expected full list tool"); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toHaveLength(1); + expect(sends[0]).toContain("Which port?"); + const afterWake = await list.handler( + { id: "list-after", name: "list_agents", arguments: {} }, + new AbortController().signal, + ); + const afterContent = + typeof afterWake.content === "string" + ? afterWake.content + : JSON.stringify(afterWake.content); + expect(afterWake.isError).toBe(true); + expect(afterContent.startsWith("Error:")).toBe(true); + expect(afterContent).toContain("send_input"); + expect(afterContent).toContain(worker.id); + expect(afterContent).toContain("q1"); + expect(afterContent).toContain("Which port?"); + } 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 40fcf09b4..18de53712 100644 --- a/src/tui/runner/wiring.ts +++ b/src/tui/runner/wiring.ts @@ -272,6 +272,14 @@ export function wirePostStartup( }); }), ); + sessionBridge.setOnAskWakeSent((asks) => { + services.toolset.fleetRecords?.noteParkedAsksSurfaced( + asks.map((ask) => ({ + id: ask.sessionId, + questionId: ask.questionId, + })), + ); + }); sessionBridge.setWaitYieldWake(() => { services.subAgentSessions.wake(); }); @@ -297,6 +305,7 @@ export function wirePostStartup( unsubscribeFleetReport(); sessionBridge.setDryOpenTaskDriver(undefined); sessionBridge.setMailboxMailDriver(undefined); + sessionBridge.setOnAskWakeSent(undefined); sessionBridge.setWaitYieldWake(undefined); }; diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index b5848906a..71a1612e0 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -274,6 +274,13 @@ export interface SessionBridge { * settle (like flushPendingAskWake) and from the store-subscribe driver. */ setMailboxMailDriver: (driver: (() => boolean) | undefined) => void; + /** + * Called after flushPendingAskWake actually sendInternalText. Wiring stamps + * the fleet mailbox so list_agents fails closed until send_input. + */ + setOnAskWakeSent: ( + handler: ((asks: readonly PendingAskWake[]) => void) | undefined, + ) => void; /** * Wake in-flight wait_agents when the operator queues a steer. Timeout-shaped * yield — workers are not interrupted. @@ -482,6 +489,8 @@ export interface BridgeBag { * edge — that stays on dryOpenTaskDriver. */ mailboxMailDriver: (() => boolean) | undefined; + /** After a pending ask wake is actually sent. Independent of deliveredAskWake. */ + onAskWakeSent: ((asks: readonly PendingAskWake[]) => void) | undefined; /** Wake in-flight wait_agents when a steer is queued (timeout-shaped yield). */ waitYieldWake: (() => void) | undefined; /** Last prompt actually sent — replay source for the quota auto-retry. */ @@ -1321,6 +1330,7 @@ export function attachSessionBridge( awaitingContinuationInference: false, dryOpenTaskDriver: undefined, mailboxMailDriver: undefined, + onAskWakeSent: undefined, waitYieldWake: undefined, lastSentMessage: "", lastSentOrigin: null, @@ -1677,6 +1687,7 @@ export function attachSessionBridge( for (const ask of asks) bag.deliveredAskWake.set(ask.sessionId, ask.questionId); sendInternalText(asks.map((ask) => pendingAskWakeText(ask)).join("\n\n")); + bag.onAskWakeSent?.(asks); }; bag.flushPendingAskWake = flushPendingAskWake; @@ -1904,6 +1915,9 @@ export function attachSessionBridge( setMailboxMailDriver: (driver) => { bag.mailboxMailDriver = driver; }, + setOnAskWakeSent: (handler) => { + bag.onAskWakeSent = handler; + }, setWaitYieldWake: (wake) => { bag.waitYieldWake = wake; },