fix(queue): 立即发送追加到当前轮并保证消息落盘顺序

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<string>();
 const scheduledRuntime = createScheduledRuntime({
   dataDir,
@@ -1140,6 +1138,7 @@ const planRuntime = createPlanRuntime({
   resolveAgentRuntimeLaunch,
   isQuitting: () => quitting,
   onTurnSettled: sessionCollaboration.settle,
+  settleTranscript,
 });
 const {
   finishTurn,
@@ -1248,6 +1247,7 @@ const { bootHostStatus, runtimeArch, bootBackends } = runtimeLifecycle;
 
 function registerIpc() {
   return registerIpcHandlers({
+    settleTranscript,
     traySessions: applicationLifecycle!.traySessions,
     ipcMain,
     getMainWindow: () => mainWindow,

apps/desktop/electron/main/ipc/agent-ipc.ts

@@ -34,6 +34,7 @@ export type AgentIpcDependencies = {
   resolveAgentRuntimeLaunch: (...args: any[]) => Promise<any>;
   acquireSessionOperation: (sessionId: string) => Promise<() => void>;
   finishTurn: FinishTurn;
+  settleTranscript: (sessionId: string) => Promise<boolean>;
   /**
    * 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

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,

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;
@@ -11,7 +12,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 +27,7 @@ export class PersistenceOutbox {
   private flushing: Promise<void> | null = null;
   private persistChain = Promise.resolve();
   private readonly loaded: Promise<void>;
+  private readonly enqueuing = new Map<Promise<void>, string>();
 
   constructor(dataDir: string, logger: OutboxLogger) {
     this.path = join(dataDir, "session-message-outbox.json");
@@ -34,35 +36,72 @@ export class PersistenceOutbox {
     this.loaded = this.load();
   }
 
-  async enqueue(
+  enqueue(entry: MessageAppend, getHost: () => HostProcess | null): Promise<void> {
+    // 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<void> {
     await this.loaded;
     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);
     }
     await this.persist();
     void this.flush(getHost);
   }
 
-  async flush(getHost: () => HostProcess | null): Promise<void> {
+  async flush(getHost: () => HostProcess | null, sessionId?: string): Promise<void> {
     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<boolean> {
+    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<void>((resolve) => { setTimeout(resolve, 1000).unref(); });
+    }
   }
 
   /**
@@ -81,9 +120,10 @@ export class PersistenceOutbox {
     return this.entries.length;
   }
 
-  private async flushLoop(getHost: () => HostProcess | null): Promise<void> {
-    while (this.entries.length > 0) {
-      const current = this.entries[0];
+  private async flushLoop(getHost: () => HostProcess | null, sessionId?: string): Promise<void> {
+    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 +160,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();
     }
   }
@@ -179,3 +220,17 @@ function isDuplicateMessageIdError(error: unknown): boolean {
 function isPoisonMessageError(error: unknown): boolean {
   return /(?<![A-Z_])PERMISSION_DENIED:/i.test(String(error));
 }
+
+export function createPersistenceRuntime(
+  dataDir: string,
+  logger: Pick<Logger, "app">,
+  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 };
+}

apps/desktop/electron/main/runtime/plans.ts

@@ -65,6 +65,7 @@ export type PlanRuntimeDependencies = {
   acquireSessionOperation: (sessionId: string) => Promise<() => void>;
   resolveAgentRuntimeLaunch: (...args: any[]) => Promise<any>;
   isQuitting: () => boolean;
+  settleTranscript: (sessionId: string) => Promise<boolean>;
   onTurnSettled?: (sessionId: string, turnId: string) => Promise<void>;
 };
 
@@ -91,6 +92,7 @@ export function createPlanRuntime({
   resolveAgentRuntimeLaunch,
   isQuitting,
   onTurnSettled,
+  settleTranscript,
 }: PlanRuntimeDependencies): {
   finishTurn: FinishTurn;
   finishApprovedExecution: (executionId: string, status: PlanExecutionFinishStatus, errorCode?: string) => Promise<void>;
@@ -196,6 +198,9 @@ function finishTurn(
 
   const runFinalization = async (): Promise<void> => {
     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<{

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(

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(

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;

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", () => {

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
 

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