fix(claude): steers no longer abort context compaction

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

@@ -811,9 +811,11 @@ function makeReplayQueryRunner(
         messages: Stream.fromAsyncIterable(replayMessagesWithGateCleanup(input.options), (cause) =>
           replayQueryRunnerError(transcript, cause),
         ),
-        offer: (message) =>
+        offer: (message, onQueued, canQueue) =>
           replayEffect(() => {
+            if (canQueue?.() === false) return false;
             assertNextOutboundFrame(makeClaudePromptOfferFrame(message));
+            onQueued?.();
           }),
         setModel: (model) =>
           replayEffect(() => {

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

@@ -140,6 +140,86 @@ const resumedMcpQueryInput = {
   },
 } satisfies ClaudeAgentSdkQueryOpenInput;
 
+it.effect.each(["closed", "gate", "log"] as const)(
+  "acknowledges actual prompt admission: %s",
+  (failure) =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        let prompts: AsyncIterable<ClaudeSdk.SDKUserMessage> | undefined;
+        let queued = 0;
+        const querySpy = vi.spyOn(ClaudeSdk, "query").mockImplementation(({ prompt }) => {
+          if (typeof prompt === "string") throw new Error("Expected streaming input");
+          prompts = prompt;
+          return {
+            setMcpServers: async () => ({ added: [], removed: [], errors: {} }),
+            close: () => {},
+            [Symbol.asyncDispose]: async () => {},
+          } as unknown as ClaudeSdk.Query;
+        });
+        yield* Effect.addFinalizer(() => Effect.sync(() => querySpy.mockRestore()));
+        const runner = yield* ClaudeAgentSdkQueryRunner;
+        const session = yield* runner.open(resumedMcpQueryInput);
+        if (failure === "closed") yield* session.close;
+        const message = {
+          type: "user",
+          message: { role: "user", content: "Steer" },
+          parent_tool_use_id: null,
+          session_id: FORK_SESSION_ID,
+        } satisfies ClaudeSdk.SDKUserMessage;
+        const result = yield* session
+          .offer(
+            message,
+            () => {
+              queued += 1;
+            },
+            () => failure !== "gate",
+          )
+          .pipe(Effect.exit);
+        if (failure === "gate") {
+          assert.isTrue(Exit.isSuccess(result));
+          if (Exit.isSuccess(result)) assert.equal(result.value, false);
+          assert.equal(queued, 0);
+        } else {
+          assert.isTrue(Exit.isFailure(result));
+          assert.equal(queued, failure === "log" ? 1 : 0);
+        }
+        if (failure === "log") {
+          if (prompts === undefined) throw new Error("Missing prompt iterator");
+          const delivered = yield* Effect.promise(() => prompts![Symbol.asyncIterator]().next());
+          assert.deepEqual(delivered.value, message);
+        }
+        yield* session.close;
+      }).pipe(
+        Effect.provide(
+          layerQueryRunner.pipe(
+            Layer.provide(PortScanner.layerTest),
+            Layer.provide(
+              Layer.succeed(ProviderEventLoggers, {
+                canonical: undefined,
+                native:
+                  failure === "log"
+                    ? {
+                        filePath: "unused",
+                        close: () => Effect.void,
+                        write: (event) => {
+                          const protocol = event as {
+                            event?: { direction?: string; payload?: { type?: string } };
+                          };
+                          return protocol.event?.payload?.type === "prompt.offer"
+                            ? Effect.die(new Error("Log unavailable"))
+                            : Effect.void;
+                        },
+                      }
+                    : undefined,
+              }),
+            ),
+            Layer.provide(NodeServices.layer),
+          ),
+        ),
+      ),
+    ),
+);
+
 it.effect("reopens a stalled MCP registration before delivering the resumed prompt", () =>
   Effect.scoped(
     Effect.gen(function* () {

apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts

@@ -155,14 +155,23 @@ it.effect.each(deliveries)(
             open: () =>
               Effect.succeed({
                 messages: Stream.fromQueue(sdkMessages),
-                offer: (message) =>
-                  Effect.sync(() => {
-                    offers.push(message);
-                    nativeQueue.push(message);
-                    // Claude Code 2.1.289 aborts on now-priority commands even while Bash blocks reads.
-                    if (nativeQueue.some((command) => command.priority === "now")) {
-                      batchAbort.abort({ kind: "interrupt" });
-                    }
+                offer: (message, onQueued, canQueue) =>
+                  Effect.try({
+                    try: () => {
+                      if (canQueue?.() === false) return false;
+                      offers.push(message);
+                      nativeQueue.push(message);
+                      onQueued?.();
+                      // Claude Code 2.1.289 aborts on now-priority commands even while Bash blocks reads.
+                      if (nativeQueue.some((command) => command.priority === "now")) {
+                        batchAbort.abort({ kind: "interrupt" });
+                      }
+                    },
+                    catch: (cause) =>
+                      new ClaudeAdapterV2.ClaudeAgentSdkQueryRunnerError({
+                        method: "offer",
+                        cause,
+                      }),
                   }),
                 setModel: () => Effect.void,
                 applyFlagSettings: () => Effect.void,
@@ -395,9 +404,18 @@ it.effect.each(deliveries)(
             assert.ok(steer?.type === "user_message");
             assert.equal(steer.inputIntent, "promoted_queued_to_steer");
           }
-          yield* worker.drain();
           const explicitSteer =
             delivery === "user steering" || delivery === "promoted queued message";
+          const admitted = explicitSteer
+            ? yield* watch(
+                (event) =>
+                  event.type === "turn-item.updated" &&
+                  event.payload.id === IdAllocator.derive.userTurnItem({ messageId }) &&
+                  event.payload.status === "completed",
+              )
+            : null;
+          yield* worker.drain();
+          if (admitted !== null) yield* Fiber.join(admitted);
           assert.equal(batchAbort.signal.aborted, explicitSteer);
           assert.equal(offers.length, explicitSteer ? 2 : 1);
           const finished = yield* watch(

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

@@ -9,16 +9,20 @@ import {
   MessageId,
   type ModelSelection,
   NodeId,
+  type OrchestrationV2DomainEvent,
   type OrchestrationV2Run,
   ProjectId,
   ProviderDriverKind,
   ProviderInstanceId,
   ProviderThreadId,
+  ProviderTurnId,
+  RunAttemptId,
   RunId,
   RuntimeRequestId,
   ThreadId,
   TurnItemId,
 } from "@t3tools/contracts";
+import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
 import * as DateTime from "effect/DateTime";
 import * as Effect from "effect/Effect";
 import * as Deferred from "effect/Deferred";
@@ -988,6 +992,7 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => {
     { command: "acknowledge", before: "running", after: "failed", capture: false },
     { command: "acknowledge", before: "waiting", after: "completed", capture: true },
     { command: "accept", before: "waiting", after: "completed", capture: true },
+    { command: "settle", before: "waiting", after: "completed", capture: true },
   ] as const)(
     "keeps a $after run that settled between $command's read and commit",
     ({ command, before, after, capture }) =>
@@ -1001,17 +1006,93 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => {
         const taskId = NodeId.make(`node:${race}-task`);
         const checkpointId = CheckpointId.make(`checkpoint:${race}`);
         const messageId = MessageId.make(`message:delegated-delivery:${threadId}`);
+        const rootNodeId = NodeId.make(`node:${race}-root`);
+        const attemptId = RunAttemptId.make(`attempt:${race}`);
+        const providerTurnId = ProviderTurnId.make(`provider-turn:${race}`);
         yield* seedParentWithTerminalTask({
           threadId,
           projectId: ProjectId.make(`project:${race}`),
           runId,
-          rootNodeId: NodeId.make(`node:${race}-root`),
+          rootNodeId,
           taskId,
           deliveryState: "claimed",
           deliveryTaskIds: [taskId],
           now,
         });
-        const parent = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!;
+        const seeded = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!;
+        // A held Claude steer settles against the attempt and provider turn it was offered to.
+        const parent = command === "settle" ? { ...seeded, activeAttemptId: attemptId } : seeded;
+        const providerThreadId = parent.providerThreadId!;
+        const offeredTo: ReadonlyArray<OrchestrationV2DomainEvent> =
+          command === "settle"
+            ? [
+                {
+                  id: EventId.make(`event:${race}-attempt`),
+                  type: "run-attempt.created",
+                  threadId,
+                  runId,
+                  occurredAt: now,
+                  payload: {
+                    id: attemptId,
+                    runId,
+                    attemptOrdinal: 1,
+                    rootNodeId: rootNodeId,
+                    providerInstanceId: modelSelection.instanceId,
+                    providerThreadId,
+                    providerTurnId,
+                    reason: "initial",
+                    status: "completed",
+                    startedAt: now,
+                    completedAt: now,
+                  },
+                },
+                {
+                  id: EventId.make(`event:${race}-turn`),
+                  type: "provider-turn.updated",
+                  threadId,
+                  runId,
+                  occurredAt: now,
+                  payload: {
+                    id: providerTurnId,
+                    providerThreadId,
+                    nodeId: rootNodeId,
+                    runAttemptId: attemptId,
+                    nativeTurnRef: null,
+                    ordinal: 1,
+                    status: "completed",
+                    startedAt: now,
+                    completedAt: now,
+                  },
+                },
+                {
+                  id: EventId.make(`event:${race}-item`),
+                  type: "turn-item.updated",
+                  threadId,
+                  runId,
+                  occurredAt: now,
+                  payload: {
+                    id: IdAllocator.derive.userTurnItem({ messageId }),
+                    threadId,
+                    runId,
+                    nodeId: rootNodeId,
+                    providerThreadId,
+                    providerTurnId,
+                    nativeItemRef: null,
+                    parentItemId: null,
+                    ordinal: 1,
+                    type: "notification",
+                    source: { kind: "delegated_task", taskIds: [taskId] },
+                    outcome: "completed",
+                    summary: "Background task finished",
+                    status: "pending",
+                    title: null,
+                    startedAt: now,
+                    completedAt: null,
+                    updatedAt: now,
+                  },
+                },
+              ]
+            : [];
         yield* sink.write({
           events: [
             {
@@ -1022,6 +1103,7 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => {
               occurredAt: now,
               payload: { ...parent, status: before },
             },
+            ...offeredTo,
             {
               id: EventId.make(`event:${race}-message`),
               type: "message.updated",
@@ -1058,7 +1140,9 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => {
                   input.commandType ===
                     (command === "accept"
                       ? "notification.delivery.accept"
-                      : "delegated_task.completion-delivery.acknowledge")
+                      : command === "settle"
+                        ? "message.steer.settle"
+                        : "delegated_task.completion-delivery.acknowledge")
                 ) {
                   settled = true;
                   // The provider turn ends (or its checkpoint lands) after the command read the run.
@@ -1098,21 +1182,32 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => {
                     threadId,
                     messageId,
                   }
-                : {
-                    type: "delegated_task.completion-delivery.acknowledge",
-                    commandId: CommandId.make(`command:${race}:task-status`),
-                    parentThreadId: threadId,
-                    taskId,
-                    observedByRunId: runId,
-                  },
+                : command === "settle"
+                  ? {
+                      type: "message.steer.settle",
+                      commandId: CommandId.make(`command:${race}:settle`),
+                      threadId,
+                      messageId,
+                      runId,
+                      attemptId,
+                      providerTurnId,
+                      outcome: "offered",
+                    }
+                  : {
+                      type: "delegated_task.completion-delivery.acknowledge",
+                      commandId: CommandId.make(`command:${race}:task-status`),
+                      parentThreadId: threadId,
+                      taskId,
+                      observedByRunId: runId,
+                    },
             ),
           (spy) => Effect.sync(() => spy.mockRestore()),
         );
         assert.isTrue(settled);
         const finished = yield* orchestrator.getThreadProjection(threadId);
         assert.equal(
           finished.subagents.find((row) => row.id === taskId)?.completionDelivery?.state,
-          command === "accept" ? "delivered" : "acknowledged",
+          command === "acknowledge" ? "acknowledged" : "delivered",
         );
         const run = finished.runs.find((candidate) => candidate.id === runId)!;
         assert.deepEqual(

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

@@ -30,6 +30,14 @@ import * as SqlClient from "effect/sql/SqlClient";
 import { forkParked } from "../serverActivation.ts";
 
 export const OrchestrationEffectRequestV2 = Schema.Union([
+  Schema.Struct({
+    type: Schema.Literal("message.steer.settle"),
+    messageId: MessageId,
+    runId: RunId,
+    attemptId: RunAttemptId,
+    providerTurnId: ProviderTurnId,
+    outcome: Schema.Literals(["offered", "failed", "cancelled", "completed"]),
+  }),
   Schema.Struct({
     type: Schema.Literal("provider-runtime.continue"),
     sourceRunId: RunId,
@@ -123,6 +131,7 @@ export const OrchestrationEffectRequestV2 = Schema.Union([
 export type OrchestrationEffectRequestV2 = typeof OrchestrationEffectRequestV2.Type;
 
 export const REPLAY_SAFE_EFFECT_TYPES_AFTER_PROCESS_LOSS = [
+  "message.steer.settle",
   "provider-runtime.continue",
   "provider-session.detach",
   "provider-thread.rollback",