diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index a7655e36a9159..95df58c2bff32 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -4958,6 +4958,154 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + // regression: #15708 Stop must not silently drop a steer Claude queued behind a foreground task + it.effect.each([ + { + readBy: "none", + echoes: true, + title: "warns that a steer Claude never read before Stop was not delivered", + }, + { + readBy: "none", + echoes: false, + title: "does not warn about a steer on a CLI that never echoes prompt uuids", + }, + { + readBy: "assistant", + echoes: true, + title: "does not warn about a steer Claude read before Stop", + }, + ] as const)("$title", ({ readBy, echoes }) => + Effect.gen(function* () { + const harness = yield* makeWakeHarnessWithOptions({ + close: (sdkMessages) => Queue.shutdown(sdkMessages), + freshQueueOnReopen: true, + }); + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const firstInput = makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-claude-unread-steer-1"), + text: "Run the slow curl in the foreground.", + attachments: [], + }); + yield* harness.runtime.startTurn(firstInput); + const providerTurnId = idAllocator.derive.providerTurn({ + driver: ClaudeAdapterV2.CLAUDE_PROVIDER, + nativeTurnId: `turn:${firstInput.attemptId}`, + }); + yield* harness.offerAndWait( + claudeSdkFrame({ + type: "assistant", + uuid: "00000000-0000-4000-8000-000000015701", + session_id: WAKE_NATIVE_SESSION, + parent_tool_use_id: null, + ...(echoes ? { user_message_uuids: [harness.promptUuid(0)] } : {}), + message: { + id: "msg_unread_steer_bash", + type: "message", + role: "assistant", + model: "claude-sonnet-4-6", + content: [ + { + type: "tool_use", + id: "toolu-unread-steer-bash", + name: "Bash", + input: { command: "curl http://127.0.0.1:9/hang" }, + }, + ], + }, + }), + ); + // Past task_started the CLI queues a "now" steer until the call returns. + yield* harness.offerAndWait( + claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: "task-unread-steer-bash", + tool_use_id: "toolu-unread-steer-bash", + description: "Slow curl", + is_backgrounded: false, + task_type: "local_bash", + uuid: "00000000-0000-4000-8000-000000015702", + session_id: WAKE_NATIVE_SESSION, + }), + ); + const steerText = "Codeword: PELICAN. Remember it, I will ask for it later."; + yield* harness.runtime.steerTurn({ + threadId: harness.threadId, + runId: firstInput.runId, + providerThread: harness.providerThread, + providerTurnId, + message: { + createdBy: "user", + creationSource: "web", + messageId: MessageId.make("unread-steer"), + text: steerText, + attachments: [], + }, + }); + const steerUuid = harness.offeredMessages.at(-1)?.uuid; + assert.isDefined(steerUuid); + if (readBy === "assistant") { + yield* harness.offerAndWait( + claudeSdkFrame({ + type: "assistant", + uuid: "00000000-0000-4000-8000-000000015703", + session_id: WAKE_NATIVE_SESSION, + parent_tool_use_id: null, + user_message_uuids: [steerUuid], + message: { + id: "msg_unread_steer_reply", + type: "message", + role: "assistant", + model: "claude-sonnet-4-6", + content: [{ type: "text", text: "Noted." }], + }, + }), + ); + } + const noticesBefore = harness.events.length; + yield* harness.runtime.interruptTurn({ + providerThread: harness.providerThread, + providerTurnId, + }); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "interrupted"); + + const notices = harness.events + .slice(noticesBefore) + .flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "system_notice" + ? [event.turnItem.message] + : [], + ); + const undelivered = notices.find((message) => + /not (have )?(been )?delivered|never (reached|delivered)/i.test(message ?? ""), + ); + if (readBy !== "none" || !echoes) { + assert.isUndefined(undelivered); + return; + } + assert.isDefined( + undelivered, + `expected an undelivered-steer notice, got ${JSON.stringify(notices)}`, + ); + assert.match(undelivered ?? "", /resend|send it again|re-send/i); + assert.include(undelivered, "PELICAN"); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + // regression: #23948 Claude holds a steer behind a running foreground tool // call and goes silent, so each steer stays pending until Claude starts it. it.effect("keeps a steer pending until Claude starts it", () => diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index f29ed3cb65e51..378e399e2532b 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -4453,6 +4453,9 @@ interface ActiveClaudeTurnContext { compactionFailure: { readonly message: string | undefined } | null; // Steers end segments whose usage is excluded from the final result; bill them too, oldest first. readonly steeredSegmentUsage: Array; + // Set as finalization starts: a steer admitted later would stay pending + // forever, so it fails and the orchestrator follows up instead. + steersClosed: boolean; // Ledger growth of the steered segments; undefined once any segment's growth is unknown. steeredLedgerUsage: ClaudeLedgerUsage | undefined; readonly rootModels: Set; @@ -4465,8 +4468,16 @@ interface ActiveClaudeTurnContext { gatedFramesBeforeEcho: number; readonly heldRootFrames: Array; // Steers Claude has not started or echoed yet, by prompt uuid. Claude holds - // a steer behind a running foreground tool call without saying so. - readonly pendingSteers: Map; + // a steer behind a running foreground tool call without saying so, and Stop + // drops it unread. + readonly pendingSteers: Map< + string, + { + readonly messageId: MessageId; + readonly text: string; + readonly liveQuery: ClaudeLiveQueryContext; + } + >; // Claude may settle a turn with a result that echoes only a steer's uuid. readonly steerUuids: Set; // Claude can run another model than requested, such as when an organization disallows it. @@ -8124,10 +8135,10 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ? [lifecycle.uuid] : []; for (const uuid of accepted) { - const messageId = context.pendingSteers.get(uuid); - if (messageId === undefined) continue; + const steer = context.pendingSteers.get(uuid); + if (steer === undefined) continue; context.pendingSteers.delete(uuid); - yield* emitSteerStatus(context, messageId, "completed"); + yield* emitSteerStatus(context, steer.messageId, "completed"); } }); @@ -8142,16 +8153,18 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( readonly threadDisposition?: "reusable" | "broken"; readonly result?: SDKResultMessage; }) { + input.context.steersClosed = true; yield* reasoningDeltas.flushTurn(input.context.nativeTurnId); // A turn that completes has taken in every steer; one that stops drops them unread. - for (const messageId of input.context.pendingSteers.values()) { + const droppedSteers = [...input.context.pendingSteers]; + input.context.pendingSteers.clear(); + for (const [, { messageId }] of droppedSteers) { yield* emitSteerStatus( input.context, messageId, input.status === "completed" ? "completed" : "cancelled", ); } - input.context.pendingSteers.clear(); if (input.context.compaction?.status === "running") { const reason = input.result?.subtype === "success" ? input.result.result.trim() : ""; yield* emitProviderEvent({ @@ -8289,6 +8302,49 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }); } + if (input.status === "interrupted") { + for (const [uuid, { text, liveQuery }] of droppedSteers) { + // Only a CLI proven to echo prompt uuids can leave a steer unconfirmed. + if ( + liveQuery.promptEchoMode !== "early" && + liveQuery.promptEchoMode !== "result_only" + ) { + continue; + } + const nativeItemId = `unread-steer:${uuid}`; + const notice = `Claude did not confirm reading your steer before Stop. It may not have been delivered. Resend it if needed: "${text}"`; + yield* emitProviderEvent({ + type: "turn_item.updated", + driver: CLAUDE_PROVIDER, + turnItem: { + id: idAllocator.derive.turnItemFromProviderItem({ + driver: CLAUDE_PROVIDER, + nativeItemId, + }), + threadId: input.context.input.threadId, + runId: input.context.input.runId, + nodeId: input.context.input.rootNodeId, + providerThreadId: input.context.input.providerThread.id, + providerTurnId: input.context.providerTurnId, + nativeItemRef: { + driver: CLAUDE_PROVIDER, + nativeId: nativeItemId, + strength: "weak", + }, + parentItemId: null, + ordinal: yield* resolveItemOrdinal(input.context, nativeItemId), + type: "system_notice", + status: "completed", + title: "Steer delivery not confirmed", + message: notice, + startedAt: input.completedAt, + completedAt: input.completedAt, + updatedAt: input.completedAt, + }, + }); + } + } + const threadDisposition = input.threadDisposition ?? "reusable"; const terminalEvent: ProviderAdapter.ProviderAdapterV2Event = input.status === "failed" @@ -11538,6 +11594,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( latestAssistantRateLimit: null, compactionFailure: null, steeredSegmentUsage: [], + steersClosed: false, steeredLedgerUsage: { inputTokens: 0, outputTokens: 0 }, rootModels: new Set(), subagentsByTaskId: new Map(), @@ -11978,22 +12035,28 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( // the steer fails and the orchestrator starts the follow-up turn. if ( (yield* Ref.get(activeTurn))?.providerTurnId !== turnInput.providerTurnId || - (yield* Ref.get(queryContext)) !== existing + (yield* Ref.get(queryContext)) !== existing || + // No yield between this check and the set below. + currentTurn.steersClosed ) { return yield* new ProviderAdapter.ProviderAdapterProtocolError({ driver: CLAUDE_PROVIDER, detail: `Claude provider turn ${turnInput.providerTurnId} ended before the steer was sent.`, }); } + // Register before offering: Claude can start the steer and end the turn before offer returns. + const messageId = turnInput.message.messageId; + currentTurn.steerUuids.add(steerUuid); + currentTurn.pendingSteers.set(steerUuid, { + messageId, + text: turnInput.message.text, + liveQuery: existing, + }); yield* Ref.update(steeredTurns, (current) => { const next = new Set(current); next.add(turnInput.providerTurnId); return next; }); - // Register before offering: Claude can start the steer and end the turn before offer returns. - const messageId = turnInput.message.messageId; - currentTurn.steerUuids.add(steerUuid); - currentTurn.pendingSteers.set(steerUuid, messageId); yield* emitSteerStatus(currentTurn, messageId, "pending"); yield* Effect.uninterruptibleMask((restore) => restore(