[agent] 🤖 fix: preserve queue-cut continuations

src/browser/stores/WorkspaceStore.test.ts

@@ -2230,6 +2230,75 @@ describe("WorkspaceStore", () => {
       expect(store.getStreamingMessage(workspaceId, secondRow.id, secondMessageId)).not.toBeNull();
     });
 
+    it("keeps the replacement channel for a cleanup-only abort", async () => {
+      const workspaceId = "cleanup-only-abort-channel";
+      const oldMessageId = "old-stream";
+      const replacementMessageId = "replacement-stream";
+      createAndAddWorkspace(store, workspaceId);
+      const rawStore = getInternal<{
+        streamingMessageStore: { has: (key: string) => boolean };
+        handleChatMessage: (id: string, event: WorkspaceChatMessage) => void;
+        processStreamEvent: (
+          id: string,
+          aggregator: ReturnType<WorkspaceStore["getAggregator"]>,
+          event: WorkspaceChatMessage
+        ) => void;
+      }>(store);
+      const dispatch = (event: WorkspaceChatMessage) =>
+        rawStore.processStreamEvent(workspaceId, store.getAggregator(workspaceId), event);
+
+      rawStore.handleChatMessage(workspaceId, {
+        type: "stream-start",
+        workspaceId,
+        messageId: oldMessageId,
+        historySequence: 1,
+        model: TEST_MODEL,
+        startTime: 1,
+      });
+      rawStore.handleChatMessage(workspaceId, caughtUpEvent());
+      dispatch({
+        type: "stream-start",
+        workspaceId,
+        messageId: replacementMessageId,
+        historySequence: 2,
+        model: TEST_MODEL,
+        startTime: 2,
+      });
+      for (const [delta, timestamp] of [
+        ["hello", 3],
+        [" world", 4],
+      ] as const) {
+        dispatch({
+          type: "stream-delta",
+          workspaceId,
+          messageId: replacementMessageId,
+          delta,
+          tokens: 1,
+          timestamp,
+        });
+      }
+      await new Promise<void>((resolve) => queueMicrotask(resolve));
+
+      const replacementRow = store
+        .getAggregator(workspaceId)!
+        .getDisplayedMessages()
+        .find((message) => "historyId" in message && message.historyId === replacementMessageId)!;
+      const replacementKey = `${workspaceId}\u0000${replacementRow.id}`;
+      expect(rawStore.streamingMessageStore.has(replacementKey)).toBe(true);
+
+      dispatch({
+        type: "stream-abort",
+        workspaceId,
+        messageId: oldMessageId,
+        abortReason: "user",
+        rendererCleanupOnly: true,
+      });
+
+      expect(rawStore.streamingMessageStore.has(replacementKey)).toBe(true);
+      expect(store.getAggregator(workspaceId)!.isStreamActive(oldMessageId)).toBe(false);
+      expect(store.getAggregator(workspaceId)!.isStreamActive(replacementMessageId)).toBe(true);
+    });
+
     it("releases the keyed channel when a background activity stop clears the stream", async () => {
       const workspaceId = "keyed-channel-background-stop";
       createAndAddWorkspace(store, workspaceId);

src/browser/stores/WorkspaceStore.ts

@@ -979,6 +979,13 @@ export class WorkspaceStore {
     "stream-abort": (workspaceId, aggregator, data) => {
       const streamAbortData = data as StreamAbortEvent;
       applyWorkspaceChatEventToAggregator(aggregator, streamAbortData);
+      if (streamAbortData.rendererCleanupOnly === true) {
+        // This delayed event only closes its old message. Keep replacement stream state intact.
+        this.cancelPendingStreamingBump(workspaceId, streamAbortData.messageId);
+        this.states.bump(workspaceId);
+        this.streamingStatsStore.bump(workspaceId);
+        return;
+      }
       this.releaseStreamingMessageChannel(workspaceId);
 
       // Track stream interruption telemetry (get model from aggregator)
@@ -1735,7 +1742,13 @@ export class WorkspaceStore {
     });
   }
 
-  private cancelPendingStreamingBump(workspaceId: string): void {
+  private cancelPendingStreamingBump(workspaceId: string, messageId?: string): void {
+    if (
+      messageId !== undefined &&
+      this.pendingStreamingMessageBump.get(workspaceId) !== messageId
+    ) {
+      return;
+    }
     this.pendingStreamingMessageBump.delete(workspaceId);
   }
 

src/browser/utils/messages/StreamingMessageAggregator.test.ts

@@ -3932,6 +3932,47 @@ describe("StreamingMessageAggregator", () => {
   });
 
   describe("abort reason tracking", () => {
+    test("cleanup-only abort preserves a replacement stream", () => {
+      const aggregator = new StreamingMessageAggregator(TEST_CREATED_AT);
+      aggregator.handleStreamStart({
+        type: "stream-start",
+        workspaceId: "test-workspace",
+        messageId: "old-stream",
+        historySequence: 1,
+        model: "claude-3-5-sonnet-20241022",
+        startTime: 1,
+      });
+      aggregator.setInterrupting();
+      aggregator.handleStreamStart({
+        type: "stream-start",
+        workspaceId: "test-workspace",
+        messageId: "replacement-stream",
+        historySequence: 2,
+        model: "claude-3-5-sonnet-20241022",
+        startTime: 2,
+      });
+      aggregator.handleStreamLifecycle({
+        type: "stream-lifecycle",
+        workspaceId: "test-workspace",
+        phase: "streaming",
+        hadAnyOutput: false,
+      });
+
+      aggregator.handleStreamAbort({
+        type: "stream-abort",
+        workspaceId: "test-workspace",
+        messageId: "old-stream",
+        abortReason: "user",
+        rendererCleanupOnly: true,
+      });
+
+      expect(aggregator.isStreamActive("old-stream")).toBe(false);
+      expect(aggregator.isStreamActive("replacement-stream")).toBe(true);
+      expect(aggregator.hasInterruptingStream()).toBe(false);
+      expect(aggregator.getStreamLifecycle()?.phase).toBe("streaming");
+      expect(aggregator.getLastAbortReason()).toBeNull();
+    });
+
     test("stores last abort reason and clears on stream-start", () => {
       const aggregator = new StreamingMessageAggregator(TEST_CREATED_AT);
 

src/browser/utils/messages/StreamingMessageAggregator.ts

@@ -2402,13 +2402,16 @@ export class StreamingMessageAggregator {
   }
 
   handleStreamAbort(data: StreamAbortEvent): void {
-    // Abort can arrive before stream-start. Clear pending lifecycle UI immediately.
-    this.clearPendingStreamLifecycleState();
-    this.clearInFlightStreamLifecycle();
-    this.lastAbortReason = {
-      reason: data.abortReason ?? "system",
-      at: Date.now(),
-    };
+    const rendererCleanupOnly = data.rendererCleanupOnly === true;
+    if (!rendererCleanupOnly) {
+      // Abort can arrive before stream-start. Clear pending lifecycle UI immediately.
+      this.clearPendingStreamLifecycleState();
+      this.clearInFlightStreamLifecycle();
+      this.lastAbortReason = {
+        reason: data.abortReason ?? "system",
+        at: Date.now(),
+      };
+    }
 
     // Clear "interrupting" state - stream is now fully "interrupted"
     if (this.interruptingMessageId === data.messageId) {

src/common/constants/paths.ts

@@ -37,6 +37,12 @@ export const TIMELINE_FILE_NAME = "timeline.jsonl";
  */
 export const CHAT_ARCHIVE_FILE_NAME = "chat-archive.jsonl";
 
+/**
+ * Per-workspace tombstones for synthetic history rows that must stay out of
+ * provider requests when the primary history rewrite fails.
+ */
+export const PROVIDER_EXCLUDED_MESSAGE_IDS_FILE_NAME = "provider-excluded-message-ids.jsonl";
+
 /**
  * Per-workspace sidecar recording headless AI usage (status generation,
  * memory consolidation/harvest) that produces no chat.jsonl assistant row.

src/common/orpc/schemas/stream.ts

@@ -336,12 +336,14 @@ export const StreamAbortEventSchema = z.object({
       // Last step's provider metadata (for context window cache display)
       contextProviderMetadata: z.record(z.string(), z.unknown()).optional(),
       duration: z.number().optional(),
+      abortTurnGeneration: z.number().int().nonnegative().optional(),
     })
     .optional()
     .meta({
       description: "Metadata may contain usage if abort occurred after stream completed processing",
     }),
   abandonPartial: z.boolean().optional(),
+  rendererCleanupOnly: z.boolean().optional(),
   acpPromptId: z
     .string()
     .optional()

src/common/types/message.ts

@@ -478,6 +478,24 @@ export function filterOrphanedMcpPromptSnapshots(messages: MuxMessage[]): MuxMes
   });
 }
 
+/** Identify a synthetic row that a user Stop canceled after its rollback boundary. */
+export function isProviderExcludedMessage(message: MuxMessage): boolean {
+  return (
+    message.metadata?.providerExcluded === true &&
+    message.metadata.synthetic === true &&
+    message.metadata.contextBoundaryKind == null &&
+    message.metadata.compactionBoundary !== true
+  );
+}
+
+/** Remove durable rows that a user Stop canceled after their rollback boundary. */
+export function filterProviderExcludedMessages(messages: MuxMessage[]): MuxMessage[] {
+  if (!messages.some(isProviderExcludedMessage)) {
+    return messages;
+  }
+  return messages.filter((message) => !isProviderExcludedMessage(message));
+}
+
 export function dedupeMcpPromptRefs(refs: MCPPromptReference[]): MCPPromptReference[] {
   const deduped = new Map<string, MCPPromptReference>();
   for (const ref of refs) {
@@ -929,6 +947,8 @@ export interface MuxMetadata {
   systemMessageTokens?: number; // Token count for system message sent with this request (calculated by AIService)
   partial?: boolean; // Whether this message was interrupted and is incomplete
   synthetic?: boolean; // Whether this message was synthetically generated (e.g., [CONTINUE] sentinel)
+  /** Keep a canceled durable admission in history, but exclude it from every provider request. */
+  providerExcluded?: boolean;
   /**
    * For queue-dispatched user turns: when the user last added to the queued
    * entry. The row `timestamp` is stamped at dispatch (after the blocking turn

src/common/utils/messages/compactionBoundary.ts

@@ -6,7 +6,7 @@ import {
 import { isPositiveInteger } from "@/common/utils/numbers";
 import { hasProviderReplayableContent } from "@/common/utils/messages/providerEligibility";
 
-import type { MuxMessage } from "@/common/types/message";
+import { isProviderExcludedMessage, type MuxMessage } from "@/common/types/message";
 
 export { CONTEXT_BOUNDARY_KINDS, type ContextBoundaryKind };
 
@@ -129,7 +129,7 @@ export function sliceMessagesFromLatestCompactionBoundary(messages: MuxMessage[]
 }
 
 export function isProviderEligibleMessage(message: MuxMessage): boolean {
-  if (isDurableContextResetBoundaryMarker(message)) {
+  if (isProviderExcludedMessage(message) || isDurableContextResetBoundaryMarker(message)) {
     return false;
   }