🤖 fix: resume turns stranded by a withdrawn queued tool-end message

src/common/orpc/schemas.ts

@@ -253,6 +253,7 @@ export {
   ErrorEventSchema,
   GoalBudgetLimitedEventSchema,
   LanguageModelV2UsageSchema,
+  ModelFallbackProgressSchema,
   OnChatDowngradeReasonSchema,
   QueuedMessageChangedEventSchema,
   ReasoningDeltaEventSchema,

src/common/orpc/schemas/stream.ts

@@ -292,7 +292,10 @@ export const StreamEndEventSchema = z.object({
   }),
 });
 
-export const StreamAbortReasonSchema = z.enum(["user", "startup", "system"]);
+// "queued-message": the backend's own soft stop at a provider-executed tool boundary so a queued
+// tool-end message can dispatch; distinct from "system" so a concurrent hard stop cannot be
+// mistaken for it.
+export const StreamAbortReasonSchema = z.enum(["user", "startup", "system", "queued-message"]);
 
 export const StreamLifecyclePhaseSchema = z.enum([
   "idle",
@@ -320,6 +323,12 @@ export const StreamLifecycleEventSchema = StreamLifecycleSnapshotSchema.extend({
   workspaceId: z.string(),
 });
 
+// Refusal-fallback chain a turn runs under and how far along it is. A stream that resumes a cut
+// turn continues this chain instead of resolving one from the model it resumes on.
+export const ModelFallbackProgressSchema = ModelFallbackRecordSchema.extend({
+  chain: z.array(z.string()),
+});
+
 export const StreamAbortEventSchema = z.object({
   type: z.literal("stream-abort"),
   workspaceId: z.string(),
@@ -336,6 +345,13 @@ 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(),
+      // Model active at the abort (a configured fallback may differ from the requested model)
+      model: z.string().optional(),
+      // Steps left under the stream's ceiling at the abort; a turn cut for a queued message
+      // resumes under this budget rather than a fresh one.
+      stepsRemaining: z.number().int().nonnegative().optional(),
+      // Fallback chain state at the abort, carried into the resumed stream for the same reason.
+      modelFallbackProgress: ModelFallbackProgressSchema.optional(),
     })
     .optional()
     .meta({

src/common/types/message.ts

@@ -9,6 +9,7 @@ import type {
 } from "@/common/constants/contextBoundary";
 import type { GoalSyntheticMessageKind } from "@/constants/goals";
 import type { SendMessageOptions } from "@/common/orpc/types";
+import type { ModelFallbackProgress } from "./stream";
 import { withLegacyPtcExclusiveMirror } from "@/common/constants/experiments";
 import type { z } from "zod";
 import type { AgentMode } from "./mode";
@@ -220,6 +221,14 @@ export interface CompactionFollowUpRequest extends CompactionFollowUpInput, Pres
   goalId?: string;
   /** Internal dispatch guardrails for crash-safe follow-up recovery. */
   dispatchOptions?: CompactionFollowUpDispatchOptions;
+  /**
+   * What the turn interrupted for mid-stream compaction had left of its step ceiling, the
+   * fallback chain state it reached, and whether it ran under admission revalidation: the
+   * follow-up continues that turn, not a fresh one.
+   */
+  stepBudget?: number;
+  modelFallbackProgress?: ModelFallbackProgress;
+  revalidateAdmission?: boolean;
   /**
    * Open delegated workspace-turn correlation captured before on-send
    * compaction consumed this follow-up (e.g. a bash-monitor wake continuing a

src/common/types/stream.ts

@@ -11,6 +11,7 @@ import type {
   AutoRetryScheduledEventSchema,
   AutoRetryStartingEventSchema,
   ErrorEventSchema,
+  ModelFallbackProgressSchema,
   ReasoningDeltaEventSchema,
   ReasoningEndEventSchema,
   StreamAbortReasonSchema,
@@ -45,6 +46,7 @@ export type StreamStartEvent = z.infer<typeof StreamStartEventSchema>;
 export type StreamDeltaEvent = z.infer<typeof StreamDeltaEventSchema>;
 export type StreamEndEvent = z.infer<typeof StreamEndEventSchema>;
 export type StreamAbortReason = z.infer<typeof StreamAbortReasonSchema>;
+export type ModelFallbackProgress = z.infer<typeof ModelFallbackProgressSchema>;
 export type StreamLifecyclePhase = z.infer<typeof StreamLifecyclePhaseSchema>;
 export type StreamLifecycleSnapshot = z.infer<typeof StreamLifecycleSnapshotSchema>;
 export type StreamLifecycleEvent = z.infer<typeof StreamLifecycleEventSchema>;

src/node/services/agentSession.autoCompaction.test.ts

@@ -1160,8 +1160,14 @@ describe("AgentSession on-send auto-compaction snapshot deferral", () => {
     session.dispose();
   });
 
-  test("hides default follow-up sentinel in mid-stream auto-compaction prompts", async () => {
+  test("mid-stream auto-compaction hides the default follow-up sentinel and hands over the interrupted turn's remainder", async () => {
     const workspaceId = "ws-auto-compaction-mid-stream-sentinel";
+    // The interrupted stream had already moved down its fallback chain and spent steps.
+    const interruptedProgress = {
+      requestedModel: "openai:gpt-4o",
+      refusedModels: ["openai:gpt-4o"],
+      chain: ["openai:gpt-4o-fallback"],
+    };
 
     const { historyService, cleanup } = await createTestHistoryService();
     historyCleanup = cleanup;
@@ -1211,6 +1217,11 @@ describe("AgentSession on-send auto-compaction snapshot deferral", () => {
         workspaceId,
         messageId: "assistant-mid-stream",
         abortReason: "system",
+        metadata: {
+          model: "openai:gpt-4o-fallback",
+          stepsRemaining: 7,
+          modelFallbackProgress: interruptedProgress,
+        },
       });
 
       return Promise.resolve(Ok(undefined));
@@ -1275,14 +1286,18 @@ describe("AgentSession on-send auto-compaction snapshot deferral", () => {
       ownerWorkspaceId: "parent-mid-stream-compaction",
       turnId: "turn-mid-stream-compaction",
     } as const;
-    const result = await session.sendMessage(
-      "hello",
+    // The interrupted turn is a revalidated resume (a stranded delegated turn's continuation).
+    await historyService.appendToHistory(
+      workspaceId,
+      createMuxMessage("user-hello", "user", "hello", { timestamp: Date.now() })
+    );
+    const result = await session.resumeStream(
       {
         model: "openai:gpt-4o",
         agentId: "exec",
         muxMetadata: workspaceTurnMetadata,
       },
-      { agentInitiated: true }
+      { agentInitiated: true, revalidateAdmission: true }
     );
 
     expect(result.success).toBe(true);
@@ -1309,6 +1324,14 @@ describe("AgentSession on-send auto-compaction snapshot deferral", () => {
       workspaceTurnMetadata
     );
     expect(compactionRequestMetadata.parsed.followUpContent?.agentInitiated).toBe(true);
+    // The follow-up continues the interrupted turn: on the model it reached, under what it had
+    // left of the ceiling, with the refusals so far.
+    expect(compactionRequestMetadata.parsed.followUpContent).toMatchObject({
+      model: "openai:gpt-4o-fallback",
+      stepBudget: 7,
+      modelFallbackProgress: interruptedProgress,
+      revalidateAdmission: true,
+    });
 
     const compactionRequestText =
       compactionRequestMessage?.parts.find((part) => part.type === "text")?.text ?? "";

src/node/services/agentSession.continueMessageAgentId.test.ts

@@ -22,14 +22,26 @@ type SendMessageResult =
 interface AutoRetryResumeRequest {
   options: SendMessageOptions;
   agentInitiated?: boolean;
+  stepBudget?: number;
+  modelFallbackProgress?: unknown;
+  revalidateAdmission?: boolean;
+}
+
+interface SendInternal {
+  synthetic?: boolean;
+  agentInitiated?: boolean;
+  stepBudget?: number;
+  modelFallbackProgress?: unknown;
+  revalidateAdmission?: boolean;
+  refuseStreamStart?: () => boolean;
 }
 
 interface SessionInternals {
   dispatchPendingFollowUp: () => Promise<boolean>;
   sendMessage: (
     message: string,
     options?: SendOptions,
-    internal?: { synthetic?: boolean; agentInitiated?: boolean }
+    internal?: SendInternal
   ) => Promise<SendMessageResult>;
   scheduleStartupRecovery: () => void;
   startupRecoveryPromise: Promise<void> | null;
@@ -150,7 +162,13 @@ describe("AgentSession continue-message agentId fallback", () => {
     historyCleanup = undefined;
   });
 
-  const createSession = async (messages: MuxMessage[] = []) => {
+  const createSession = async (
+    messages: MuxMessage[] = [],
+    turnOptions?: Pick<
+      ConstructorParameters<typeof AgentSession>[0],
+      "admitStrandedTurnResume" | "settleForfeitedWorkspaceTurnContinuation"
+    >
+  ) => {
     const { historyService, cleanup } = await createTestHistoryService();
     historyCleanup = cleanup;
     for (const message of messages) {
@@ -164,6 +182,7 @@ describe("AgentSession continue-message agentId fallback", () => {
       aiService: createAiService(),
       initStateManager: createInitStateManager(),
       backgroundProcessManager: createBackgroundProcessManager(),
+      ...turnOptions,
     });
     sessions.push(session);
 
@@ -266,6 +285,173 @@ describe("AgentSession continue-message agentId fallback", () => {
     expect(internals.lastAutoRetryResumeRequest?.agentInitiated).toBe(true);
   });
 
+  test("dispatchPendingFollowUp continues the interrupted turn's step budget, fallback chain, and admission revalidation", async () => {
+    const progress = {
+      requestedModel: "anthropic:claude-sonnet-4-5",
+      refusedModels: ["anthropic:claude-sonnet-4-5"],
+      chain: ["openai:gpt-4o", "google:gemini-fallback"],
+    };
+    const dispatched: SendInternal[] = [];
+    const { internals } = await createSession([
+      compactionSummaryMessage("summary-remainder", {
+        text: "Continue",
+        model: "openai:gpt-4o",
+        agentId: "exec",
+        stepBudget: 7,
+        modelFallbackProgress: progress,
+        revalidateAdmission: true,
+      }),
+    ]);
+    internals.sendMessage = mock(
+      (_message: string, _options?: SendOptions, internal?: SendInternal) => {
+        dispatched.push(internal ?? {});
+        return Promise.resolve({ success: true as const });
+      }
+    );
+
+    await internals.dispatchPendingFollowUp();
+
+    expect(dispatched[0]).toMatchObject({
+      stepBudget: 7,
+      modelFallbackProgress: progress,
+      revalidateAdmission: true,
+    });
+    expect(internals.lastAutoRetryResumeRequest).toMatchObject({
+      stepBudget: 7,
+      modelFallbackProgress: progress,
+      revalidateAdmission: true,
+    });
+  });
+
+  const DELEGATED_TURN = {
+    type: "workspace-turn-task",
+    taskHandleId: "wst_follow_up",
+    ownerWorkspaceId: "owner-ws",
+    turnId: "turn-follow-up",
+  } as const;
+
+  test("dispatchPendingFollowUp discards a follow-up whose interrupted turn spent its step budget", async () => {
+    const settle = mock((_correlation: unknown, _reason: string) => Promise.resolve());
+    const { internals, historyService } = await createSession(
+      [
+        compactionSummaryMessage("summary-spent", {
+          text: "Continue",
+          model: "openai:gpt-4o",
+          agentId: "exec",
+          stepBudget: 0,
+          muxMetadata: DELEGATED_TURN,
+        }),
+      ],
+      { settleForfeitedWorkspaceTurnContinuation: settle }
+    );
+    const sendMessage = mock(() => Promise.resolve({ success: true as const }));
+    internals.sendMessage = sendMessage;
+
+    expect(await internals.dispatchPendingFollowUp()).toBe(false);
+
+    // The ceiling ended the turn; the follow-up is dropped rather than left to redispatch later,
+    // and the delegated turn it continued is settled since no successor stream will end it.
+    expect(sendMessage).not.toHaveBeenCalled();
+    expect(settle).toHaveBeenCalledTimes(1);
+    expect(settle.mock.calls[0]?.[0]).toEqual(DELEGATED_TURN);
+    const tail = await historyService.getLastMessages("ws", 1);
+    expect(tail.success).toBe(true);
+    const summary = tail.success ? tail.data[0] : undefined;
+    expect(summary?.id).toBe("summary-spent");
+    expect(summary?.metadata?.muxMetadata).toEqual({ type: "compaction-summary" });
+  });
+
+  test("dispatchPendingFollowUp admits a delegated turn's follow-up like a stranded resume", async () => {
+    const settle = mock((_correlation: unknown, _reason: string) => Promise.resolve());
+    let stale = false;
+    const admit = mock((_correlation: unknown) =>
+      Promise.resolve({ admissible: true, admissionStale: () => stale })
+    );
+    const dispatched: SendInternal[] = [];
+    const { internals } = await createSession(
+      [
+        compactionSummaryMessage("summary-delegated", {
+          text: "Continue",
+          model: "openai:gpt-4o",
+          agentId: "exec",
+          muxMetadata: DELEGATED_TURN,
+        }),
+      ],
+      { admitStrandedTurnResume: admit, settleForfeitedWorkspaceTurnContinuation: settle }
+    );
+    internals.sendMessage = mock(
+      (_message: string, _options?: SendOptions, internal?: SendInternal) => {
+        dispatched.push(internal ?? {});
+        return Promise.resolve({ success: true as const });
+      }
+    );
+
+    expect(await internals.dispatchPendingFollowUp()).toBe(true);
+
+    // Admitted against the delegated turn, with the handle probe carried to the launch boundary.
+    expect(admit.mock.calls[0]?.[0]).toEqual(DELEGATED_TURN);
+    expect(dispatched[0]?.refuseStreamStart?.()).toBe(false);
+    stale = true;
+    expect(dispatched[0]?.refuseStreamStart?.()).toBe(true);
+    expect(settle).not.toHaveBeenCalled();
+  });
+
+  test("dispatchPendingFollowUp settles and drops a delegated turn's follow-up its owner no longer admits", async () => {
+    const settle = mock((_correlation: unknown, _reason: string) => Promise.resolve());
+    const { internals, historyService } = await createSession(
+      [
+        compactionSummaryMessage("summary-refused", {
+          text: "Continue",
+          model: "openai:gpt-4o",
+          agentId: "exec",
+          muxMetadata: DELEGATED_TURN,
+        }),
+      ],
+      {
+        admitStrandedTurnResume: mock(() => Promise.resolve({ admissible: false })),
+        settleForfeitedWorkspaceTurnContinuation: settle,
+      }
+    );
+    const sendMessage = mock(() => Promise.resolve({ success: true as const }));
+    internals.sendMessage = sendMessage;
+
+    expect(await internals.dispatchPendingFollowUp()).toBe(false);
+
+    expect(sendMessage).not.toHaveBeenCalled();
+    expect(settle).toHaveBeenCalledTimes(1);
+    expect(settle.mock.calls[0]?.[0]).toEqual(DELEGATED_TURN);
+    const tail = await historyService.getLastMessages("ws", 1);
+    const summary = tail.success ? tail.data[0] : undefined;
+    expect(summary?.metadata?.muxMetadata).toEqual({ type: "compaction-summary" });
+  });
+
+  test("dispatchPendingFollowUp drops a malformed persisted remainder", async () => {
+    const dispatched: SendInternal[] = [];
+    const { internals } = await createSession([
+      compactionSummaryMessage("summary-malformed-remainder", {
+        text: "Continue",
+        model: "openai:gpt-4o",
+        agentId: "exec",
+        stepBudget: "seven" as unknown as number,
+        modelFallbackProgress: {
+          requestedModel: 1,
+        } as unknown as CompactionFollowUpRequest["modelFallbackProgress"],
+      }),
+    ]);
+    internals.sendMessage = mock(
+      (_message: string, _options?: SendOptions, internal?: SendInternal) => {
+        dispatched.push(internal ?? {});
+        return Promise.resolve({ success: true as const });
+      }
+    );
+
+    await internals.dispatchPendingFollowUp();
+
+    expect(dispatched).toHaveLength(1);
+    expect(dispatched[0]?.stepBudget).toBeUndefined();
+    expect(dispatched[0]?.modelFallbackProgress).toBeUndefined();
+  });
+
   test("dispatchPendingFollowUp forwards strictAgentResolution to the resumed turn", async () => {
     let dispatchedOptions: SendOptions | undefined;
     const { internals } = await createSession([