diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 8b476ef4d..603a1afe8 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -211,6 +211,8 @@ The **`spawn_agent`** tool starts a fleet agent on a separate inference source ( Workers ask the spawning parent with **`ask_director`** (not the human). That parks a question while the worker stays `running`. **`wait_agents`** returns `awaiting_director` with a question payload — that is not terminal. The parent answers with **`send_input`**, then **`wait_agents`** again. Escalate to the human with **`ask_operator`** only when the parent cannot resolve it. +When the parent TUI is not blocked in `wait_agents`, the runner publishes an authoritative snapshot of currently pending top-level questions on each store notification, including empty snapshots before fleet-count updates. During synchronous session rotation, a runner-owned barrier suppresses both publications before delivery-generation invalidation, transcript clearing, and worker cancellation; successful reset reconciles a fresh snapshot before resuming asynchronous backend rebuild. The bridge drops resolved, cancelled, replaced, terminal, and removed asks and delivers each session/question identity once while pending. A coalesced wake starts only when the parent is not processing and every operator gate is closed, including parent-idle fleet holds where the shell stays busy. Worker gates do not manufacture parent processing. Replies use `send_input`'s `target` field with the worker session ID, never its shared catalog ID. Synthetic wakes use `SessionPort.deliver` through queued-delivery's idle-send path without entering the user follow-up queue or composer `/feedback` capture. + When profiles exist (local `.agents/agents/` and/or enabled **`kind: "agent"`** plugins, including **data-only** markdown plugins with no `index.ts`), the chat model also receives **`search_agents`** — a lexical index over profile id, description, and role text so the model can discover ids before calling `spawn_agent(agent=...)`. Results include each match's full loaded system prompt / body so the parent can inspect plugin or Claude marketplace agents without `read_file` on paths outside the session cwd (path-escape blocks those roots by design; writes remain blocked). `spawn_agent` and `search_agents` are core tools on the primary session. Built-in directors with `spawn.maySpawn` may themselves call `spawn_agent` (one hop only): nested dispatch installs the mailbox-scoped fleet verbs (`spawn_agent`, `wait_agents`, `list_agents`, …) with `allowOrchestrator: false` so the tree bottoms out. Profile-sourced `orchestrator: true` is rejected before a session starts because it has no trusted tier/authority semantics today. Fleet discovery (`search_agents`) stays Tier 1 only. Unknown `agent` ids fail closed. diff --git a/docs/TUI.md b/docs/TUI.md index d44e42db9..bfcbb451f 100644 --- a/docs/TUI.md +++ b/docs/TUI.md @@ -221,6 +221,14 @@ status / current tool) — Amp/Codex-style lanes without a FLEET header board: ``` An `ask_director` lane stays live and reads as waiting on the director, not stalled. +The runner snapshots currently pending root-worker questions, dropping resolved, +cancelled, replaced, terminal, or removed asks before delivery. It sends one +coalesced wake when the parent is not processing and all operator gates are closed, +even while live workers hold the shell busy. Replies use `send_input`'s `target` +field with the worker's session ID, not its shared catalog ID. Each session/question identity is +delivered once while pending; the strip never re-delivers it. Synthetic wakes use +the idle delivery path, bypassing composer `/feedback` capture and leaving queued +user follow-ups untouched. `formatChromeZones` → `formatAgentsPanel` owns that paint. Geometry stays stack-only (`layoutMode: "stack"`, `railWidth: 0`); the zone max is diff --git a/src/subagent/fleet-report.ask-wake.test.ts b/src/subagent/fleet-report.ask-wake.test.ts new file mode 100644 index 000000000..a08ba5354 --- /dev/null +++ b/src/subagent/fleet-report.ask-wake.test.ts @@ -0,0 +1,83 @@ +import { describe, expect, test } from "bun:test"; +import { pendingAskSnapshot, pendingAskWakeText, type FleetLane } from "./fleet-report.js"; + +function lane(overrides: Partial & { id: string }): FleetLane { + return { + description: overrides.id, + status: "running", + startedAt: 0, + lastActivityAt: 0, + currentToolName: null, + currentToolPreview: null, + currentToolStartedAt: null, + ...overrides, + }; +} + +describe("pendingAskSnapshot", () => { + test("repeated calls return complete identical snapshots for distinct sessions sharing a catalog", () => { + const lanes = [ + lane({ id: "a1", agentId: "builder", description: "Build the thing" }), + lane({ id: "a2", agentId: "builder" }), + ]; + const peek = () => ({ question: "Which port?", questionId: "q1" }); + const expected = lanes.map((worker) => ({ + sessionId: worker.id, + agentId: "builder", + description: worker.description, + question: "Which port?", + questionId: "q1", + })); + expect(pendingAskSnapshot(lanes, peek)).toEqual(expected); + expect(pendingAskSnapshot(lanes, peek)).toEqual(expected); + }); + + test("resolution, removal and replacement are reflected without prior watch state", () => { + const lanes = [lane({ id: "a1" })]; + const asks = new Map([["a1", { question: "A?", questionId: "q1" }]]); + const peek = (id: string) => asks.get(id); + expect(pendingAskSnapshot(lanes, peek)[0]?.questionId).toBe("q1"); + expect(pendingAskSnapshot([], peek)).toEqual([]); + asks.clear(); + expect(pendingAskSnapshot(lanes, peek)).toEqual([]); + asks.set("a1", { question: "B?", questionId: "q2" }); + expect(pendingAskSnapshot(lanes, peek)[0]).toMatchObject({ + sessionId: "a1", + question: "B?", + questionId: "q2", + }); + }); + + test("only running root workers with a pending question are included", () => { + const lanes = [ + lane({ id: "root" }), + lane({ id: "child", parentSessionId: "orchestrator" }), + lane({ id: "done", status: "done" }), + lane({ id: "cancelled", status: "cancelled" }), + lane({ id: "no-ask" }), + ]; + const asks = pendingAskSnapshot(lanes, (id) => + id === "no-ask" ? undefined : { question: "Q?", questionId: "q1" }, + ); + expect(asks.map((ask) => ask.sessionId)).toEqual(["root"]); + }); +}); + +describe("pendingAskWakeText", () => { + test("names agent, description, question and question id, and routes to send_input", () => { + const text = pendingAskWakeText({ + sessionId: "a1", + agentId: "builder", + description: "Build the thing", + question: "Which port?", + questionId: "q1", + }); + expect(text).toContain("builder"); + expect(text).toContain("Build the thing"); + expect(text).toContain("Which port?"); + expect(text).toContain("q1"); + expect(text).toContain("send_input"); + expect(text).toContain("using target a1"); + expect(text.toLowerCase()).toContain("worker"); + }); +}); diff --git a/src/subagent/fleet-report.ts b/src/subagent/fleet-report.ts index 50b60816c..c02d4c4a6 100644 --- a/src/subagent/fleet-report.ts +++ b/src/subagent/fleet-report.ts @@ -29,6 +29,10 @@ export interface FleetLane { readonly error?: string; /** Machine-readable forced-stop reason (see SubAgentSession.stopReason). */ readonly stopReason?: string; + /** Catalog agent id (SubAgentSession.agentId); the wake message names it. */ + readonly agentId?: string; + /** Set on nested (one-hop) dispatches; such asks never wake the root. */ + readonly parentSessionId?: string; } interface LaneMark { @@ -117,6 +121,54 @@ export function liveFleetCount(lanes: readonly FleetLane[]): number { return lanes.filter((lane) => lane.status === "running").length; } +/** + * One parked ask_director question. Replies target the unique `sessionId`; + * `agentId` is only the descriptive catalog identity shared by workers. + */ +export interface PendingAskWake { + readonly sessionId: string; + readonly agentId: string; + readonly description: string; + readonly question: string; + readonly questionId: string; +} + +/** Nested orchestrators own their children's questions; only root workers wake the TUI. */ +export function pendingAskSnapshot( + lanes: readonly FleetLane[], + peekAsk: (sessionId: string) => { question: string; questionId: string } | undefined, +): readonly PendingAskWake[] { + const asks: PendingAskWake[] = []; + for (const lane of lanes) { + if (lane.parentSessionId !== undefined || lane.status !== "running") continue; + const ask = peekAsk(lane.id); + if (ask === undefined) continue; + asks.push({ + sessionId: lane.id, + agentId: lane.agentId ?? lane.id, + description: lane.description, + question: ask.question, + questionId: ask.questionId, + }); + } + return asks; +} + +/** + * The wake turn text. It must read as the worker's question reaching the + * parent, not as the operator being asked — the parent answers via + * send_input itself and only escalates when it genuinely cannot. + */ +export function pendingAskWakeText(wake: PendingAskWake): string { + return [ + `ask_director wake — worker ${wake.agentId} (${wake.description}) parked question ${wake.questionId} while this session was not collecting:`, + "", + wake.question, + "", + `The worker — not the operator — raised this. Answer it with send_input (soft) using target ${wake.sessionId}; do not relay to the operator unless it genuinely needs them.`, + ].join("\n"); +} + type Change = | { readonly kind: "dispatched"; readonly line: string } | { readonly kind: "done"; readonly line: string } diff --git a/src/subagent/index.ts b/src/subagent/index.ts index bcdc1b861..22393f125 100644 --- a/src/subagent/index.ts +++ b/src/subagent/index.ts @@ -18,9 +18,12 @@ export { FLEET_STALL_POLL_MS, liveFleetCount, observeFleet, + pendingAskSnapshot, + pendingAskWakeText, type FleetLane, type FleetObservation, type FleetWatch, + type PendingAskWake, } from "./fleet-report.js"; export { EMPTY_THRASH_STATE, diff --git a/src/tui/agent-ask-wake.test.ts b/src/tui/agent-ask-wake.test.ts new file mode 100644 index 000000000..a7fcac888 --- /dev/null +++ b/src/tui/agent-ask-wake.test.ts @@ -0,0 +1,381 @@ +import { describe, expect, test } from "bun:test"; +import { attachSessionBridge } from "./runtime-bridge"; +import { createLiveSessionPort } from "./live-session-port"; +import { createAppShell } from "./shell/index"; +import { withTestRenderer } from "./harness"; +import type { PendingAskWake } from "../subagent/fleet-report.js"; +import { classifySubmission, createSubmitHandler } from "./runner/submit.js"; +import { routeQueuedDelivery } from "./queued-delivery.js"; +import { + armFeedbackCapture, + cancelFeedbackCapture, + isFeedbackCapturePending, + resetFeedbackStateForTests, +} from "../telemetry/feedback.js"; + +function wake(id: string, questionId: string): PendingAskWake { + return { + sessionId: id, + agentId: "builder", + description: `worker ${id}`, + question: "Which port?", + questionId, + }; +} + +async function withWakeBridge( + run: (bridge: ReturnType, sends: string[]) => void, +) { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }); + const sends: string[] = []; + const send = (text: string) => { + sends.push(text); + }; + const bridge = attachSessionBridge( + shell, + createLiveSessionPort({ send, deliver: send, interrupt: () => {} }), + ); + try { + run(bridge, sends); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); +} + +for (const action of ["retry", "interrupt", "reset", "dispose", "composer", "ordinary"] as const) { + test(`quota replay preserves submission origin (${action})`, async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }); + const sends: string[] = []; + const composerSends: string[] = []; + const feedback: string[] = []; + let cancellations = 0; + let nowMs = 0; + let tick = () => {}; + const submit = createSubmitHandler({ + dispatchCommand: () => {}, + sendPrompt: (text) => { + composerSends.push(text); + sends.push(text); + }, + isFeedbackCapturePending, + onFeedbackText: (text) => { + feedback.push(text); + cancelFeedbackCapture(); + return "Thanks"; + }, + cancelFeedbackCapture: () => { + cancellations++; + cancelFeedbackCapture(); + }, + }); + const port = createLiveSessionPort({ + send: submit, + classifySubmit: (text) => + classifySubmission(text, { + feedbackPending: isFeedbackCapturePending(), + feedbackCaptureEnabled: true, + }), + interrupt: () => {}, + deliver: routeQueuedDelivery({ + send: (text) => { + sends.push(text); + }, + deliverSteer: () => { + throw new Error("quota replay must not live-inject"); + }, + parentCycleLive: () => bridge.parentCycleLive, + }), + }); + const bridge = attachSessionBridge(shell, port, { + now: () => nowMs, + schedule: (fn) => { + tick = fn; + // Retain the callback to exercise even a stale timer after disposal. + return () => {}; + }, + }); + try { + if (action === "ordinary") { + bridge.submit("operator prompt", "immediate"); + } else { + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "fleet", running: 1 }); + bridge.submit("held first", "queue"); + bridge.submit("held second", "queue"); + bridge.handle({ type: "inference.done", data: {} }); + if (action !== "composer") armFeedbackCapture(); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + if (action === "composer") bridge.submit("operator prompt", "immediate"); + } + const queued = shell.session.items; + const before = sends.length; + const replay = sends.at(-1); + bridge.handle({ + type: "inference.error", + data: { error: { category: "quota_exhausted", retryAfterMs: 1000 } }, + }); + if (action === "interrupt") bridge.interrupt(); + if (action === "reset") bridge.clearQueuedDelivery(); + if (action === "dispose") bridge.dispose(); + const afterCleanup = sends.length; + nowMs = 999; + tick(); + expect(sends).toHaveLength(afterCleanup); + nowMs = 1000; + tick(); + tick(); + if (action === "interrupt" || action === "reset" || action === "dispose") { + expect(sends).toHaveLength(afterCleanup); + expect(feedback).toEqual([]); + } else { + expect(sends).toHaveLength(before + 1); + expect(sends.at(-1)).toBe(replay); + expect(shell.session.items).toBe(queued); + if (action === "retry") { + expect(sends[0]).toBe(sends[1]); + expect(composerSends).toEqual([]); + expect(feedback).toEqual([]); + expect(cancellations).toBe(0); + expect(isFeedbackCapturePending()).toBe(true); + expect(queued.map((item) => item.text)).toEqual(["held first", "held second"]); + bridge.submit("actual feedback", "immediate"); + expect(feedback).toEqual(["actual feedback"]); + expect(sends).toHaveLength(2); + } else { + expect(composerSends).toEqual(["operator prompt", "operator prompt"]); + } + } + } finally { + resetFeedbackStateForTests(); + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); +} + +describe("agent ask wake delivery", () => { + test("synthetic wake bypasses armed feedback and leaves user followups held", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + }); + const sends: string[] = []; + const feedback: string[] = []; + let cancellations = 0; + const submit = createSubmitHandler({ + dispatchCommand: () => {}, + sendPrompt: (text) => { + sends.push(text); + }, + isFeedbackCapturePending, + onFeedbackText: (text) => { + feedback.push(text); + cancelFeedbackCapture(); + return "Thanks"; + }, + cancelFeedbackCapture: () => { + cancellations++; + cancelFeedbackCapture(); + }, + }); + const port = createLiveSessionPort({ + send: submit, + classifySubmit: (text) => + classifySubmission(text, { + feedbackPending: isFeedbackCapturePending(), + feedbackCaptureEnabled: true, + }), + interrupt: () => {}, + deliver: routeQueuedDelivery({ + send: (text) => { + sends.push(text); + // Real delivery can synchronously settle and notify the bridge again. + bridge.handle({ type: "inference.done", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + }, + deliverSteer: () => { + throw new Error("wake must not live-inject"); + }, + parentCycleLive: () => bridge.parentCycleLive, + }), + }); + const bridge = attachSessionBridge(shell, port); + try { + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "fleet", running: 1 }); + bridge.submit("held followup", "queue"); + bridge.handle({ type: "inference.done", data: {} }); + const held = shell.session; + armFeedbackCapture(); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toHaveLength(1); + expect(sends[0]).toContain("q1"); + expect(feedback).toEqual([]); + expect(cancellations).toBe(0); + expect(isFeedbackCapturePending()).toBe(true); + expect(shell.session.items).toEqual(held.items); + expect(held.items).toHaveLength(1); + expect(bridge.turn.isProcessing).toBe(false); + bridge.submit("actual feedback", "immediate"); + expect(feedback).toEqual(["actual feedback"]); + expect(sends).toHaveLength(1); + } finally { + resetFeedbackStateForTests(); + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("resolved snapshots remove deferred questions", async () => { + await withWakeBridge((bridge, sends) => { + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + bridge.handle({ type: "agent-ask", asks: [] }); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toEqual([]); + }); + }); + + test("final worker gate closes over an idle parent fleet hold", async () => { + await withWakeBridge((bridge, sends) => { + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "fleet", running: 2 }); + bridge.handle({ type: "inference.done", data: {} }); + bridge.gateOpened(); + bridge.gateOpened(); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + bridge.gateClosed(); + expect(sends).toEqual([]); + bridge.gateClosed(); + expect(sends).toHaveLength(1); + bridge.handle({ type: "inference.done", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toHaveLength(1); + }); + }); + + for (const settleFirst of [false, true]) { + test(`live parent and two gates wait for both conditions (settle first: ${settleFirst})`, async () => { + await withWakeBridge((bridge, sends) => { + bridge.gateOpened(); + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "fleet", running: 1 }); + bridge.gateOpened(); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + bridge.gateClosed(); + if (settleFirst) bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toEqual([]); + bridge.gateClosed(); + if (!settleFirst) { + expect(sends).toEqual([]); + expect(bridge.turn.isProcessing).toBe(true); + bridge.handle({ type: "inference.done", data: {} }); + } + expect(sends).toHaveLength(1); + }); + }); + } + + test("a parked ask wakes an idle parent exactly once, including notifications after settle", async () => { + await withWakeBridge((bridge, sends) => { + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toHaveLength(1); + expect(sends[0]).toContain("a1"); + expect(sends[0]).toContain("Which port?"); + expect(sends[0]).toContain("q1"); + expect(sends[0]).toContain("send_input"); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + bridge.handle({ type: "inference.done", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toHaveLength(1); + bridge.handle({ type: "agent-ask", asks: [] }); + expect(sends).toHaveLength(1); + }); + }); + + test("an ask parking mid-cycle defers to settle, then flushes once", async () => { + await withWakeBridge((bridge, sends) => { + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toEqual([]); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toHaveLength(1); + }); + }); + + test("multiple parked asks coalesce into one wake message", async () => { + await withWakeBridge((bridge, sends) => { + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1"), wake("a2", "q2")] }); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toHaveLength(1); + expect(sends[0]).toContain("a1"); + expect(sends[0]).toContain("a2"); + }); + }); + + test("an open gate defers the wake; gate close flushes once", async () => { + await withWakeBridge((bridge, sends) => { + bridge.gateOpened(); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toEqual([]); + bridge.gateClosed(); + expect(sends).toHaveLength(1); + }); + }); + + test("clearQueuedDelivery drops stashed wakes and resets delivered identities", async () => { + await withWakeBridge((bridge, sends) => { + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + bridge.clearQueuedDelivery(); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toEqual([]); + bridge.handle({ type: "agent-ask", asks: [wake("a2", "q2")] }); + expect(sends).toHaveLength(1); + bridge.clearQueuedDelivery(); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toHaveLength(1); + bridge.handle({ type: "agent-ask", asks: [wake("a2", "q2")] }); + expect(sends).toHaveLength(2); + }); + }); + + test("disposed bridges cannot revive deferred wakes", async () => { + await withWakeBridge((bridge, sends) => { + bridge.gateOpened(); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + bridge.dispose(); + bridge.gateClosed(); + bridge.handle({ type: "inference.done", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toEqual([]); + }); + }); +}); diff --git a/src/tui/runner/exit.ts b/src/tui/runner/exit.ts index 36f76427d..f4786330c 100644 --- a/src/tui/runner/exit.ts +++ b/src/tui/runner/exit.ts @@ -39,6 +39,23 @@ import { const tuiLogger = getLogger([LOG_NAMESPACE_ROOT, "tui"]); +export function resetSessionForRotation( + state: Pick, + services: Pick, +): void { + const reset = (): void => { + services.deliveryGeneration.bump(); + cancelFeedbackCapture(); + services.emitter.emit("session.clear"); + services.subAgentSessions.cancelAll("Session cleared"); + }; + if (state.withFleetPublicationSuspended === undefined) { + reset(); + } else { + state.withFleetPublicationSuspended(reset); + } +} + export interface ResolveExitCodeArgs { runError: string | undefined; sinkError: string | undefined; @@ -427,15 +444,7 @@ export async function createRunLifecycle( // abort handles → child agent.close) before clearing the session store so // /clear does not leave orphaned child reactors burning tokens. const newSession = (): void => { - services.deliveryGeneration.bump(); - cancelFeedbackCapture(); - // Wipe the painted transcript immediately. The product host listens for - // session.clear; the Ink App used to clear its own stream unconditionally - // and that path never moved to OpenTUI. - services.emitter.emit("session.clear"); - // Cancel live workers before rotation so /clear does not leave orphaned - // child reactors burning tokens under the old session id. - services.subAgentSessions.cancelAll("Session cleared"); + resetSessionForRotation(state, services); // Backend rotation is always enqueued regardless of contention; the queue // serialises it behind any in-progress op. Sub-agents nest under the new // session automatically because getWorkdirBase reads the live sessionId. diff --git a/src/tui/runner/state.ts b/src/tui/runner/state.ts index 0db63a254..523680f0b 100644 --- a/src/tui/runner/state.ts +++ b/src/tui/runner/state.ts @@ -238,6 +238,7 @@ export interface RunnerState { ) => Promise; shutdownRuntime?: () => Promise; stopFleetReporting?: () => void; + withFleetPublicationSuspended?: (reset: () => void) => void; } export function recordRunError(state: RunnerState, err: unknown): void { diff --git a/src/tui/runner/wiring.ask-wake.test.ts b/src/tui/runner/wiring.ask-wake.test.ts new file mode 100644 index 000000000..c7b3c7318 --- /dev/null +++ b/src/tui/runner/wiring.ask-wake.test.ts @@ -0,0 +1,323 @@ +import { expect, test } from "bun:test"; +import { EventEmitter } from "node:events"; +import { createSubAgentSessionStore } from "../../subagent/session-store.js"; +import { createFleetMailbox, createWaitAgentsTool } from "../../subagent/agent-fleet.js"; +import { createFleetWakePublisher } from "./wiring.js"; +import { attachSessionBridge, type BridgeInboundEvent } from "../runtime-bridge.js"; +import { createLiveSessionPort } from "../live-session-port.js"; +import { createAppShell } from "../shell/index.js"; +import { withTestRenderer } from "../harness.js"; +import { resetSessionForRotation } from "./exit.js"; +import { clearTranscript } from "../shell/chrome.js"; +import { createDeliveryGeneration, createLeftoverSend } from "../queued-delivery.js"; +import { createSessionOperationQueue } from "../session-operation-queue.js"; + +test("failed reset releases publication without flushing partially cancelled workers", () => { + const store = createSubAgentSessionStore(); + const emitter = new EventEmitter(); + const publisher = createFleetWakePublisher(store, emitter); + const events: BridgeInboundEvent[] = []; + emitter.on("event", (event: BridgeInboundEvent) => events.push(event)); + const unsubscribe = store.subscribe(publisher.publish); + try { + store.start({ id: "old", agentId: "builder", description: "old", brief: "build" }); + store.markRunning("old"); + store.registerAsk("old", { + question: "Old question?", + questionId: "old-question", + resolve: () => {}, + reject: () => {}, + }); + events.length = 0; + const error = new Error("reset failed"); + expect(() => + publisher.withSuspended(() => { + store.wake(); + throw error; + }), + ).toThrow(error); + expect(events).toEqual([]); + publisher.withSuspended(() => store.cancelAll("Session cleared")); + expect(events).toEqual([ + { type: "agent-ask", asks: [] }, + { type: "fleet", running: 0 }, + ]); + events.length = 0; + publisher.withSuspended(() => store.cancelAll("Session cleared")); + expect(events).toEqual([ + { type: "agent-ask", asks: [] }, + { type: "fleet", running: 0 }, + ]); + } finally { + unsubscribe(); + } +}); + +for (const phase of ["settled", "prequeued", "deferred"] as const) { + test(`rotation suppresses old worker snapshots (${phase}), including repeated resets`, async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + }); + const store = createSubAgentSessionStore(); + const emitter = new EventEmitter(); + const deliveryGeneration = createDeliveryGeneration(); + const queue = createSessionOperationQueue(); + const sends: string[] = []; + const scheduled: string[] = []; + const deliver = createLeftoverSend({ + enqueue: queue.enqueue, + ingest: async (text, attachments) => ({ text, attachments }), + send: (text) => { + sends.push(text); + }, + captureGeneration: deliveryGeneration.capture, + onFailure: (error) => { + throw error; + }, + }); + const bridge = attachSessionBridge( + shell, + createLiveSessionPort({ + send: (text) => { + sends.push(text); + }, + deliver: (text) => { + scheduled.push(text); + deliver(text); + }, + interrupt: () => {}, + }), + ); + let resetting = false; + const resetEvents: BridgeInboundEvent[] = []; + const repainted: string[] = []; + emitter.on("event", (event: BridgeInboundEvent) => { + bridge.handle(event); + if (resetting) { + resetEvents.push(event); + repainted.push(...shell.streamLog.map((row) => row.text)); + } + }); + emitter.on("session.clear", () => { + clearTranscript(shell); + bridge.clearQueuedDelivery(); + }); + const publisher = createFleetWakePublisher(store, emitter); + const unsubscribe = store.subscribe(publisher.publish); + const start = (id: string) => { + store.start({ id, agentId: "builder", description: id, brief: "build" }); + store.markRunning(id); + store.registerAsk(id, { + question: `question ${id}`, + questionId: `question-${id}`, + resolve: () => {}, + reject: () => {}, + }); + }; + try { + for (let round = 0; round < 2; round++) { + bridge.handle({ type: "inference.start", data: {} }); + start(`old-${round}-one`); + start(`old-${round}-two`); + if (phase !== "deferred") bridge.handle({ type: "inference.done", data: {} }); + if (phase === "settled") { + await queue.awaitTail(); + bridge.handle({ type: "inference.done", data: {} }); + } + const beforeScheduled = scheduled.length; + const beforeSent = sends.length; + resetting = true; + resetSessionForRotation( + { withFleetPublicationSuspended: publisher.withSuspended }, + { deliveryGeneration, emitter, subAgentSessions: store }, + ); + resetting = false; + expect(scheduled).toHaveLength(beforeScheduled); + expect(repainted).toEqual([]); + expect(resetEvents).toEqual([ + { type: "agent-ask", asks: [] }, + { type: "fleet", running: 0 }, + ]); + expect(store.list().every((worker) => worker.status === "cancelled")).toBe(true); + await queue.awaitTail(); + expect(sends).toHaveLength(beforeSent); + bridge.handle({ type: "inference.done", data: {} }); + resetEvents.length = 0; + } + const before = sends.length; + start("new-worker"); + await queue.awaitTail(); + expect(sends).toHaveLength(before + 1); + expect(sends.at(-1)).toContain("question new-worker"); + bridge.handle({ type: "inference.done", data: {} }); + store.wake(); + await queue.awaitTail(); + expect(sends).toHaveLength(before + 1); + } finally { + unsubscribe(); + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); +} + +for (const removal of ["answer", "cancel", "terminal", "remove", "replace"] as const) { + test(`production pending snapshot drops a deferred ask on ${removal}`, async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + }); + const sends: string[] = []; + const send = (text: string) => { + sends.push(text); + }; + const bridge = attachSessionBridge( + shell, + createLiveSessionPort({ send, deliver: send, interrupt: () => {} }), + ); + const store = createSubAgentSessionStore(); + const emitter = new EventEmitter(); + const events: BridgeInboundEvent[] = []; + emitter.on("event", (event: BridgeInboundEvent) => { + events.push(event); + bridge.handle(event); + }); + const publisher = createFleetWakePublisher(store, emitter); + const unsubscribe = store.subscribe(publisher.publish); + try { + bridge.handle({ type: "inference.start", data: {} }); + const worker = store.start({ + id: "worker-session", + agentId: "builder", + description: "work", + brief: "build", + }); + store.markRunning(worker.id); + store.registerAsk(worker.id, { + question: "Which port?", + questionId: "q1", + resolve: () => {}, + reject: () => {}, + }); + if (removal === "answer") { + const mailbox = createFleetMailbox(store); + mailbox.register(worker.id); + const wait = createWaitAgentsTool({ sessions: store, fleetRecords: mailbox }); + if (wait.kind !== "full") throw new Error("expected full wait tool"); + const result = await wait.handler( + { + id: "wait-call", + name: "wait_agents", + arguments: { targets: [worker.id], timeout_ms: 1000 }, + }, + new AbortController().signal, + ); + expect(result.content).toContain("awaiting_director"); + store.sendInputOne(worker.id, "8080"); + } + if (removal === "cancel") store.cancelAsk(worker.id); + if (removal === "terminal") store.complete(worker.id, "done"); + if (removal === "remove") store.clear(); + if (removal === "replace") { + store.start({ + id: worker.id, + agentId: "builder", + description: "replacement", + brief: "build", + }); + } + expect(events.at(-1)?.type === "fleet" ? events.at(-2) : events.at(-1)).toEqual({ + type: "agent-ask", + asks: [], + }); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toEqual([]); + } finally { + unsubscribe(); + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); +} + +test("same catalog workers answer by session, reconcile one resolution and replacement, exclude nested asks", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + }); + const sends: string[] = []; + const send = (text: string) => { + sends.push(text); + }; + const bridge = attachSessionBridge( + shell, + createLiveSessionPort({ send, deliver: send, interrupt: () => {} }), + ); + const store = createSubAgentSessionStore(); + const emitter = new EventEmitter(); + emitter.on("event", (event: BridgeInboundEvent) => bridge.handle(event)); + const publisher = createFleetWakePublisher(store, emitter); + const unsubscribe = store.subscribe(publisher.publish); + const answers: string[] = []; + const ask = (id: string, questionId: string) => + store.registerAsk(id, { + question: questionId, + questionId, + resolve: (answer) => { + answers.push(`${id}:${answer}`); + }, + reject: () => {}, + }); + try { + bridge.handle({ type: "inference.start", data: {} }); + for (const id of ["session-one", "session-two", "nested"]) { + store.start({ + id, + agentId: "builder", + description: id, + brief: "build", + ...(id === "nested" ? { parentSessionId: "session-one" } : {}), + }); + store.markRunning(id); + ask(id, `question-${id}`); + } + expect(store.sendInputOne("session-one", "8080").ok).toBe(true); + expect(answers).toEqual(["session-one:8080"]); + expect(store.hasPendingAsk("session-two")).toBe(true); + store.cancelAsk("session-two"); + ask("session-two", "replacement-question"); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toHaveLength(1); + expect(sends[0]).toContain("using target session-two"); + expect(sends[0]).toContain("replacement-question"); + expect(sends[0]).not.toContain("question-session-two"); + expect(sends[0]).not.toContain("question-session-one"); + expect(sends[0]).not.toContain("question-nested"); + bridge.handle({ type: "inference.done", data: {} }); + store.wake(); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toHaveLength(1); + expect(store.sendInputOne("session-two", "9090").ok).toBe(true); + expect(answers).toEqual(["session-one:8080", "session-two:9090"]); + } finally { + unsubscribe(); + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); +}); diff --git a/src/tui/runner/wiring.ts b/src/tui/runner/wiring.ts index 687b45cfc..46cf0b827 100644 --- a/src/tui/runner/wiring.ts +++ b/src/tui/runner/wiring.ts @@ -21,6 +21,7 @@ import { FLEET_STALL_POLL_MS, liveFleetCount, observeFleet, + pendingAskSnapshot, } from "../../subagent/index.js"; import { scheduleUpgradeNotice } from "../../upgrade/index.js"; import pkg from "../../../package.json" with { type: "json" }; @@ -51,6 +52,39 @@ import { LOG_NAMESPACE_ROOT } from "../../branding.js"; const tuiLogger = getLogger([LOG_NAMESPACE_ROOT, "tui"]); +export function createFleetWakePublisher( + sessions: RunnerServices["subAgentSessions"], + emitter: RunnerServices["emitter"], +) { + let lastLiveFleet = 0; + let suspended = false; + const publish = (): void => { + if (suspended) return; + const lanes = sessions.list(); + // Reconcile even an empty snapshot before a fleet drop can settle the parent. + const asks = pendingAskSnapshot(lanes, (id) => sessions.peekAsk(id)); + emitter.emit("event", { type: "agent-ask", asks }); + const fleet = liveFleetCount(lanes); + if (fleet !== lastLiveFleet) { + lastLiveFleet = fleet; + emitter.emit("event", { type: "fleet", running: fleet }); + } + }; + const withSuspended = (reset: () => void): void => { + suspended = true; + // The bridge clears its fleet count even when the store's count is unchanged. + lastLiveFleet = -1; + try { + reset(); + } finally { + suspended = false; + } + // A failed cancellation must not publish its partially reset snapshot. + publish(); + }; + return { publish, withSuspended }; +} + export function wirePostStartup( state: RunnerState, services: RunnerServices, @@ -125,17 +159,10 @@ export function wirePostStartup( for (const update of observation.updates) surfaceSystemNotice(hostOf(state).shell, update); }; let fleetSettle: ReturnType | null = null; - // Live-lane count feeds the bridge's idle-with-fleet hold (CL-7057): the - // run stays busy after the parent turn settles until the last lane - // terminalizes. Store notifications fire per child event, not per status - // flip, so emit only when the count itself moves. - let lastLiveFleet = 0; + const fleetWakePublisher = createFleetWakePublisher(services.subAgentSessions, services.emitter); + state.withFleetPublicationSuspended = fleetWakePublisher.withSuspended; const unsubscribeFleetReport = services.subAgentSessions.subscribe(() => { - const fleet = liveFleetCount(services.subAgentSessions.list()); - if (fleet !== lastLiveFleet) { - lastLiveFleet = fleet; - services.emitter.emit("event", { type: "fleet", running: fleet }); - } + fleetWakePublisher.publish(); if (fleetSettle !== null) return; fleetSettle = setTimeout(() => { fleetSettle = null; diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index c105ca6dd..f8bb3fa50 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -67,6 +67,7 @@ import { fleetProgress, type AgentProgressSession, } from "./agent-progress.js"; +import { pendingAskWakeText, type PendingAskWake } from "../subagent/fleet-report.js"; /** Tool name a sub-agent dispatch call carries — its row gets live progress. */ const SPAWN_AGENT_TOOL_NAME = "spawn_agent"; @@ -268,6 +269,7 @@ function isBridgeInbound(event: { type: string }): event is BridgeInboundEvent { case "system": case "run": case "fleet": + case "agent-ask": case "tool.boundary": case "error": return true; @@ -353,8 +355,22 @@ export interface BridgeBag { * session-idle — and the hold releases when the count lands back at zero. */ liveFleet: number; + /** + * Worker asks parked in ask_director, keyed by session, waiting for a + * moment the parent can act on them (idle settle or last gate closing). + * Keying by session is what stops a repeat emitter notification from + * stashing the same question twice. + */ + pendingAskWake: Map; + deliveredAskWake: Map; + /** + * Set inside `attachSessionBridge`; the module-scope + * settle path (`settleRunToIdle`) and `gateClosed` re-enter through it. + */ + flushPendingAskWake: (() => void) | null; /** Last prompt actually sent — replay source for the quota auto-retry. */ lastSentMessage: string; + lastSentOrigin: "composer" | "internal" | null; /** One auto-retry per rate-limit window. */ quotaFired: boolean; now: () => number; @@ -900,11 +916,13 @@ function settleRunToIdle(shell: AppShell, bag: BridgeBag): void { // pending send now — the parent they were steering has stopped, so // each one starts its own turn — while follow-ups keep waiting. drainSteersAtBoundary(shell, bag); + bag.flushPendingAskWake?.(); return; } shell.session = setRunState(shell.session, "idle"); // Full drain: soft steers first, then follow-ups (drainOrder). drainAtBoundary(shell, bag); + bag.flushPendingAskWake?.(); } function applyInbound(shell: AppShell, bag: BridgeBag, event: BridgeInboundEvent): void { @@ -913,16 +931,26 @@ function applyInbound(shell: AppShell, bag: BridgeBag, event: BridgeInboundEvent // Fleet liveness owns no transcript row state, so it is handled before the // open-row machinery — a lane terminalizing mid-parent-stream must not // close the assistant row the parent's own deltas are growing. - if (event.type === "fleet") { + if (event.type === "fleet" || event.type === "agent-ask") { // Idle-with-fleet bookkeeping. A transition to zero while the parent is // already idle releases the hold: that moment is true session-idle, so // queued follow-ups drain now. While the parent is still working the // count just updates — the ordinary turn settle does the draining. - bag.liveFleet = event.running; - if (event.running === 0 && !bag.turn.isProcessing) { - settleRunToIdle(shell, bag); + if (event.type === "fleet") { + bag.liveFleet = event.running; + if (event.running === 0 && !bag.turn.isProcessing) { + settleRunToIdle(shell, bag); + } + paintChrome(shell); + return; } - paintChrome(shell); + bag.pendingAskWake = new Map(event.asks.map((ask) => [ask.sessionId, ask])); + for (const [sessionId, questionId] of bag.deliveredAskWake) { + if (bag.pendingAskWake.get(sessionId)?.questionId !== questionId) { + bag.deliveredAskWake.delete(sessionId); + } + } + bag.flushPendingAskWake?.(); return; } @@ -1011,7 +1039,11 @@ export function attachSessionBridge( disposed: false, turn: initialTurnState(now()), liveFleet: 0, + pendingAskWake: new Map(), + deliveredAskWake: new Map(), + flushPendingAskWake: null, lastSentMessage: "", + lastSentOrigin: null, quotaFired: false, now, toolRows: new Map(), @@ -1209,6 +1241,13 @@ export function attachSessionBridge( if (settled) settleRun(); }; + const recordLastSent = ( + replay: { text: string; origin: "composer" | "internal" } | null, + ): void => { + bag.lastSentMessage = replay === null ? "" : replay.text; + bag.lastSentOrigin = replay === null ? null : replay.origin; + }; + const submit = ( text: string, kind: "queue" | "steer" | "immediate" | "reinject", @@ -1260,7 +1299,7 @@ export function attachSessionBridge( meta: "stop", }); bag.port.interrupt(); - bag.lastSentMessage = ""; + recordLastSent(null); bag.turn = turnStateOnInterrupt(bag.turn, now()); } @@ -1281,12 +1320,12 @@ export function attachSessionBridge( ...(kind === "reinject" ? { meta: "reinject" } : {}), }); bag.pendingEchoes.push(t); - bag.port.sendImmediate(t, attachments); shell.session = setRunState(shell.session, "busy"); - bag.lastSentMessage = t; + recordLastSent({ text: t, origin: "composer" }); bag.turn = turnStateOnSubmit(bag.turn, now()); paintChrome(shell); paintPhase(); + bag.port.sendImmediate(t, attachments); return; } @@ -1306,6 +1345,29 @@ export function attachSessionBridge( paintChrome(shell); }; + const sendInternalText = (text: string): void => { + appendStreamRow(shell, { role: "user", text }); + bag.pendingEchoes.push(text); + shell.session = setRunState(shell.session, "busy"); + recordLastSent({ text, origin: "internal" }); + bag.turn = turnStateOnSubmit(bag.turn, now()); + paintChrome(shell); + paintPhase(); + // Harness turns are not composer input: /feedback must never consume them. + bag.port.deliver({ id: crypto.randomUUID(), text, kind: "queue", enqueuedAt: now() }); + }; + const flushPendingAskWake = (): void => { + if (bag.disposed || bag.turn.isProcessing || bag.turn.blockedGateCount > 0) return; + const asks = [...bag.pendingAskWake.values()].filter( + (ask) => bag.deliveredAskWake.get(ask.sessionId) !== ask.questionId, + ); + if (asks.length === 0) return; + // Outbound delivery can synchronously re-enter through store/stream events. + for (const ask of asks) bag.deliveredAskWake.set(ask.sessionId, ask.questionId); + sendInternalText(asks.map((ask) => pendingAskWakeText(ask)).join("\n\n")); + }; + bag.flushPendingAskWake = () => flushPendingAskWake(); + const doInterrupt = (): void => { if (bag.disposed) return; closeOpenRow(shell, bag); @@ -1324,15 +1386,18 @@ export function attachSessionBridge( drainAtBoundary(shell, bag); // Clearing the last prompt is what stops the quota loop from replaying a // turn the operator (or the watchdog) deliberately stopped. - bag.lastSentMessage = ""; + recordLastSent(null); bag.turn = turnStateOnInterrupt(bag.turn, now()); paintPhase(); }; const clearQueuedDelivery = (): void => { if (bag.disposed) return; shell.session = createSessionQueue("idle"); + recordLastSent(null); bag.pendingEchoes.length = 0; bag.liveFleet = 0; + bag.pendingAskWake.clear(); + bag.deliveredAskWake.clear(); bag.pendingRowUpdates.clear(); paintChrome(shell); }; @@ -1353,6 +1418,7 @@ export function attachSessionBridge( if (bag.disposed) return; bag.turn = turnStateGateClosed(bag.turn, now()); paintPhase(); + flushPendingAskWake(); }; const tick = (): void => { @@ -1369,12 +1435,13 @@ export function attachSessionBridge( }) ) { bag.quotaFired = true; - const replay = bag.lastSentMessage; + const replay = { text: bag.lastSentMessage, origin: bag.lastSentOrigin }; bag.turn = clearQuotaWait(bag.turn); setStatusFlash(shell, "rate limit cleared — resubmitting", { ttlMs: RUNTIME_FLASH_MS, }); - submit(replay, "immediate"); + if (replay.origin === "internal") sendInternalText(replay.text); + else submit(replay.text, "immediate"); return; } @@ -1475,6 +1542,10 @@ export function attachSessionBridge( dispose: () => { flushOpenRow(shell, bag); bag.disposed = true; + recordLastSent(null); + bag.pendingAskWake.clear(); + bag.deliveredAskWake.clear(); + bag.flushPendingAskWake = null; applyCadence(null); clearShellBridgeHooks(shell); bridges.delete(shell); diff --git a/src/tui/stream-event-map.ts b/src/tui/stream-event-map.ts index 965cdaae5..b7f3d3b5b 100644 --- a/src/tui/stream-event-map.ts +++ b/src/tui/stream-event-map.ts @@ -16,6 +16,7 @@ import { } from "../inference-gateway-error.js"; import { isProviderFailurePresentationSuppressed } from "./provider/failure-attempt.js"; import type { RunState } from "./session-queue.js"; +import type { PendingAskWake } from "../subagent/fleet-report.js"; /** Canonical inbound events the bridge understands (fixtures + mapped reactor). */ export type BridgeInboundEvent = @@ -48,6 +49,11 @@ export type BridgeInboundEvent = * new turn rather than a queued steer. */ | { readonly type: "fleet"; readonly running: number } + /** + * Authoritative snapshot of currently pending top-level ask_director + * questions, including empty. The bridge reconciles and dedups delivery. + */ + | { readonly type: "agent-ask"; readonly asks: readonly PendingAskWake[] } | { readonly type: "tool.boundary" } | { readonly type: "error"; readonly message: string } /** diff --git a/src/tui/turn-state.test.ts b/src/tui/turn-state.test.ts index cfc095abf..f6d60da99 100644 --- a/src/tui/turn-state.test.ts +++ b/src/tui/turn-state.test.ts @@ -23,6 +23,16 @@ const fold = ( initialTurnState(startMs), ); +test("an idle worker gate does not manufacture parent processing", () => { + const opened = turnStateGateOpened(initialTurnState(0)); + expect(opened.isProcessing).toBe(false); + expect(opened.status).toBe("blocked"); + expect(turnStateGateClosed(opened, 1).status).toBe("idle"); + const started = turnStateFromEvent(opened, { type: "inference.start" }, 2); + expect(started.isProcessing).toBe(true); + expect(turnStateGateClosed(started, 3).status).toBe("running"); +}); + describe("turnStateFromEvent", () => { test("start awaits the first token", () => { const s = fold([{ type: "inference.start" }]); diff --git a/src/tui/turn-state.ts b/src/tui/turn-state.ts index ba6e0b339..33038b7ac 100644 --- a/src/tui/turn-state.ts +++ b/src/tui/turn-state.ts @@ -214,14 +214,14 @@ export function turnStateGateOpened(state: TurnState): TurnState { return { ...state, status: "blocked", - isProcessing: true, blockedGateCount, }; } /** * A gate resolved. Only the last outstanding gate clearing returns the turn - * to "running" — earlier ones just decrement the count. `lastActivityAt` + * to "running" if the parent is processing, otherwise "idle" — earlier ones + * just decrement the count. `lastActivityAt` * moves to `nowMs` so the stall clock restarts from the moment the operator * actually answered, rather than crediting silence spent reading the prompt. */ @@ -230,7 +230,7 @@ export function turnStateGateClosed(state: TurnState, nowMs: number): TurnState if (blockedGateCount > 0) return { ...state, blockedGateCount }; return { ...state, - status: state.status === "blocked" ? "running" : state.status, + status: state.status === "blocked" ? (state.isProcessing ? "running" : "idle") : state.status, lastActivityAt: nowMs, blockedGateCount, };