🤖 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/message.ts

@@ -190,6 +190,10 @@ export const MuxMessageSchema = z.object({
       retrySendOptions: z.any().optional(),
       agentId: AgentIdSchema.optional().catch(undefined),
       partial: z.boolean().optional(),
+      // Steps the cut turn had left under its ceiling when a queued message interrupted it
+      // (stamped on the committed partial): a startup retry of that turn runs under this budget
+      // instead of a fresh ceiling. Self-healing read path: a malformed value reads as absent.
+      stepsRemaining: z.number().int().nonnegative().optional().catch(undefined),
       synthetic: z.boolean().optional(),
       uiVisible: z.boolean().optional(),
       // RLM keep-recent floor: sanitized post-boundary copy of a pre-compaction row.

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,16 @@ 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(),
+      // A required completion tool succeeded in the interrupted step: the turn was complete, so a
+      // queued-message soft stop owes it no continuation.
+      requiredToolSatisfied: z.boolean().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
@@ -928,6 +937,8 @@ export interface MuxMetadata {
   contextProviderMetadata?: Record<string, unknown>;
   systemMessageTokens?: number; // Token count for system message sent with this request (calculated by AIService)
   partial?: boolean; // Whether this message was interrupted and is incomplete
+  /** Steps a queued-message cut left under the turn's ceiling; a startup retry runs under it. */
+  stepsRemaining?: number;
   synthetic?: boolean; // Whether this message was synthetically generated (e.g., [CONTINUE] sentinel)
   /**
    * For queue-dispatched user turns: when the user last added to the queued

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 ?? "";