fix(server): Stop returns an unread Claude steer to the queue

apps/mobile/src/features/threads/ThreadQueueControl.tsx

@@ -191,7 +191,7 @@ export function ThreadQueueSheet({ route }: StaticScreenProps<QueueTarget>) {
     >
       {workflow?.isHeld && queuedRuns.length > 0 ? (
         <View className="gap-2 py-3">
-          <Text className="text-sm text-foreground-muted">Queue held after restart</Text>
+          <Text className="text-sm text-foreground-muted">Queue held</Text>
           <MaterialButton
             label="Resume queue"
             disabled={resuming || busyRunId !== null}

apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts

@@ -2317,7 +2317,7 @@ describe("ClaudeAdapterV2 background wake turns", () => {
 
   const makeWakeHarnessWithOptions = (options?: {
     readonly close?: (sdkMessages: Queue.Queue<SDKMessage>) => Effect.Effect<void>;
-    readonly interrupt?: Effect.Effect<void>;
+    readonly interrupt?: ClaudeAdapterV2.ClaudeAgentSdkQuerySession["interrupt"];
     readonly stopTask?: (
       taskId: string,
     ) => Effect.Effect<void, ClaudeAdapterV2.ClaudeAgentSdkQueryRunnerError>;
@@ -3026,6 +3026,64 @@ describe("ClaudeAdapterV2 background wake turns", () => {
     ),
   );
 
+  it.effect("returns the steers an interrupt finds still queued in Claude", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const read = MessageId.make("message-steer-read");
+        const unread = MessageId.make("message-steer-unread");
+        // Claude also lists queued commands T3 never sent; those are ignored.
+        const stillQueued = ["00000000-0000-4000-8000-000000000903"];
+        const harness = yield* makeWakeHarnessWithOptions({
+          close: (sdkMessages) => Queue.shutdown(sdkMessages),
+          interrupt: Effect.sync(() => ({ still_queued: stillQueued })),
+        });
+        const idAllocator = yield* IdAllocator.IdAllocatorV2;
+        const attemptId = RunAttemptId.make("attempt-unread-steer");
+        const input = makeClaudeTestTurnInput({
+          threadId: harness.threadId,
+          providerThread: harness.providerThread,
+          now: yield* DateTime.now,
+          attemptId,
+          text: "Run the slow command.",
+          attachments: [],
+        });
+        const providerTurnId = idAllocator.derive.providerTurn({
+          driver: ClaudeAdapterV2.CLAUDE_PROVIDER,
+          nativeTurnId: `turn:${attemptId}`,
+        });
+        yield* harness.runtime.startTurn(input);
+        for (const messageId of [read, unread]) {
+          yield* harness.runtime.steerTurn({
+            threadId: harness.threadId,
+            runId: input.runId,
+            providerThread: harness.providerThread,
+            providerTurnId,
+            message: {
+              createdBy: "user",
+              creationSource: "web",
+              messageId,
+              text: messageId,
+              attachments: [],
+            },
+          });
+        }
+        // Claude still holds the second steer, under the uuid it was offered with.
+        stillQueued.push(harness.offeredMessages[2]!.uuid!);
+
+        const result = yield* harness.runtime.interruptTurn({
+          providerThread: harness.providerThread,
+          providerTurnId,
+        });
+
+        assert.deepEqual(result, { unreadSteerMessageIds: [unread] });
+      }).pipe(
+        Effect.provide(
+          Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer),
+        ),
+      ),
+    ),
+  );
+
   it.effect("announces usage-limit pauses once per window and again on a new turn", () =>
     Effect.gen(function* () {
       const harness = yield* makeWakeHarness;
@@ -6710,7 +6768,7 @@ describe("ClaudeAdapterV2 background wake turns", () => {
             close: (sdkMessages) =>
               Deferred.await(closeGate).pipe(Effect.andThen(Queue.shutdown(sdkMessages))),
             stopTask: () => Effect.void,
-            interrupt: Deferred.succeed(interruptStarted, undefined),
+            interrupt: Deferred.succeed(interruptStarted, undefined).pipe(Effect.asVoid),
           });
           const idAllocator = yield* IdAllocator.IdAllocatorV2;
           const now = yield* DateTime.now;
@@ -6772,7 +6830,7 @@ describe("ClaudeAdapterV2 background wake turns", () => {
           close: (sdkMessages) =>
             Deferred.await(closeGate).pipe(Effect.andThen(Queue.shutdown(sdkMessages))),
           stopTask: () => Effect.void,
-          interrupt: Deferred.succeed(interruptStarted, undefined),
+          interrupt: Deferred.succeed(interruptStarted, undefined).pipe(Effect.asVoid),
         });
         const idAllocator = yield* IdAllocator.IdAllocatorV2;
         const now = yield* DateTime.now;

apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts

@@ -30,6 +30,7 @@ import {
   type Settings as ClaudeSdkSettings,
   type SDKAssistantMessage,
   type SDKAPIRetryMessage,
+  type SDKControlInterruptResponse,
   type SDKMessage,
   type SDKRateLimitInfo,
   type SDKResultMessage,
@@ -51,6 +52,7 @@ import {
   type ChatAttachment,
   ClaudeSettings,
   defaultInstanceIdForDriver,
+  type MessageId,
   type ModelSelection,
   type OrchestrationV2ConversationMessage,
   type OrchestrationV2ExecutionNode,
@@ -382,7 +384,11 @@ export interface ClaudeAgentSdkQuerySession {
     mode: PermissionMode,
   ) => Effect.Effect<void, ClaudeAgentSdkQueryRunnerError>;
   readonly stopTask: (taskId: string) => Effect.Effect<void, ClaudeAgentSdkQueryRunnerError>;
-  readonly interrupt: Effect.Effect<void, ClaudeAgentSdkQueryRunnerError>;
+  /** CLIs with the interrupt_receipt_v1 capability name the prompts they still held unread. */
+  readonly interrupt: Effect.Effect<
+    SDKControlInterruptResponse | void,
+    ClaudeAgentSdkQueryRunnerError
+  >;
   readonly close: Effect.Effect<void, ClaudeAgentSdkQueryRunnerError>;
 }
 
@@ -846,12 +852,13 @@ export const layerQueryRunner: Layer.Layer<
             try: () => queryRuntime.interrupt(),
             catch: (cause) => queryRunnerError(cause, "interrupt"),
           }).pipe(
-            Effect.tap(() =>
+            Effect.tap((receipt) =>
               logProtocolEvent({
                 direction: "outgoing",
                 stage: "decoded",
                 payload: {
                   type: "query.interrupt",
+                  ...(receipt === undefined ? {} : { stillQueued: receipt.still_queued }),
                 },
               }),
             ),
@@ -2931,6 +2938,8 @@ interface ActiveClaudeTurnContext {
   // Root frames seen before the echo; held only when the CLI echoes early.
   gatedFramesBeforeEcho: number;
   readonly heldRootFrames: Array<SDKMessage>;
+  // Steers offered into this turn, by the uuid each was offered with.
+  readonly steerMessageIds: Map<string, MessageId>;
 }
 
 interface ActiveClaudeProviderRetry {
@@ -8571,6 +8580,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
               promptStarted: false,
               gatedFramesBeforeEcho: 0,
               heldRootFrames: [],
+              steerMessageIds: new Map(),
             };
             // Continuation turns attach to the wake output the CLI already
             // produced instead of prompting it again: drain the buffered wake
@@ -8768,13 +8778,20 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
               next.add(turnInput.providerTurnId);
               return next;
             });
-            yield* existing.query.interrupt;
+            const receipt = yield* existing.query.interrupt;
+            // A steer Claude still queues behind a running tool dies with the
+            // process closed below, so the thread takes it back.
+            const result: ProviderAdapter.ProviderAdapterV2InterruptResult = {
+              unreadSteerMessageIds: (receipt?.still_queued ?? []).flatMap(
+                (uuid) => currentTurn.steerMessageIds.get(uuid) ?? [],
+              ),
+            };
             yield* existing.query.close.pipe(Effect.ignore);
             const closed = yield* Deferred.await(existing.closed).pipe(
               Effect.timeoutOption("10 seconds"),
             );
             if (Option.isSome(closed)) {
-              return;
+              return result;
             }
 
             const completedAt = yield* DateTime.now;
@@ -8799,6 +8816,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
               });
             }
             yield* Deferred.succeed(existing.closed, undefined);
+            return result;
           },
           (effect, turnInput) =>
             effect.pipe(
@@ -8830,6 +8848,19 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
                 detail: `Claude provider turn ${turnInput.providerTurnId} is not the active turn.`,
               });
             }
+            // Fresh for every offer, like a prompt's: an interrupt receipt that
+            // lists it as still queued names the message Claude never read.
+            const uuid = yield* crypto.randomUUIDv4.pipe(
+              Effect.filterOrFail(isClaudePromptUuid),
+              Effect.mapError(
+                (cause) =>
+                  new ProviderAdapter.ProviderAdapterProtocolError({
+                    driver: CLAUDE_PROVIDER,
+                    detail: "Failed to allocate a Claude steer uuid.",
+                    payload: cause,
+                  }),
+              ),
+            );
             const userMessage = yield* makeClaudeUserMessageWithAttachments({
               text: applyClaudePromptEffortPrefix(
                 turnInput.message.text,
@@ -8840,12 +8871,14 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
               attachmentsDir,
               fileSystem,
               skillNames: yield* userInvocableSkillNames(currentTurn.input.runtimePolicy.cwd),
+              uuid,
             });
             yield* Ref.update(steeredTurns, (current) => {
               const next = new Set(current);
               next.add(turnInput.providerTurnId);
               return next;
             });
+            currentTurn.steerMessageIds.set(uuid, turnInput.message.messageId);
             yield* existing.query.offer(userMessage);
           },
           (effect, turnInput) =>

apps/server/src/orchestration-v2/EffectOutbox.ts

@@ -50,6 +50,8 @@ export const OrchestrationEffectRequestV2 = Schema.Union([
     providerThreadId: ProviderThreadId,
     providerTurnId: ProviderTurnId,
     subagent: Schema.optional(Schema.Struct({ id: NodeId, nativeTaskId: Schema.String })),
+    /** The Stop held the queue, so steers the provider never read return to it held. */
+    holdQueue: Schema.optional(Schema.Boolean),
   }),
   Schema.Struct({
     type: Schema.Literal("provider-turn.steer"),

apps/server/src/orchestration-v2/EffectWorker.test.ts

@@ -89,14 +89,14 @@ function layerExecutorFor(input: {
     Layer.succeed(
       ProviderTurnControlService.ProviderTurnControlServiceV2,
       ProviderTurnControlService.ProviderTurnControlServiceV2.of({
-        interrupt: input.interrupt ?? (() => Effect.void),
+        interrupt: input.interrupt ?? (() => Effect.succeed({ unreadSteerMessageIds: [] })),
         steer: () => Effect.void,
         interruptAndAwaitTerminal: (request) =>
           record(
             request.replacementProviderSessionId === undefined
               ? "interrupt"
               : `interrupt:${request.replacementProviderSessionId}`,
-          ),
+          ).pipe(Effect.as([])),
       }),
     ),
     Layer.succeed(
@@ -861,11 +861,14 @@ it.effect.each(["confirmed", "unconfirmed", "failed"] as const)(
                 : Effect.succeed(
                     outcome === "confirmed"
                       ? {
-                          status: "cancelled" as const,
-                          completedAt: now,
-                          result: "Stopped by Claude",
+                          unreadSteerMessageIds: [],
+                          subagentOutcome: {
+                            status: "cancelled" as const,
+                            completedAt: now,
+                            result: "Stopped by Claude",
+                          },
                         }
-                      : undefined,
+                      : { unreadSteerMessageIds: [] },
                   ),
             ),
           );

apps/server/src/orchestration-v2/EffectWorker.ts

@@ -1,4 +1,4 @@
-import { CommandId } from "@t3tools/contracts";
+import { CommandId, type MessageId } from "@t3tools/contracts";
 import * as Cause from "effect/Cause";
 import * as Context from "effect/Context";
 import * as DateTime from "effect/DateTime";
@@ -104,6 +104,24 @@ export const layerExecutor: Layer.Layer<
       yield* ThreadTitleRegenerationService.ThreadTitleRegenerationService;
     const threads = yield* ThreadManagementService.ThreadManagementService;
     const settings = yield* ServerSettings.ServerSettingsService;
+    // Steers the provider never read would end with its process, so they go
+    // back to the queue, where the user can still send or drop them.
+    const requeueUnreadSteers = (
+      effect: EffectOutbox.OrchestrationEffectV2,
+      messageIds: ReadonlyArray<MessageId>,
+      holdQueue: boolean,
+    ) =>
+      messageIds.length === 0
+        ? Effect.void
+        : threads
+            .dispatch({
+              type: "thread.unread-steers.requeue",
+              commandId: CommandId.make(`${effect.id}:unread-steers-requeued`),
+              threadId: effect.threadId,
+              messageIds,
+              holdQueue,
+            })
+            .pipe(Effect.asVoid);
     return OrchestrationEffectExecutorV2.of({
       execute: (effect, options) => {
         const willRetry = options?.willRetry ?? false;
@@ -164,6 +182,7 @@ export const layerExecutor: Layer.Layer<
               );
           case "provider-turn.interrupt": {
             const request = effect.request;
+            const holdQueue = request.holdQueue === true;
             return providerTurnControl
               .interrupt({
                 threadId: effect.threadId,
@@ -194,12 +213,17 @@ export const layerExecutor: Layer.Layer<
                     request.type,
                     Cause.pretty(Cause.fail(cause)),
                   )
-                    ? Effect.void
+                    ? Effect.succeed<ProviderTurnControlService.ProviderTurnInterruptResult>({
+                        unreadSteerMessageIds: [],
+                      })
                     : Effect.fail(cause),
                 ),
+                Effect.tap(({ unreadSteerMessageIds }) =>
+                  requeueUnreadSteers(effect, unreadSteerMessageIds, holdQueue),
+                ),
                 // A targeted Stop repairs only an outcome the provider actually observed.
                 // A full-turn Stop retains its existing process-teardown repair.
-                Effect.andThen((outcome) =>
+                Effect.andThen(({ subagentOutcome: outcome }) =>
                   request.subagent !== undefined && outcome === undefined
                     ? Effect.void
                     : threads.dispatch({
@@ -360,6 +384,9 @@ export const layerExecutor: Layer.Layer<
                   : {}),
               })
               .pipe(
+                // The restarted run keeps the thread busy, so the requeued
+                // steers follow it like any queued message.
+                Effect.flatMap((unread) => requeueUnreadSteers(effect, unread, false)),
                 Effect.andThen(
                   effect.request.sessionTransition?.type === "replace"
                     ? providerSessions.detach({