fix(server): agent messages no longer cancel Claude tools#21158

@@ -9150,6 +9241,43 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
                 detail: `Claude provider turn ${turnInput.providerTurnId} is not the active turn.`,
               });
             }
+            if (turnInput.message.createdBy === "agent") {
+              // `priority: next` also folds after tools, but can start an unowned
+              // native turn if the answer ends first. Hooks keep delivery owned
+              // by this turn. Allow ordinary commands to finish before falling
+              // back to T3's queue.
+              if (turnInput.message.attachments.length > 0) {
+                return yield* new ProviderAdapter.ProviderAdapterSteerRunDeferredError({
+                  driver: CLAUDE_PROVIDER,
+                  providerThreadId: turnInput.providerThread.id,
+                  providerTurnId: turnInput.providerTurnId,
+                });
+              }
+              const pending: PendingClaudeAgentSteer = {
+                text: turnInput.message.text,
+                completion: yield* Deferred.make<ClaudeAgentSteerOutcome>(),
+                ...(turnInput.resolveMessageText === undefined
+                  ? {}
+                  : { resolveMessageText: turnInput.resolveMessageText }),
+              };
+              currentTurn.pendingAgentSteers.set(turnInput.message.messageId, pending);
+              yield* Deferred.await(pending.completion).pipe(
+                Effect.timeoutOption("1 minute"),

@@ -9150,6 +9241,43 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
                 detail: `Claude provider turn ${turnInput.providerTurnId} is not the active turn.`,
               });
             }
+            if (turnInput.message.createdBy === "agent") {
+              // `priority: next` also folds after tools, but can start an unowned
+              // native turn if the answer ends first. Hooks keep delivery owned
+              // by this turn. Allow ordinary commands to finish before falling
+              // back to T3's queue.
+              if (turnInput.message.attachments.length > 0) {
+                return yield* new ProviderAdapter.ProviderAdapterSteerRunDeferredError({
+                  driver: CLAUDE_PROVIDER,
+                  providerThreadId: turnInput.providerThread.id,
+                  providerTurnId: turnInput.providerTurnId,
+                });
+              }
+              const pending: PendingClaudeAgentSteer = {
+                text: turnInput.message.text,
+                completion: yield* Deferred.make<ClaudeAgentSteerOutcome>(),
+                ...(turnInput.resolveMessageText === undefined
+                  ? {}
+                  : { resolveMessageText: turnInput.resolveMessageText }),
+              };
+              currentTurn.pendingAgentSteers.set(turnInput.message.messageId, pending);
+              yield* Deferred.await(pending.completion).pipe(
+                Effect.timeoutOption("1 minute"),

@@ -242,6 +244,60 @@ export const layer = Layer.effectDiscard(
       },
     );
 
+    // The provider waits on this from a tool-batch or stop hook, so each update
+    // joins its running turn before the hook returns the text. An update that
+    // cannot join (edited away, held by Stop, turn no longer running) stays
+    // queued for the next turn.
+    const claimAgentMessages: ProviderContinuationRequests.ClaimAgentMessages = (input) =>
+      Effect.gen(function* () {
+        const queued = yield* threads.getThreadRecords(input.threadId, ["runs", "messages"], {
+          messageRoles: ["user"],
+        });
+        const texts: Array<string> = [];
+        for (const run of queuedRunsInDeliveryOrder(queued)) {
+          const message = queued.messages.find((row) => row.id === run.userMessageId);
+          if (message?.atToolBoundary !== true) continue;
+          const result = yield* threads
+            .dispatch({
+              type: "queued-message.promote-to-steer",

@@ -242,6 +244,68 @@ export const layer = Layer.effectDiscard(
       },
     );
 
+    // The provider waits on this from a tool-batch or stop hook, so each update
+    // joins its running turn before the hook returns the text. An update that
+    // cannot join (edited away, held by Stop, turn no longer running) stays
+    // queued for the next turn.
+    const claimAgentMessages: ProviderContinuationRequests.ClaimAgentMessages = (input) =>
+      Effect.gen(function* () {
+        // This runs at every tool batch: read only queued runs and their messages.
+        const { runs } = yield* threads.getThreadRecords(input.threadId, ["runs"], {
+          runStatuses: ["queued"],
+        });
+        if (runs.length === 0) return [];
+        const { messages } = yield* threads.getThreadRecords(input.threadId, ["messages"], {
+          messageIds: runs.map((run) => run.userMessageId),
+        });
+        const queued = { runs, messages };
+        const texts: Array<string> = [];
+        for (const run of queuedRunsInDeliveryOrder(queued)) {
+          const message = queued.messages.find((row) => row.id === run.userMessageId);
+          if (message?.atToolBoundary !== true) continue;
+          const result = yield* threads
+            .dispatch({
+              type: "queued-message.promote-to-steer",

@@ -8644,11 +8675,68 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
             onUserDialog,
             supportedDialogKinds: ["resume_return"],
           });
+          let boundaryQuery: ClaudeAgentSdkQuerySession | undefined;
+          const agentMessageBoundary: HookCallback = (hookInput, _toolUseId, options) =>
+            runPromise(
+              Effect.gen(function* () {
+                const context = yield* Ref.get(activeTurn);
+                const live = yield* Ref.get(queryContext);
+                if (
+                  options.signal.aborted ||
+                  hookInput.agent_id !== undefined ||
+                  context === null ||
+                  live?.query !== boundaryQuery ||
+                  live?.stopping ||
+                  context.nativeThreadId !== hookInput.session_id ||
+                  (yield* Ref.get(interruptedTurns)).has(context.providerTurnId) ||
+                  (hookInput.hook_event_name !== "PostToolBatch" &&
+                    hookInput.hook_event_name !== "Stop")
+                ) {
+                  return {};
+                }
+                // Resolve before claiming so an aborted hook consumes no partial batch.
+                const pendingSteers = yield* Effect.forEach(
+                  Array.from(context.pendingAgentSteers),
+                  ([id, pending]) =>
+                    (pending.resolveMessageText ?? Effect.succeed(pending.text)).pipe(
+                      Effect.map((text) => ({ id, pending, text })),
+                    ),
+                );

@@ -242,6 +244,60 @@ export const layer = Layer.effectDiscard(
       },
     );
 
+    // The provider waits on this from a tool-batch or stop hook, so each update
+    // joins its running turn before the hook returns the text. An update that
+    // cannot join (edited away, held by Stop, turn no longer running) stays
+    // queued for the next turn.
+    const claimAgentMessages: ProviderContinuationRequests.ClaimAgentMessages = (input) =>
+      Effect.gen(function* () {
+        const queued = yield* threads.getThreadRecords(input.threadId, ["runs", "messages"], {
+          messageRoles: ["user"],
+        });
+        const texts: Array<string> = [];
+        for (const run of queuedRunsInDeliveryOrder(queued)) {
+          const message = queued.messages.find((row) => row.id === run.userMessageId);
+          if (message?.atToolBoundary !== true) continue;
+          const result = yield* threads
+            .dispatch({
+              type: "queued-message.promote-to-steer",
+              // Fresh per try: a refusal now must not be replayed at a later boundary.
+              commandId: yield* ids.allocate.command({
+                fixtureName: "tool-boundary",
+                commandName: "promote",
+              }),
+              threadId: input.threadId,
+              queuedRunId: run.id,
+              targetRunId: input.runId,
+              atToolBoundary: true,
+            })
+            .pipe(Effect.catchCause(() => Effect.succeed(undefined)));

@@ -242,6 +244,60 @@ export const layer = Layer.effectDiscard(
       },
     );
 
+    // The provider waits on this from a tool-batch or stop hook, so each update
+    // joins its running turn before the hook returns the text. An update that
+    // cannot join (edited away, held by Stop, turn no longer running) stays
+    // queued for the next turn.
+    const claimAgentMessages: ProviderContinuationRequests.ClaimAgentMessages = (input) =>
+      Effect.gen(function* () {
+        const queued = yield* threads.getThreadRecords(input.threadId, ["runs", "messages"], {
+          messageRoles: ["user"],
+        });

@@ -384,11 +443,21 @@ export const layer: Layer.Layer<
                     operation: "steer",
                     providerTurnId: input.providerTurnId,
                     turnCompleted: current.providerTurn?.status === "completed",
+                    deferToQueue:
+                      cause._tag === "ProviderAdapterSteerRunError" && cause.deferToQueue === true,
                     cause,
                   });
                 }),
               ),
             );
+          if (input.isCurrent !== undefined && !(yield* input.isCurrent)) {
+            return yield* new ProviderTurnControlError({
+              threadId: input.threadId,
+              operation: "steer",
+              providerTurnId: input.providerTurnId,
+              deliveryCancelled: true,
+            });
+          }

@@ -234,7 +234,7 @@ export const ThreadUpdateTool = Tool.make("t3_thread_update", {
 
 const ThreadSendTool = Tool.make("t3_thread_send", {
   description:
-    "Send a message to any T3 thread in this environment. The target cannot have broader permission modes than the caller. Use delegate_task with the full review context and a new clientRequestId for independent review rounds. For a follow-up that needs the same child's conversation, send to its childThreadId. After the original task completes, a requested follow-up creates a fresh task in that child thread and returns taskId when it starts a turn or restarts a turn without its own task; its completion notifies the parent, and the original result stays unchanged. A steer that joins a running turn returns that turn's task, if it has one. mode='auto' starts an idle thread, steers a fully active turn, or queues behind a turn that is not yet steerable. Use queue for a separate follow-up turn, steer for an in-flight update, or restart to interrupt-and-restart the active turn. clientRequestId makes retries idempotent.",
+    "Send a message to any T3 thread in this environment. The target cannot have broader permission modes than the caller. Use delegate_task with the full review context and a new clientRequestId for independent review rounds. For a follow-up that needs the same child's conversation, send to its childThreadId. After the original task completes, a requested follow-up creates a fresh task in that child thread and returns taskId when it starts a turn or restarts a turn without its own task; its completion notifies the parent, and the original result stays unchanged. A steer that joins a running turn returns that turn's task, if it has one. mode='auto' starts an idle thread, safely steers a fully active turn, or queues behind a turn that is not yet steerable. When the target session's steering interrupts tools (including Claude), both auto and explicit steer queue a separate turn instead; delivery='queued' confirms this fallback. Use queue for a separate follow-up turn, steer for an in-flight update, or restart to explicitly interrupt-and-restart the active turn. clientRequestId makes retries idempotent.",

@@ -3136,6 +3140,127 @@ describe("ClaudeAdapterV2 background wake turns", () => {
       ),
   );
 
+  it.effect("delivers claimed agent updates at root tool-batch and stop hooks", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const claims: Array<{ readonly threadId: string; readonly runId: string }> = [];
+        const harness = yield* makeWakeHarnessWithOptions({
+          claimAgentMessages: (input) =>
+            Effect.sync(() => {
+              claims.push(input);
+              return ["Child finished."];
+            }),
+        });
+        const input = makeClaudeTestTurnInput({
+          threadId: harness.threadId,
+          providerThread: harness.providerThread,
+          now: yield* DateTime.now,
+          attemptId: RunAttemptId.make("attempt-agent-boundary"),
+          text: "Run the slow build.",
+          attachments: [],
+        });
+        yield* harness.runtime.startTurn(input);
+        const hooks = harness.getOpenedOptions()?.hooks;
+        const signal = new AbortController().signal;
+        const base = {
+          session_id: WAKE_NATIVE_SESSION,
+          transcript_path: "",
+          cwd: "",
+        };
+        const batch = yield* Effect.promise(() =>
+          hooks!.PostToolBatch![0]!.hooks[0]!(
+            { ...base, hook_event_name: "PostToolBatch", tool_calls: [] },
+            undefined,
+            { signal },
+          ),
+        );
+        const stop = yield* Effect.promise(() =>
+          hooks!.Stop![0]!.hooks[0]!(

@@ -4721,6 +4721,24 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
               turn.status === "completed"))
         ) {
           dispatchMode = { type: "start_immediately" };
+        } else if (command.createdBy === "agent") {
+          const providerThread = projection.providerThreads.find(
+            (candidate) => candidate.id === target?.providerThreadId,
+          );
+          if (providerThread?.providerSessionId != null) {
+            const session = yield* providerSessions
+              .get(providerThread.providerSessionId)
+              .pipe(mapDispatchError(command));
+            if (
+              Option.isSome(session) &&
+              session.value.providerSession.capabilities.turns.activeSteeringInterruptsTools ===
+                true
+            ) {
+              // Claude steering cancels pending tools, which appear as user refusals.
+              // Agent updates can wait for the durable queue's next-turn boundary.
+              dispatchMode = { type: "queue_after_active" };