# Files changed · PR \#19196 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/19196/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [fix\(server\): a Claude wake run settles when Claude never opens the wake turn](https://git.cafe/pingdotgg/t3code/pull/19196.md) Version: 5 Head commit: 9ae3de2f875d2e29cbd47c5ef650bec453f5a3a4 Base commit: 64a76c84ef795b41cffa884491a4c961f9ac0af4 Comparison base: 64a76c84ef795b41cffa884491a4c961f9ac0af4 Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts Status: modified; +204; −0 ```diff @@ -9465,6 +9465,210 @@ ), ); + const startSubagentWakeWithoutWakeTurn = ( + options?: Parameters[0], + bufferedFrames: ReadonlyArray = [], + ) => + Effect.gen(function* () { + const TASK_ID = "task-wake-never-opened"; + const TOOL_USE_ID = "toolu-wake-never-opened"; + const harness = yield* makeWakeHarnessWithOptions(options); + const now = yield* DateTime.now; + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-never-opened-1"), + text: "Audit in the background.", + attachments: [], + }), + ); + for (const frame of [ + makeSubagentTaskStartedFrame({ + taskId: TASK_ID, + toolUseId: TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000981", + }), + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000982", + result: "Auditing in the background.", + }), + ]) { + yield* harness.offerAndWait(frame); + } + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "completed"); + yield* harness.offerAndWait( + makeSubagentNotificationFrame({ + taskId: TASK_ID, + toolUseId: TOOL_USE_ID, + summary: "Audit done.", + uuid: "00000000-0000-4000-8000-000000000983", + }), + ); + yield* Queue.take(harness.continuationReceipts); + for (const frame of bufferedFrames) { + yield* harness.offerAndWait(frame); + } + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-never-opened-2"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + assert.lengthOf(harness.terminalEvents(), 1); + return harness; + }); + + it.effect("settles a subagent wake run although a subagent reports progress meanwhile", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* startSubagentWakeWithoutWakeTurn(); + + yield* TestClock.adjust("1 minute"); + assert.lengthOf(harness.terminalEvents(), 1); + yield* harness.offerAndWait( + claudeSdkFrame({ + type: "system", + subtype: "task_progress", + task_id: "task-still-running-elsewhere", + description: "Reading more files", + uuid: "00000000-0000-4000-8000-000000000984", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* TestClock.adjust("1 minute"); + assert.lengthOf(harness.terminalEvents(), 2); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "completed"); + assert.lengthOf(harness.offeredMessages, 1); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + + it.effect("settles a subagent wake run with only subagent messages buffered", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* startSubagentWakeWithoutWakeTurn(undefined, [ + ...makeSubagentAssistantFrames({ + parentToolUseId: "toolu-wake-never-opened", + text: "Audit done.", + uuid: "00000000-0000-4000-8000-000000000985", + }), + makeSubagentToolResultFrame({ + parentToolUseId: "toolu-wake-never-opened", + toolUseId: "toolu-wake-child-bash", + uuid: "00000000-0000-4000-8000-000000000986", + }), + ]); + + yield* TestClock.adjust("2 minutes"); + assert.lengthOf(harness.terminalEvents(), 2); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "completed"); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + + it.effect("keeps a subagent wake run open while Claude works on the wake turn", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* startSubagentWakeWithoutWakeTurn(); + + yield* harness.offerAndWait(wakeTurnInit); + yield* TestClock.adjust("2 minutes"); + assert.lengthOf(harness.terminalEvents(), 1); + + yield* harness.offerAndWait(wakeResult); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "completed"); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + + it.effect("keeps a subagent wake run open when root output starts streaming", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* startSubagentWakeWithoutWakeTurn(); + yield* harness.offerAndWait( + claudeSdkFrame({ + type: "stream_event", + event: { + type: "content_block_delta", + index: 0, + delta: { type: "text_delta", text: "Working" }, + }, + parent_tool_use_id: null, + uuid: "00000000-0000-4000-8000-000000000987", + session_id: WAKE_NATIVE_SESSION, + }), + ); + + yield* TestClock.adjust("2 minutes"); + assert.lengthOf(harness.terminalEvents(), 1); + yield* harness.offerAndWait(wakeResult); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "completed"); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + + it.effect("does not complete a subagent wake run while Stop is pending", () => + Effect.scoped( + Effect.gen(function* () { + const stopRequested = yield* Deferred.make(); + const releaseStop = yield* Deferred.make(); + const harness = yield* startSubagentWakeWithoutWakeTurn({ + interrupt: Deferred.succeed(stopRequested, undefined).pipe( + Effect.andThen(Deferred.await(releaseStop)), + ), + close: Queue.shutdown, + }); + const stop = yield* harness.runtime + .interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: ClaudeAdapterV2.CLAUDE_PROVIDER, + nativeTurnId: "turn:attempt-claude-wake-never-opened-2", + }), + }) + .pipe(Effect.forkScoped); + yield* Deferred.await(stopRequested); + + yield* TestClock.adjust("2 minutes"); + assert.lengthOf(harness.terminalEvents(), 1); + + yield* Deferred.succeed(releaseStop, undefined); + yield* Fiber.join(stop); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "interrupted"); + assert.lengthOf(harness.terminalEvents(), 2); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + it.effect("keeps a nested subagent's launch in order while the root turn is idle", () => Effect.scoped( Effect.gen(function* () { ``` ## apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts Status: modified; +48; −7 ```diff @@ -273,6 +273,9 @@ // The SDK puts no deadline on control requests. The CLI answers // mcp_set_servers once each server connected or hit its own 30 s timeout. const CLAUDE_MCP_REGISTRATION_TIMEOUT = "90 seconds"; +// Claude can consume a subagent result before the launching turn ends, leaving +// its notification with no native wake turn to answer the continuation. +const CLAUDE_WAKE_TURN_START_DEADLINE = "2 minutes"; // Claude Code flushes transcript writes every 100 ms, so a workflow member's // reply can land after the progress frame that settles it. Its transcript // stays the source of its answer this long after it settles. @@ -2229,6 +2232,14 @@ return message.type === "system" && message.subtype === "init"; } +function isClaudeRootTurnFrame(message: SDKMessage): boolean { + return ( + isClaudeTurnStartMessage(message) || + ((message.type === "user" || message.type === "assistant" || message.type === "stream_event") && + message.parent_tool_use_id == null) + ); +} + function claudePendingBackgroundTasksFromRoster( roster: ReadonlyMap, ): ReadonlyArray { @@ -3200,6 +3211,7 @@ readonly promptUuid: string | null; promptEcho: "pending" | "confirmed"; promptStarted: boolean; + nativeWakeStarted: boolean; gatedFramesBeforeEcho: number; readonly heldRootFrames: Array; } @@ -8289,6 +8301,13 @@ const context = yield* Ref.get(activeTurn); const liveQuery = yield* Ref.get(queryContext); if ( + context?.promptUuid === null && + liveQuery?.query === input.query && + isClaudeRootTurnFrame(message) + ) { + context.nativeWakeStarted = true; + } + if ( context === null || context.promptUuid === null || context.promptEcho === "confirmed" || @@ -9071,6 +9090,7 @@ promptUuid, promptEcho: isClaudeProviderContinuationTurn(turnInput) ? "confirmed" : "pending", promptStarted: false, + nativeWakeStarted: false, gatedFramesBeforeEcho: 0, heldRootFrames: [], }; @@ -9187,15 +9207,36 @@ yield* handleSdkMessage({ query: querySession.query, message: lastResult }); return; } - const hasNativeWakeFrame = drained.some( - (entry) => - entry.type === "user" || - entry.type === "assistant" || + if (hasOpaqueTaskNotification && !context.nativeWakeStarted) { - isClaudeTurnStartMessage(entry), - ); - if (hasOpaqueTaskNotification && !hasNativeWakeFrame) { const completedAt = yield* DateTime.now; yield* finalizeActiveTurn({ context, status: "completed", completedAt }); + } else if (!context.nativeWakeStarted) { + yield* Effect.sleep(CLAUDE_WAKE_TURN_START_DEADLINE).pipe( + Effect.andThen( + sdkMessageLock.withPermits(1)( + Effect.gen(function* () { + if ( + (yield* Ref.get(activeTurn)) !== context || + (yield* Ref.get(queryContext)) !== querySession || + querySession.stopping || + context.nativeWakeStarted || + (yield* Ref.get(interruptedTurns)).has(context.providerTurnId) + ) + return; + yield* Effect.logWarning( + "orchestration-v2.claude-wake-turn-never-opened", + { + providerTurnId: context.providerTurnId, + nativeThreadId, + }, + ); + const completedAt = yield* DateTime.now; + yield* finalizeActiveTurn({ context, status: "completed", completedAt }); + }), + ), + ), + Effect.forkIn(sessionScope), + ); } }), ); ```