From 3047f62d23830847e54d3ba916ff9441615a6240 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 8 Sep 2026 21:59:48 -0700 Subject: [PATCH 1/6] Resume the parent when the fleet goes dry with open tasks A session with todo/doing tasks was settling idle just because the fleet printed nothing running. Re-enter with collected reports on the wentDry edge. Idle-with-live-fleet stays. --- docs/ARCHITECTURE.md | 2 +- docs/TUI.md | 9 +- src/agent/director.ts | 11 + src/agent/directors/skywalker/package.test.ts | 2 + src/agent/directors/skywalker/package.ts | 2 +- src/agent/tools.ts | 11 +- src/director.test.ts | 42 +++ src/session/assemble-runtime.ts | 6 + src/session/runtime-assembly.ts | 20 ++ src/subagent/fleet-dry-drive.test.ts | 267 ++++++++++++++++++ src/subagent/fleet-dry-drive.ts | 139 +++++++++ src/subagent/index.ts | 10 + src/tui/runner-host.test.ts | 32 +++ src/tui/runner/session.ts | 3 +- src/tui/runner/wiring.ts | 28 +- src/tui/runtime-bridge.test.ts | 114 ++++++++ src/tui/runtime-bridge.ts | 17 ++ 17 files changed, 706 insertions(+), 9 deletions(-) create mode 100644 src/subagent/fleet-dry-drive.test.ts create mode 100644 src/subagent/fleet-dry-drive.ts diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 5dd0fdb9a..fe8635de6 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -112,7 +112,7 @@ In TUI chat mode there is no completion gate — the session stays open across t Two directors, selected by role: -- **ChatDirector** (interactive, `src/agent/director.ts`) — Extends `DefaultDirector` with task list tracking, workflow nudges, LSP auto-activation, and multi-turn chat semantics. It never terminates the session: operator declines are surfaced as replies and the reactor stays alive for the next message. Auto mode is toggled by CLI flags (`--auto` / `--no-auto`); there is currently no in-session key to toggle it (default on; constrained envelope — workspace writes and unconstrained shell auto-allow; installs, recursive rm, force/uncontained worktree changes, sensitive-path and opaque-wrapper shell still ask; contained non-force `git worktree add`/`remove`/`prune` and `list` auto-allow; shell file-mutation denied). It is not a separate edit/plan mode. +- **ChatDirector** (interactive, `src/agent/director.ts`) — Extends `DefaultDirector` with task list tracking, workflow nudges, LSP auto-activation, and multi-turn chat semantics. It never terminates the session: operator declines are surfaced as replies and the reactor stays alive for the next message. Yielding while a live fleet is running is allowed (idle-with-fleet); the open-task nudge does not rewrite that wait/reply. When the fleet goes dry with tasks still todo/doing, the TUI runtime re-enters the parent with collected worker reports rather than settling idle. Auto mode is toggled by CLI flags (`--auto` / `--no-auto`); there is currently no in-session key to toggle it (default on; constrained envelope — workspace writes and unconstrained shell auto-allow; installs, recursive rm, force/uncontained worktree changes, sensitive-path and opaque-wrapper shell still ask; contained non-force `git worktree add`/`remove`/`prune` and `list` auto-allow; shell file-mutation denied). It is not a separate edit/plan mode. - **SubAgentDirector** (delegated work, `src/subagent/index.ts`) — Drives a dispatched worker until a turn arrives with no tool calls, then replies with the final assistant text and ends the run. A tool-less turn **after tools** completes only with the four-heading envelope (Summary, Findings, Blockers, Paths); a missing envelope nudges once (**incomplete-report**) and a second tool-less turn still without the envelope salvages as **incomplete-report-stop**. Explore/read-only workers that used tools then replied with findings remain normal completes; `requireEvidence` (off by default, set per director) additionally requires at least one read before a tool-less spawn-only reply can complete. Reads done through `run_shell` count as evidence too — `src/subagent/shell-evidence.ts` classifies shell reads (`cat`, `grep`, `sed` without `-i`, …) over the same subject expansion the auto-shell policy uses — but there is no corresponding shell-write evidence or file-write requirement: a run that never touches a file still completes normally once it replies with the envelope. There is no turn budget. Operator/parent cancel after any progress returns a **cancelled** salvage report (partial findings + tool activity) instead of a bare cancel string; cancel before progress still surfaces as cancelled-by-operator. There is no repetition/no-progress/never-acted/never-edited hard stop and no fingerprint-based re-dispatch block — a genuinely stuck worker runs until it completes, stalls, hits an opt-in wall-clock deadline, or is cancelled. `spawn_agent` starts each worker and records it in the caller's fleet mailbox; `wait_agents` collects terminal reports from that mailbox. Wait JSON includes `stop_reason` from the session when present so a salvage that is wait-`done` is not mistaken for a clean complete, and so parent-initiated interrupt (`interrupted`) is not mistaken for operator-cancel (`cancelled`). Deadline salvage prepends an advisory parent hint suggesting continuation plus a longer deadline if more wall-clock time is warranted. Failed and incomplete-report salvage tell the parent to diagnose from the report or error and MAY spawn one successor with a changed brief. A parent-initiated interrupt is a resumable pause: wait unblocks with `stop_reason: interrupted` (often while the session is still running and has no report); the parent should `resume_agent` or re-wait, and must not spawn a successor against a still-live worker. Successor only if that session is no longer resumable. Operator-cancelled salvage asks the parent to synthesize Findings and Paths and wait for the operator instead of auto-starting another specialist. Identical re-dispatch of the same brief stays refused at the prompt / spawn-handoff layer; there is no fingerprint-based re-dispatch hard-block. Deadline hints are advisory only — an identical re-dispatch is still admitted at runtime. Parent hints are prepended on salvage reports returned to the parent. The runtime does not auto-spawn successors. diff --git a/docs/TUI.md b/docs/TUI.md index 1b711b866..3e0452ade 100644 --- a/docs/TUI.md +++ b/docs/TUI.md @@ -258,7 +258,10 @@ lane finishes (`N done · nothing running`; failed and cancelled counts appear only when non-zero, e.g. `N done, M failed, K cancelled · nothing running`). Per-lane `done — summary` walls and live `dispatched` re-announcements -are never printed. +are never printed. That dry-fleet line stays operator-facing. If tasks +are still todo/doing, the runtime re-enters the parent with collected +reports as a system continuation — it does not paint the report wall as +a user message. `src/subagent/fleet-report.ts` is pure: it reads the same fleet-agent session store and the same `agentProgress()` stall definition. Store changes drive it; @@ -582,7 +585,9 @@ there is no parent tool left to steer — while Alt+Enter follow-ups keep waiting for true session-idle. A steer still pending when the hold engages sends at once (the parent it was steering has stopped), and the last lane terminalizing releases the hold, drains follow-ups, and returns the session -to idle. +to idle — unless todo/doing tasks remain, in which case a system +continuation starts before the fleet-0 event so the run stays busy and +follow-ups wait one more turn. Interrupting (Ctrl+C) never discards a queued or steered message. It used to — the transcript literally said `interrupt — discarded N pending`, and an diff --git a/src/agent/director.ts b/src/agent/director.ts index 95036d8f9..81e07b5fa 100644 --- a/src/agent/director.ts +++ b/src/agent/director.ts @@ -377,6 +377,12 @@ export interface ChatDirectorOptions { getProviderId?: (() => string | undefined) | undefined; /** Explicit retry policy; when set, skips the default Corbits policy. */ retryPolicy?: RetryPolicy | undefined; + /** + * Live `status === "running"` fleet-lane count. When greater than zero the + * director allows a terminal wait/reply with open tasks (idle-with-fleet). + * Omitted or 0 keeps the open-task nudge. Exec omits this. + */ + getLiveFleetCount?: (() => number) | undefined; } // The constructor takes the resolved ModelFamilyPolicy rather than the raw @@ -415,6 +421,7 @@ class ChatDirectorImpl extends DefaultDirector { private readonly compaction: CompactionGovernor; private readonly modelFamilyPolicy: ModelFamilyPolicy; private readonly retryPolicy: RetryPolicy; + private readonly getLiveFleetCount: (() => number) | undefined; // Consecutive assistant turns that contain tool calls and no text. Reset on // any turn with text and on every fresh user message — a weak model that // spins in place on one thread of tool calls still converges to the @@ -448,6 +455,7 @@ class ChatDirectorImpl extends DefaultDirector { this.modelFamilyPolicy = options.modelFamilyPolicy ?? resolveModelFamilyPolicy({ providerName: "" }); this.retryPolicy = options.retryPolicy ?? createCorbitsRetryPolicy(); + this.getLiveFleetCount = options.getLiveFleetCount; } setWorkflowCoordinator(coordinator: WorkflowCoordinator | undefined): void { @@ -883,6 +891,9 @@ class ChatDirectorImpl extends DefaultDirector { if (!atWorkflowGate && hasActiveTasks(this.tasks)) { const hasTerminal = baseActions.some((a) => a.type === "wait" || a.type === "reply"); if (hasTerminal) { + if ((this.getLiveFleetCount?.() ?? 0) > 0) { + return base; + } if (this.idleTerminationNudges < MAX_OPEN_TASK_NUDGES) { this.idleTerminationNudges++; const passThrough = baseActions.filter( diff --git a/src/agent/directors/skywalker/package.test.ts b/src/agent/directors/skywalker/package.test.ts index 73cbe39d9..745134c0e 100644 --- a/src/agent/directors/skywalker/package.test.ts +++ b/src/agent/directors/skywalker/package.test.ts @@ -117,6 +117,8 @@ describe("skywalkerPackage", () => { expect(p).not.toContain("task()"); expect(p).toContain('mode="all"'); expect(p).toContain("uncollected spawns"); + expect(p).toContain("When the fleet goes dry the runtime re-enters with collected reports"); + expect(p).toContain("do not tight-loop wait_agents"); expect(p).not.toContain("Present the plan when the change is large or ambiguous"); }); diff --git a/src/agent/directors/skywalker/package.ts b/src/agent/directors/skywalker/package.ts index 6ed508804..f0c77f784 100644 --- a/src/agent/directors/skywalker/package.ts +++ b/src/agent/directors/skywalker/package.ts @@ -16,7 +16,7 @@ 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 what happens next before you block. Prefer ending that turn (or calling wait_agents with a short timeout_ms) so Enter can land; do not immediately fuse into a long wait_agents right after spawn. wait_agents later on the targets you need (or omit targets to wait on this session's own uncollected spawns — never a sibling's). list_agents shows that same fleet without blocking. Use mode="all" when you need every target to finish; interrupt_agent unblocks wait_agents immediately. A timeout means still running — do not tight-loop wait_agents hoping for a different answer. Enter mid-run delivers at the next parent tool.boundary — a long parent run_shell or awaiting wait_agents holds those steers. A bare spawn_agent does not. +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 what happens next before you block. Prefer ending that turn (or calling wait_agents with a short timeout_ms) so Enter can land; do not immediately fuse into a long wait_agents right after spawn. wait_agents later on the targets you need (or omit targets to wait on this session's own uncollected spawns — never a sibling's). list_agents shows that same fleet without blocking. Use mode="all" when you need every target to finish; interrupt_agent unblocks wait_agents immediately. A timeout means still running — do not tight-loop wait_agents hoping for a different answer. Enter mid-run delivers at the next parent tool.boundary — a long parent run_shell or awaiting wait_agents holds those steers. A bare spawn_agent does not. When the fleet goes dry the runtime re-enters with collected reports; do not tight-loop wait_agents. # Operator updates (mandatory while fleet is live) diff --git a/src/agent/tools.ts b/src/agent/tools.ts index 42901aaa8..bc7cd08d4 100644 --- a/src/agent/tools.ts +++ b/src/agent/tools.ts @@ -59,6 +59,7 @@ import { createSpawnAgentTool, createWaitAgentsTool, createListAgentsTool, + type FleetMailboxHandle, } from "../subagent/agent-fleet.js"; import { DEFAULT_CLOSE_DEADLINE_MS } from "../subagent/dispose.js"; import { @@ -252,6 +253,12 @@ export interface AgentToolset { setToolPromoter: (promote: (names: string[]) => void) => void; // Session-start skill snapshot shared with the prompt listing. skills: SkillSummary[]; + /** + * The live wait mailbox this toolset already built for spawn_agent / + * wait_agents. Optional because a session without sub-agents has none. + * Callers must read this each time — do not capture a startup snapshot. + */ + fleetRecords?: FleetMailboxHandle; dispose: () => Promise; } @@ -377,9 +384,10 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise { @@ -977,6 +985,7 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise { expect(hasInfer(exhausted)).toBe(false); }); + test("live fleet with open tasks allows terminal wait/reply and does not spend the nudge budget", async () => { + let live = 1; + const director = createChatDirector("base", [], { + onTasksChange: () => {}, + getLiveFleetCount: () => live, + }); + await director.decide(manageTasksEvent("doing"), mockState, mockCapabilities); + + for (let i = 0; i < 4; i++) { + const actions = actionsArray(await director.decide(textTurn(), mockState, mockCapabilities)); + expect(hasInfer(actions)).toBe(false); + expect(hasReply(actions)).toBe(true); + } + + live = 0; + for (let i = 0; i < 3; i++) { + const nudged = actionsArray(await director.decide(textTurn(), mockState, mockCapabilities)); + expect(hasInfer(nudged)).toBe(true); + expect(hasReply(nudged)).toBe(false); + } + const exhausted = actionsArray(await director.decide(textTurn(), mockState, mockCapabilities)); + expect(hasReply(exhausted)).toBe(true); + expect(hasInfer(exhausted)).toBe(false); + }); + + test("omitted or zero live fleet count still nudges while a task is open", async () => { + const omitted = createChatDirector("base", [], { onTasksChange: () => {} }); + await omitted.decide(manageTasksEvent("doing"), mockState, mockCapabilities); + expect( + hasInfer(actionsArray(await omitted.decide(textTurn(), mockState, mockCapabilities))), + ).toBe(true); + + const zero = createChatDirector("base", [], { + onTasksChange: () => {}, + getLiveFleetCount: () => 0, + }); + await zero.decide(manageTasksEvent("doing"), mockState, mockCapabilities); + expect(hasInfer(actionsArray(await zero.decide(textTurn(), mockState, mockCapabilities)))).toBe( + true, + ); + }); + test("empty model turn settles with a valid empty reply", async () => { // DefaultDirector ends empty responses with bare wait; without a reply, // agent.send hangs and the TUI Working spinner sticks forever. diff --git a/src/session/assemble-runtime.ts b/src/session/assemble-runtime.ts index 23c2f598e..5cd9167f9 100644 --- a/src/session/assemble-runtime.ts +++ b/src/session/assemble-runtime.ts @@ -335,6 +335,11 @@ export interface ChatAgentWiring { inactivityTimeoutMs: number; totalTimeoutMs?: number | undefined; onTasksChange: (tasks: Task[]) => void; + /** + * Live running-lane count for ChatDirector idle-with-fleet. Omitted in exec + * (treated as 0). + */ + getLiveFleetCount?: () => number; /** Compaction governor re-entry (the reactor emits no event after compact). */ requestContinuation: () => void; getProvider: () => { providerName: string; model: string }; @@ -392,6 +397,7 @@ export function assembleChatAgent(wiring: ChatAgentWiring): AssembledChatAgent { requestContinuation: wiring.requestContinuation, provider: { ...wiring.getProvider() }, getProviderId: wiring.getProviderId, + getLiveFleetCount: wiring.getLiveFleetCount, }, ); directorHolder.instance = d; diff --git a/src/session/runtime-assembly.ts b/src/session/runtime-assembly.ts index d499e7545..4cae9d76c 100644 --- a/src/session/runtime-assembly.ts +++ b/src/session/runtime-assembly.ts @@ -379,3 +379,23 @@ export function buildCompactionContinuationMessage(): InboundMessage { signatureStatus: "missing", }; } + +/** + * System-originated inbound that re-enters the parent after the fleet goes dry + * with todo/doing tasks still open. No OPERATOR_ORIGINATED_FLAG — this is not + * an operator prompt and must not reset the tool-only loop-protection backstop. + */ +export function buildFleetDryContinuationMessage(text: string): InboundMessage { + return { + ref: { uid: 0, mailbox: "system" }, + headers: { + from: "user@local", + to: ["agent@local"], + date: new Date().toISOString(), + messageId: `fleet-dry-continue-${Date.now()}@local`, + }, + flags: [], + content: text, + signatureStatus: "missing", + }; +} diff --git a/src/subagent/fleet-dry-drive.test.ts b/src/subagent/fleet-dry-drive.test.ts new file mode 100644 index 000000000..8e077a1f1 --- /dev/null +++ b/src/subagent/fleet-dry-drive.test.ts @@ -0,0 +1,267 @@ +import { describe, expect, test } from "bun:test"; +import { createFleetMailbox } from "./agent-fleet.js"; +import { + buildFleetDryContinuationPrompt, + collectUncollectedTerminals, + driveOpenTasksAfterFleetDry, + FLEET_DRY_CONTINUATION_PREFIX, + FLEET_DRY_REPORT_CHARS, + shouldDriveOpenTasks, + type FleetDryMailbox, + type FleetDryMailboxRecord, +} from "./fleet-dry-drive.js"; +import { createSubAgentSessionStore } from "./session-store.js"; +import type { Task } from "../agent/tasks.js"; + +const openTask: Task = { id: "t1", title: "keep going", status: "todo" }; + +describe("shouldDriveOpenTasks", () => { + test("is true only on wentDry && open tasks && !parentProcessing", () => { + expect( + shouldDriveOpenTasks({ + previousRunning: 1, + running: 0, + hasOpenTasks: true, + parentProcessing: false, + }), + ).toBe(true); + }); + + test("is false for dry+terminal, live+open, parentProcessing, and already-dry", () => { + expect( + shouldDriveOpenTasks({ + previousRunning: 1, + running: 0, + hasOpenTasks: false, + parentProcessing: false, + }), + ).toBe(false); + expect( + shouldDriveOpenTasks({ + previousRunning: 2, + running: 1, + hasOpenTasks: true, + parentProcessing: false, + }), + ).toBe(false); + expect( + shouldDriveOpenTasks({ + previousRunning: 1, + running: 0, + hasOpenTasks: true, + parentProcessing: true, + }), + ).toBe(false); + expect( + shouldDriveOpenTasks({ + previousRunning: 0, + running: 0, + hasOpenTasks: true, + parentProcessing: false, + }), + ).toBe(false); + }); +}); + +describe("buildFleetDryContinuationPrompt", () => { + test("contains the prefix, open-task ids, and collected JSON", () => { + const prompt = buildFleetDryContinuationPrompt( + [openTask, { id: "t2", title: "done already", status: "done" }], + [{ agent_id: "worker-1", status: "done", report: "shipped", description: "lane" }], + ); + expect(prompt.startsWith(FLEET_DRY_CONTINUATION_PREFIX)).toBe(true); + expect(prompt).toContain("- t1: keep going (todo)"); + expect(prompt).not.toContain("t2:"); + expect(prompt).toContain("worker-1"); + expect(prompt).toContain("shipped"); + expect(prompt).toContain("already collected — do not call wait_agents for these agent_ids"); + }); + + test("empty reports still produce the prefix and an empty JSON array", () => { + const prompt = buildFleetDryContinuationPrompt([openTask], []); + expect(prompt.startsWith(FLEET_DRY_CONTINUATION_PREFIX)).toBe(true); + expect(prompt).toContain("- t1: keep going (todo)"); + expect(prompt).toContain("[]"); + }); +}); + +describe("collectUncollectedTerminals", () => { + test("take()s terminals and leaves live / awaiting_director / already-collected", () => { + const sessions = createSubAgentSessionStore(); + const mailbox = createFleetMailbox(sessions); + const start = (id: string, description: string) => { + const session = sessions.start({ + id, + description, + agentId: "builder", + brief: "brief", + }); + mailbox.register(session.id); + return session; + }; + + start("live", "still running"); + start("done", "finished lane"); + sessions.complete("done", "worker finished"); + start("fail", "failed lane"); + sessions.fail("fail", "boom"); + start("coll", "already collected"); + sessions.complete("coll", "already taken"); + mailbox.take("coll"); + start("ask", "waiting on director"); + sessions.markRunning("ask"); + expect( + sessions.registerAsk("ask", { + question: "which path?", + questionId: "q1", + resolve: () => undefined, + reject: () => undefined, + }), + ).toBe(true); + + const reports = collectUncollectedTerminals(mailbox, sessions.list()); + expect(reports.map((r) => r.agent_id).sort()).toEqual(["done", "fail"]); + expect(reports.find((r) => r.agent_id === "done")).toEqual({ + agent_id: "done", + status: "done", + description: "finished lane", + report: "worker finished", + }); + expect(reports.find((r) => r.agent_id === "fail")).toEqual({ + agent_id: "fail", + status: "failed", + description: "failed lane", + error: "boom", + }); + expect(mailbox.peek("done")?.collected).toBe(true); + expect(mailbox.peek("fail")?.collected).toBe(true); + expect(mailbox.peek("live")?.collected).not.toBe(true); + expect(mailbox.peek("live")?.status).toBe("running"); + expect(mailbox.peek("ask")?.status).toBe("awaiting_director"); + expect(mailbox.peek("ask")?.collected).not.toBe(true); + expect(mailbox.peek("coll")?.collected).toBe(true); + }); + + test("fills report/error from the session-store lane when the mailbox snapshot is empty", () => { + const records = new Map([["ghost", { status: "done" }]]); + const mailbox: FleetDryMailbox = { + ids: () => [...records.keys()], + peek: (id) => records.get(id), + take: (id) => { + const existing = records.get(id); + if (existing === undefined) return undefined; + const taken = { ...existing, collected: true }; + records.set(id, taken); + return taken; + }, + }; + const reports = collectUncollectedTerminals(mailbox, [ + { id: "ghost", description: "from store", report: "store report" }, + ]); + expect(reports).toEqual([ + { + agent_id: "ghost", + status: "done", + description: "from store", + report: "store report", + }, + ]); + expect(records.get("ghost")?.collected).toBe(true); + }); + + test("clips oversized reports", () => { + const records = new Map([ + ["big", { status: "done", report: "x".repeat(FLEET_DRY_REPORT_CHARS + 40) }], + ]); + const mailbox: FleetDryMailbox = { + ids: () => [...records.keys()], + peek: (id) => records.get(id), + take: (id) => records.get(id), + }; + const reports = collectUncollectedTerminals(mailbox, []); + expect(reports[0]?.report?.length).toBe(FLEET_DRY_REPORT_CHARS); + expect(reports[0]?.report?.endsWith("…")).toBe(true); + }); +}); + +describe("driveOpenTasksAfterFleetDry", () => { + test("dry+open collects, begins continuation, then sends", () => { + const order: string[] = []; + const records = new Map([ + ["w1", { status: "done", report: "ok", description: "lane" }], + ]); + const mailbox: FleetDryMailbox = { + ids: () => [...records.keys()], + peek: (id) => records.get(id), + take: (id) => { + const existing = records.get(id); + if (existing === undefined) return undefined; + const taken = { ...existing, collected: true }; + records.set(id, taken); + return taken; + }, + }; + const sent: string[] = []; + const driven = driveOpenTasksAfterFleetDry({ + previousRunning: 1, + running: 0, + openTasks: [openTask], + parentProcessing: false, + mailbox, + lanes: [], + beginSystemContinuation: (prompt) => { + order.push("begin"); + sent.push(prompt); + }, + send: (prompt) => { + order.push("send"); + sent.push(prompt); + }, + }); + expect(driven).toBe(true); + expect(order).toEqual(["begin", "send"]); + expect(sent[0]).toContain(FLEET_DRY_CONTINUATION_PREFIX); + expect(sent[0]).toContain("w1"); + expect(records.get("w1")?.collected).toBe(true); + }); + + test("dry+terminal, live+open, and parentProcessing only skip", () => { + const noop = { + mailbox: undefined, + lanes: [], + beginSystemContinuation: () => { + throw new Error("must not begin"); + }, + send: () => { + throw new Error("must not send"); + }, + }; + expect( + driveOpenTasksAfterFleetDry({ + previousRunning: 1, + running: 0, + openTasks: [{ id: "t1", title: "done", status: "done" }], + parentProcessing: false, + ...noop, + }), + ).toBe(false); + expect( + driveOpenTasksAfterFleetDry({ + previousRunning: 2, + running: 2, + openTasks: [openTask], + parentProcessing: false, + ...noop, + }), + ).toBe(false); + expect( + driveOpenTasksAfterFleetDry({ + previousRunning: 1, + running: 0, + openTasks: [openTask], + parentProcessing: true, + ...noop, + }), + ).toBe(false); + }); +}); diff --git a/src/subagent/fleet-dry-drive.ts b/src/subagent/fleet-dry-drive.ts new file mode 100644 index 000000000..4da23c687 --- /dev/null +++ b/src/subagent/fleet-dry-drive.ts @@ -0,0 +1,139 @@ +/** + * Drive the parent back into a turn when the live fleet goes dry while + * todo/doing tasks remain. Pure: the TUI subscriber decides when to call, + * this module decides whether to drive and what to send. + */ + +import { hasActiveTasks, type Task } from "../agent/tasks.js"; +import { isLiveWaitStatus, type WaitJSONStatus } from "./lifecycle.js"; + +/** Enough of a lane report for a parent continuation; traces stay on disk. */ +export const FLEET_DRY_REPORT_CHARS = 8_192; + +export const FLEET_DRY_CONTINUATION_PREFIX = "The fleet has gone dry. Remaining open tasks:"; + +export interface FleetDryMailboxRecord { + readonly status: WaitJSONStatus; + readonly collected?: boolean; + readonly report?: string; + readonly error?: string; + readonly description?: string; + readonly hint?: string; + readonly providerFailure?: true; +} + +export interface FleetDryMailbox { + ids(): readonly string[]; + peek(id: string): FleetDryMailboxRecord | undefined; + take(id: string): FleetDryMailboxRecord | undefined; +} + +export interface FleetDryLane { + readonly id: string; + readonly description?: string; + readonly report?: string; + readonly error?: string; +} + +export interface CollectedWorkerReport { + agent_id: string; + status: string; + description?: string; + report?: string; + error?: string; + hint?: string; + provider_failure?: true; +} + +export function shouldDriveOpenTasks(input: { + previousRunning: number; + running: number; + hasOpenTasks: boolean; + parentProcessing: boolean; +}): boolean { + const wentDry = input.running === 0 && input.previousRunning > 0; + return wentDry && input.hasOpenTasks && !input.parentProcessing; +} + +function clipField(text: string | undefined): string | undefined { + if (text === undefined) return undefined; + if (text.length <= FLEET_DRY_REPORT_CHARS) return text; + return `${text.slice(0, FLEET_DRY_REPORT_CHARS - 1).trimEnd()}…`; +} + +export function collectUncollectedTerminals( + mailbox: FleetDryMailbox | undefined, + lanes: readonly FleetDryLane[], +): CollectedWorkerReport[] { + if (mailbox === undefined) return []; + const byId = new Map(lanes.map((lane) => [lane.id, lane])); + const reports: CollectedWorkerReport[] = []; + for (const id of mailbox.ids()) { + const peeked = mailbox.peek(id); + if (peeked === undefined) continue; + if (peeked.collected === true) continue; + if (isLiveWaitStatus(peeked.status)) continue; + const taken = mailbox.take(id) ?? peeked; + const lane = byId.get(id); + const report = clipField(taken.report ?? lane?.report); + const error = clipField(taken.error ?? lane?.error); + const description = taken.description ?? lane?.description; + reports.push({ + agent_id: id, + status: taken.status, + ...(description !== undefined && description.length > 0 ? { description } : {}), + ...(taken.status !== "failed" && report !== undefined ? { report } : {}), + ...(error !== undefined ? { error } : {}), + ...(taken.hint !== undefined ? { hint: taken.hint } : {}), + ...(taken.providerFailure === true ? { provider_failure: true } : {}), + }); + } + return reports; +} + +export function buildFleetDryContinuationPrompt( + tasks: readonly Task[], + reports: readonly CollectedWorkerReport[], +): string { + const open = tasks.filter((task) => task.status === "todo" || task.status === "doing"); + const taskLines = open.map((task) => `- ${task.id}: ${task.title} (${task.status})`).join("\n"); + return [ + FLEET_DRY_CONTINUATION_PREFIX, + taskLines, + "", + "Collected worker reports (already collected — do not call wait_agents for these agent_ids):", + JSON.stringify(reports), + "", + "Continue the remaining work. Mark each task done or cancelled with manage_tasks", + "when finished, or spawn_agent the next specialist. Do not end this turn while", + "tasks are still todo/doing unless you dispatch live workers.", + ].join("\n"); +} + +export function driveOpenTasksAfterFleetDry(args: { + previousRunning: number; + running: number; + openTasks: readonly Task[]; + parentProcessing: boolean; + mailbox: FleetDryMailbox | undefined; + lanes: readonly FleetDryLane[]; + beginSystemContinuation: (prompt: string) => void; + send: (prompt: string) => void; +}): boolean { + const tasks = [...args.openTasks]; + if ( + !shouldDriveOpenTasks({ + previousRunning: args.previousRunning, + running: args.running, + hasOpenTasks: hasActiveTasks(tasks), + parentProcessing: args.parentProcessing, + }) + ) { + return false; + } + const reports = collectUncollectedTerminals(args.mailbox, args.lanes); + const prompt = buildFleetDryContinuationPrompt(tasks, reports); + args.beginSystemContinuation(prompt); + args.send(prompt); + return true; +} diff --git a/src/subagent/index.ts b/src/subagent/index.ts index 22393f125..906f4703b 100644 --- a/src/subagent/index.ts +++ b/src/subagent/index.ts @@ -25,6 +25,16 @@ export { type FleetWatch, type PendingAskWake, } from "./fleet-report.js"; +export { + buildFleetDryContinuationPrompt, + collectUncollectedTerminals, + driveOpenTasksAfterFleetDry, + FLEET_DRY_CONTINUATION_PREFIX, + shouldDriveOpenTasks, + type CollectedWorkerReport, + type FleetDryLane, + type FleetDryMailbox, +} from "./fleet-dry-drive.js"; export { EMPTY_THRASH_STATE, nextThrashState, diff --git a/src/tui/runner-host.test.ts b/src/tui/runner-host.test.ts index fd17b851e..381a43fff 100644 --- a/src/tui/runner-host.test.ts +++ b/src/tui/runner-host.test.ts @@ -133,6 +133,38 @@ describe("observeSessionFromSubAgents", () => { }); }); +describe("mountRunnerHost session bridge", () => { + test("exposes the live session bridge so a system continuation can mark the run busy", async () => { + const harness = await createHarness({ width: 80, height: 24 }); + const host = await mountRunnerHost({ + title: "test", + eventEmitter: new EventEmitter(), + send: () => {}, + interrupt: () => {}, + deliver: () => {}, + providers: {}, + onModelSelect: () => {}, + commands: [], + onCommand: () => {}, + chrome: () => ({ agents: [] }), + subscribeChrome: () => () => {}, + subAgentSessions: () => [], + createRenderer: async () => harness.renderer, + }); + try { + expect(typeof host.bridge.beginSystemContinuation).toBe("function"); + expect(host.shell.session.run).toBe("idle"); + host.bridge.beginSystemContinuation( + "The fleet has gone dry. Remaining open tasks:\n- t1: keep going (todo)", + ); + expect(host.shell.session.run).toBe("busy"); + } finally { + host.dispose(); + harness.destroy(); + } + }); +}); + describe("mountRunnerHost chrome wiring", () => { test("reads the current command catalog on every palette access", async () => { const harness = await createHarness({ width: 80, height: 24 }); diff --git a/src/tui/runner/session.ts b/src/tui/runner/session.ts index 70c680568..bbef5dc6d 100644 --- a/src/tui/runner/session.ts +++ b/src/tui/runner/session.ts @@ -19,7 +19,7 @@ import { import { isCodexProviderName } from "../../config/codex-providers.js"; import { createGlobalSettingsWriter, createLocalSettingsWriter } from "../../mcp/add-server.js"; import { getProcessAdmissionQueue } from "../../subagent/admission.js"; -import { createSubAgentSessionStore } from "../../subagent/index.js"; +import { createSubAgentSessionStore, liveFleetCount } from "../../subagent/index.js"; import { buildPluginDescriptor, createPluginsAdmin, @@ -418,6 +418,7 @@ export async function assembleTUISession( inactivityTimeoutMs: config.inactivityTimeoutMs ?? 750_000, totalTimeoutMs: config.totalTimeoutMs, onTasksChange: (tasks) => emitter.emit("tasks", tasks), + getLiveFleetCount: () => liveFleetCount(subAgentSessions.list()), requestContinuation: () => { state.enqueueAgentDeliver?.(() => liveAgent(state).deliver(buildCompactionContinuationMessage()), diff --git a/src/tui/runner/wiring.ts b/src/tui/runner/wiring.ts index 567bbb5bb..876978644 100644 --- a/src/tui/runner/wiring.ts +++ b/src/tui/runner/wiring.ts @@ -17,6 +17,7 @@ import { loadSentMessages } from "../../session/sent-messages.js"; import { setActiveDisposeHost } from "../../session/active-host.js"; import { createFleetWatch, + driveOpenTasksAfterFleetDry, FLEET_REPORT_SETTLE_MS, FLEET_STALL_POLL_MS, liveFleetCount, @@ -49,6 +50,7 @@ import { resumeTranscriptLoadErrorBlock } from "./exit.js"; import { userInboundMessage } from "./submit.js"; import { hostOf, liveAgent, type RunnerServices, type RunnerState } from "./state.js"; import { LOG_NAMESPACE_ROOT } from "../../branding.js"; +import { buildFleetDryContinuationMessage } from "../../session/runtime-assembly.js"; const tuiLogger = getLogger([LOG_NAMESPACE_ROOT, "tui"]); @@ -58,17 +60,19 @@ export function createFleetWakePublisher( ) { let lastLiveFleet = 0; let suspended = false; - const publish = (): void => { - if (suspended) return; + const publish = (): { previousRunning: number; running: number } => { + if (suspended) return { previousRunning: lastLiveFleet, running: lastLiveFleet }; 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 previousRunning = lastLiveFleet; const fleet = liveFleetCount(lanes); if (fleet !== lastLiveFleet) { lastLiveFleet = fleet; emitter.emit("event", { type: "fleet", running: fleet }); } + return { previousRunning, running: fleet }; }; const withSuspended = (reset: () => void): void => { suspended = true; @@ -153,6 +157,7 @@ export function wirePostStartup( // boundary. The settle timer coalesces a parallel burst into one observation; // the stall poll re-runs so a lane that goes quiet with no further store // event is still announced once. `observeFleet` decides what is worth saying. + const sessionBridge = hostOf(state).bridge; let fleetWatch = createFleetWatch(); const reportFleet = (): void => { const observation = observeFleet(fleetWatch, services.subAgentSessions.list(), Date.now()); @@ -163,7 +168,24 @@ export function wirePostStartup( const fleetWakePublisher = createFleetWakePublisher(services.subAgentSessions, services.emitter); state.withFleetPublicationSuspended = fleetWakePublisher.withSuspended; const unsubscribeFleetReport = services.subAgentSessions.subscribe(() => { - fleetWakePublisher.publish(); + const { previousRunning, running } = fleetWakePublisher.publish(); + const send = state.sendWithAttemptIdentity; + if (send !== undefined) { + driveOpenTasksAfterFleetDry({ + previousRunning, + running, + openTasks: services.directorHolder.instance?.getTasks() ?? [], + parentProcessing: sessionBridge.turn.isProcessing, + mailbox: services.toolset.fleetRecords, + lanes: services.subAgentSessions.list(), + beginSystemContinuation: (prompt) => { + sessionBridge.beginSystemContinuation(prompt); + }, + send: (prompt) => { + void send(buildFleetDryContinuationMessage(prompt)); + }, + }); + } if (fleetSettle !== null) return; fleetSettle = setTimeout(() => { fleetSettle = null; diff --git a/src/tui/runtime-bridge.test.ts b/src/tui/runtime-bridge.test.ts index 7ec4262d8..6339f9e42 100644 --- a/src/tui/runtime-bridge.test.ts +++ b/src/tui/runtime-bridge.test.ts @@ -1399,6 +1399,120 @@ describe("idle-with-fleet (CL-7057)", () => { }); }); +describe("fleet-dry open-task drive (CL-7540)", () => { + function settleToollessTurn(bridge: ReturnType): void { + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "inference.done", data: {} }); + } + + test("dry+open: beginSystemContinuation then fleet-0 keeps the run busy and swallows the prompt", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }); + const port = createRecordingPort(); + const bridge = attachSessionBridge(shell, port); + try { + bridge.submit("dispatch workers", "immediate"); + bridge.handle({ type: "fleet", running: 1 }); + settleToollessTurn(bridge); + expect(shell.session.run).toBe("busy"); + port.clear(); + const prompt = "The fleet has gone dry. Remaining open tasks:\n- t1: keep going (todo)\n"; + const userRowsBefore = shell.streamLog.filter((r) => r.role === "user").length; + bridge.beginSystemContinuation(prompt); + expect(shell.streamLog.filter((r) => r.role === "user").length).toBe(userRowsBefore); + bridge.handle({ type: "fleet", running: 0 }); + expect(shell.session.run).toBe("busy"); + expect(port.calls.some((c) => c.op === "sendImmediate")).toBe(false); + bridge.handle({ + type: "message.received", + data: { message: { content: prompt } }, + }); + expect(shell.streamLog.filter((r) => r.role === "user").length).toBe(userRowsBefore); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("dry+terminal: fleet 1→0 without continuation idles and drains a queued follow-up", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "busy", + }); + const port = createRecordingPort(); + const bridge = attachSessionBridge(shell, port); + try { + bridge.handle({ type: "fleet", running: 1 }); + settleToollessTurn(bridge); + expect(shell.session.run).toBe("busy"); + bridge.submit("when it finishes, summarize", "queue"); + expect(port.calls.some((c) => c.op === "sendImmediate")).toBe(false); + expect(badgeCount(shell.session)).toBe(1); + port.clear(); + bridge.handle({ type: "fleet", running: 0 }); + expect(shell.session.run).toBe("idle"); + expect(badgeCount(shell.session)).toBe(0); + const deliver = port.calls.find((c) => c.op === "deliver"); + expect(deliver).toEqual({ + op: "deliver", + item: expect.objectContaining({ + text: "when it finishes, summarize", + kind: "queue", + }), + }); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("live+open: fleet running 2 without continuation holds busy; Enter still sendImmediate", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: true, + run: "busy", + }); + const port = createRecordingPort(); + const bridge = attachSessionBridge(shell, port); + try { + bridge.handle({ type: "fleet", running: 2 }); + settleToollessTurn(bridge); + expect(shell.session.run).toBe("busy"); + bridge.submit("follow up later", "queue"); + expect(badgeCount(shell.session)).toBe(1); + port.clear(); + shell.prompt.value = "also update the docs"; + shell.prompt.submit(); + expect(port.calls.some((c) => c.op === "enqueue")).toBe(false); + expect(port.calls.some((c) => c.op === "sendImmediate")).toBe(true); + expect(badgeCount(shell.session)).toBe(1); + expect(shell.session.run).toBe("busy"); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); +}); + describe("syncAgentProgress", () => { function taskSession(over: Partial): TaskProgressSession { return { diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index 16a3274f1..1a2f74330 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -213,6 +213,13 @@ export interface SessionBridge { * when the harness event omits `providerId`. */ setInferenceProviderId: (id: string | undefined, displayLabel?: string) => void; + /** + * Mark the run busy for a system-originated continuation (fleet-dry open-task + * drive). Pushes `text` onto pendingEchoes so the inbound `message.received` + * is not painted as a user row. Does not send — the caller uses + * sendWithAttemptIdentity with a system mailbox message. + */ + beginSystemContinuation: (text: string) => void; } const NOOP_PORT: SessionPort = { @@ -1540,6 +1547,16 @@ export function attachSessionBridge( } } }, + beginSystemContinuation: (text) => { + if (bag.disposed) return; + const t = text.trim(); + if (t.length === 0) return; + bag.pendingEchoes.push(t); + shell.session = setRunState(shell.session, "busy"); + bag.turn = turnStateOnSubmit(bag.turn, now()); + paintChrome(shell); + paintPhase(); + }, dispose: () => { flushOpenRow(shell, bag); bag.disposed = true; From 415d2bc4770fd400299b31adb35a46e121a7c9a2 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Wed, 9 Sep 2026 10:26:02 -0700 Subject: [PATCH 2/6] Fire one occupancy shot after a missed fleet-dry edge A last-lane terminal while the parent is still processing dropped the wentDry edge, then settle idled with open tasks. Occupancy now latches that edge and drives once from settle. Send failure no longer consumes mailbox reports. --- docs/ARCHITECTURE.md | 5 +- src/agent/directors/skywalker/package.ts | 2 +- src/session/runtime-assembly.ts | 6 +- src/subagent/agent-fleet.ts | 21 ++--- src/subagent/fleet-dry-drive.test.ts | 104 +++++++++++++++++++++++ src/subagent/fleet-dry-drive.ts | 89 ++++++++++++++----- src/subagent/index.ts | 2 + src/tui/runner/wiring.ts | 39 +++++---- src/tui/runtime-bridge.test.ts | 51 ++++++++++- src/tui/runtime-bridge.ts | 40 ++++++++- 10 files changed, 299 insertions(+), 60 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index fe8635de6..1b1f8f494 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -112,7 +112,10 @@ In TUI chat mode there is no completion gate — the session stays open across t Two directors, selected by role: -- **ChatDirector** (interactive, `src/agent/director.ts`) — Extends `DefaultDirector` with task list tracking, workflow nudges, LSP auto-activation, and multi-turn chat semantics. It never terminates the session: operator declines are surfaced as replies and the reactor stays alive for the next message. Yielding while a live fleet is running is allowed (idle-with-fleet); the open-task nudge does not rewrite that wait/reply. When the fleet goes dry with tasks still todo/doing, the TUI runtime re-enters the parent with collected worker reports rather than settling idle. Auto mode is toggled by CLI flags (`--auto` / `--no-auto`); there is currently no in-session key to toggle it (default on; constrained envelope — workspace writes and unconstrained shell auto-allow; installs, recursive rm, force/uncontained worktree changes, sensitive-path and opaque-wrapper shell still ask; contained non-force `git worktree add`/`remove`/`prune` and `list` auto-allow; shell file-mutation denied). It is not a separate edit/plan mode. +- **ChatDirector** (interactive, `src/agent/director.ts`) — Extends `DefaultDirector` with task list tracking, workflow nudges, LSP auto-activation, and multi-turn chat semantics. It never terminates the session: operator declines are surfaced as replies and the reactor stays alive for the next message. Yielding while a live fleet is running is allowed (idle-with-fleet); the open-task nudge does not rewrite that wait/reply. When the fleet goes dry with tasks still todo/doing, the TUI runtime re-enters the parent with collected worker reports rather than settling idle. + + Auto mode is toggled by CLI flags (`--auto` / `--no-auto`); there is currently no in-session key to toggle it (default on; constrained envelope — workspace writes and unconstrained shell auto-allow; installs, recursive rm, force/uncontained worktree changes, sensitive-path and opaque-wrapper shell still ask; contained non-force `git worktree add`/`remove`/`prune` and `list` auto-allow; shell file-mutation denied). It is not a separate edit/plan mode. + - **SubAgentDirector** (delegated work, `src/subagent/index.ts`) — Drives a dispatched worker until a turn arrives with no tool calls, then replies with the final assistant text and ends the run. A tool-less turn **after tools** completes only with the four-heading envelope (Summary, Findings, Blockers, Paths); a missing envelope nudges once (**incomplete-report**) and a second tool-less turn still without the envelope salvages as **incomplete-report-stop**. Explore/read-only workers that used tools then replied with findings remain normal completes; `requireEvidence` (off by default, set per director) additionally requires at least one read before a tool-less spawn-only reply can complete. Reads done through `run_shell` count as evidence too — `src/subagent/shell-evidence.ts` classifies shell reads (`cat`, `grep`, `sed` without `-i`, …) over the same subject expansion the auto-shell policy uses — but there is no corresponding shell-write evidence or file-write requirement: a run that never touches a file still completes normally once it replies with the envelope. There is no turn budget. Operator/parent cancel after any progress returns a **cancelled** salvage report (partial findings + tool activity) instead of a bare cancel string; cancel before progress still surfaces as cancelled-by-operator. There is no repetition/no-progress/never-acted/never-edited hard stop and no fingerprint-based re-dispatch block — a genuinely stuck worker runs until it completes, stalls, hits an opt-in wall-clock deadline, or is cancelled. `spawn_agent` starts each worker and records it in the caller's fleet mailbox; `wait_agents` collects terminal reports from that mailbox. Wait JSON includes `stop_reason` from the session when present so a salvage that is wait-`done` is not mistaken for a clean complete, and so parent-initiated interrupt (`interrupted`) is not mistaken for operator-cancel (`cancelled`). Deadline salvage prepends an advisory parent hint suggesting continuation plus a longer deadline if more wall-clock time is warranted. Failed and incomplete-report salvage tell the parent to diagnose from the report or error and MAY spawn one successor with a changed brief. A parent-initiated interrupt is a resumable pause: wait unblocks with `stop_reason: interrupted` (often while the session is still running and has no report); the parent should `resume_agent` or re-wait, and must not spawn a successor against a still-live worker. Successor only if that session is no longer resumable. Operator-cancelled salvage asks the parent to synthesize Findings and Paths and wait for the operator instead of auto-starting another specialist. Identical re-dispatch of the same brief stays refused at the prompt / spawn-handoff layer; there is no fingerprint-based re-dispatch hard-block. Deadline hints are advisory only — an identical re-dispatch is still admitted at runtime. Parent hints are prepended on salvage reports returned to the parent. The runtime does not auto-spawn successors. diff --git a/src/agent/directors/skywalker/package.ts b/src/agent/directors/skywalker/package.ts index f0c77f784..46113bd94 100644 --- a/src/agent/directors/skywalker/package.ts +++ b/src/agent/directors/skywalker/package.ts @@ -16,7 +16,7 @@ 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 what happens next before you block. Prefer ending that turn (or calling wait_agents with a short timeout_ms) so Enter can land; do not immediately fuse into a long wait_agents right after spawn. wait_agents later on the targets you need (or omit targets to wait on this session's own uncollected spawns — never a sibling's). list_agents shows that same fleet without blocking. Use mode="all" when you need every target to finish; interrupt_agent unblocks wait_agents immediately. A timeout means still running — do not tight-loop wait_agents hoping for a different answer. Enter mid-run delivers at the next parent tool.boundary — a long parent run_shell or awaiting wait_agents holds those steers. A bare spawn_agent does not. When the fleet goes dry the runtime re-enters with collected reports; do not tight-loop wait_agents. +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 what happens next before you block. Prefer ending that turn (or calling wait_agents with a short timeout_ms) so Enter can land; do not immediately fuse into a long wait_agents right after spawn. wait_agents later on the targets you need (or omit targets to wait on this session's own uncollected spawns — never a sibling's). list_agents shows that same fleet without blocking. Use mode="all" when you need every target to finish; interrupt_agent unblocks wait_agents immediately. A timeout means still running — do not tight-loop wait_agents hoping for a different answer. Enter mid-run delivers at the next parent tool.boundary — a long parent run_shell or awaiting wait_agents holds those steers. 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) diff --git a/src/session/runtime-assembly.ts b/src/session/runtime-assembly.ts index 4cae9d76c..d5cbbc3c9 100644 --- a/src/session/runtime-assembly.ts +++ b/src/session/runtime-assembly.ts @@ -382,8 +382,10 @@ export function buildCompactionContinuationMessage(): InboundMessage { /** * System-originated inbound that re-enters the parent after the fleet goes dry - * with todo/doing tasks still open. No OPERATOR_ORIGINATED_FLAG — this is not - * an operator prompt and must not reset the tool-only loop-protection backstop. + * with todo/doing tasks still open. Not operator input, so no + * OPERATOR_ORIGINATED_FLAG. ChatDirector still resets idle and tool-only + * nudge counters on any message.received — occupancy therefore fires one + * deferred shot per dry edge rather than re-driving on every settle. */ export function buildFleetDryContinuationMessage(text: string): InboundMessage { return { diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index b5a2f7046..b35ee785a 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -94,6 +94,7 @@ import { formatSubAgentSpawnAuthFailureMessage } from "./inference-auth-failure. import { isResolvedProviderFailureError } from "../inference-error-message.js"; import { isSubAgentCancelError } from "./dispose.js"; import { createInterventionLog, type InterventionSink } from "./intervention-log.js"; +import { takeAndProjectMailboxRecord } from "./fleet-dry-drive.js"; const log = getLogger([LOG_NAMESPACE_ROOT, "subagent", "agent-fleet"]); @@ -1398,20 +1399,14 @@ export function createWaitAgentsTool(deps: WaitAgentsDeps): AgentTool { if (isLiveWaitStatus(record.status)) { return { agent_id: id, status: record.status }; } - const taken = deps.fleetRecords.take(id) ?? record; + const projected = takeAndProjectMailboxRecord(deps.fleetRecords, id); + if (projected === undefined) { + return { agent_id: id, status: "unknown" as const }; + } return { - agent_id: id, - status: taken.status, - ...(taken.question !== undefined ? { question: taken.question } : {}), - ...(taken.questionId !== undefined ? { question_id: taken.questionId } : {}), - ...(taken.description !== undefined ? { description: taken.description } : {}), - ...(taken.status !== "failed" && taken.report !== undefined - ? { report: taken.report } - : {}), - ...(taken.error !== undefined ? { error: taken.error } : {}), - ...(taken.stopReason !== undefined ? { stop_reason: taken.stopReason } : {}), - ...(taken.providerFailure === true ? { provider_failure: true } : {}), - ...(taken.hint !== undefined ? { hint: taken.hint } : {}), + ...projected, + ...(record.question !== undefined ? { question: record.question } : {}), + ...(record.questionId !== undefined ? { question_id: record.questionId } : {}), }; }); diff --git a/src/subagent/fleet-dry-drive.test.ts b/src/subagent/fleet-dry-drive.test.ts index 8e077a1f1..6a82fc965 100644 --- a/src/subagent/fleet-dry-drive.test.ts +++ b/src/subagent/fleet-dry-drive.test.ts @@ -61,6 +61,45 @@ describe("shouldDriveOpenTasks", () => { }), ).toBe(false); }); + + test("deferred dry edge fires once when still dry+open and parent is idle", () => { + expect( + shouldDriveOpenTasks({ + previousRunning: 0, + running: 0, + hasOpenTasks: true, + parentProcessing: false, + deferredDryEdge: true, + }), + ).toBe(true); + expect( + shouldDriveOpenTasks({ + previousRunning: 0, + running: 0, + hasOpenTasks: true, + parentProcessing: true, + deferredDryEdge: true, + }), + ).toBe(false); + expect( + shouldDriveOpenTasks({ + previousRunning: 0, + running: 0, + hasOpenTasks: false, + parentProcessing: false, + deferredDryEdge: true, + }), + ).toBe(false); + expect( + shouldDriveOpenTasks({ + previousRunning: 0, + running: 1, + hasOpenTasks: true, + parentProcessing: false, + deferredDryEdge: true, + }), + ).toBe(false); + }); }); describe("buildFleetDryContinuationPrompt", () => { @@ -264,4 +303,69 @@ describe("driveOpenTasksAfterFleetDry", () => { }), ).toBe(false); }); + + test("deferred dry edge after parentProcessing still collects and sends", () => { + const records = new Map([ + ["w1", { status: "done", report: "ok" }], + ]); + const mailbox: FleetDryMailbox = { + ids: () => [...records.keys()], + peek: (id) => records.get(id), + take: (id) => { + const existing = records.get(id); + if (existing === undefined) return undefined; + const taken = { ...existing, collected: true }; + records.set(id, taken); + return taken; + }, + }; + const sent: string[] = []; + const driven = driveOpenTasksAfterFleetDry({ + previousRunning: 0, + running: 0, + deferredDryEdge: true, + openTasks: [openTask], + parentProcessing: false, + mailbox, + lanes: [], + beginSystemContinuation: () => undefined, + send: (prompt) => { + sent.push(prompt); + }, + }); + expect(driven).toBe(true); + expect(sent[0]).toContain("w1"); + expect(records.get("w1")?.collected).toBe(true); + }); + + test("send failure after take leaves reports waitable", () => { + const records = new Map([ + ["w1", { status: "done", report: "ok" }], + ]); + const mailbox: FleetDryMailbox = { + ids: () => [...records.keys()], + peek: (id) => records.get(id), + take: (id) => { + const existing = records.get(id); + if (existing === undefined) return undefined; + const taken = { ...existing, collected: true }; + records.set(id, taken); + return taken; + }, + }; + const driven = driveOpenTasksAfterFleetDry({ + previousRunning: 1, + running: 0, + openTasks: [openTask], + parentProcessing: false, + mailbox, + lanes: [], + beginSystemContinuation: () => undefined, + send: () => { + throw new Error("send failed"); + }, + }); + expect(driven).toBe(false); + expect(records.get("w1")?.collected).not.toBe(true); + }); }); diff --git a/src/subagent/fleet-dry-drive.ts b/src/subagent/fleet-dry-drive.ts index 4da23c687..916ad2963 100644 --- a/src/subagent/fleet-dry-drive.ts +++ b/src/subagent/fleet-dry-drive.ts @@ -1,7 +1,7 @@ /** * Drive the parent back into a turn when the live fleet goes dry while - * todo/doing tasks remain. Pure: the TUI subscriber decides when to call, - * this module decides whether to drive and what to send. + * todo/doing tasks remain. Pure: occupancy (settleRunToIdle) decides when + * to call; this module decides whether to drive and what to send. */ import { hasActiveTasks, type Task } from "../agent/tasks.js"; @@ -50,9 +50,11 @@ export function shouldDriveOpenTasks(input: { running: number; hasOpenTasks: boolean; parentProcessing: boolean; + deferredDryEdge?: boolean; }): boolean { const wentDry = input.running === 0 && input.previousRunning > 0; - return wentDry && input.hasOpenTasks && !input.parentProcessing; + const dryEdge = wentDry || input.deferredDryEdge === true; + return dryEdge && input.running === 0 && input.hasOpenTasks && !input.parentProcessing; } function clipField(text: string | undefined): string | undefined { @@ -61,9 +63,56 @@ function clipField(text: string | undefined): string | undefined { return `${text.slice(0, FLEET_DRY_REPORT_CHARS - 1).trimEnd()}…`; } +export function projectMailboxRecord( + id: string, + taken: FleetDryMailboxRecord, + lane?: FleetDryLane, +): CollectedWorkerReport { + const report = taken.report ?? lane?.report; + const error = taken.error ?? lane?.error; + const description = taken.description ?? lane?.description; + return { + agent_id: id, + status: taken.status, + ...(description !== undefined && description.length > 0 ? { description } : {}), + ...(taken.status !== "failed" && report !== undefined ? { report } : {}), + ...(error !== undefined ? { error } : {}), + ...(taken.hint !== undefined ? { hint: taken.hint } : {}), + ...(taken.providerFailure === true ? { provider_failure: true } : {}), + }; +} + +/** + * Mailbox take plus the wait_agents/fleet-dry projection. Live statuses are + * not collected. Callers that need question fields (wait_agents) spread them + * from the pre-take peek. + */ +export function takeAndProjectMailboxRecord( + mailbox: FleetDryMailbox, + id: string, + lane?: FleetDryLane, +): CollectedWorkerReport | undefined { + const peeked = mailbox.peek(id); + if (peeked === undefined) return undefined; + if (isLiveWaitStatus(peeked.status)) return undefined; + const taken = mailbox.take(id) ?? peeked; + return projectMailboxRecord(id, taken, lane); +} + +function clipCollectedReport(report: CollectedWorkerReport): CollectedWorkerReport { + const clippedReport = clipField(report.report); + const clippedError = clipField(report.error); + return { + ...report, + ...(clippedReport !== undefined ? { report: clippedReport } : {}), + ...(clippedError !== undefined ? { error: clippedError } : {}), + }; +} + export function collectUncollectedTerminals( mailbox: FleetDryMailbox | undefined, lanes: readonly FleetDryLane[], + consume = true, ): CollectedWorkerReport[] { if (mailbox === undefined) return []; const byId = new Map(lanes.map((lane) => [lane.id, lane])); @@ -73,20 +122,11 @@ export function collectUncollectedTerminals( if (peeked === undefined) continue; if (peeked.collected === true) continue; if (isLiveWaitStatus(peeked.status)) continue; - const taken = mailbox.take(id) ?? peeked; - const lane = byId.get(id); - const report = clipField(taken.report ?? lane?.report); - const error = clipField(taken.error ?? lane?.error); - const description = taken.description ?? lane?.description; - reports.push({ - agent_id: id, - status: taken.status, - ...(description !== undefined && description.length > 0 ? { description } : {}), - ...(taken.status !== "failed" && report !== undefined ? { report } : {}), - ...(error !== undefined ? { error } : {}), - ...(taken.hint !== undefined ? { hint: taken.hint } : {}), - ...(taken.providerFailure === true ? { provider_failure: true } : {}), - }); + const projected = consume + ? takeAndProjectMailboxRecord(mailbox, id, byId.get(id)) + : projectMailboxRecord(id, peeked, byId.get(id)); + if (projected === undefined) continue; + reports.push(clipCollectedReport(projected)); } return reports; } @@ -115,6 +155,7 @@ export function driveOpenTasksAfterFleetDry(args: { running: number; openTasks: readonly Task[]; parentProcessing: boolean; + deferredDryEdge?: boolean; mailbox: FleetDryMailbox | undefined; lanes: readonly FleetDryLane[]; beginSystemContinuation: (prompt: string) => void; @@ -127,13 +168,21 @@ export function driveOpenTasksAfterFleetDry(args: { running: args.running, hasOpenTasks: hasActiveTasks(tasks), parentProcessing: args.parentProcessing, + ...(args.deferredDryEdge === true ? { deferredDryEdge: true } : {}), }) ) { return false; } - const reports = collectUncollectedTerminals(args.mailbox, args.lanes); + const reports = collectUncollectedTerminals(args.mailbox, args.lanes, false); const prompt = buildFleetDryContinuationPrompt(tasks, reports); - args.beginSystemContinuation(prompt); - args.send(prompt); + try { + args.beginSystemContinuation(prompt); + args.send(prompt); + } catch { + return false; + } + for (const report of reports) { + args.mailbox?.take(report.agent_id); + } return true; } diff --git a/src/subagent/index.ts b/src/subagent/index.ts index 906f4703b..4736f978a 100644 --- a/src/subagent/index.ts +++ b/src/subagent/index.ts @@ -30,7 +30,9 @@ export { collectUncollectedTerminals, driveOpenTasksAfterFleetDry, FLEET_DRY_CONTINUATION_PREFIX, + projectMailboxRecord, shouldDriveOpenTasks, + takeAndProjectMailboxRecord, type CollectedWorkerReport, type FleetDryLane, type FleetDryMailbox, diff --git a/src/tui/runner/wiring.ts b/src/tui/runner/wiring.ts index 876978644..ac82c54f4 100644 --- a/src/tui/runner/wiring.ts +++ b/src/tui/runner/wiring.ts @@ -167,25 +167,27 @@ export function wirePostStartup( let fleetSettle: ReturnType | null = null; const fleetWakePublisher = createFleetWakePublisher(services.subAgentSessions, services.emitter); state.withFleetPublicationSuspended = fleetWakePublisher.withSuspended; - const unsubscribeFleetReport = services.subAgentSessions.subscribe(() => { - const { previousRunning, running } = fleetWakePublisher.publish(); + sessionBridge.setDryOpenTaskDriver(() => { const send = state.sendWithAttemptIdentity; - if (send !== undefined) { - driveOpenTasksAfterFleetDry({ - previousRunning, - running, - openTasks: services.directorHolder.instance?.getTasks() ?? [], - parentProcessing: sessionBridge.turn.isProcessing, - mailbox: services.toolset.fleetRecords, - lanes: services.subAgentSessions.list(), - beginSystemContinuation: (prompt) => { - sessionBridge.beginSystemContinuation(prompt); - }, - send: (prompt) => { - void send(buildFleetDryContinuationMessage(prompt)); - }, - }); - } + if (send === undefined) return false; + return driveOpenTasksAfterFleetDry({ + previousRunning: 0, + running: 0, + deferredDryEdge: true, + openTasks: services.directorHolder.instance?.getTasks() ?? [], + parentProcessing: false, + mailbox: services.toolset.fleetRecords, + lanes: services.subAgentSessions.list(), + beginSystemContinuation: (prompt) => { + sessionBridge.beginSystemContinuation(prompt); + }, + send: (prompt) => { + void send(buildFleetDryContinuationMessage(prompt)); + }, + }); + }); + const unsubscribeFleetReport = services.subAgentSessions.subscribe(() => { + fleetWakePublisher.publish(); if (fleetSettle !== null) return; fleetSettle = setTimeout(() => { fleetSettle = null; @@ -199,6 +201,7 @@ export function wirePostStartup( clearInterval(fleetStallPoll); if (fleetSettle !== null) clearTimeout(fleetSettle); unsubscribeFleetReport(); + sessionBridge.setDryOpenTaskDriver(undefined); }; // Registered slash-command names only — bare skill/agent words stay unstyled. diff --git a/src/tui/runtime-bridge.test.ts b/src/tui/runtime-bridge.test.ts index 6339f9e42..fa1378617 100644 --- a/src/tui/runtime-bridge.test.ts +++ b/src/tui/runtime-bridge.test.ts @@ -1405,7 +1405,7 @@ describe("fleet-dry open-task drive (CL-7540)", () => { bridge.handle({ type: "inference.done", data: {} }); } - test("dry+open: beginSystemContinuation then fleet-0 keeps the run busy and swallows the prompt", async () => { + test("dry+open: fleet-0 settle drives once, keeps the run busy, and swallows the prompt", async () => { await withTestRenderer( async (h) => { const shell = createAppShell(h.renderer, { @@ -1416,16 +1416,21 @@ describe("fleet-dry open-task drive (CL-7540)", () => { const port = createRecordingPort(); const bridge = attachSessionBridge(shell, port); try { + const prompt = "The fleet has gone dry. Remaining open tasks:\n- t1: keep going (todo)\n"; + let drives = 0; + bridge.setDryOpenTaskDriver(() => { + drives += 1; + bridge.beginSystemContinuation(prompt); + return true; + }); bridge.submit("dispatch workers", "immediate"); bridge.handle({ type: "fleet", running: 1 }); settleToollessTurn(bridge); expect(shell.session.run).toBe("busy"); port.clear(); - const prompt = "The fleet has gone dry. Remaining open tasks:\n- t1: keep going (todo)\n"; const userRowsBefore = shell.streamLog.filter((r) => r.role === "user").length; - bridge.beginSystemContinuation(prompt); - expect(shell.streamLog.filter((r) => r.role === "user").length).toBe(userRowsBefore); bridge.handle({ type: "fleet", running: 0 }); + expect(drives).toBe(1); expect(shell.session.run).toBe("busy"); expect(port.calls.some((c) => c.op === "sendImmediate")).toBe(false); bridge.handle({ @@ -1433,6 +1438,44 @@ describe("fleet-dry open-task drive (CL-7540)", () => { data: { message: { content: prompt } }, }); expect(shell.streamLog.filter((r) => r.role === "user").length).toBe(userRowsBefore); + settleToollessTurn(bridge); + expect(drives).toBe(1); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("wentDry during processing then settle drives after settle, not idle", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }); + const port = createRecordingPort(); + const bridge = attachSessionBridge(shell, port); + try { + const prompt = "The fleet has gone dry. Remaining open tasks:\n- t1: keep going (todo)\n"; + let drives = 0; + bridge.setDryOpenTaskDriver(() => { + drives += 1; + bridge.beginSystemContinuation(prompt); + return true; + }); + bridge.submit("dispatch workers", "immediate"); + bridge.handle({ type: "fleet", running: 1 }); + expect(shell.session.run).toBe("busy"); + bridge.handle({ type: "fleet", running: 0 }); + expect(drives).toBe(0); + expect(shell.session.run).toBe("busy"); + settleToollessTurn(bridge); + expect(drives).toBe(1); + expect(shell.session.run).toBe("busy"); } finally { bridge.dispose(); shell.dispose(); diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index 1a2f74330..250a0a74b 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -220,6 +220,12 @@ export interface SessionBridge { * sendWithAttemptIdentity with a system mailbox message. */ beginSystemContinuation: (text: string) => void; + /** + * Occupancy owner for dry+open continuation. Called once from + * settleRunToIdle when a latched fleet-dry edge is still dry. Return true + * if a continuation was sent (run stays busy). + */ + setDryOpenTaskDriver: (driver: (() => boolean) | undefined) => void; } const NOOP_PORT: SessionPort = { @@ -375,6 +381,14 @@ export interface BridgeBag { * settle path (`settleRunToIdle`) and `gateClosed` re-enter through it. */ flushPendingAskWake: (() => void) | null; + /** + * One deferred occupancy shot for the last live-fleet → 0 edge. Consumed on + * settle so a missed wentDry while the parent was processing still drives + * once, and a later settle cannot loop. + */ + pendingDryOpenDrive: boolean; + /** Occupancy driver: collect+send when settle takes the deferred dry shot. */ + dryOpenTaskDriver: (() => boolean) | undefined; /** Last prompt actually sent — replay source for the quota auto-retry. */ lastSentMessage: string; lastSentOrigin: "composer" | "internal" | null; @@ -908,7 +922,9 @@ function drainLiveSteersAtBoundary(shell: AppShell, bag: BridgeBag): void { * session-idle. A live fleet holds the run busy after the parent turn settles * (idle-with-fleet): Enter upgrades to a new primary turn during the hold and * follow-ups keep waiting; the fleet event landing at zero re-enters here to - * release the hold. + * release the hold. A latched dry edge with open tasks takes one occupancy + * shot here instead of idling, so a wentDry missed while processing cannot + * disagree with settle. */ function settleRunToIdle(shell: AppShell, bag: BridgeBag): void { if (shell.session.run !== "busy") return; @@ -926,6 +942,16 @@ function settleRunToIdle(shell: AppShell, bag: BridgeBag): void { bag.flushPendingAskWake?.(); return; } + if (bag.pendingDryOpenDrive) { + bag.pendingDryOpenDrive = false; + let driven = false; + try { + driven = bag.dryOpenTaskDriver?.() === true; + } catch { + driven = false; + } + if (driven) return; + } shell.session = setRunState(shell.session, "idle"); // Full drain: soft steers first, then follow-ups (drainOrder). drainAtBoundary(shell, bag); @@ -944,7 +970,13 @@ function applyInbound(shell: AppShell, bag: BridgeBag, event: BridgeInboundEvent // queued follow-ups drain now. While the parent is still working the // count just updates — the ordinary turn settle does the draining. if (event.type === "fleet") { + const previous = bag.liveFleet; bag.liveFleet = event.running; + if (event.running > 0) { + bag.pendingDryOpenDrive = false; + } else if (previous > 0) { + bag.pendingDryOpenDrive = true; + } if (event.running === 0 && !bag.turn.isProcessing) { settleRunToIdle(shell, bag); } @@ -1049,6 +1081,8 @@ export function attachSessionBridge( pendingAskWake: new Map(), deliveredAskWake: new Map(), flushPendingAskWake: null, + pendingDryOpenDrive: false, + dryOpenTaskDriver: undefined, lastSentMessage: "", lastSentOrigin: null, quotaFired: false, @@ -1406,6 +1440,7 @@ export function attachSessionBridge( bag.liveFleet = 0; bag.pendingAskWake.clear(); bag.deliveredAskWake.clear(); + bag.pendingDryOpenDrive = false; bag.pendingRowUpdates.clear(); paintChrome(shell); }; @@ -1557,6 +1592,9 @@ export function attachSessionBridge( paintChrome(shell); paintPhase(); }, + setDryOpenTaskDriver: (driver) => { + bag.dryOpenTaskDriver = driver; + }, dispose: () => { flushOpenRow(shell, bag); bag.disposed = true; From 21b60e97c9e56297ebc5bd3bb232cd9cab765784 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Wed, 9 Sep 2026 11:46:43 -0700 Subject: [PATCH 3/6] Hold occupancy continuation through a stale connector reply The occupancy shot re-arms processing during inference.done, so the same turn's connector.reply settled idle and drained queued follow-ups. Hold until the continuation's inference.start. Take mailbox reports only after send actually succeeds on the TUI path. --- src/subagent/fleet-dry-drive.test.ts | 144 ++++++++++++++++++++++++++- src/subagent/fleet-dry-drive.ts | 42 +++++--- src/subagent/index.ts | 13 +-- src/tui/runner/state.ts | 2 +- src/tui/runner/submit.ts | 4 +- src/tui/runner/wiring.ts | 6 +- src/tui/runtime-bridge.test.ts | 11 ++ src/tui/runtime-bridge.ts | 18 +++- 8 files changed, 203 insertions(+), 37 deletions(-) diff --git a/src/subagent/fleet-dry-drive.test.ts b/src/subagent/fleet-dry-drive.test.ts index 6a82fc965..5079908d9 100644 --- a/src/subagent/fleet-dry-drive.test.ts +++ b/src/subagent/fleet-dry-drive.test.ts @@ -72,6 +72,13 @@ describe("shouldDriveOpenTasks", () => { deferredDryEdge: true, }), ).toBe(true); + expect( + shouldDriveOpenTasks({ + hasOpenTasks: true, + parentProcessing: false, + deferredDryEdge: true, + }), + ).toBe(true); expect( shouldDriveOpenTasks({ previousRunning: 0, @@ -158,7 +165,7 @@ describe("collectUncollectedTerminals", () => { }), ).toBe(true); - const reports = collectUncollectedTerminals(mailbox, sessions.list()); + const reports = collectUncollectedTerminals(mailbox, sessions.list(), true); expect(reports.map((r) => r.agent_id).sort()).toEqual(["done", "fail"]); expect(reports.find((r) => r.agent_id === "done")).toEqual({ agent_id: "done", @@ -194,9 +201,11 @@ describe("collectUncollectedTerminals", () => { return taken; }, }; - const reports = collectUncollectedTerminals(mailbox, [ - { id: "ghost", description: "from store", report: "store report" }, - ]); + const reports = collectUncollectedTerminals( + mailbox, + [{ id: "ghost", description: "from store", report: "store report" }], + true, + ); expect(reports).toEqual([ { agent_id: "ghost", @@ -217,10 +226,30 @@ describe("collectUncollectedTerminals", () => { peek: (id) => records.get(id), take: (id) => records.get(id), }; - const reports = collectUncollectedTerminals(mailbox, []); + const reports = collectUncollectedTerminals(mailbox, [], true); expect(reports[0]?.report?.length).toBe(FLEET_DRY_REPORT_CHARS); expect(reports[0]?.report?.endsWith("…")).toBe(true); }); + + test("consume false peeks without take", () => { + const records = new Map([ + ["w1", { status: "done", report: "ok" }], + ]); + const mailbox: FleetDryMailbox = { + ids: () => [...records.keys()], + peek: (id) => records.get(id), + take: (id) => { + const existing = records.get(id); + if (existing === undefined) return undefined; + const taken = { ...existing, collected: true }; + records.set(id, taken); + return taken; + }, + }; + const reports = collectUncollectedTerminals(mailbox, [], false); + expect(reports).toEqual([{ agent_id: "w1", status: "done", report: "ok" }]); + expect(records.get("w1")?.collected).not.toBe(true); + }); }); describe("driveOpenTasksAfterFleetDry", () => { @@ -368,4 +397,109 @@ describe("driveOpenTasksAfterFleetDry", () => { expect(driven).toBe(false); expect(records.get("w1")?.collected).not.toBe(true); }); + + test("TUI sendWithAttemptIdentity rejection leaves mailbox uncollected", async () => { + const records = new Map([ + ["w1", { status: "done", report: "ok" }], + ]); + const mailbox: FleetDryMailbox = { + ids: () => [...records.keys()], + peek: (id) => records.get(id), + take: (id) => { + const existing = records.get(id); + if (existing === undefined) return undefined; + const taken = { ...existing, collected: true }; + records.set(id, taken); + return taken; + }, + }; + const sendWithAttemptIdentity = async (): Promise => { + await Promise.resolve(); + throw new Error("agentProxy.send failed"); + }; + const driven = driveOpenTasksAfterFleetDry({ + deferredDryEdge: true, + openTasks: [openTask], + parentProcessing: false, + mailbox, + lanes: [], + beginSystemContinuation: () => undefined, + send: () => sendWithAttemptIdentity(), + }); + expect(driven).toBe(true); + expect(records.get("w1")?.collected).not.toBe(true); + await Promise.resolve(); + await Promise.resolve(); + expect(records.get("w1")?.collected).not.toBe(true); + }); + + test("TUI sendWithAttemptIdentity false after handleSendFailure leaves mailbox uncollected", async () => { + const records = new Map([ + ["w1", { status: "done", report: "ok" }], + ]); + const mailbox: FleetDryMailbox = { + ids: () => [...records.keys()], + peek: (id) => records.get(id), + take: (id) => { + const existing = records.get(id); + if (existing === undefined) return undefined; + const taken = { ...existing, collected: true }; + records.set(id, taken); + return taken; + }, + }; + const sendWithAttemptIdentity = async (): Promise => { + await Promise.resolve(); + return false; + }; + const driven = driveOpenTasksAfterFleetDry({ + deferredDryEdge: true, + openTasks: [openTask], + parentProcessing: false, + mailbox, + lanes: [], + beginSystemContinuation: () => undefined, + send: () => sendWithAttemptIdentity(), + }); + expect(driven).toBe(true); + await Promise.resolve(); + await Promise.resolve(); + expect(records.get("w1")?.collected).not.toBe(true); + }); + + test("TUI sendWithAttemptIdentity true takes mailbox after send resolves", async () => { + const records = new Map([ + ["w1", { status: "done", report: "ok" }], + ]); + const mailbox: FleetDryMailbox = { + ids: () => [...records.keys()], + peek: (id) => records.get(id), + take: (id) => { + const existing = records.get(id); + if (existing === undefined) return undefined; + const taken = { ...existing, collected: true }; + records.set(id, taken); + return taken; + }, + }; + let resolveSend: ((ok: boolean) => void) | undefined; + const sendWithAttemptIdentity = (): Promise => + new Promise((resolve) => { + resolveSend = resolve; + }); + const driven = driveOpenTasksAfterFleetDry({ + deferredDryEdge: true, + openTasks: [openTask], + parentProcessing: false, + mailbox, + lanes: [], + beginSystemContinuation: () => undefined, + send: () => sendWithAttemptIdentity(), + }); + expect(driven).toBe(true); + expect(records.get("w1")?.collected).not.toBe(true); + resolveSend?.(true); + await Promise.resolve(); + expect(records.get("w1")?.collected).toBe(true); + }); }); diff --git a/src/subagent/fleet-dry-drive.ts b/src/subagent/fleet-dry-drive.ts index 916ad2963..ed7554cce 100644 --- a/src/subagent/fleet-dry-drive.ts +++ b/src/subagent/fleet-dry-drive.ts @@ -46,15 +46,21 @@ export interface CollectedWorkerReport { } export function shouldDriveOpenTasks(input: { - previousRunning: number; - running: number; + previousRunning?: number | undefined; + running?: number | undefined; hasOpenTasks: boolean; parentProcessing: boolean; deferredDryEdge?: boolean; }): boolean { - const wentDry = input.running === 0 && input.previousRunning > 0; + const running = input.running ?? 0; + const previousRunning = input.previousRunning ?? 0; + const wentDry = running === 0 && previousRunning > 0; const dryEdge = wentDry || input.deferredDryEdge === true; - return dryEdge && input.running === 0 && input.hasOpenTasks && !input.parentProcessing; + return dryEdge && running === 0 && input.hasOpenTasks && !input.parentProcessing; +} + +function isPromiseLike(value: unknown): value is Promise { + return typeof value === "object" && value !== null && "then" in value; } function clipField(text: string | undefined): string | undefined { @@ -112,7 +118,7 @@ function clipCollectedReport(report: CollectedWorkerReport): CollectedWorkerRepo export function collectUncollectedTerminals( mailbox: FleetDryMailbox | undefined, lanes: readonly FleetDryLane[], - consume = true, + consume: boolean, ): CollectedWorkerReport[] { if (mailbox === undefined) return []; const byId = new Map(lanes.map((lane) => [lane.id, lane])); @@ -151,15 +157,15 @@ export function buildFleetDryContinuationPrompt( } export function driveOpenTasksAfterFleetDry(args: { - previousRunning: number; - running: number; + previousRunning?: number | undefined; + running?: number | undefined; openTasks: readonly Task[]; parentProcessing: boolean; deferredDryEdge?: boolean; mailbox: FleetDryMailbox | undefined; lanes: readonly FleetDryLane[]; beginSystemContinuation: (prompt: string) => void; - send: (prompt: string) => void; + send: (prompt: string) => unknown; }): boolean { const tasks = [...args.openTasks]; if ( @@ -175,14 +181,26 @@ export function driveOpenTasksAfterFleetDry(args: { } const reports = collectUncollectedTerminals(args.mailbox, args.lanes, false); const prompt = buildFleetDryContinuationPrompt(tasks, reports); + const takeReports = (): void => { + for (const report of reports) { + args.mailbox?.take(report.agent_id); + } + }; try { args.beginSystemContinuation(prompt); - args.send(prompt); + const sent = args.send(prompt); + if (isPromiseLike(sent)) { + void sent.then( + (result) => { + if (result !== false) takeReports(); + }, + () => undefined, + ); + return true; + } + if (sent !== false) takeReports(); } catch { return false; } - for (const report of reports) { - args.mailbox?.take(report.agent_id); - } return true; } diff --git a/src/subagent/index.ts b/src/subagent/index.ts index 4736f978a..7729d62f9 100644 --- a/src/subagent/index.ts +++ b/src/subagent/index.ts @@ -25,18 +25,7 @@ export { type FleetWatch, type PendingAskWake, } from "./fleet-report.js"; -export { - buildFleetDryContinuationPrompt, - collectUncollectedTerminals, - driveOpenTasksAfterFleetDry, - FLEET_DRY_CONTINUATION_PREFIX, - projectMailboxRecord, - shouldDriveOpenTasks, - takeAndProjectMailboxRecord, - type CollectedWorkerReport, - type FleetDryLane, - type FleetDryMailbox, -} from "./fleet-dry-drive.js"; +export { driveOpenTasksAfterFleetDry } from "./fleet-dry-drive.js"; export { EMPTY_THRASH_STATE, nextThrashState, diff --git a/src/tui/runner/state.ts b/src/tui/runner/state.ts index 523680f0b..b09674c8a 100644 --- a/src/tui/runner/state.ts +++ b/src/tui/runner/state.ts @@ -222,7 +222,7 @@ export interface RunnerState { attempt: InferenceAttemptIdentity, providerFailure: ProviderFailureAttempt, ) => void; - sendWithAttemptIdentity?: (message: InboundMessage) => Promise; + sendWithAttemptIdentity?: (message: InboundMessage) => Promise; sendUserPrompt?: (text: string, pending: readonly PendingImageAttachment[]) => Promise; dispatchCommand?: (name: string, args: string) => void; newSession?: () => void; diff --git a/src/tui/runner/submit.ts b/src/tui/runner/submit.ts index edea2a307..ecd1d9f2f 100644 --- a/src/tui/runner/submit.ts +++ b/src/tui/runner/submit.ts @@ -255,7 +255,7 @@ export function createSubmitPath( }; state.handleSendFailure = handleSendFailure; - const sendWithAttemptIdentity = async (message: InboundMessage): Promise => { + const sendWithAttemptIdentity = async (message: InboundMessage): Promise => { const attempt = live.attemptIdentity(); const providerFailure = services.providerFailureAttempts.begin(attempt); try { @@ -265,8 +265,10 @@ export function createSubmitPath( // decision on the correlationId signal channel so the parked run // resumes. await services.approvalResume.handle(result); + return true; } catch (error) { handleSendFailure(error, attempt, providerFailure); + return false; } finally { services.providerFailureAttempts.sendSettled(providerFailure); } diff --git a/src/tui/runner/wiring.ts b/src/tui/runner/wiring.ts index ac82c54f4..af9a20559 100644 --- a/src/tui/runner/wiring.ts +++ b/src/tui/runner/wiring.ts @@ -171,8 +171,6 @@ export function wirePostStartup( const send = state.sendWithAttemptIdentity; if (send === undefined) return false; return driveOpenTasksAfterFleetDry({ - previousRunning: 0, - running: 0, deferredDryEdge: true, openTasks: services.directorHolder.instance?.getTasks() ?? [], parentProcessing: false, @@ -181,9 +179,7 @@ export function wirePostStartup( beginSystemContinuation: (prompt) => { sessionBridge.beginSystemContinuation(prompt); }, - send: (prompt) => { - void send(buildFleetDryContinuationMessage(prompt)); - }, + send: (prompt) => send(buildFleetDryContinuationMessage(prompt)), }); }); const unsubscribeFleetReport = services.subAgentSessions.subscribe(() => { diff --git a/src/tui/runtime-bridge.test.ts b/src/tui/runtime-bridge.test.ts index fa1378617..5fac8b04d 100644 --- a/src/tui/runtime-bridge.test.ts +++ b/src/tui/runtime-bridge.test.ts @@ -1476,6 +1476,17 @@ describe("fleet-dry open-task drive (CL-7540)", () => { settleToollessTurn(bridge); expect(drives).toBe(1); expect(shell.session.run).toBe("busy"); + bridge.submit("when it finishes, summarize", "queue"); + expect(badgeCount(shell.session)).toBe(1); + port.clear(); + bridge.handle({ type: "connector.reply", data: { content: "" } }); + expect(shell.session.run).toBe("busy"); + expect(badgeCount(shell.session)).toBe(1); + expect(port.calls.some((c) => c.op === "deliver")).toBe(false); + settleToollessTurn(bridge); + expect(drives).toBe(1); + expect(shell.session.run).toBe("idle"); + expect(badgeCount(shell.session)).toBe(0); } finally { bridge.dispose(); shell.dispose(); diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index 250a0a74b..8b76aa29a 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -387,6 +387,12 @@ export interface BridgeBag { * once, and a later settle cannot loop. */ pendingDryOpenDrive: boolean; + /** + * beginSystemContinuation re-armed the turn during the previous cycle's + * settle. Late connector.reply from that cycle must not settle this one + * until its own inference.start arrives. + */ + awaitingContinuationInference: boolean; /** Occupancy driver: collect+send when settle takes the deferred dry shot. */ dryOpenTaskDriver: (() => boolean) | undefined; /** Last prompt actually sent — replay source for the quota auto-retry. */ @@ -953,6 +959,7 @@ function settleRunToIdle(shell: AppShell, bag: BridgeBag): void { if (driven) return; } shell.session = setRunState(shell.session, "idle"); + bag.awaitingContinuationInference = false; // Full drain: soft steers first, then follow-ups (drainOrder). drainAtBoundary(shell, bag); bag.flushPendingAskWake?.(); @@ -1082,6 +1089,7 @@ export function attachSessionBridge( deliveredAskWake: new Map(), flushPendingAskWake: null, pendingDryOpenDrive: false, + awaitingContinuationInference: false, dryOpenTaskDriver: undefined, lastSentMessage: "", lastSentOrigin: null, @@ -1254,7 +1262,12 @@ export function attachSessionBridge( const handle = (event: BridgeInboundEvent | ReactorLikeEvent): void => { if (bag.disposed) return; - const settled = noteEvent(event); + if (event.type === "inference.start") { + bag.awaitingContinuationInference = false; + } + const staleContinuationReply = + event.type === "connector.reply" && bag.awaitingContinuationInference; + const settled = staleContinuationReply ? false : noteEvent(event); // Reactor-shaped types always map first (avoids tool.done name collision). if (PRODUCTION_REACTOR_TYPES.has(event.type)) { if (consumePendingEchoEvent(bag, event)) { @@ -1428,6 +1441,7 @@ export function attachSessionBridge( // Clearing the last prompt is what stops the quota loop from replaying a // turn the operator (or the watchdog) deliberately stopped. recordLastSent(null); + bag.awaitingContinuationInference = false; bag.turn = turnStateOnInterrupt(bag.turn, now()); paintPhase(); flushPendingAskWake(); @@ -1441,6 +1455,7 @@ export function attachSessionBridge( bag.pendingAskWake.clear(); bag.deliveredAskWake.clear(); bag.pendingDryOpenDrive = false; + bag.awaitingContinuationInference = false; bag.pendingRowUpdates.clear(); paintChrome(shell); }; @@ -1587,6 +1602,7 @@ export function attachSessionBridge( const t = text.trim(); if (t.length === 0) return; bag.pendingEchoes.push(t); + bag.awaitingContinuationInference = true; shell.session = setRunState(shell.session, "busy"); bag.turn = turnStateOnSubmit(bag.turn, now()); paintChrome(shell); From ec464a58faa3a8acf7339ca90849c147e50f28f4 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Wed, 9 Sep 2026 13:15:51 -0700 Subject: [PATCH 4/6] Re-arm occupancy send failure and stamp lastSentMessage A failed occupancy send consumed the dry-open latch and left the turn running. Abort now re-arms, drains, and resets cadence so a later settle can retry; a successful occupancy stamps lastSentMessage so quota retry resends the continuation. --- src/subagent/fleet-dry-drive.test.ts | 99 +++++++++++++++++ src/subagent/fleet-dry-drive.ts | 15 ++- src/tui/runner/wiring.ts | 3 + src/tui/runtime-bridge.test.ts | 160 +++++++++++++++++++++++++++ src/tui/runtime-bridge.ts | 20 ++++ 5 files changed, 294 insertions(+), 3 deletions(-) diff --git a/src/subagent/fleet-dry-drive.test.ts b/src/subagent/fleet-dry-drive.test.ts index 5079908d9..57598f530 100644 --- a/src/subagent/fleet-dry-drive.test.ts +++ b/src/subagent/fleet-dry-drive.test.ts @@ -502,4 +502,103 @@ describe("driveOpenTasksAfterFleetDry", () => { await Promise.resolve(); expect(records.get("w1")?.collected).toBe(true); }); + + test("sync send false returns false, calls onSendFailure, and leaves mailbox uncollected", () => { + const records = new Map([ + ["w1", { status: "done", report: "ok" }], + ]); + const mailbox: FleetDryMailbox = { + ids: () => [...records.keys()], + peek: (id) => records.get(id), + take: (id) => { + const existing = records.get(id); + if (existing === undefined) return undefined; + const taken = { ...existing, collected: true }; + records.set(id, taken); + return taken; + }, + }; + let failures = 0; + const driven = driveOpenTasksAfterFleetDry({ + previousRunning: 1, + running: 0, + openTasks: [openTask], + parentProcessing: false, + mailbox, + lanes: [], + beginSystemContinuation: () => undefined, + send: () => false, + onSendFailure: () => { + failures += 1; + }, + }); + expect(driven).toBe(false); + expect(failures).toBe(1); + expect(records.get("w1")?.collected).not.toBe(true); + }); + + test("sync send throw calls onSendFailure", () => { + let failures = 0; + const driven = driveOpenTasksAfterFleetDry({ + previousRunning: 1, + running: 0, + openTasks: [openTask], + parentProcessing: false, + mailbox: undefined, + lanes: [], + beginSystemContinuation: () => undefined, + send: () => { + throw new Error("send failed"); + }, + onSendFailure: () => { + failures += 1; + }, + }); + expect(driven).toBe(false); + expect(failures).toBe(1); + }); + + test("TUI send false after handleSendFailure calls onSendFailure", async () => { + let failures = 0; + const driven = driveOpenTasksAfterFleetDry({ + deferredDryEdge: true, + openTasks: [openTask], + parentProcessing: false, + mailbox: undefined, + lanes: [], + beginSystemContinuation: () => undefined, + send: async () => false, + onSendFailure: () => { + failures += 1; + }, + }); + expect(driven).toBe(true); + expect(failures).toBe(0); + await Promise.resolve(); + await Promise.resolve(); + expect(failures).toBe(1); + }); + + test("TUI send rejection calls onSendFailure", async () => { + let failures = 0; + const driven = driveOpenTasksAfterFleetDry({ + deferredDryEdge: true, + openTasks: [openTask], + parentProcessing: false, + mailbox: undefined, + lanes: [], + beginSystemContinuation: () => undefined, + send: async () => { + await Promise.resolve(); + throw new Error("agentProxy.send failed"); + }, + onSendFailure: () => { + failures += 1; + }, + }); + expect(driven).toBe(true); + await Promise.resolve(); + await Promise.resolve(); + expect(failures).toBe(1); + }); }); diff --git a/src/subagent/fleet-dry-drive.ts b/src/subagent/fleet-dry-drive.ts index ed7554cce..1c1a3576d 100644 --- a/src/subagent/fleet-dry-drive.ts +++ b/src/subagent/fleet-dry-drive.ts @@ -166,6 +166,7 @@ export function driveOpenTasksAfterFleetDry(args: { lanes: readonly FleetDryLane[]; beginSystemContinuation: (prompt: string) => void; send: (prompt: string) => unknown; + onSendFailure?: () => void; }): boolean { const tasks = [...args.openTasks]; if ( @@ -186,6 +187,10 @@ export function driveOpenTasksAfterFleetDry(args: { args.mailbox?.take(report.agent_id); } }; + const fail = (): boolean => { + args.onSendFailure?.(); + return false; + }; try { args.beginSystemContinuation(prompt); const sent = args.send(prompt); @@ -193,14 +198,18 @@ export function driveOpenTasksAfterFleetDry(args: { void sent.then( (result) => { if (result !== false) takeReports(); + else args.onSendFailure?.(); + }, + () => { + args.onSendFailure?.(); }, - () => undefined, ); return true; } - if (sent !== false) takeReports(); + if (sent === false) return fail(); + takeReports(); } catch { - return false; + return fail(); } return true; } diff --git a/src/tui/runner/wiring.ts b/src/tui/runner/wiring.ts index af9a20559..752135666 100644 --- a/src/tui/runner/wiring.ts +++ b/src/tui/runner/wiring.ts @@ -180,6 +180,9 @@ export function wirePostStartup( sessionBridge.beginSystemContinuation(prompt); }, send: (prompt) => send(buildFleetDryContinuationMessage(prompt)), + onSendFailure: () => { + sessionBridge.abortSystemContinuation(); + }, }); }); const unsubscribeFleetReport = services.subAgentSessions.subscribe(() => { diff --git a/src/tui/runtime-bridge.test.ts b/src/tui/runtime-bridge.test.ts index 5fac8b04d..e58fbc6da 100644 --- a/src/tui/runtime-bridge.test.ts +++ b/src/tui/runtime-bridge.test.ts @@ -1565,6 +1565,166 @@ describe("fleet-dry open-task drive (CL-7540)", () => { { width: 80, height: 24 }, ); }); + + test("occupancy send failure re-arms, clears continuation hold, and drains follow-ups", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }); + const port = createRecordingPort(); + const bridge = attachSessionBridge(shell, port); + try { + const prompt = "The fleet has gone dry. Remaining open tasks:\n- t1: keep going (todo)\n"; + let drives = 0; + bridge.setDryOpenTaskDriver(() => { + drives += 1; + bridge.beginSystemContinuation(prompt); + if (drives === 1) { + void Promise.resolve().then(() => { + bridge.abortSystemContinuation(); + }); + } + return true; + }); + bridge.submit("dispatch workers", "immediate"); + bridge.handle({ type: "fleet", running: 1 }); + settleToollessTurn(bridge); + expect(shell.session.run).toBe("busy"); + bridge.handle({ type: "fleet", running: 0 }); + expect(drives).toBe(1); + expect(shell.session.run).toBe("busy"); + bridge.submit("when it finishes, summarize", "queue"); + expect(badgeCount(shell.session)).toBe(1); + port.clear(); + await Promise.resolve(); + expect(shell.session.run).toBe("idle"); + expect(badgeCount(shell.session)).toBe(0); + const deliver = port.calls.find((c) => c.op === "deliver"); + expect(deliver).toEqual({ + op: "deliver", + item: expect.objectContaining({ + text: "when it finishes, summarize", + kind: "queue", + }), + }); + bridge.handle({ type: "connector.reply", data: { content: "" } }); + expect(shell.session.run).toBe("idle"); + bridge.submit("continue the remaining work", "immediate"); + expect(shell.session.run).toBe("busy"); + settleToollessTurn(bridge); + expect(drives).toBe(2); + expect(shell.session.run).toBe("busy"); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("occupancy send abort resets the turn without a following reply", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }); + const port = createRecordingPort(); + const nowMs = 0; + let tick: (() => void) | undefined; + const prompt = "The fleet has gone dry. Remaining open tasks:\n- t1: keep going (todo)\n"; + const bridge = attachSessionBridge(shell, port, { + now: () => nowMs, + stallNoticeMs: 400, + stallTimeoutMs: 1_000, + schedule: (fn) => { + tick = fn; + return () => { + tick = undefined; + }; + }, + }); + try { + bridge.setDryOpenTaskDriver(() => { + bridge.beginSystemContinuation(prompt); + void Promise.resolve().then(() => { + bridge.abortSystemContinuation(); + }); + return true; + }); + bridge.submit("dispatch workers", "immediate"); + bridge.handle({ type: "fleet", running: 1 }); + settleToollessTurn(bridge); + bridge.handle({ type: "fleet", running: 0 }); + expect(bridge.turn.isProcessing).toBe(true); + expect(tick).toBeDefined(); + await Promise.resolve(); + expect(bridge.turn.isProcessing).toBe(false); + expect(bridge.turn.awaitingResponse).toBe(false); + expect(bridge.turn.status).not.toBe("running"); + expect(shell.lockupPhase).toBeNull(); + expect(tick).toBeUndefined(); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("occupancy continuation is what quota auto-retry resubmits", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }); + const port = createRecordingPort(); + let nowMs = 0; + let tick: (() => void) | undefined; + const continuation = + "The fleet has gone dry. Remaining open tasks:\n- t1: keep going (todo)\n"; + const bridge = attachSessionBridge(shell, port, { + now: () => nowMs, + schedule: (fn) => { + tick = fn; + return () => { + tick = undefined; + }; + }, + }); + try { + bridge.setDryOpenTaskDriver(() => { + bridge.beginSystemContinuation(continuation); + return true; + }); + bridge.submit("dispatch workers", "immediate"); + bridge.handle({ type: "fleet", running: 1 }); + settleToollessTurn(bridge); + bridge.handle({ type: "fleet", running: 0 }); + port.clear(); + bridge.handle({ + type: "inference.error", + data: { error: { category: "quota_exhausted", retryAfterMs: 1_000 } }, + }); + nowMs += 10_000; + tick?.(); + expect(port.calls).toEqual([{ op: "sendImmediate", text: continuation.trim() }]); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); }); describe("syncAgentProgress", () => { diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index 8b76aa29a..41c1564f0 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -220,6 +220,12 @@ export interface SessionBridge { * sendWithAttemptIdentity with a system mailbox message. */ beginSystemContinuation: (text: string) => void; + /** + * Occupancy send failed after beginSystemContinuation. Re-arm the dry-open + * latch, drop the continuation hold, and idle so follow-ups can drain and a + * later settle can take another occupancy shot. + */ + abortSystemContinuation: () => void; /** * Occupancy owner for dry+open continuation. Called once from * settleRunToIdle when a latched fleet-dry edge is still dry. Return true @@ -1602,12 +1608,26 @@ export function attachSessionBridge( const t = text.trim(); if (t.length === 0) return; bag.pendingEchoes.push(t); + bag.lastSentMessage = t; bag.awaitingContinuationInference = true; shell.session = setRunState(shell.session, "busy"); bag.turn = turnStateOnSubmit(bag.turn, now()); paintChrome(shell); paintPhase(); }, + abortSystemContinuation: () => { + if (bag.disposed) return; + bag.awaitingContinuationInference = false; + bag.pendingDryOpenDrive = true; + bag.lastSentMessage = ""; + flushOpenRow(shell, bag); + bag.turnThinking = null; + shell.inFlightTool = null; + shell.session = setRunState(shell.session, "idle"); + drainAtBoundary(shell, bag); + bag.turn = turnStateOnInterrupt(bag.turn, now()); + paintPhase(); + }, setDryOpenTaskDriver: (driver) => { bag.dryOpenTaskDriver = driver; }, From 2c67f433d281200840ab31ac978edf7a650089ae Mon Sep 17 00:00:00 2001 From: Sawyer Date: Wed, 9 Sep 2026 15:12:40 -0700 Subject: [PATCH 5/6] Drop the occupancy continuation echo when the send aborts beginSystemContinuation queues the occupancy string on pendingEchoes. Abort without a matching message.received left that echo in place, so a later inbound with the same text could be swallowed. Pop that continuation echo on abort without wiping unrelated operator echoes. --- src/tui/runtime-bridge.test.ts | 42 ++++++++++++++++++++++++++++++++++ src/tui/runtime-bridge.ts | 15 +++++++++++- 2 files changed, 56 insertions(+), 1 deletion(-) diff --git a/src/tui/runtime-bridge.test.ts b/src/tui/runtime-bridge.test.ts index e58fbc6da..9d710f662 100644 --- a/src/tui/runtime-bridge.test.ts +++ b/src/tui/runtime-bridge.test.ts @@ -1626,6 +1626,48 @@ describe("fleet-dry open-task drive (CL-7540)", () => { ); }); + test("occupancy send abort drops the continuation echo so a later matching inbound paints", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }); + const port = createRecordingPort(); + const bridge = attachSessionBridge(shell, port); + try { + const occupancy = + "The fleet has gone dry. Remaining open tasks:\n- t1: keep going (todo)\n"; + const operator = "dispatch workers"; + bridge.submit(operator, "immediate"); + const userRowsAfterSubmit = shell.streamLog.filter((r) => r.role === "user").length; + bridge.beginSystemContinuation(occupancy); + bridge.abortSystemContinuation(); + expect(shell.session.run).toBe("idle"); + + bridge.handle({ + type: "message.received", + data: { message: { content: operator } }, + }); + expect(shell.streamLog.filter((r) => r.role === "user").length).toBe(userRowsAfterSubmit); + + bridge.handle({ + type: "message.received", + data: { message: { content: occupancy } }, + }); + expect(shell.streamLog.filter((r) => r.role === "user").length).toBe( + userRowsAfterSubmit + 1, + ); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + test("occupancy send abort resets the turn without a following reply", async () => { await withTestRenderer( async (h) => { diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index 41c1564f0..217965b75 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -221,7 +221,8 @@ export interface SessionBridge { */ beginSystemContinuation: (text: string) => void; /** - * Occupancy send failed after beginSystemContinuation. Re-arm the dry-open + * Occupancy send failed after beginSystemContinuation. Drop the occupancy + * echo so a later matching inbound is not swallowed, re-arm the dry-open * latch, drop the continuation hold, and idle so follow-ups can drain and a * later settle can take another occupancy shot. */ @@ -1617,6 +1618,18 @@ export function attachSessionBridge( }, abortSystemContinuation: () => { if (bag.disposed) return; + if (bag.awaitingContinuationInference) { + const occupancy = bag.lastSentMessage; + if (occupancy.length > 0) { + const last = bag.pendingEchoes.length - 1; + if (last >= 0 && bag.pendingEchoes[last] === occupancy) { + bag.pendingEchoes.pop(); + } else { + const index = bag.pendingEchoes.lastIndexOf(occupancy); + if (index !== -1) bag.pendingEchoes.splice(index, 1); + } + } + } bag.awaitingContinuationInference = false; bag.pendingDryOpenDrive = true; bag.lastSentMessage = ""; From 459fff42b32265e3b7e0c9a45ff2c855094f2159 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Wed, 9 Sep 2026 16:32:00 -0700 Subject: [PATCH 6/6] Keep stop_reason on collected wait_agents projections The occupancy collector is the wait_agents JSON shape. Dropping stopReason there made interrupt and incomplete-report waits look clean after collection. --- src/subagent/fleet-dry-drive.test.ts | 19 +++++++++++++++++++ src/subagent/fleet-dry-drive.ts | 3 +++ 2 files changed, 22 insertions(+) diff --git a/src/subagent/fleet-dry-drive.test.ts b/src/subagent/fleet-dry-drive.test.ts index 57598f530..500821ec8 100644 --- a/src/subagent/fleet-dry-drive.test.ts +++ b/src/subagent/fleet-dry-drive.test.ts @@ -217,6 +217,25 @@ describe("collectUncollectedTerminals", () => { expect(records.get("ghost")?.collected).toBe(true); }); + test("projects mailbox stopReason as stop_reason", () => { + const records = new Map([ + ["w1", { status: "interrupted", report: "salvage", stopReason: "interrupted" }], + ]); + const mailbox: FleetDryMailbox = { + ids: () => [...records.keys()], + peek: (id) => records.get(id), + take: (id) => records.get(id), + }; + expect(collectUncollectedTerminals(mailbox, [], true)).toEqual([ + { + agent_id: "w1", + status: "interrupted", + report: "salvage", + stop_reason: "interrupted", + }, + ]); + }); + test("clips oversized reports", () => { const records = new Map([ ["big", { status: "done", report: "x".repeat(FLEET_DRY_REPORT_CHARS + 40) }], diff --git a/src/subagent/fleet-dry-drive.ts b/src/subagent/fleet-dry-drive.ts index 1c1a3576d..9cf89ac8f 100644 --- a/src/subagent/fleet-dry-drive.ts +++ b/src/subagent/fleet-dry-drive.ts @@ -20,6 +20,7 @@ export interface FleetDryMailboxRecord { readonly description?: string; readonly hint?: string; readonly providerFailure?: true; + readonly stopReason?: string; } export interface FleetDryMailbox { @@ -43,6 +44,7 @@ export interface CollectedWorkerReport { error?: string; hint?: string; provider_failure?: true; + stop_reason?: string; } export function shouldDriveOpenTasks(input: { @@ -85,6 +87,7 @@ export function projectMailboxRecord( ...(error !== undefined ? { error } : {}), ...(taken.hint !== undefined ? { hint: taken.hint } : {}), ...(taken.providerFailure === true ? { provider_failure: true } : {}), + ...(taken.stopReason !== undefined ? { stop_reason: taken.stopReason } : {}), }; }