From 33971ad6bcd12db49d7327cbbc4d02de57884461 Mon Sep 17 00:00:00 2001 From: zszz3 <91608029+zszz3@users.noreply.github.com> Date: Sat, 19 Sep 2026 21:43:10 +0800 Subject: [PATCH 1/2] fix(queue): steer promoted input and preserve transcript ordering Deliver promoted messages into the active turn and wake delegate waits so corrective input does not wait for background work to finish. Hold turn settlement until transcript writes reach the host, preserving overflow across restart. Keep graceful stop latched through recovery and retry boundaries without aborting work already in progress. fixes #597 --- apps/desktop/electron/main/index.ts | 4 + apps/desktop/electron/main/ipc/agent-ipc.ts | 10 ++ apps/desktop/electron/main/ipc/register.ts | 2 + .../electron/main/persistence-outbox.ts | 74 +++++++--- apps/desktop/electron/main/runtime/plans.ts | 5 + apps/desktop/src/stores/slices/queue-slice.ts | 3 - .../test/queued-turn-finalization.test.mjs | 8 +- .../test/session-message-input.test.mjs | 11 +- ...ded-host-runtime-and-persistence-outbox.md | 8 +- ...rn-queue-priority-block-and-row-actions.md | 25 ++-- docs/spec/03-runtime/01-ipc-protocol.md | 18 +-- docs/spec/03-runtime/02-agent-runtime.md | 28 +++- docs/spec/03-runtime/04-data-storage.md | 16 +++ docs/spec/04-ux/08-component-spec.md | 9 +- docs/spec/04-ux/09-interaction-patterns.md | 21 ++- docs/spec/06-delivery/04-e2e-test-plan.md | 65 ++++++++- packages/agent-host/src/agent-host.test.ts | 27 ++-- packages/agent-host/src/agent-host.ts | 33 +++-- packages/agent-runtime/src/provider-retry.ts | 35 ++++- packages/agent-runtime/src/runtime.test.ts | 4 +- packages/agent-runtime/src/runtime.ts | 136 ++++++++++++------ 21 files changed, 396 insertions(+), 146 deletions(-) diff --git a/apps/desktop/electron/main/index.ts b/apps/desktop/electron/main/index.ts index 0062e5a770..b9968d2f5d 100644 --- a/apps/desktop/electron/main/index.ts +++ b/apps/desktop/electron/main/index.ts @@ -1117,6 +1117,8 @@ const sessionCollaboration = createSessionCollaborationService({ log: (message, data) => logger.app("runtime", "warn", message, { data }), }); +const settleTranscript = (sessionId: string) => + persistenceOutbox.drainSession(sessionId, () => host, () => quitting); const planRuntime = createPlanRuntime({ runtimeState, planState: planRuntimeState, @@ -1140,6 +1142,7 @@ const planRuntime = createPlanRuntime({ resolveAgentRuntimeLaunch, isQuitting: () => quitting, onTurnSettled: sessionCollaboration.settle, + settleTranscript, }); const { finishTurn, @@ -1248,6 +1251,7 @@ const { bootHostStatus, runtimeArch, bootBackends } = runtimeLifecycle; function registerIpc() { return registerIpcHandlers({ + settleTranscript, traySessions: applicationLifecycle!.traySessions, ipcMain, getMainWindow: () => mainWindow, diff --git a/apps/desktop/electron/main/ipc/agent-ipc.ts b/apps/desktop/electron/main/ipc/agent-ipc.ts index 809a243541..b47f4e5584 100644 --- a/apps/desktop/electron/main/ipc/agent-ipc.ts +++ b/apps/desktop/electron/main/ipc/agent-ipc.ts @@ -34,6 +34,7 @@ export type AgentIpcDependencies = { resolveAgentRuntimeLaunch: (...args: any[]) => Promise; acquireSessionOperation: (sessionId: string) => Promise<() => void>; finishTurn: FinishTurn; + settleTranscript: (sessionId: string) => Promise; /** * Record a cancellation before the cancel request is issued, so a terminal * event arriving while it is in flight cannot restate the abort as a @@ -76,6 +77,7 @@ export function registerAgentIpc({ resolveAgentRuntimeLaunch, acquireSessionOperation, finishTurn, + settleTranscript, lockAbortReason, finishApprovedExecution, dispatchApprovedPlan, @@ -311,6 +313,14 @@ export function registerAgentIpc({ if (!host) throw new Error("host unavailable"); const releaseSessionOperation = await acquireSessionOperation(req.sessionId); try { + // Restart may have restored an outbox without a live finalization record. + // Flush it before beginning a new turn or reading the prompt's history. + if (!(await settleTranscript(req.sessionId))) { + throw Object.assign(new Error("Application is shutting down"), { errorCode: "TURN_ABORTED" }); + } + host = getHost(); + sidecar = getSidecar(); + if (!host || !sidecar) throw new Error("backend unavailable after transcript settlement"); const sessionMessage = await resolveSessionMessageInput(host, req); // Install the renderer's prompt-time snapshot before any asynchronous // setup. This closes the gap where a fast completion could beat the diff --git a/apps/desktop/electron/main/ipc/register.ts b/apps/desktop/electron/main/ipc/register.ts index a7f83bbde0..6afe61c023 100644 --- a/apps/desktop/electron/main/ipc/register.ts +++ b/apps/desktop/electron/main/ipc/register.ts @@ -133,6 +133,7 @@ export function registerIpcHandlers(dependencies: RegisterIpcDependencies) { claimedExecutionSessions, resolveAgentRuntimeLaunch, finishTurn, + settleTranscript, lockAbortReason, finishApprovedExecution, dispatchApprovedPlan, @@ -353,6 +354,7 @@ export function registerIpcHandlers(dependencies: RegisterIpcDependencies) { resolveAgentRuntimeLaunch, acquireSessionOperation, finishTurn, + settleTranscript, lockAbortReason, finishApprovedExecution, dispatchApprovedPlan, diff --git a/apps/desktop/electron/main/persistence-outbox.ts b/apps/desktop/electron/main/persistence-outbox.ts index 1c0ab34a43..44bfae3b6d 100644 --- a/apps/desktop/electron/main/persistence-outbox.ts +++ b/apps/desktop/electron/main/persistence-outbox.ts @@ -11,7 +11,7 @@ type MessageAppend = { type OutboxLogger = (level: "warn" | "error", message: string, data?: unknown) => void; -const MAX_ENTRIES = 1024; +const BACKLOG_WARNING_THRESHOLD = 1024; /** * Keeps transcript appends away from a dead host pipe. The file is an @@ -26,6 +26,7 @@ export class PersistenceOutbox { private flushing: Promise | null = null; private persistChain = Promise.resolve(); private readonly loaded: Promise; + private readonly enqueuing = new Map, string>(); constructor(dataDir: string, logger: OutboxLogger) { this.path = join(dataDir, "session-message-outbox.json"); @@ -34,7 +35,14 @@ export class PersistenceOutbox { this.loaded = this.load(); } - async enqueue( + enqueue(entry: MessageAppend, getHost: () => HostProcess | null): Promise { + // Reserve synchronously: a terminal event can follow before load or disk I/O settles. + const pending = this.enqueueEntry(entry, getHost).finally(() => this.enqueuing.delete(pending)); + this.enqueuing.set(pending, entry.sessionId); + return pending; + } + + private async enqueueEntry( entry: MessageAppend, getHost: () => HostProcess | null, ): Promise { @@ -42,13 +50,15 @@ export class PersistenceOutbox { const existing = this.entries.findIndex((item) => item.key === entry.key); if (existing >= 0) this.entries[existing] = entry; else { - if (this.entries.length >= MAX_ENTRIES) await this.flush(getHost); - if (this.entries.length >= MAX_ENTRIES) { - this.logger("error", "session persistence outbox is full", { + // Completed messages must remain recoverable until the host acknowledges them. + // Persist overflow in the same recovery file instead of acknowledging a + // dropped row. Turn settlement prevents subsequent prompts from adding + // work in this session until its backlog reaches the host. + if (this.entries.length === BACKLOG_WARNING_THRESHOLD) { + this.logger("warn", "session persistence outbox backlog is high", { size: this.entries.length, - max: MAX_ENTRIES, + threshold: BACKLOG_WARNING_THRESHOLD, }); - return; } this.entries.push(entry); } @@ -56,13 +66,41 @@ export class PersistenceOutbox { void this.flush(getHost); } - async flush(getHost: () => HostProcess | null): Promise { + async flush(getHost: () => HostProcess | null, sessionId?: string): Promise { await this.loaded; - if (this.flushing) return this.flushing; - this.flushing = this.flushLoop(getHost).finally(() => { - this.flushing = null; - }); - return this.flushing; + while (this.flushing) await this.flushing; + const pending = this.flushLoop(getHost, sessionId); + this.flushing = pending; + try { + await pending; + } finally { + if (this.flushing === pending) this.flushing = null; + } + } + + /** Hold turn ownership until its transcript is durable; quit leaves the outbox for restart. */ + async drainSession( + sessionId: string, + getHost: () => HostProcess | null, + isStopping: () => boolean, + ): Promise { + await this.loaded; + for (;;) { + if (isStopping()) return false; + try { + const pending = [...this.enqueuing].filter(([, id]) => id === sessionId).map(([write]) => write); + await Promise.all(pending); + await this.flush(getHost, sessionId); + if (!this.entries.some((entry) => entry.sessionId === sessionId) && + ![...this.enqueuing.values()].includes(sessionId)) return true; + } catch (error) { + // A failed local write must not release the next prompt either. + this.logger("warn", "session transcript settlement retrying", { sessionId, error: String(error) }); + } + // This finalization owns the retry timer. It cannot keep the process alive; + // shutdown ends the wait on the next iteration and preserves unsaved rows. + await new Promise((resolve) => { setTimeout(resolve, 1000).unref(); }); + } } /** @@ -81,9 +119,10 @@ export class PersistenceOutbox { return this.entries.length; } - private async flushLoop(getHost: () => HostProcess | null): Promise { - while (this.entries.length > 0) { - const current = this.entries[0]; + private async flushLoop(getHost: () => HostProcess | null, sessionId?: string): Promise { + for (;;) { + const current = this.entries.find((entry) => sessionId === undefined || entry.sessionId === sessionId); + if (!current) return; const currentHost = getHost(); if (!currentHost || !currentHost.isAvailable()) return; try { @@ -120,7 +159,8 @@ export class PersistenceOutbox { } // A newer snapshot may have replaced this key while the host wrote it. // Only remove the exact entry acknowledged by that write. - if (this.entries[0] === current) this.entries.shift(); + const index = this.entries.indexOf(current); + if (index >= 0) this.entries.splice(index, 1); await this.persist(); } } diff --git a/apps/desktop/electron/main/runtime/plans.ts b/apps/desktop/electron/main/runtime/plans.ts index 5f53b32af7..bb609119c5 100644 --- a/apps/desktop/electron/main/runtime/plans.ts +++ b/apps/desktop/electron/main/runtime/plans.ts @@ -65,6 +65,7 @@ export type PlanRuntimeDependencies = { acquireSessionOperation: (sessionId: string) => Promise<() => void>; resolveAgentRuntimeLaunch: (...args: any[]) => Promise; isQuitting: () => boolean; + settleTranscript: (sessionId: string) => Promise; onTurnSettled?: (sessionId: string, turnId: string) => Promise; }; @@ -91,6 +92,7 @@ export function createPlanRuntime({ resolveAgentRuntimeLaunch, isQuitting, onTurnSettled, + settleTranscript, }: PlanRuntimeDependencies): { finishTurn: FinishTurn; finishApprovedExecution: (executionId: string, status: PlanExecutionFinishStatus, errorCode?: string) => Promise; @@ -196,6 +198,9 @@ function finishTurn( const runFinalization = async (): Promise => { try { + // A completed runtime is not yet a durable transcript. Keep queue ownership + // while old replies retry, so the next user's direct append cannot overtake them. + if (!(await settleTranscript(id))) return; if (runtimeState.host) { try { const result = await runtimeState.host.call<{ diff --git a/apps/desktop/src/stores/slices/queue-slice.ts b/apps/desktop/src/stores/slices/queue-slice.ts index d4efd4cac3..ee589f0e59 100644 --- a/apps/desktop/src/stores/slices/queue-slice.ts +++ b/apps/desktop/src/stores/slices/queue-slice.ts @@ -273,9 +273,6 @@ export function createQueueSlice({ })); try { await api.prioritizeQueuedPrompt(promptId); - // Send now keeps its graceful stop: the active turn reaches its - // boundary before the promoted row starts. - if (get().runningSessions[sessionId]) await api.stop(sessionId); } catch (error) { void get().refreshQueuedPrompts(sessionId); get().showToast( diff --git a/apps/desktop/test/queued-turn-finalization.test.mjs b/apps/desktop/test/queued-turn-finalization.test.mjs index a4bb6164a7..708a62d497 100644 --- a/apps/desktop/test/queued-turn-finalization.test.mjs +++ b/apps/desktop/test/queued-turn-finalization.test.mjs @@ -34,7 +34,7 @@ function deferred() { const SESSION = "s1"; const FIRST_TURN = "initial"; -function fixture() { +function fixture({ settleTranscript = async () => true } = {}) { const activeTurns = new Map([[SESSION, FIRST_TURN]]); const persistedQueue = new Map(); const prompts = []; @@ -110,6 +110,7 @@ function fixture() { const planRuntime = createPlanRuntime({ runtimeState: { host, agentHostBridge: bridge }, planState: { approvedExecutionDrain: null }, + settleTranscript, logger: { app() {} }, sendToRenderer() {}, coordination, @@ -360,7 +361,8 @@ test("a turn whose session moved on releases its waiters and its cancellation lo ); }); -test("a promoted queue resumes even when no terminal event reaches Agent Host", async () => { +test("a promoted queue resumes even when no terminal event reaches Agent Host", async (t) => { + t.mock.timers.enable({ apis: ["setTimeout"] }); const f = fixture(); await f.bridge.queue.push({ sessionId: SESSION, content: "promoted follow-up" }); const [entry] = f.bridge.queue.list(SESSION); @@ -375,6 +377,8 @@ test("a promoted queue resumes even when no terminal event reaches Agent Host", await setImmediate(); f.writes[0].resolve({ ok: true }); await pending; + // A rejected steering attempt may already be waiting for its bounded retry. + t.mock.timers.tick(150); await setImmediate(); assert.deepEqual( diff --git a/apps/desktop/test/session-message-input.test.mjs b/apps/desktop/test/session-message-input.test.mjs index a61a7d95ae..2736979b00 100644 --- a/apps/desktop/test/session-message-input.test.mjs +++ b/apps/desktop/test/session-message-input.test.mjs @@ -1,6 +1,7 @@ import assert from "node:assert/strict"; import { register } from "node:module"; import test from "node:test"; +import { setImmediate } from "node:timers/promises"; import { IPC } from "@pi-desktop/shared"; register(new URL("./helpers/ts-import-hooks.mjs", import.meta.url)); @@ -76,6 +77,8 @@ test("prompt IPC persists original session text, skips slash expansion and binds }, }; let released = false; + let releaseTranscript; + const transcriptReady = new Promise((resolve) => { releaseTranscript = resolve; }); registerAgentIpc({ registrar: { handle: (channel, handler) => handlers.set(channel, handler) }, getHost: () => host, @@ -93,12 +96,18 @@ test("prompt IPC persists original session text, skips slash expansion and binds }), acquireSessionOperation: async () => () => { released = true; }, finishTurn: async () => { assert.fail("the prompt should succeed"); }, + settleTranscript: async () => { await transcriptReady; return true; }, emitAgentEvent: (event) => events.push(event), setNotificationViewingSessionId() {}, optionalWorkspaceRoot: async () => { assert.fail("session text must not expand slash commands"); }, composerCommandService: { buildComposerCommands: async () => { assert.fail("session text must not expand slash commands"); } }, loadComposerTemplatesCached: async () => { assert.fail("session text must not expand slash commands"); }, }); - assert.deepEqual(await handlers.get(IPC.invoke.agentPrompt)(request), { accepted: true, turnId: "turn-1" }); + const pendingPrompt = handlers.get(IPC.invoke.agentPrompt)(request); + await setImmediate(); + assert.equal(calls.length, 0, "restored transcript appends must settle before history reads or a new turn"); + assert.equal(sidecarCalls.length, 0); + releaseTranscript(); + assert.deepEqual(await pendingPrompt, { accepted: true, turnId: "turn-1" }); const begin = calls.find((entry) => entry.method === "session.beginTurn"); assert.equal(begin.params.sessionMessageId, message.id); const row = calls.find((entry) => entry.method === "session.appendMessage").params.message; diff --git a/docs/adr/0041-bounded-host-runtime-and-persistence-outbox.md b/docs/adr/0041-bounded-host-runtime-and-persistence-outbox.md index fc8d4e51e0..72eed45c9f 100644 --- a/docs/adr/0041-bounded-host-runtime-and-persistence-outbox.md +++ b/docs/adr/0041-bounded-host-runtime-and-persistence-outbox.md @@ -30,7 +30,13 @@ to another session is remapped to `{sessionId}:{id}` before the JSONL write (D444). The outbox treats `UNIQUE constraint failed: messages.id` as an ack, not a pause. A permanently rejected append (`PERMISSION_DENIED:` provenance or permission on that row) is dropped the same way so one poison head cannot -fill the 1024-entry cap and discard every later row (D597). +block later transcript rows (D597). The outbox's 1024-entry threshold is a +backlog warning, not permission to drop an already-produced message. Overflow +is persisted in the same recovery file; per-session turn settlement blocks +subsequent prompts until that session's writes finish. This preserves restart +recovery without introducing an in-memory-only overflow queue. A sustained +outage can grow the recovery file with already-running work; protecting those +completed messages takes precedence over the former silent-drop cap (#597). ## Consequences diff --git a/docs/adr/0265-turn-queue-priority-block-and-row-actions.md b/docs/adr/0265-turn-queue-priority-block-and-row-actions.md index bac7fee616..204a378273 100644 --- a/docs/adr/0265-turn-queue-priority-block-and-row-actions.md +++ b/docs/adr/0265-turn-queue-priority-block-and-row-actions.md @@ -50,19 +50,18 @@ queue at all even though the Host already orders it by a durable `position`. refused with a toast while the input is non-empty, and the emptiness check runs against the live editor read because the draft cache is not written per keystroke. -6. **The promoted block is delivered as adjacent messages, not as separate - turns.** Promotion still does not touch the running turn by itself: the - renderer requests the existing graceful `agent/stop`, and the block leaves at - the next boundary. The first promoted entry then starts the turn, and every - later promoted entry is injected into that same turn as user input over the - Composer's existing steering channel. The transcript therefore reads - `user: first`, `user: second` and the model answers once. The injection is - retried a bounded number of times because the runtime only accepts input for - a live run; an entry that is still undelivered stays queued and leaves at the - next boundary as its own turn, which is the previous behavior and never a - lost prompt. An injected entry's own RACP turn is canceled: its input was - delivered by another turn, and no client may be left believing it is still - waiting. +6. **The promoted block steers the active turn.** Send now delivers promoted + entries into the current turn through the existing steering channel, in + click order, without requesting `agent/stop`. The session admission lock + serializes delivery with ordinary queue dispatch. Each entry stays queued + until the runtime acknowledges it. A target that ends or refuses steering + leaves the entry queued for ordinary dispatch after finalization. If the + session is idle, the first entry starts a turn and the rest join it. An + injected entry's own RACP turn is canceled because its input was delivered + into another turn. This supersedes the original graceful-stop-first behavior + for #597: that behavior delayed corrective input behind long delegate waits. + Steering wakes TaskWait and idle delegate waits; it does not stop the child + agents. Other tools complete before the next model request consumes input. 7. **The turn's owner is authoritative about its end.** A runtime terminal event is not a reliable release: Main drops one that names a turn it no longer owns (`isStaleTerminalEvent`), and an abort need not produce one at all. A turn diff --git a/docs/spec/03-runtime/01-ipc-protocol.md b/docs/spec/03-runtime/01-ipc-protocol.md index d8e5a49404..9d9721dcc6 100644 --- a/docs/spec/03-runtime/01-ipc-protocol.md +++ b/docs/spec/03-runtime/01-ipc-protocol.md @@ -545,21 +545,21 @@ entries and `IDEMPOTENCY_CONFLICT` when a key is reused with other input. `entries` arrive in delivery order: promoted entries first in ascending `priority` (the order they were promoted), then every remaining entry by `position`. `prioritize` appends an entry to the end of that priority block -without touching the running turn, refuses an entry that already carries a -priority with `CONFLICT`, and refuses a turn that is no longer queued. The -renderer's "send now" then requests a graceful stop so the entry starts at -the next boundary. `reorder` swaps one non-promoted entry with its adjacent +and delivers promoted entries through steering when a regular turn is active. +It refuses an entry that already carries a priority with `CONFLICT`, and +refuses a turn that is no longer queued. The renderer does not request a stop. +The Host removes each row only after acceptance; a refused or ended target +leaves the row queued for normal dispatch after finalization. `reorder` swaps one non-promoted entry with its adjacent non-promoted neighbour and reports `moved: false` for a promoted entry, a missing entry, or a block/queue edge; a promoted entry is never a neighbour. `remove` cancels an entry that has not started. A restored queue stays held until the desktop attaches as the owner, so a reboot never starts work unattended. -The promoted block is delivered as adjacent messages rather than as separate -turns: the first promoted entry starts the turn at the boundary and every later -promoted entry is injected into that same turn through the steering channel -(`pi-desktop/agent/steer` with the running turn's id), so the transcript shows -the user rows one after another and the model answers once. An injected entry +The promoted block joins the active turn through the steering channel +(`pi-desktop/agent/steer` with that turn's id). When idle, the first entry starts +a turn and subsequent promoted entries join it. Input is consumed at the next +model boundary, not necessarily at acknowledgement time. An injected entry leaves the queue and its own turn is canceled because it never runs on its own. An entry the runtime refuses to accept stays queued and leaves at the next boundary as its own turn. diff --git a/docs/spec/03-runtime/02-agent-runtime.md b/docs/spec/03-runtime/02-agent-runtime.md index c39871745c..d20d9ab9c7 100644 --- a/docs/spec/03-runtime/02-agent-runtime.md +++ b/docs/spec/03-runtime/02-agent-runtime.md @@ -55,12 +55,22 @@ interface AgentRuntime { } ``` -`requestGracefulStop()` is a one-shot request for the active runtime. The pi -loop evaluates it after `turn_end`, once the current assistant response and -tool batch have completed, and emits a normal `agent_end` before another model -request. It does not cancel an active provider stream or running tool. An idle -runtime returns `{ requested: false }`; immediate `abort()` remains the -separate cancellation path. +`requestGracefulStop()` applies to the active durable turn, including a parent +waiting for its delegates. The request stays latched across pi loop boundaries +until the next durable turn starts. The current assistant response and tool +batch finish normally; already-started delegates may finish, but the runtime +must not start another model request to integrate their reports or recover a +silent/progress-only response. Terminal events are released after that existing +work settles, so the next queued prompt can start without `AGENT_BUSY`. +It does not cancel an active provider stream or running tool. An idle runtime +returns `{ requested: false }`; immediate `abort()` remains the separate +cancellation path. Starting, retry-wait, and compaction phases remain active +for stop admission even while pi is not streaming. A retry that already claimed +its budget rechecks the stop after its delay; an opaque-400 repair and every +provider dispatch check before starting fresh work. A failed setup attempt's original error +remains visible when its retry is suppressed. Graceful stop wakes an existing +retry timer without cancelling an active provider stream; immediate abort +retains precedence when both requests arrive. ### 4.0 Active-turn steering @@ -79,6 +89,12 @@ a second public `agent_start`. Existing context/provider recovery takes precedence over that continuation. Steering also wakes a parent that is idle waiting for background delegates; it does not cancel those delegates. +For active-turn steering, `TaskWait` returns early when accepted user input is +pending. Its result has `status: "interrupted"` and explains that the unfinished +delegates keep running. The parent consumes that input in its next model +request without changing the durable turn id. An already-pending steer also +prevents a newly entered TaskWait from sleeping through the input (#597). + Abort, graceful stop, fatal errors and terminal settlement close admission. Accepted but unconsumed input remains transcript/context history and is removed from pi's steering queue so it cannot execute independently on a later turn. diff --git a/docs/spec/03-runtime/04-data-storage.md b/docs/spec/03-runtime/04-data-storage.md index 6b0fc29a9d..5abf72dea8 100644 --- a/docs/spec/03-runtime/04-data-storage.md +++ b/docs/spec/03-runtime/04-data-storage.md @@ -1134,6 +1134,22 @@ sweep promotes a leftover checkpoint whose final row never landed: as checkpoint. A user Stop does not touch the checkpoint, because the runtime's own aborted final row is still on its way and removes it on arrival. Electron handshake awaits the outbox drain before a cold `session.get`. +Turn finalization also waits for every admitted append in that session before +`session.endTurn` releases the follow-up queue. Admission is tracked before +asynchronous file loading or writes, so a following terminal event cannot pass +an append that is not yet visible in the in-memory queue. A paused append is +retried every second while finalization owns the session; another session's +failed head does not block this session's drain. Shutdown ends the wait and +leaves unsaved entries in the outbox for startup recovery. The next user's +direct append must never overtake the previous reply during a transient failure +(#597). Prompt admission applies the same barrier before reading history or +beginning a turn, including after restart when no in-memory finalization exists. +The 1024-entry backlog threshold emits a warning; it must not discard completed +messages. Overflow remains in the same recovery file, including across restart, +and participates in the same per-session settlement barrier. During a sustained +host outage the recovery file may grow with already-running work; subsequent +prompts in each affected session remain blocked until its writes settle. + Renderer-side Stop never rewrites a transcript that has a started reply (spec 01 ยง5.3); its only rewrite is the undo of an unanswered prompt, computed from the full durable transcript merged with the live rows. diff --git a/docs/spec/04-ux/08-component-spec.md b/docs/spec/04-ux/08-component-spec.md index 813750a7e7..c4d8bb4580 100644 --- a/docs/spec/04-ux/08-component-spec.md +++ b/docs/spec/04-ux/08-component-spec.md @@ -2639,10 +2639,11 @@ reasoning-level control. a toast and nothing changes. Remove drops the row immediately. - Send now: promotes the row to the end of the session's priority block, so a second Send now leaves behind the first instead of replacing it at the head. - It then requests `agent/stop`, and the promoted block is released after the - current reply/tool batch completes normally, before every waiting row. The - first promoted row starts the turn and the rest join it as adjacent user - messages, so the block is answered once. When idle it starts immediately. + The Host injects the promoted block into the active turn through steering, + without requesting a stop, and removes each row after acceptance. Rejected + input remains queued for normal dispatch. When idle the first row starts a + turn and the remaining promoted rows join it. Acceptance does not mean the + model has already consumed the input; it reads it at the next boundary. - A promoted row is locked: move up/down, edit, and remove are disabled with their tooltip and `aria-disabled` state intact, and the Send now button reads as already decided (`chat.sendNowPending`). The row carries a distinct diff --git a/docs/spec/04-ux/09-interaction-patterns.md b/docs/spec/04-ux/09-interaction-patterns.md index 8c86d108c4..6303d40ea5 100644 --- a/docs/spec/04-ux/09-interaction-patterns.md +++ b/docs/spec/04-ux/09-interaction-patterns.md @@ -715,11 +715,17 @@ may be retained while exactly one workspace supplies the visible shell context. never moves or clears another session's queue. - The queue renders above the composer. Each row has an independently keyboard-reachable Remove action and a Send now action. -- Send now moves its row to the head and requests the new `agent/stop` channel. - The current assistant response and completed tool batch finish normally; - after `agent_end` and durable turn finalization, the promoted row is - dispatched through the normal `agent/prompt` flow before the remaining rows. - An idle Send now dispatches immediately. +- Send now promotes the row and asks the Host to deliver the promoted block + into the active turn through steering, in click order. It does not request a + graceful stop or change the active turn's identity. A row leaves the durable + queue only after the runtime accepts it. If the target ends or refuses input, + the row stays queued and starts normally after finalization; an idle Send now + dispatches immediately. Delivery is serialized with ordinary queue dispatch. +- Acceptance is not model consumption: an in-flight assistant response stays + in its existing transcript row before the new user input. The input reaches + the next model request after the current response/tool batch. `TaskWait` and + idle delegate waits wake on steering without canceling the delegated work; + ordinary tools retain their completion/cancellation semantics. - Without Send now, the next FIFO row starts automatically after the active turn completes, fails, or is aborted. A terminal event can arrive before persistence releases the session; finalization must wake the queue again @@ -737,7 +743,8 @@ may be retained while exactly one workspace supplies the visible shell context. IME candidate (`isComposing` or key code 229) never sends or steers. - Steering appears as a user message in the current transcript, clears the draft immediately, and reaches the next model request after the current - response/tool batch. It creates no FIFO row and does not interrupt tools. + response/tool batch. It creates no FIFO row and does not cancel tools. + Delegate waits can return early with an interrupted status for new input. - Submission captures the session and current turn identity. If that target ends, rejects input, or is awaiting approval, the draft is restored in its own session and a concise error is shown. New text typed after submission @@ -1502,7 +1509,7 @@ This does not prevent state changes โ€” it makes them instant. 2. Enter sends when Enter-to-send is on; when it is off, Cmd/Ctrl+Enter sends and Enter/Shift+Enter insert a newline 3. Abort immediately cancels running turn and pending permissions without confirmation dialog 3a. Send stays enabled while running, queues prompts per session, and Send now - finishes the current boundary before releasing its prioritized prompt + steers promoted prompts into the active turn without stopping it 4. Long content (>50 lines for messages, >10 for args, >20 for results) is collapsed by default with expand link 5. Tool results that were cut short show a truncation marker or chip per D306; a filled Read window of a longer file does not 6. Permission interrupt inserts inline card, disables composer, shows countdown, and re-enables after resolution diff --git a/docs/spec/06-delivery/04-e2e-test-plan.md b/docs/spec/06-delivery/04-e2e-test-plan.md index 3a3e219ca5..a60e08c120 100644 --- a/docs/spec/06-delivery/04-e2e-test-plan.md +++ b/docs/spec/06-delivery/04-e2e-test-plan.md @@ -1318,11 +1318,10 @@ identify the platform validation still needed. state: disabled Send while idle and empty, enabled Send while running with content (which queues the prompt), and Stop while running with an empty draft. A's two prompts appear in FIFO order, the removed row never sends, - and B's queue remains independent. Send now requests a graceful stop: the - current batch completes with a normal `agent_end`/completed turn, then the - promoted rows are delivered in the order they were promoted, before any - waiting row, without `AGENT_BUSY`: the first starts the turn and the rest join - it as adjacent user messages, so the model answers once for the whole block. + and B's queue remains independent. Send now delivers promoted rows in click + order into the current turn through steering, without a stop request. The + turn id stays unchanged. An unavailable target leaves the row queued for + normal dispatch after finalization. A promoted row locks move/edit/ remove, and its Send now button reads as already decided; promotion is one-way. Move up/down swaps only waiting rows, never crosses the promoted block, and persists. Edit is refused with a visible message while the input @@ -1346,6 +1345,29 @@ identify the platform validation still needed. promotion, reorder, and edit contracts covered at source level (`composer-send-state.test.mjs`); full UI scenario Draft +#### E2E-QUEUE-graceful-stop-across-runtime-loops (#597) + +- **Boundary**: Real runtime and pi loop against a loopback HTTP/SSE provider; + no paid provider or running desktop instance. +- **Steps**: Hold the first response, request graceful stop, and finish with + a silent response. Repeat with a Task that starts a held child response, + stopping both during the parent response and while the parent waits. Release + the child, then submit the next queued prompt through the runtime API. +- **Expected**: The stopped durable turn makes no additional parent request + for response recovery or report integration. Already-started child work + finishes, one parent `agent_end` is emitted, and the next prompt succeeds. + Without a stop, existing recovery and delegate-report behavior is preserved. +- **Additional races**: Request stop after HTTP 429/503 retry admission and + after the retry delay begins; no fresh provider request may follow. Pause an + old reply's append, finish the runtime, and queue new input: neither durable + `endTurn` nor the next prompt may overtake the reply. Recover the host without + enqueueing anything else and verify automatic draining. A different session's + append failure must not block this session, and shutdown must preserve the + paused outbox without dispatching follow-ups. +- **Validation**: Run the isolated runtime scenario above and verify queue + release after durable finalization with + `apps/desktop/test/queued-turn-finalization.test.mjs`. + #### E2E-QUEUE-promote-orders-delivery-by-click: Two Send now clicks deliver in click order - **Preconditions**: Provider configured; session A is running a turn with at @@ -1356,7 +1378,8 @@ identify the platform validation still needed. - **Expected**: The first click is delivered first and the second second โ€” the click order is the delivery order, not "last click wins" and not the original queue order. Both rows appear as adjacent user messages in one turn and the - model answers once; the queue no longer lists either promoted row. Both + active turn consumes them at its next available model boundary; the queue no + longer lists either acknowledged row. Both promoted rows show as already decided and cannot be edited, removed, or reordered. The waiting row keeps its actions and is not delivered before either promoted row. @@ -13515,3 +13538,33 @@ frames must produce no further renders or pending callbacks. Card geometry is read at press time, and move/release/cancel/unmount paths must clear transient transforms and queued frames. Release before the scheduled frame must still save the latest destination. These assertions measure work counts, not device FPS. + + +#### E2E-597-overflow: Full recovery outbox preserves queued-turn ordering + +- Seed the recovery file with 1024 entries from another session and make host + appends fail transiently. Complete the active reply and queue the next input. +- Verify the reply is retained on disk and the next input does not dispatch. + Restore host writes for the active session; its reply must persist before + `session.endTurn` and the next user's append, even if the other session fails. +- Restart with overflow entries still on disk and verify every overflow reply + drains once, in order, without dropping unrelated backlog entries. +- Validate with isolated files, controlled host failures, and controlled retry + timers. Existing outbox tests do not cover the complete overflow scenario. + + +#### E2E-597-send-now-steers-active-turn + +- While a regular turn is active, queue two messages and click Send now on + each. Verify click-order delivery into the same active turn, no graceful + stop, and removal from the queue only after acceptance. +- During TaskWait or an idle delegate wait, send corrective input. Verify the + parent wakes without waiting for all children, and children remain running. + Ordinary tool calls retain their completion/cancellation behavior. +- If the active turn ends during delivery, verify refused input stays queued + and dispatches after finalization. It must neither disappear nor run twice. +- The in-flight assistant row remains before the accepted user row; its late + deltas update that row. Acceptance does not imply immediate model consumption. +- Automated coverage: `packages/agent-host/src/agent-host.test.ts` covers + promotion and fallback. The runtime delegation suites cover normal waits; + steering during TaskWait also requires the interrupted-wait scenario above. diff --git a/packages/agent-host/src/agent-host.test.ts b/packages/agent-host/src/agent-host.test.ts index 1040b2db7c..3fc480f172 100644 --- a/packages/agent-host/src/agent-host.test.ts +++ b/packages/agent-host/src/agent-host.test.ts @@ -384,31 +384,22 @@ describe("AgentHost turns", () => { }); - it("folds the rest of the promoted block into the started turn, in click order", async () => { + it("steers the promoted block into the active turn in click order", async () => { const { host, runtime } = build(); const first = await host.startTurn(controller, { sessionId: "s1", input: { text: "one" }, context: { requestId: "r1" } }); host.ingest(envelope("s1", first.turn.id, { type: "agent_start" })); const second = await host.startTurn(controller, { sessionId: "s1", admission: "queue", input: { text: "two" }, context: { requestId: "r2" } }); const third = await host.startTurn(controller, { sessionId: "s1", admission: "queue", input: { text: "three" }, context: { requestId: "r3" } }); - // Send now on the later row first, then on the earlier one: click order is - // delivery order. await host.prioritizeTurn(controller, third.turn.id); await host.prioritizeTurn(controller, second.turn.id); - expect(host.queueEntries("s1").map((entry) => entry.content)).toEqual(["three", "two"]); - - host.ingest(envelope("s1", first.turn.id, { type: "agent_end", messageIds: [] })); - await new Promise((resolve) => setTimeout(resolve, 0)); - await new Promise((resolve) => setTimeout(resolve, 0)); - - // One turn carries both rows: the first click starts it, the second joins it - // as input. - expect(runtime.prompts.map((prompt) => prompt.content)).toEqual(["one", "three"]); - expect(runtime.steers.map((steer) => steer.content)).toEqual(["two"]); - expect(runtime.steers[0]?.turnId).toBe("rt_2"); - expect(host.queueEntries("s1")).toHaveLength(0); - // The injected row never runs its own turn. + await vi.waitFor(() => expect(host.queueEntries("s1")).toHaveLength(0)); + expect(runtime.prompts.map((prompt) => prompt.content)).toEqual(["one"]); + expect(runtime.steers.map((steer) => steer.content)).toEqual(["three", "two"]); + expect(runtime.steers.map((steer) => steer.turnId)).toEqual([first.turn.id, first.turn.id]); + expect(runtime.stops).toHaveLength(0); + expect(host.getTurn(first.turn.id).status).toBe("running"); expect(host.getTurn(second.turn.id).status).toBe("canceled"); - expect(host.getTurn(third.turn.id).status).toBe("running"); + expect(host.getTurn(third.turn.id).status).toBe("canceled"); }); it("keeps a promoted row queued when the runtime cannot steer it", async () => { @@ -422,7 +413,7 @@ describe("AgentHost turns", () => { await host.prioritizeTurn(controller, third.turn.id); host.ingest(envelope("s1", first.turn.id, { type: "agent_end", messageIds: [] })); - await new Promise((resolve) => setTimeout(resolve, 0)); + await vi.waitFor(() => expect(runtime.prompts).toHaveLength(2)); // Nothing is lost: the refused row is still the next queued turn. expect(runtime.prompts.map((prompt) => prompt.content)).toEqual(["one", "two"]); diff --git a/packages/agent-host/src/agent-host.ts b/packages/agent-host/src/agent-host.ts index babd591a83..efd9a6b529 100644 --- a/packages/agent-host/src/agent-host.ts +++ b/packages/agent-host/src/agent-host.ts @@ -804,7 +804,15 @@ export class AgentHost { private async drainAdmitted(sessionId: string): Promise { while (true) { const state = this.state(sessionId); - if (this.queue.isHeld(sessionId) || this.isOccupied(state)) return; + if (this.queue.isHeld(sessionId)) return; + if (this.isOccupied(state)) { + const active = state.activeTurnId ? state.turns.get(state.activeTurnId) : undefined; + if (active && isActive(active.status) && + state.planningState !== "awaiting_approval") { + await this.deliverPromotedBlock(state, active.runtimeTurnId ?? active.id); + } + return; + } const record = await this.queue.shift(sessionId); if (!record) return; const turn = this.ensureTurn(state, record.id); @@ -829,11 +837,6 @@ export class AgentHost { this.renumberQueue(state); this.renumberQueue(state); this.notifyQueue(sessionId); - // The rest of the promoted block joins this turn as user input, so the - // messages stay adjacent instead of waiting for their own turns - // (ADR 0265). A runtime without steering keeps the previous behavior. - void this.deliverPromotedBlock(state, started.turnId); - return; } catch (error) { turn.status = "failed"; turn.endedAt = new Date(this.clock.now()).toISOString(); @@ -846,7 +849,12 @@ export class AgentHost { this.emit(state, "turn.failed", { turn: this.toRacpTurn(state, turn) }, { turnId: turn.id }); this.renumberQueue(state); this.notifyQueue(sessionId); + continue; } + // A later delivery failure must not restate a successfully started turn + // as a prompt failure. Keep delivery under the same admission lock. + if (turn.runtimeTurnId) await this.deliverPromotedBlock(state, turn.runtimeTurnId); + return; } } @@ -871,14 +879,14 @@ export class AgentHost { } /** - * Deliver the promoted entries that are still queued into the turn that just - * started, so "Send now" twice puts both messages in front of the model - * together instead of spreading them over two turns (ADR 0265). + * Deliver promoted entries into the active turn in click order. The admission + * lock serializes this with normal dispatch and other promotions, so an entry + * remains queued until steering acknowledges it and cannot be sent twice by + * overlapping drain passes (ADR 0265). * * The runtime only accepts input for a turn whose run is live, so a refusal is * retried a bounded number of times. Anything still undelivered stays queued - * and leaves at the next boundary as its own turn: the previous behavior is - * the fallback, never a lost prompt. + * and starts its own turn after the active turn is finalized. */ private async deliverPromotedBlock(state: SessionState, runtimeTurnId: string): Promise { const steer = this.runtime.steer?.bind(this.runtime); @@ -887,7 +895,7 @@ export class AgentHost { const head = this.queue.peek(state.id); if (!head || head.priority === undefined) return; const active = state.activeTurnId ? state.turns.get(state.activeTurnId) : undefined; - if (!active || active.runtimeTurnId !== runtimeTurnId) { + if (!active || (active.runtimeTurnId ?? active.id) !== runtimeTurnId) { // The turn ended (or moved on) while the block was being delivered. return; } @@ -905,6 +913,7 @@ export class AgentHost { accepted = false; } if (!accepted) { + if (state.activeTurnId !== active.id || !isActive(active.status)) return; await delay(PROMOTED_DELIVERY_RETRY_MS); continue; } diff --git a/packages/agent-runtime/src/provider-retry.ts b/packages/agent-runtime/src/provider-retry.ts index ac816ac2b8..78e2f4b909 100644 --- a/packages/agent-runtime/src/provider-retry.ts +++ b/packages/agent-runtime/src/provider-retry.ts @@ -144,6 +144,10 @@ export type ProviderRetryController = { attempt: number; delayMs: number; }) => void; + /** A completed failed attempt must not start fresh work after a graceful stop. */ + shouldStop?: () => boolean; + /** Wakes a retry delay without aborting an in-flight provider stream. */ + stopSignal?: AbortSignal; /** Test hook; production uses the abortable timer below. */ sleep?: (ms: number, signal?: AbortSignal) => Promise; }; @@ -421,6 +425,7 @@ export function createProviderRetryStream( void context; const outer = createAssistantMessageEventStream(); const sleep = controller.sleep ?? delayWithAbort; + const shouldStop = () => controller.stopSignal?.aborted || controller.shouldStop?.() === true; void (async () => { // One repair per logical turn: after an opaque 400/422 the next attempt @@ -428,7 +433,7 @@ export function createProviderRetryStream( // transient budget, and a second opaque failure surfaces untouched. let limitRepairTried = false; for (;;) { - if (options.signal?.aborted) throw requestAbortedError(); + if (options.signal?.aborted || shouldStop()) throw requestAbortedError(); const inner = createStream({ ...(limitRepairTried ? withoutDerivedOutputLimit(options) : options), maxRetries: 0, @@ -480,8 +485,13 @@ export function createProviderRetryStream( if (opaqueLimitRejection) { // Drain the ended stream so providers with deferred cleanup do not // overlap the repair request, mirroring the retry path below. - await inner.result(); + const failed = await inner.result(); if (options.signal?.aborted) throw requestAbortedError(); + if (shouldStop()) { + outer.push({ type: "error", reason: "error", error: failed }); + outer.end(failed); + return; + } limitRepairTried = true; continue; } @@ -498,7 +508,7 @@ export function createProviderRetryStream( // The failed event has already ended this inner stream. Awaiting its // result keeps providers with deferred cleanup from overlapping retries. - await inner.result(); + const failed = await inner.result(); const delayMs = retry.error.code === "PROVIDER_RATE_LIMITED" ? providerRateLimitDelayMs( @@ -516,7 +526,24 @@ export function createProviderRetryStream( attempt: retry.attempt, delayMs, }); - await sleep(delayMs, options.signal); + if (!shouldStop()) { + const signal = controller.stopSignal + ? AbortSignal.any([controller.stopSignal, ...(options.signal ? [options.signal] : [])]) + : options.signal; + try { + await sleep(delayMs, signal); + } catch (error) { + if (!shouldStop() || options.signal?.aborted || + !(error instanceof Error) || error.name !== "AbortError") throw error; + } + } + if (options.signal?.aborted) throw requestAbortedError(); + if (shouldStop()) { + const terminal = controller.status?.() === 429 ? normalizeRateLimitMessage(failed) : failed; + outer.push({ type: "error", reason: "error", error: terminal }); + outer.end(terminal); + return; + } } })().catch((error) => { const aborted = diff --git a/packages/agent-runtime/src/runtime.test.ts b/packages/agent-runtime/src/runtime.test.ts index 45daff8ae1..1ab53a5046 100644 --- a/packages/agent-runtime/src/runtime.test.ts +++ b/packages/agent-runtime/src/runtime.test.ts @@ -257,14 +257,14 @@ describe("DesktopAgentRuntime configuration matching", () => { await runtime.dispose(); }); - it("stops once at the next completed turn boundary", async () => { + it("keeps a stop request across loop boundaries until the next durable turn", async () => { const runtime = createRuntime(); const agent = (runtime as any).agent; agent.state.isStreaming = true; expect(runtime.requestGracefulStop()).toEqual({ requested: true }); expect(await agent.shouldStopAfterTurn({})).toBe(true); - expect(await agent.shouldStopAfterTurn({})).toBe(false); + expect(await agent.shouldStopAfterTurn({})).toBe(true); agent.state.isStreaming = false; expect(runtime.requestGracefulStop()).toEqual({ requested: false }); diff --git a/packages/agent-runtime/src/runtime.ts b/packages/agent-runtime/src/runtime.ts index 572831bf86..27d24481cb 100644 --- a/packages/agent-runtime/src/runtime.ts +++ b/packages/agent-runtime/src/runtime.ts @@ -1659,8 +1659,9 @@ export class DesktopAgentRuntime { private compactionAborted = false; /** Set by the `new_context` tool, consumed at the next turn boundary. */ private pendingModelCompaction = false; - /** One-shot request to finish the current turn at the next boundary. */ - private gracefulStopRequested = false; + /** Stop remains latched until the next durable turn starts. */ + private gracefulStop = new AbortController(); + private get gracefulStopRequested(): boolean { return this.gracefulStop.signal.aborted; } /** Codex's `claim_*` flags: one of each reminder per context window. */ private contextReminderClaimed = false; private contextFallbackReminderClaimed = false; @@ -1843,6 +1844,8 @@ Delegation rules: : this.models.streamSimple(m, context, retryOptions), { claim: (error, phase) => this.claimProviderRetry(error, phase), + stopSignal: this.gracefulStop.signal, + shouldStop: () => this.gracefulStopRequested, headers: () => this.providerRetryHeaders, status: () => this.providerResponseStatus, failure: () => this.providerFetchFailure, @@ -1891,11 +1894,7 @@ Delegation rules: // the next turn boundary. pi-agent-core evaluates this after the // assistant response and completed tool batch, before another provider // request, so no second concurrent durable turn is created. - shouldStopAfterTurn: async () => { - if (!this.gracefulStopRequested) return false; - this.gracefulStopRequested = false; - return true; - }, + shouldStopAfterTurn: async () => this.gracefulStopRequested, }); // pi awaits every listener, so a throw here would reject the run in @@ -4542,6 +4541,9 @@ Delegation rules: if (this.turnHadError) this.terminateParentTurn(); return; } + // Graceful stop lets delegates finish without another parent request + // to integrate their reports. + if (this.gracefulStopRequested) return; const settled = targets.filter( (record) => record.status !== "running" && !record.reportDelivered, ); @@ -4654,6 +4656,9 @@ Delegation rules: ? targets.length : Math.min(Math.max(minCompleted, 1), targets.length); const deadline = Date.now() + timeoutSeconds * 1000; + const waitAbort = new AbortController(); + this.steeringWaitAbort = waitAbort; + if (this.pendingSteering.size) waitAbort.abort(); this.beginDelegationWait(targets); let timedOut = false; try { @@ -4661,11 +4666,13 @@ Delegation rules: targets, targetCompleted, deadline, - signal, + AbortSignal.any([waitAbort.signal, ...(signal ? [signal] : [])]), ); } finally { + if (this.steeringWaitAbort === waitAbort) this.steeringWaitAbort = undefined; this.endDelegationWait(); } + const steered = waitAbort.signal.aborted && !signal?.aborted; // A settled report included in this bounded result reached the parent; // the idle resume must not deliver it a second time. Omitted reports // stay pending so the idle resume can deliver them later. Running @@ -4685,7 +4692,9 @@ Delegation rules: ? formatDelegationHeartbeat(record) : (record.result?.report ?? `(${record.status} without a report)`), })); - const note = timedOut + const note = steered + ? "Wait interrupted by new user input. Process that input now; unfinished delegates keep working." + : timedOut ? `Still running after ${timeoutSeconds}s: ${results.filter((r) => r.status !== "running").length}/${targets.length} finished. This is not a failure โ€” unfinished delegates keep working and the runtime will deliver their reports when they finish. Call TaskStop only to cancel.\n${targets .filter((record) => record.status === "running") .map(formatDelegationHeartbeat) @@ -4717,7 +4726,7 @@ Delegation rules: }, ], details: { - status: timedOut ? "timeout" : "completed", + status: steered ? "interrupted" : timedOut ? "timeout" : "completed", ...(unknownIds.length ? { unknownIds } : {}), delegations: results, }, @@ -4740,6 +4749,7 @@ Delegation rules: const settledCount = () => targets.filter((record) => record.status !== "running").length; if (settledCount() >= targetCompleted) return Promise.resolve(false); + if (signal?.aborted) return Promise.resolve(true); return new Promise((resolve) => { let done = false; const finish = (timedOut: boolean) => { @@ -5258,7 +5268,7 @@ Delegation rules: error: ReturnType, phase: "request" | "stream", ): number | undefined { - if (!error.retriable) return undefined; + if (!error.retriable || this.gracefulStopRequested) return undefined; if (error.code === "PROVIDER_RATE_LIMITED") { if ( this.providerRateLimitRetryAttempt >= @@ -5448,7 +5458,16 @@ Delegation rules: retryDelayMs: delayMs, error: this.retryActivityError(retryError), }); - await delayWithAbort(delayMs, this.providerRetryAbort.signal); + try { + await delayWithAbort(delayMs, AbortSignal.any([ + this.providerRetryAbort.signal, this.gracefulStop.signal, + ])); + } catch (error) { + if (!this.gracefulStopRequested || this.runCancelled || this.disposed || + !(error instanceof Error) || error.name !== "AbortError") throw error; + return; + } + if (this.gracefulStopRequested) return; if (this.disposed) throw new Error("runtime disposed"); // The failed attempt has already finished. Only its lifecycle events // are suppressed; the retry must close the visible run normally. @@ -5521,6 +5540,19 @@ Delegation rules: this.pendingSilentTurnRerun || this.pendingProgressTurnRerun ) { + if (this.gracefulStopRequested) { + if (this.pendingOverflow) { + this.terminateParentTurn(); + this.emit({ type: "error", error: { + code: "CONTEXT_TOO_LARGE", + message: "The provider rejected the model context before the turn was stopped", + retriable: false, + } }); + this.finishAgentRun(); + return false; + } + return true; + } if (this.pendingProviderRetry) { await this.retryPendingProviderFailure(); continue; @@ -5559,6 +5591,7 @@ Delegation rules: return false; } this.turnHadError = false; + if (this.gracefulStopRequested) return true; this.requestStartedAt = Date.now(); await this.agent.continue(); await this.waitForIdleAndSteering(); @@ -6862,6 +6895,7 @@ Delegation rules: // answer is never rendered. Re-run once with a nudge before letting // that surface as a finished turn. const silence = + !this.gracefulStopRequested && !failed && !aborted && responseText.trim().length === 0 && @@ -6918,6 +6952,7 @@ Delegation rules: // Autonomous plan/goal: clearly forward-looking text without a tool // call is probably progress, not a final answer. Nudge once (#43). const progressOnlyTurn = + !this.gracefulStopRequested && !failed && !aborted && !silentTurn && @@ -7017,6 +7052,7 @@ Delegation rules: this.streamStartedAt = undefined; this.currentAssistant = undefined; const canRecoverOverflow = + !this.gracefulStopRequested && this.compactionEnabled && overflow && !this.overflowRecoveryAttempted; @@ -7033,7 +7069,7 @@ Delegation rules: } } else if (!failed && !aborted && !emptyResponse) { this.appendLiveEntry(assistantId, event.message); - } else { + } else if (!(aborted && this.gracefulStopRequested && !this.runCancelled)) { this.turnHadError = true; } if (canRecoverOverflow) { @@ -7123,6 +7159,7 @@ Delegation rules: } break; case "turn_end": + if (this.gracefulStopRequested && !this.turnHadError) break; if ( this.suppressOverflowRunEnd || this.suppressProviderRetryRunEnd || @@ -7139,6 +7176,7 @@ Delegation rules: }); break; case "agent_end": + if (this.gracefulStopRequested && !this.turnHadError) break; // Input admitted after pi's last queue poll still belongs to this turn. // Continue after the current run settles; never wake the follow-up FIFO. if (this.pendingSteering.size && this.acceptingSteering && !this.runCancelled && !this.turnHadError) break; @@ -7150,21 +7188,40 @@ Delegation rules: this.keepTurnOpenForDelegates() ) break; - this.acceptingSteering = false; - this.retainPendingSteering(); - this.autonomousExecution = false; - this.clearAgentActivity(); - this.reportMutationTermination(); - this.emit({ - type: "agent_end", - messageIds: [], - }); + this.finishAgentRun(); break; default: break; } } + private finishAgentRun(): void { + this.acceptingSteering = false; + this.retainPendingSteering(); + this.autonomousExecution = false; + this.clearAgentActivity(); + this.reportMutationTermination(); + this.emit({ type: "agent_end", messageIds: [] }); + } + + /** Settle the durable turn across pi loops, recovery, and delegate reports. */ + private async settlePendingRun(): Promise { + if (!(await this.runPendingRecoveries())) return; + if (this.turnHadError) { + this.terminateParentTurn(); + return; + } + await this.resumeAfterDelegations(); + if ( + !this.gracefulStopRequested || this.runCancelled || this.turnHadError || this.disposed + ) return; + this.finalizeCurrentAssistant("aborted"); + const subagentUsage = this.turnSubagentUsage; + this.turnSubagentUsage = undefined; + this.emit({ type: "turn_end", ...(subagentUsage ? { subagentUsage } : {}) }); + this.finishAgentRun(); + } + /** * The recovery guard stops the agent loop by terminating the tool batch, so * the turn would otherwise end on a failed tool card with nothing said. Give @@ -7304,7 +7361,7 @@ Delegation rules: this.turnId = durableTurnId; this.acceptingSteering = true; this.pendingUserMessageId = undefined; - this.gracefulStopRequested = false; + this.gracefulStop = new AbortController(); this.runCancelled = false; this.resetRunRecoveryState(); this.turnEpoch += 1; @@ -7355,12 +7412,7 @@ Delegation rules: // Same recovery contract as a user prompt: a plan execution that overflows, // hits a retriable stream failure, or comes back silent must not end as a // run with no end events at all. - if (!(await this.runPendingRecoveries())) return { turnId: this.turnId }; - if (this.turnHadError) { - this.terminateParentTurn(); - return { turnId: this.turnId }; - } - await this.resumeAfterDelegations(); + await this.settlePendingRun(); return { turnId: this.turnId }; } @@ -7379,7 +7431,7 @@ Delegation rules: this.hostTurnId = nextTurnId; this.turnId = nextTurnId; this.acceptingSteering = true; - this.gracefulStopRequested = false; + this.gracefulStop = new AbortController(); this.runCancelled = false; this.turnSubagentUsage = undefined; // Capabilities and path-scoped instruction claims belong to one prompt. @@ -7432,7 +7484,12 @@ Delegation rules: return { turnId: this.turnId }; } } - await this.extensionBeforeAgentStart(modelInput); + if (!this.gracefulStopRequested) await this.extensionBeforeAgentStart(modelInput); + if (this.gracefulStopRequested) { + this.keepPreflightUserMessage(incomingUserMessage); + await this.settlePendingRun(); + return { turnId: this.turnId }; + } if (typeof modelInput === "string") { await this.agent.prompt(modelInput); } else { @@ -7441,12 +7498,7 @@ Delegation rules: await this.waitForIdleAndSteering(); void this.extensionRunner?.emit("agent_settled", { type: "agent_settled" }); - if (!(await this.runPendingRecoveries())) return { turnId: this.turnId }; - if (this.turnHadError) { - this.terminateParentTurn(); - return { turnId: this.turnId }; - } - await this.resumeAfterDelegations(); + await this.settlePendingRun(); } catch (err) { const classifiedError = classifyAgentError(err); const diagnosticError = @@ -7614,7 +7666,7 @@ Delegation rules: async abort(): Promise { this.acceptingSteering = false; - this.gracefulStopRequested = false; + this.gracefulStop = new AbortController(); this.runCancelled = true; this.resolvePendingAskTools(); this.abortRunningDelegations(); @@ -7627,11 +7679,13 @@ Delegation rules: /** Ask pi-agent-core to stop after the current assistant/tool turn. */ requestGracefulStop(): { requested: boolean } { - if (this.disposed || !this.agent.state.isStreaming) { + if ( + this.disposed || !this.getStatus().isRunning + ) { return { requested: false }; } this.acceptingSteering = false; - this.gracefulStopRequested = true; + this.gracefulStop.abort(); return { requested: true }; } @@ -7671,7 +7725,7 @@ Delegation rules: this.terminatingToolCalls.clear(); this.delegateToolCalls.clear(); this.appendedDelegationRowIds.clear(); - this.gracefulStopRequested = false; + this.gracefulStop = new AbortController(); this.hostCloseUnsubscribe?.(); this.hostCloseUnsubscribe = undefined; this.agent.abort(); From fa1442797d0e0fc23a4dbb8015164e7487568875 Mon Sep 17 00:00:00 2001 From: zszz3 <91608029+zszz3@users.noreply.github.com> Date: Sat, 19 Sep 2026 21:52:34 +0800 Subject: [PATCH 2/2] refactor(persistence): move outbox wiring out of main entry Keep persistence construction and transcript settlement together so the main entry remains within its architecture budget. Preserve lazy host and shutdown reads and the single-instance initialization boundary. --- apps/desktop/electron/main/index.ts | 8 ++------ apps/desktop/electron/main/persistence-outbox.ts | 15 +++++++++++++++ apps/desktop/test/single-instance.test.mjs | 2 +- 3 files changed, 18 insertions(+), 7 deletions(-) diff --git a/apps/desktop/electron/main/index.ts b/apps/desktop/electron/main/index.ts index b9968d2f5d..a109fa64ad 100644 --- a/apps/desktop/electron/main/index.ts +++ b/apps/desktop/electron/main/index.ts @@ -49,7 +49,7 @@ import { shouldCreateTaskNotification as shouldCreateTaskNotificationPolicy, shouldShowNativeNotification, } from "./notification-policy"; -import { PersistenceOutbox } from "./persistence-outbox"; +import { createPersistenceRuntime } from "./persistence-outbox"; import { AgentSidecar } from "./agent-sidecar"; import { Logger, ignoreBrokenStdio } from "./logger"; import { installMainProcessErrorHandlers } from "./main-process-errors"; @@ -552,9 +552,7 @@ installMainProcessErrorHandlers({ }, }); -const persistenceOutbox = new PersistenceOutbox(dataDir, (level, message, data) => { - logger.app("persistence", level, message, { data }); -}); +const { persistenceOutbox, settleTranscript } = createPersistenceRuntime(dataDir, logger, () => host, () => quitting); const steeringReplies = new Set(); const scheduledRuntime = createScheduledRuntime({ dataDir, @@ -1117,8 +1115,6 @@ const sessionCollaboration = createSessionCollaborationService({ log: (message, data) => logger.app("runtime", "warn", message, { data }), }); -const settleTranscript = (sessionId: string) => - persistenceOutbox.drainSession(sessionId, () => host, () => quitting); const planRuntime = createPlanRuntime({ runtimeState, planState: planRuntimeState, diff --git a/apps/desktop/electron/main/persistence-outbox.ts b/apps/desktop/electron/main/persistence-outbox.ts index 44bfae3b6d..d6a9a85062 100644 --- a/apps/desktop/electron/main/persistence-outbox.ts +++ b/apps/desktop/electron/main/persistence-outbox.ts @@ -1,6 +1,7 @@ import { mkdir, readFile, rename, writeFile } from "node:fs/promises"; import { dirname, join } from "node:path"; import type { HostProcess } from "./host-process"; +import type { Logger } from "./logger"; type MessageAppend = { key: string; @@ -219,3 +220,17 @@ function isDuplicateMessageIdError(error: unknown): boolean { function isPoisonMessageError(error: unknown): boolean { return /(?, + getHost: () => HostProcess | null, + isStopping: () => boolean, +) { + const persistenceOutbox = new PersistenceOutbox(dataDir, (level, message, data) => { + logger.app("persistence", level, message, { data }); + }); + const settleTranscript = (sessionId: string) => + persistenceOutbox.drainSession(sessionId, getHost, isStopping); + return { persistenceOutbox, settleTranscript }; +} diff --git a/apps/desktop/test/single-instance.test.mjs b/apps/desktop/test/single-instance.test.mjs index c8213fa25b..39770e2a82 100644 --- a/apps/desktop/test/single-instance.test.mjs +++ b/apps/desktop/test/single-instance.test.mjs @@ -19,7 +19,7 @@ test("the single-instance lock is taken before anything touches the data directo assert.ok(lock > 0, "main must request the single-instance lock"); assert.ok(mainSource.indexOf("app.setName(APP_NAME)") < lock); assert.ok(lock < mainSource.indexOf("new Logger(")); - assert.ok(lock < mainSource.indexOf("new PersistenceOutbox(")); + assert.ok(lock < mainSource.indexOf("createPersistenceRuntime(")); }); test("a launch that loses the lock quits and boots nothing", () => {