diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 5dd0fdb9a..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. 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..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. +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/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..d5cbbc3c9 100644 --- a/src/session/runtime-assembly.ts +++ b/src/session/runtime-assembly.ts @@ -379,3 +379,25 @@ 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. 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 { + 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/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 new file mode 100644 index 000000000..500821ec8 --- /dev/null +++ b/src/subagent/fleet-dry-drive.test.ts @@ -0,0 +1,623 @@ +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); + }); + + 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({ + 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", () => { + 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(), true); + 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" }], + true, + ); + expect(reports).toEqual([ + { + agent_id: "ghost", + status: "done", + description: "from store", + report: "store report", + }, + ]); + 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) }], + ]); + const mailbox: FleetDryMailbox = { + ids: () => [...records.keys()], + peek: (id) => records.get(id), + take: (id) => records.get(id), + }; + 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", () => { + 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); + }); + + 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); + }); + + 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); + }); + + 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 new file mode 100644 index 000000000..9cf89ac8f --- /dev/null +++ b/src/subagent/fleet-dry-drive.ts @@ -0,0 +1,218 @@ +/** + * Drive the parent back into a turn when the live fleet goes dry while + * 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"; +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; + readonly stopReason?: string; +} + +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; + stop_reason?: string; +} + +export function shouldDriveOpenTasks(input: { + previousRunning?: number | undefined; + running?: number | undefined; + hasOpenTasks: boolean; + parentProcessing: boolean; + deferredDryEdge?: boolean; +}): boolean { + const running = input.running ?? 0; + const previousRunning = input.previousRunning ?? 0; + const wentDry = running === 0 && previousRunning > 0; + const dryEdge = wentDry || input.deferredDryEdge === true; + 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 { + 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 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 } : {}), + ...(taken.stopReason !== undefined ? { stop_reason: taken.stopReason } : {}), + }; +} + +/** + * 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: boolean, +): 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 projected = consume + ? takeAndProjectMailboxRecord(mailbox, id, byId.get(id)) + : projectMailboxRecord(id, peeked, byId.get(id)); + if (projected === undefined) continue; + reports.push(clipCollectedReport(projected)); + } + 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 | undefined; + running?: number | undefined; + openTasks: readonly Task[]; + parentProcessing: boolean; + deferredDryEdge?: boolean; + mailbox: FleetDryMailbox | undefined; + lanes: readonly FleetDryLane[]; + beginSystemContinuation: (prompt: string) => void; + send: (prompt: string) => unknown; + onSendFailure?: () => void; +}): boolean { + const tasks = [...args.openTasks]; + if ( + !shouldDriveOpenTasks({ + previousRunning: args.previousRunning, + running: args.running, + hasOpenTasks: hasActiveTasks(tasks), + parentProcessing: args.parentProcessing, + ...(args.deferredDryEdge === true ? { deferredDryEdge: true } : {}), + }) + ) { + return false; + } + 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); + } + }; + const fail = (): boolean => { + args.onSendFailure?.(); + return false; + }; + try { + args.beginSystemContinuation(prompt); + const sent = args.send(prompt); + if (isPromiseLike(sent)) { + void sent.then( + (result) => { + if (result !== false) takeReports(); + else args.onSendFailure?.(); + }, + () => { + args.onSendFailure?.(); + }, + ); + return true; + } + if (sent === false) return fail(); + takeReports(); + } catch { + return fail(); + } + return true; +} diff --git a/src/subagent/index.ts b/src/subagent/index.ts index 22393f125..7729d62f9 100644 --- a/src/subagent/index.ts +++ b/src/subagent/index.ts @@ -25,6 +25,7 @@ export { type FleetWatch, type PendingAskWake, } from "./fleet-report.js"; +export { driveOpenTasksAfterFleetDry } 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/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 567bbb5bb..752135666 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()); @@ -162,6 +167,24 @@ export function wirePostStartup( let fleetSettle: ReturnType | null = null; const fleetWakePublisher = createFleetWakePublisher(services.subAgentSessions, services.emitter); state.withFleetPublicationSuspended = fleetWakePublisher.withSuspended; + sessionBridge.setDryOpenTaskDriver(() => { + const send = state.sendWithAttemptIdentity; + if (send === undefined) return false; + return driveOpenTasksAfterFleetDry({ + deferredDryEdge: true, + openTasks: services.directorHolder.instance?.getTasks() ?? [], + parentProcessing: false, + mailbox: services.toolset.fleetRecords, + lanes: services.subAgentSessions.list(), + beginSystemContinuation: (prompt) => { + sessionBridge.beginSystemContinuation(prompt); + }, + send: (prompt) => send(buildFleetDryContinuationMessage(prompt)), + onSendFailure: () => { + sessionBridge.abortSystemContinuation(); + }, + }); + }); const unsubscribeFleetReport = services.subAgentSessions.subscribe(() => { fleetWakePublisher.publish(); if (fleetSettle !== null) return; @@ -177,6 +200,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 7ec4262d8..9d710f662 100644 --- a/src/tui/runtime-bridge.test.ts +++ b/src/tui/runtime-bridge.test.ts @@ -1399,6 +1399,376 @@ 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: fleet-0 settle drives once, 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 { + 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 userRowsBefore = shell.streamLog.filter((r) => r.role === "user").length; + 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({ + type: "message.received", + 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"); + 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(); + } + }, + { 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 }, + ); + }); + + 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 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) => { + 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", () => { function taskSession(over: Partial): TaskProgressSession { return { diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index 16a3274f1..217965b75 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -213,6 +213,26 @@ 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; + /** + * 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. + */ + abortSystemContinuation: () => 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 = { @@ -368,6 +388,20 @@ 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; + /** + * 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. */ lastSentMessage: string; lastSentOrigin: "composer" | "internal" | null; @@ -901,7 +935,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; @@ -919,7 +955,18 @@ 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"); + bag.awaitingContinuationInference = false; // Full drain: soft steers first, then follow-ups (drainOrder). drainAtBoundary(shell, bag); bag.flushPendingAskWake?.(); @@ -937,7 +984,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); } @@ -1042,6 +1095,9 @@ export function attachSessionBridge( pendingAskWake: new Map(), deliveredAskWake: new Map(), flushPendingAskWake: null, + pendingDryOpenDrive: false, + awaitingContinuationInference: false, + dryOpenTaskDriver: undefined, lastSentMessage: "", lastSentOrigin: null, quotaFired: false, @@ -1213,7 +1269,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)) { @@ -1387,6 +1448,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(); @@ -1399,6 +1461,8 @@ export function attachSessionBridge( bag.liveFleet = 0; bag.pendingAskWake.clear(); bag.deliveredAskWake.clear(); + bag.pendingDryOpenDrive = false; + bag.awaitingContinuationInference = false; bag.pendingRowUpdates.clear(); paintChrome(shell); }; @@ -1540,6 +1604,46 @@ export function attachSessionBridge( } } }, + beginSystemContinuation: (text) => { + if (bag.disposed) return; + 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; + 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 = ""; + 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; + }, dispose: () => { flushOpenRow(shell, bag); bag.disposed = true;