From fb7f9667fb47ae1f9baf85a666c460fdb5945f30 Mon Sep 17 00:00:00 2001 From: Mike Olson Date: Mon, 6 Jul 2026 17:26:59 -0400 Subject: [PATCH 1/3] fix(orchestrator): Resume persisted Claude sessions after provider session recycle --- .../Adapters/ClaudeAdapterV2.test.ts | 87 ++++++++++++++++++- .../Adapters/ClaudeAdapterV2.ts | 9 +- 2 files changed, 94 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 54e96ffa084..47c7f5ab639 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -101,13 +101,14 @@ function makeClaudeTestTurnInput(input: { readonly attemptId: RunAttemptId; readonly text: string; readonly attachments: ProviderAdapterV2TurnInput["message"]["attachments"]; + readonly providerTurnOrdinal?: number; }): ProviderAdapterV2TurnInput { return { appThread: makeClaudeTestAppThread(input), threadId: input.threadId, runId: RunId.make(`run-${input.attemptId}`), runOrdinal: 1, - providerTurnOrdinal: 1, + providerTurnOrdinal: input.providerTurnOrdinal ?? 1, attemptId: input.attemptId, rootNodeId: NodeId.make(`node-${input.attemptId}`), providerThread: input.providerThread, @@ -729,3 +730,87 @@ describe("ClaudeAdapterV2 native fork", () => { ), ); }); + +describe("ClaudeAdapterV2 native session identity", () => { + const openTurnWithOrdinal = (providerTurnOrdinal: number) => + Effect.scoped( + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const idAllocator = yield* IdAllocatorV2; + const attachmentsDir = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-claude-v2-session-identity-", + }); + const openedQueries: Array = []; + const adapter = makeClaudeAdapterV2({ + instanceId: CLAUDE_DEFAULT_INSTANCE_ID, + settings: DEFAULT_CLAUDE_SETTINGS, + environment: {}, + attachmentsDir, + fileSystem, + idAllocator, + queryRunner: { + allocateSessionId: Effect.succeed("native-session-identity"), + open: (input) => + Effect.sync(() => { + openedQueries.push(input); + return { + messages: Stream.empty, + offer: () => Effect.void, + setModel: () => Effect.void, + interrupt: Effect.void, + close: Effect.void, + }; + }), + forkSession: () => Effect.die("unused forkSession"), + assertComplete: Effect.void, + }, + }); + const threadId = ThreadId.make("thread-claude-session-identity"); + const providerSessionId = ProviderSessionId.make("provider-session-claude-identity"); + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId, + modelSelection: CLAUDE_TEST_MODEL_SELECTION, + runtimePolicy: CLAUDE_TEST_RUNTIME_POLICY, + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection: CLAUDE_TEST_MODEL_SELECTION, + runtimePolicy: CLAUDE_TEST_RUNTIME_POLICY, + }); + const now = yield* DateTime.now; + yield* runtime.startTurn( + makeClaudeTestTurnInput({ + threadId, + providerThread, + now, + attemptId: RunAttemptId.make("run-attempt-claude-session-identity"), + text: "Respond with identity ok", + attachments: [], + providerTurnOrdinal, + }), + ); + return openedQueries; + }).pipe(Effect.provide(Layer.merge(idAllocatorLayer, NodeServices.layer))), + ); + + it.effect("creates the native session on the first provider turn", () => + Effect.gen(function* () { + const openedQueries = yield* openTurnWithOrdinal(1); + assert.equal(openedQueries.length, 1); + assert.equal(openedQueries[0]?.options.sessionId, "native-session-identity"); + assert.equal(openedQueries[0]?.options.resume, undefined); + }), + ); + + it.effect( + "resumes the native session on a fresh session instance when prior provider turns exist", + () => + Effect.gen(function* () { + const openedQueries = yield* openTurnWithOrdinal(2); + assert.equal(openedQueries.length, 1); + assert.equal(openedQueries[0]?.options.resume, "native-session-identity"); + assert.equal(openedQueries[0]?.options.sessionId, undefined); + }), + ); +}); diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 2740ceff6d2..9539cdb4511 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -3061,7 +3061,14 @@ export function makeClaudeAdapterV2( updated.add(nativeThreadId); return [false, updated]; }); - const shouldResume = resumeSessionAt !== undefined || openedWithResume; + // openedNativeThreads is per session instance and is lost when the + // provider session is idle-released. A prior persisted provider turn + // proves the native session already exists, so the query must resume + // it; reopening with a fixed session id makes the CLI fail fast with + // "Session ID ... is already in use". + const hasPersistedProviderTurn = turnInput.providerTurnOrdinal > 1; + const shouldResume = + resumeSessionAt !== undefined || openedWithResume || hasPersistedProviderTurn; const querySession = yield* queryRunner.open({ threadId: turnInput.threadId, providerSessionId: input.providerSessionId, From 60545887c3e9900b9b62fc56fe35796a89228adc Mon Sep 17 00:00:00 2001 From: Mike Olson Date: Mon, 6 Jul 2026 18:24:28 -0400 Subject: [PATCH 2/3] fix(orchestrator): Surface Claude background wake turns as continuation runs When a background task (run_in_background Bash) settles, the Claude CLI streams a whole new turn over the still-open SDK query. handleSdkMessage dropped every message of that turn (activeTurn null), and the 30 minute idle release killed the CLI child before pending tasks could even wake. Buffer null-activeTurn messages per native thread, request one internal continuation run per wake (message.dispatch with createdBy agent / creationSource provider via a new ProviderContinuationRequests queue and worker), and drain the buffer into the continuation turn without sending a prompt to the CLI. Pin idle provider sessions while background work is pending via a new optional hasPendingBackgroundWork runtime capability, capped by maxIdlePinMs (default 4h). --- .../Adapters/ClaudeAdapterV2.test.ts | 440 +++++++++++++++++- .../Adapters/ClaudeAdapterV2.ts | 227 ++++++++- .../src/orchestration-v2/ProviderAdapter.ts | 6 + .../ProviderContinuationRequests.ts | 39 ++ .../ProviderContinuationService.ts | 72 +++ .../ProviderSessionManager.test.ts | 249 +++++++++- .../ProviderSessionManager.ts | 58 ++- .../src/orchestration-v2/runtimeLayer.ts | 8 + ...viderOrchestrationAdapterInfrastructure.ts | 9 +- 9 files changed, 1089 insertions(+), 19 deletions(-) create mode 100644 apps/server/src/orchestration-v2/ProviderContinuationRequests.ts create mode 100644 apps/server/src/orchestration-v2/ProviderContinuationService.ts diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 47c7f5ab639..2cde1fdfdd4 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -1,4 +1,4 @@ -import type { SDKUserMessage } from "@anthropic-ai/claude-agent-sdk"; +import type { SDKMessage, SDKUserMessage } from "@anthropic-ai/claude-agent-sdk"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { ChatAttachmentId, @@ -24,6 +24,7 @@ import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; +import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; @@ -32,8 +33,10 @@ import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; import type { EventNdjsonLogger } from "../../provider/Layers/EventNdjsonLogger.ts"; import { ProviderAdapterV2RuntimePolicy, + type ProviderAdapterV2Event, type ProviderAdapterV2TurnInput, } from "../ProviderAdapter.ts"; +import type { ProviderContinuationRequest } from "../ProviderContinuationRequests.ts"; import { CLAUDE_AGENT_SDK_QUERY_PROTOCOL, CLAUDE_DEFAULT_INSTANCE_ID, @@ -102,6 +105,8 @@ function makeClaudeTestTurnInput(input: { readonly text: string; readonly attachments: ProviderAdapterV2TurnInput["message"]["attachments"]; readonly providerTurnOrdinal?: number; + readonly messageCreatedBy?: ProviderAdapterV2TurnInput["message"]["createdBy"]; + readonly messageCreationSource?: ProviderAdapterV2TurnInput["message"]["creationSource"]; }): ProviderAdapterV2TurnInput { return { appThread: makeClaudeTestAppThread(input), @@ -113,8 +118,8 @@ function makeClaudeTestTurnInput(input: { rootNodeId: NodeId.make(`node-${input.attemptId}`), providerThread: input.providerThread, message: { - createdBy: "user", - creationSource: "web", + createdBy: input.messageCreatedBy ?? "user", + creationSource: input.messageCreationSource ?? "web", messageId: MessageId.make(`message-${input.attemptId}`), text: input.text, attachments: input.attachments, @@ -814,3 +819,432 @@ describe("ClaudeAdapterV2 native session identity", () => { }), ); }); + +describe("ClaudeAdapterV2 background wake turns", () => { + const WAKE_NATIVE_SESSION = "native-thread-claude-wake"; + const WAKE_TASK_ID = "task-wake-build"; + const WAKE_SUMMARY = "Background build completed successfully"; + const WAKE_RESULT_TEXT = "The background build finished; everything passed."; + + function claudeSdkFrame(frame: unknown): SDKMessage { + if ( + typeof frame !== "object" || + frame === null || + typeof Reflect.get(frame, "type") !== "string" + ) { + throw new Error("Frame is not a Claude Agent SDK message."); + } + return frame as SDKMessage; + } + + const wakeTaskStarted = claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: WAKE_TASK_ID, + description: "npm run build", + task_type: "local_bash", + uuid: "00000000-0000-4000-8000-000000000101", + session_id: WAKE_NATIVE_SESSION, + }); + const makeResultFrame = (input: { readonly uuid: string; readonly result: string }) => + claudeSdkFrame({ + type: "result", + subtype: "success", + duration_ms: 10, + duration_api_ms: 10, + is_error: false, + num_turns: 1, + result: input.result, + stop_reason: "end_turn", + total_cost_usd: 0, + usage: { + input_tokens: 1, + output_tokens: 1, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + modelUsage: {}, + permission_denials: [], + uuid: input.uuid, + session_id: WAKE_NATIVE_SESSION, + }); + const turnOneResult = makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000102", + result: "Kicked off the build in the background.", + }); + const wakeNotification = claudeSdkFrame({ + type: "system", + subtype: "task_notification", + task_id: WAKE_TASK_ID, + status: "completed", + output_file: "/tmp/task-wake-build.log", + summary: WAKE_SUMMARY, + uuid: "00000000-0000-4000-8000-000000000103", + session_id: WAKE_NATIVE_SESSION, + }); + const wakeResult = makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000104", + result: WAKE_RESULT_TEXT, + }); + + const awaitUntil = (predicate: () => boolean, label: string): Effect.Effect => + Effect.gen(function* () { + for (let attempt = 0; attempt < 5000; attempt++) { + if (predicate()) { + return; + } + yield* Effect.yieldNow; + } + return yield* Effect.die(`Timed out waiting for ${label}.`); + }); + + const makeWakeHarness = Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const idAllocator = yield* IdAllocatorV2; + const attachmentsDir = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-claude-v2-wake-", + }); + const sdkMessages = yield* Queue.unbounded(); + const offeredMessages: Array = []; + const continuationRequests: Array = []; + const adapter = makeClaudeAdapterV2({ + instanceId: CLAUDE_DEFAULT_INSTANCE_ID, + settings: DEFAULT_CLAUDE_SETTINGS, + environment: {}, + attachmentsDir, + fileSystem, + idAllocator, + continuationRequests: { + offer: (request) => + Effect.sync(() => { + continuationRequests.push(request); + }), + }, + queryRunner: { + allocateSessionId: Effect.succeed(WAKE_NATIVE_SESSION), + open: () => + Effect.succeed({ + messages: Stream.fromQueue(sdkMessages), + offer: (message) => + Effect.sync(() => { + offeredMessages.push(message); + }), + setModel: () => Effect.void, + interrupt: Effect.void, + close: Effect.void, + }), + forkSession: () => Effect.die("unused forkSession"), + assertComplete: Effect.void, + }, + }); + const threadId = ThreadId.make("thread-claude-wake"); + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("provider-session-claude-wake"), + modelSelection: CLAUDE_TEST_MODEL_SELECTION, + runtimePolicy: CLAUDE_TEST_RUNTIME_POLICY, + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection: CLAUDE_TEST_MODEL_SELECTION, + runtimePolicy: CLAUDE_TEST_RUNTIME_POLICY, + }); + const events: Array = []; + yield* runtime.events.pipe( + Stream.runForEach((event) => + Effect.sync(() => { + events.push(event); + }), + ), + Effect.forkScoped, + ); + if (runtime.hasPendingBackgroundWork === undefined) { + throw new Error("Claude adapter runtime must expose hasPendingBackgroundWork."); + } + const hasPendingBackgroundWork = runtime.hasPendingBackgroundWork; + const terminalEvents = () => + events.filter( + (event): event is Extract => + event.type === "turn.terminal", + ); + return { + runtime, + providerThread, + threadId, + sdkMessages, + offeredMessages, + continuationRequests, + events, + terminalEvents, + hasPendingBackgroundWork, + }; + }); + + it.effect("buffers wake output and requests a single continuation run", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-1"), + text: "Run the build in the background.", + attachments: [], + }), + ); + yield* Queue.offer(harness.sdkMessages, wakeTaskStarted); + yield* Queue.offer(harness.sdkMessages, turnOneResult); + yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + assert.equal(harness.terminalEvents()[0]?.status, "completed"); + assert.isTrue(yield* harness.hasPendingBackgroundWork); + assert.lengthOf(harness.continuationRequests, 0); + + yield* Queue.offer(harness.sdkMessages, wakeNotification); + yield* awaitUntil(() => harness.continuationRequests.length === 1, "continuation request"); + assert.equal(harness.continuationRequests[0]?.threadId, harness.threadId); + assert.equal(harness.continuationRequests[0]?.providerThreadId, harness.providerThread.id); + assert.equal(harness.continuationRequests[0]?.driver, CLAUDE_PROVIDER); + assert.equal(harness.continuationRequests[0]?.detail, WAKE_SUMMARY); + + yield* Queue.offer(harness.sdkMessages, wakeResult); + let settleYields = 0; + yield* awaitUntil(() => settleYields++ >= 50, "wake result to settle into the buffer"); + assert.lengthOf(harness.continuationRequests, 1); + assert.lengthOf(harness.terminalEvents(), 1); + assert.isTrue(yield* harness.hasPendingBackgroundWork); + }).pipe(Effect.provide(Layer.merge(idAllocatorLayer, NodeServices.layer))), + ), + ); + + it.effect("drains buffered wake messages into a continuation turn", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-2a"), + text: "Run the build in the background.", + attachments: [], + }), + ); + yield* Queue.offer(harness.sdkMessages, wakeTaskStarted); + yield* Queue.offer(harness.sdkMessages, turnOneResult); + yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + yield* Queue.offer(harness.sdkMessages, wakeNotification); + yield* Queue.offer(harness.sdkMessages, wakeResult); + yield* awaitUntil(() => harness.continuationRequests.length === 1, "continuation request"); + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-2b"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + + yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal"); + assert.equal(harness.terminalEvents()[1]?.status, "completed"); + // The continuation prompt never reaches the CLI; only the first turn + // offered a user message. + assert.lengthOf(harness.offeredMessages, 1); + // The wake result text surfaces as the continuation turn's assistant + // output. + assert.isTrue( + harness.events.some( + (event) => event.type === "message.updated" && event.message.text === WAKE_RESULT_TEXT, + ), + ); + // The background task never renders as a subagent node. + assert.isFalse( + harness.events.some((event) => JSON.stringify(event).includes(WAKE_TASK_ID)), + ); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + }).pipe(Effect.provide(Layer.merge(idAllocatorLayer, NodeServices.layer))), + ), + ); + + it.effect("leaves buffered wake messages for the continuation queued behind a user turn", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-4a"), + text: "Run the build in the background.", + attachments: [], + }), + ); + yield* Queue.offer(harness.sdkMessages, wakeTaskStarted); + yield* Queue.offer(harness.sdkMessages, turnOneResult); + yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + yield* Queue.offer(harness.sdkMessages, wakeNotification); + yield* Queue.offer(harness.sdkMessages, wakeResult); + yield* awaitUntil(() => harness.continuationRequests.length === 1, "continuation request"); + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-4b"), + text: "How is the build going?", + attachments: [], + providerTurnOrdinal: 2, + }), + ); + + // The user prompt reaches the CLI and the buffer stays untouched: the + // wake result must not settle the user turn or surface under it. + yield* awaitUntil(() => harness.offeredMessages.length === 2, "user prompt offered"); + assert.lengthOf(harness.terminalEvents(), 1); + assert.isTrue(yield* harness.hasPendingBackgroundWork); + + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000105", + result: "The build passed; nothing else pending.", + }), + ); + yield* awaitUntil(() => harness.terminalEvents().length === 2, "user turn terminal"); + assert.equal(harness.terminalEvents()[1]?.status, "completed"); + assert.isFalse( + harness.events.some( + (event) => event.type === "message.updated" && event.message.text === WAKE_RESULT_TEXT, + ), + ); + + // The continuation run queued behind the user turn drains the wake + // output afterwards. + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-4c"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 3, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* awaitUntil(() => harness.terminalEvents().length === 3, "continuation terminal"); + assert.equal(harness.terminalEvents()[2]?.status, "completed"); + assert.lengthOf(harness.offeredMessages, 2); + assert.isTrue( + harness.events.some( + (event) => event.type === "message.updated" && event.message.text === WAKE_RESULT_TEXT, + ), + ); + assert.isFalse( + harness.events.some((event) => JSON.stringify(event).includes(WAKE_TASK_ID)), + ); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + }).pipe(Effect.provide(Layer.merge(idAllocatorLayer, NodeServices.layer))), + ), + ); + + it.effect("clears the pending task when the wake notification carries no summary", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-5a"), + text: "Run the build in the background.", + attachments: [], + }), + ); + yield* Queue.offer(harness.sdkMessages, wakeTaskStarted); + yield* Queue.offer(harness.sdkMessages, turnOneResult); + yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + type: "system", + subtype: "task_notification", + task_id: WAKE_TASK_ID, + status: "completed", + output_file: "/tmp/task-wake-build.log", + summary: null, + uuid: "00000000-0000-4000-8000-000000000106", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* awaitUntil(() => harness.continuationRequests.length === 1, "continuation request"); + assert.isNull(harness.continuationRequests[0]?.detail); + + yield* Queue.offer(harness.sdkMessages, wakeResult); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-5b"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal"); + assert.equal(harness.terminalEvents()[1]?.status, "completed"); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + }).pipe(Effect.provide(Layer.merge(idAllocatorLayer, NodeServices.layer))), + ), + ); + + it.effect("settles a continuation turn immediately when no wake output is buffered", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-3"), + text: "Background task completed.", + attachments: [], + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + + yield* awaitUntil(() => harness.terminalEvents().length === 1, "spurious terminal"); + assert.equal(harness.terminalEvents()[0]?.status, "completed"); + assert.lengthOf(harness.offeredMessages, 0); + }).pipe(Effect.provide(Layer.merge(idAllocatorLayer, NodeServices.layer))), + ), + ); +}); diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 9539cdb4511..8c188ac6ed7 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -38,6 +38,7 @@ import { ProviderDriverKind, type ProviderInstanceId, type ProviderRequestKind, + type ProviderThreadId, type ThreadId, } from "@t3tools/contracts"; @@ -100,6 +101,10 @@ import { type ProviderAdapterDriver, type ProviderAdapterDriverCreateInput, } from "../ProviderAdapterDriver.ts"; +import { + type ProviderContinuationRequest, + ProviderContinuationRequests, +} from "../ProviderContinuationRequests.ts"; import { makeSubagentChildThread, makeSubagentConversationArtifacts, @@ -1823,12 +1828,19 @@ export interface ClaudeAdapterV2Options { readonly fileSystem: FileSystem.FileSystem; readonly idAllocator: IdAllocatorV2Shape; readonly queryRunner: ClaudeAgentSdkQueryRunnerShape; + /** Sink for wake-turn continuation requests; defaults to dropping them. */ + readonly continuationRequests?: { + readonly offer: (request: ProviderContinuationRequest) => Effect.Effect; + }; } export function makeClaudeAdapterV2( adapterOptions: ClaudeAdapterV2Options, ): ProviderAdapterV2Shape { const { attachmentsDir, fileSystem, idAllocator, queryRunner } = adapterOptions; + const continuationRequests = adapterOptions.continuationRequests ?? { + offer: () => Effect.void, + }; return ProviderAdapterV2.of({ instanceId: adapterOptions.instanceId, @@ -1857,6 +1869,26 @@ export function makeClaudeAdapterV2( const pendingRuntimeRequests = yield* Ref.make( new Map(), ); + // Background-task wake support. Claude can settle a turn while a + // local_bash background task keeps running; the CLI later re-invokes + // the model (a "wake turn") on the same query stream with no active + // provider turn. These refs track pending background tasks, buffer + // wake messages until a continuation run attaches, and remember where + // to dispatch that run. + const lastTurnRouteByNativeThread = yield* Ref.make( + new Map< + string, + { readonly threadId: ThreadId; readonly providerThreadId: ProviderThreadId } + >(), + ); + const pendingBackgroundTaskIds = yield* Ref.make(new Set()); + const wakeBuffers = yield* Ref.make( + new Map< + string, + { readonly messages: ReadonlyArray; readonly detail: string | null } + >(), + ); + const requestedContinuations = yield* Ref.make(new Set()); const runtimeContext = yield* Effect.context(); const runFork = Effect.runForkWith(runtimeContext); const runPromise = Effect.runPromiseWith(runtimeContext); @@ -2755,6 +2787,89 @@ export function makeClaudeAdapterV2( } }); + const clearPendingBackgroundTask = (taskId: string) => + Ref.modify(pendingBackgroundTaskIds, (current) => { + if (!current.has(taskId)) { + return [false, current] as const; + } + const updated = new Set(current); + updated.delete(taskId); + return [true, updated] as const; + }); + + const bufferWakeMessage = Effect.fnUntraced(function* (wakeInput: { + readonly nativeThreadId: string; + readonly message: SDKMessage; + }) { + const message = wakeInput.message; + const isNotification = + message.type === "system" && message.subtype === "task_notification"; + // Only notifications for tracked background tasks count as wake + // evidence; a stray notification (e.g. for a subagent task) is + // dropped as before instead of triggering a spurious continuation. + const isPendingTaskNotification = + isNotification && (yield* Ref.get(pendingBackgroundTaskIds)).has(message.task_id); + const isWakeEvidence = + isPendingTaskNotification || + message.type === "assistant" || + message.type === "user" || + message.type === "result"; + if (!isWakeEvidence) { + return; + } + const notificationSummary = + isNotification && typeof message.summary === "string" && message.summary.length > 0 + ? message.summary + : null; + yield* Ref.update(wakeBuffers, (current) => { + const existing = current.get(wakeInput.nativeThreadId); + const updated = new Map(current); + updated.set(wakeInput.nativeThreadId, { + messages: [...(existing?.messages ?? []), message], + detail: notificationSummary ?? existing?.detail ?? null, + }); + return updated; + }); + // Request a continuation run once per wake, when the wake turn has + // either announced the finished task or fully settled. Earlier + // messages only buffer; the continuation turn drains them. + if (!isPendingTaskNotification && message.type !== "result") { + return; + } + const route = (yield* Ref.get(lastTurnRouteByNativeThread)).get(wakeInput.nativeThreadId); + if (route === undefined) { + yield* Effect.logWarning("orchestration-v2.claude-wake-turn-unroutable", { + providerSessionId: input.providerSessionId, + nativeThreadId: wakeInput.nativeThreadId, + }); + return; + } + const shouldOffer = yield* Ref.modify(requestedContinuations, (current) => { + if (current.has(wakeInput.nativeThreadId)) { + return [false, current] as const; + } + const updated = new Set(current); + updated.add(wakeInput.nativeThreadId); + return [true, updated] as const; + }); + if (!shouldOffer) { + return; + } + const detail = + (yield* Ref.get(wakeBuffers)).get(wakeInput.nativeThreadId)?.detail ?? null; + yield* Effect.logInfo("orchestration-v2.claude-wake-turn-detected", { + providerSessionId: input.providerSessionId, + threadId: route.threadId, + providerThreadId: route.providerThreadId, + }); + yield* continuationRequests.offer({ + threadId: route.threadId, + providerThreadId: route.providerThreadId, + driver: CLAUDE_PROVIDER, + detail, + }); + }); + const handleSdkMessage = Effect.fnUntraced(function* (input: { readonly query: ClaudeAgentSdkQuerySession; readonly message: SDKMessage; @@ -2767,6 +2882,7 @@ export function makeClaudeAdapterV2( const message = input.message; const context = yield* Ref.get(activeTurn); if (context === null) { + yield* bufferWakeMessage({ nativeThreadId: liveQuery.nativeThreadId, message }); return; } @@ -2777,6 +2893,9 @@ export function makeClaudeAdapterV2( if (message.type === "system" && message.subtype === "task_started") { if (isClaudeNonSubagentTask(message)) { context.ignoredTaskIds.add(message.task_id); + yield* Ref.update(pendingBackgroundTaskIds, (current) => + new Set(current).add(message.task_id), + ); } else { yield* updateClaudeSubagentNode({ context, @@ -2791,7 +2910,14 @@ export function makeClaudeAdapterV2( if (message.type === "system" && message.subtype === "task_progress") { const progress = message.description.trim(); - if (progress.length > 0 && !context.ignoredTaskIds.has(message.task_id)) { + const isBackgroundTask = (yield* Ref.get(pendingBackgroundTaskIds)).has( + message.task_id, + ); + if ( + progress.length > 0 && + !context.ignoredTaskIds.has(message.task_id) && + !isBackgroundTask + ) { yield* updateClaudeSubagentNode({ context, taskId: message.task_id, @@ -2803,7 +2929,10 @@ export function makeClaudeAdapterV2( } if (message.type === "system" && message.subtype === "task_notification") { - if (!context.ignoredTaskIds.has(message.task_id)) { + // A wake-replay turn has empty ignoredTaskIds, so the session-level + // background registry is the durable ignore signal across turns. + const wasBackgroundTask = yield* clearPendingBackgroundTask(message.task_id); + if (!wasBackgroundTask && !context.ignoredTaskIds.has(message.task_id)) { yield* updateClaudeSubagentNode({ context, taskId: message.task_id, @@ -3136,6 +3265,14 @@ export function makeClaudeAdapterV2( detail: `Claude provider turn ${currentTurn.providerTurnId} is still active.`, }); } + yield* Ref.update(lastTurnRouteByNativeThread, (current) => { + const updated = new Map(current); + updated.set(nativeThreadId, { + threadId: turnInput.threadId, + providerThreadId: turnInput.providerThread.id, + }); + return updated; + }); const context: ActiveClaudeTurnContext = { input: turnInput, nativeTurnId, @@ -3154,15 +3291,24 @@ export function makeClaudeAdapterV2( subagentsByToolUseId: new Map(), subagentNodesByTaskId: new Map(), }; - const userMessage = yield* makeClaudeUserMessageWithAttachments({ - text: applyClaudePromptEffortPrefix( - turnInput.message.text, - compileClaudeModelSelection(turnInput.modelSelection).promptEffort, - ), - attachments: turnInput.message.attachments, - attachmentsDir, - fileSystem, - }); + // Continuation turns attach to the wake output the CLI already + // produced instead of prompting it again: drain the buffered wake + // messages into this turn and let any still-streaming messages + // follow live. The continuation prompt text never reaches the CLI. + const isContinuationTurn = + turnInput.message.createdBy === "agent" && + turnInput.message.creationSource === "provider"; + const userMessage = isContinuationTurn + ? null + : yield* makeClaudeUserMessageWithAttachments({ + text: applyClaudePromptEffortPrefix( + turnInput.message.text, + compileClaudeModelSelection(turnInput.modelSelection).promptEffort, + ), + attachments: turnInput.message.attachments, + attachmentsDir, + fileSystem, + }); const querySession = yield* openQuery(turnInput, nativeThreadId); yield* Ref.set(activeTurn, context); yield* emitProviderEvent({ @@ -3174,7 +3320,48 @@ export function makeClaudeAdapterV2( completedAt: null, }), }); - yield* querySession.query.offer(userMessage); + if (userMessage !== null) { + // A user turn that races a wake leaves the buffer alone: the + // continuation run the worker queued behind this run drains it + // afterwards with correct attribution. + yield* querySession.query.offer(userMessage); + return; + } + const drained = yield* Ref.modify(wakeBuffers, (current) => { + const entry = current.get(nativeThreadId); + if (entry === undefined) { + return [[] as ReadonlyArray, current] as const; + } + const updated = new Map(current); + updated.delete(nativeThreadId); + return [entry.messages, updated] as const; + }); + yield* Ref.update(requestedContinuations, (current) => { + const updated = new Set(current); + updated.delete(nativeThreadId); + return updated; + }); + if (drained.length === 0) { + // Spurious continuation (buffer already lost with a recycled + // session, or a duplicate request): settle immediately instead + // of leaving a run waiting on a prompt that was never sent. + const completedAt = yield* DateTime.now; + yield* finalizeActiveTurn({ context, status: "completed", completedAt }); + return; + } + // Replay any result message last: a result finalizes the turn, and + // replaying it before the rest would drop them back into the wake + // buffer and request another continuation. + const resultMessages = drained.filter((entry) => entry.type === "result"); + for (const entry of drained) { + if (entry.type !== "result") { + yield* handleSdkMessage({ query: querySession.query, message: entry }); + } + } + const lastResult = resultMessages.at(-1); + if (lastResult !== undefined) { + yield* handleSdkMessage({ query: querySession.query, message: lastResult }); + } }, (effect, turnInput) => effect.pipe( @@ -3345,6 +3532,18 @@ export function makeClaudeAdapterV2( providerSessionId: input.providerSessionId, providerSession: session, events: Stream.fromEffectRepeat(Queue.take(events)), + hasPendingBackgroundWork: Effect.gen(function* () { + if ((yield* Ref.get(pendingBackgroundTaskIds)).size > 0) { + return true; + } + const buffers = yield* Ref.get(wakeBuffers); + for (const entry of buffers.values()) { + if (entry.messages.length > 0) { + return true; + } + } + return false; + }), ensureThread: Effect.fn("ClaudeAdapterV2.ensureThread")( function* (threadInput: ProviderAdapterV2EnsureThreadInput) { const createdAt = yield* DateTime.now; @@ -3617,6 +3816,7 @@ export const ClaudeAdapterV2Driver: ProviderAdapterDriver< const idAllocator = yield* IdAllocatorV2; const queryRunner = yield* ClaudeAgentSdkQueryRunner; const serverConfig = yield* ServerConfig; + const continuationRequests = yield* ProviderContinuationRequests; const baseEnvironment = mergeProviderInstanceEnvironment(environment, hostEnvironment); const claudeEnvironment = yield* makeClaudeEnvironment(config, baseEnvironment); return makeClaudeAdapterV2({ @@ -3627,6 +3827,7 @@ export const ClaudeAdapterV2Driver: ProviderAdapterDriver< fileSystem, idAllocator, queryRunner, + continuationRequests, }); }, (effect, input) => @@ -3650,6 +3851,7 @@ const makeDefaultClaudeAdapterV2 = Effect.fn("ClaudeAdapterV2.layer")(function* const idAllocator = yield* IdAllocatorV2; const queryRunner = yield* ClaudeAgentSdkQueryRunner; const serverConfig = yield* ServerConfig; + const continuationRequests = yield* ProviderContinuationRequests; return makeClaudeAdapterV2({ instanceId: CLAUDE_DEFAULT_INSTANCE_ID, @@ -3659,6 +3861,7 @@ const makeDefaultClaudeAdapterV2 = Effect.fn("ClaudeAdapterV2.layer")(function* fileSystem, idAllocator, queryRunner, + continuationRequests, }); }); diff --git a/apps/server/src/orchestration-v2/ProviderAdapter.ts b/apps/server/src/orchestration-v2/ProviderAdapter.ts index d8cf1c9c82d..e24d221231a 100644 --- a/apps/server/src/orchestration-v2/ProviderAdapter.ts +++ b/apps/server/src/orchestration-v2/ProviderAdapter.ts @@ -474,6 +474,12 @@ export interface ProviderAdapterV2SessionRuntime { * Adapter runtimes may omit this and expose only their single-consumer event stream. */ readonly subscribeEvents?: Effect.Effect; + /** + * Adapters whose native runtime can hold pending work outside an active + * turn (for example Claude background tasks and their wake turns) report it + * here so the session manager defers idle release while it is pending. + */ + readonly hasPendingBackgroundWork?: Effect.Effect; readonly ensureThread: ( input: ProviderAdapterV2EnsureThreadInput, ) => Effect.Effect; diff --git a/apps/server/src/orchestration-v2/ProviderContinuationRequests.ts b/apps/server/src/orchestration-v2/ProviderContinuationRequests.ts new file mode 100644 index 00000000000..b972999e011 --- /dev/null +++ b/apps/server/src/orchestration-v2/ProviderContinuationRequests.ts @@ -0,0 +1,39 @@ +import { ProviderDriverKind, ProviderThreadId, ThreadId } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; + +export interface ProviderContinuationRequest { + readonly threadId: ThreadId; + readonly providerThreadId: ProviderThreadId; + readonly driver: ProviderDriverKind; + readonly detail: string | null; +} + +/** + * Adapters offer a continuation request when provider-native work completes + * outside an active turn (for example a Claude background task wake turn) so + * the orchestrator can start a run that ingests it. The default reference + * drops requests, keeping adapter construction dependency-free in tests; the + * live layer must be shared with the ProviderContinuationService worker that + * drains it. + */ +export class ProviderContinuationRequests extends Context.Reference<{ + readonly offer: (request: ProviderContinuationRequest) => Effect.Effect; + readonly take: Effect.Effect; +}>("t3/orchestration-v2/ProviderContinuationRequests", { + defaultValue: () => ({ offer: () => Effect.void, take: Effect.never }), +}) {} + +export const layer = Layer.effect( + ProviderContinuationRequests, + Effect.gen(function* () { + const queue = yield* Queue.unbounded(); + return { + offer: (request: ProviderContinuationRequest) => + Queue.offer(queue, request).pipe(Effect.asVoid), + take: Queue.take(queue), + }; + }), +); diff --git a/apps/server/src/orchestration-v2/ProviderContinuationService.ts b/apps/server/src/orchestration-v2/ProviderContinuationService.ts new file mode 100644 index 00000000000..0ad60d2eb94 --- /dev/null +++ b/apps/server/src/orchestration-v2/ProviderContinuationService.ts @@ -0,0 +1,72 @@ +import { CommandId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +import { IdAllocatorV2 } from "./IdAllocator.ts"; +import { + type ProviderContinuationRequest, + ProviderContinuationRequests, +} from "./ProviderContinuationRequests.ts"; +import { ThreadManagementService } from "./ThreadManagementService.ts"; + +const CONTINUATION_MESSAGE_TEXT = "Background task completed."; + +/** + * Drains ProviderContinuationRequests and dispatches an internal + * message.dispatch per request so the wake turn buffered by the adapter is + * ingested as a normal run. Dispatches queue_after_active, so a continuation + * racing a user run simply queues behind it and drains the wake buffer once + * that run finishes. + */ +export const workerLive = Layer.effectDiscard( + Effect.gen(function* () { + const ids = yield* IdAllocatorV2; + const requests = yield* ProviderContinuationRequests; + const threads = yield* ThreadManagementService; + + const dispatchContinuation = Effect.fn("ProviderContinuationService.dispatchContinuation")( + function* (request: ProviderContinuationRequest) { + const projection = yield* threads.getThreadProjection(request.threadId); + if (projection.thread.archivedAt !== null) { + yield* Effect.logInfo("orchestration-v2.provider-continuation.thread-archived", { + threadId: request.threadId, + providerThreadId: request.providerThreadId, + }); + return; + } + const messageId = yield* ids.allocate.message({ + threadId: request.threadId, + ordinal: projection.messages.length + 1, + }); + const commandId = CommandId.make(`provider-continuation:${messageId}`); + yield* threads.dispatch({ + type: "message.dispatch", + commandId, + threadId: request.threadId, + messageId, + text: request.detail ?? CONTINUATION_MESSAGE_TEXT, + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + creationSource: "provider", + }); + }, + ); + + yield* requests.take.pipe( + Effect.flatMap((request) => + dispatchContinuation(request).pipe( + Effect.catchCause((cause) => + Effect.logWarning("orchestration-v2.provider-continuation.dispatch-failed", { + threadId: request.threadId, + providerThreadId: request.providerThreadId, + cause, + }), + ), + ), + ), + Effect.forever, + Effect.forkScoped, + ); + }), +); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 8c163f9907f..7e8623ea96c 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -232,6 +232,7 @@ function makeProviderAdapter( readonly beforeOpen?: (input: { readonly providerSessionId: ProviderSessionId; }) => Effect.Effect; + readonly hasPendingBackgroundWork?: Effect.Effect; } = {}, ): ProviderAdapterV2Shape { return { @@ -287,6 +288,9 @@ function makeProviderAdapter( }), ) : Stream.fromQueue(events), + ...(options.hasPendingBackgroundWork === undefined + ? {} + : { hasPendingBackgroundWork: options.hasPendingBackgroundWork }), ensureThread: () => unimplemented("ensureThread unused in test"), resumeThread: (threadInput) => Ref.update(state, (current) => ({ @@ -312,6 +316,7 @@ function makeProviderAdapter( function makeTestLayer(input: { readonly state: Ref.Ref; readonly idleTimeoutMs: number; + readonly maxIdlePinMs?: number; readonly failEventStream?: boolean; readonly capabilities?: OrchestrationV2ProviderCapabilities; readonly mcpConfigs?: Ref.Ref< @@ -321,6 +326,7 @@ function makeTestLayer(input: { readonly providerSessionId: ProviderSessionId; }) => Effect.Effect; readonly failReleaseEventWrites?: boolean; + readonly hasPendingBackgroundWork?: Effect.Effect; }) { const configuredEventSinkLayer = input.failReleaseEventWrites ? FailingReleaseEventSinkLayer @@ -331,6 +337,9 @@ function makeTestLayer(input: { ...(input.capabilities === undefined ? {} : { capabilities: input.capabilities }), ...(input.mcpConfigs === undefined ? {} : { mcpConfigs: input.mcpConfigs }), ...(input.beforeOpen === undefined ? {} : { beforeOpen: input.beforeOpen }), + ...(input.hasPendingBackgroundWork === undefined + ? {} + : { hasPendingBackgroundWork: input.hasPendingBackgroundWork }), }), ); return Layer.mergeAll( @@ -338,7 +347,10 @@ function makeTestLayer(input: { configuredEventSinkLayer, idAllocatorLayer, TestMcpRegistryLayer, - providerSessionManagerLayerWithOptions({ idleTimeoutMs: input.idleTimeoutMs }).pipe( + providerSessionManagerLayerWithOptions({ + idleTimeoutMs: input.idleTimeoutMs, + ...(input.maxIdlePinMs === undefined ? {} : { maxIdlePinMs: input.maxIdlePinMs }), + }).pipe( Layer.provide( Layer.mergeAll( registryLayer, @@ -928,6 +940,241 @@ it.effect("ProviderSessionManagerV2 releases idle sessions without sweeping all }), ); +it.effect("ProviderSessionManagerV2 defers idle release while background work is pending", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const pendingWork = yield* Ref.make(true); + const effect = Effect.gen(function* () { + const eventSink = yield* EventSinkV2; + const idAllocator = yield* IdAllocatorV2; + const manager = yield* ProviderSessionManagerV2; + const now = yield* DateTime.now; + const threadId = yield* idAllocator.allocate.thread({ + fixtureName: "provider-session-manager-idle-pin", + projectId: yield* idAllocator.allocate.project({ + fixtureName: "provider-session-manager-idle-pin", + }), + }); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + yield* manager.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + + yield* TestClock.adjust("3 seconds"); + yield* Effect.yieldNow; + assert.isTrue(Option.isSome(yield* manager.get(providerSessionId))); + assert.equal((yield* Ref.get(state)).closeCount, 0); + + yield* Ref.set(pendingWork, false); + yield* TestClock.adjust("1 second"); + yield* Effect.yieldNow; + assert.isTrue(Option.isNone(yield* manager.get(providerSessionId))); + assert.equal((yield* Ref.get(state)).closeCount, 1); + }); + + yield* effect.pipe( + Effect.provide( + makeTestLayer({ + state, + idleTimeoutMs: 1000, + hasPendingBackgroundWork: Ref.get(pendingWork), + }), + ), + ); + }), +); + +it.effect("ProviderSessionManagerV2 releases pinned idle sessions once the pin cap expires", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const effect = Effect.gen(function* () { + const eventSink = yield* EventSinkV2; + const idAllocator = yield* IdAllocatorV2; + const manager = yield* ProviderSessionManagerV2; + const now = yield* DateTime.now; + const threadId = yield* idAllocator.allocate.thread({ + fixtureName: "provider-session-manager-pin-cap", + projectId: yield* idAllocator.allocate.project({ + fixtureName: "provider-session-manager-pin-cap", + }), + }); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + yield* manager.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + + yield* TestClock.adjust("3 seconds"); + yield* Effect.yieldNow; + assert.isTrue(Option.isSome(yield* manager.get(providerSessionId))); + + yield* TestClock.adjust("1 second"); + yield* Effect.yieldNow; + assert.isTrue(Option.isNone(yield* manager.get(providerSessionId))); + assert.equal((yield* Ref.get(state)).closeCount, 1); + }); + + yield* effect.pipe( + Effect.provide( + makeTestLayer({ + state, + idleTimeoutMs: 1000, + maxIdlePinMs: 3000, + hasPendingBackgroundWork: Effect.succeed(true), + }), + ), + ); + }), +); + +it.effect( + "ProviderSessionManagerV2 does not idle-release a session that turns busy during the pending-work check", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const firstCheck = yield* Ref.make(true); + const checkEntered = yield* Deferred.make(); + const checkGate = yield* Deferred.make(); + const effect = Effect.gen(function* () { + const eventSink = yield* EventSinkV2; + const idAllocator = yield* IdAllocatorV2; + const manager = yield* ProviderSessionManagerV2; + const projectionStore = yield* ProjectionStoreV2; + const now = yield* DateTime.now; + const projectId = yield* idAllocator.allocate.project({ + fixtureName: "provider-session-manager-busy-during-check", + }); + const threadId = yield* idAllocator.allocate.thread({ + fixtureName: "provider-session-manager-busy-during-check", + projectId, + }); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const providerThread = makeProviderThread({ + idAllocator, + threadId, + providerSessionId, + now, + }); + const runId = idAllocator.derive.run({ threadId, ordinal: 1 }); + const attemptId = idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 }); + const rootNodeId = idAllocator.derive.rootNode({ runId }); + const providerTurnId = idAllocator.derive.providerTurn({ + driver: CODEX_DRIVER, + nativeTurnId: "native-turn-busy-during-check", + }); + + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + const runtime = yield* manager.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + yield* runtime.events.pipe(Stream.runDrain, Effect.forkScoped); + const appThread = (yield* projectionStore.getThreadProjection(threadId)).thread; + + yield* TestClock.adjust("1 second"); + yield* Deferred.await(checkEntered); + + // The release fiber is parked inside the pending-work check, so the + // idle decision it already made is stale once this turn marks the + // session busy. + const turnFiber = yield* runtime + .startTurn({ + appThread, + threadId, + runId, + runOrdinal: 1, + providerTurnOrdinal: 1, + attemptId, + rootNodeId, + providerThread, + message: { + createdBy: "user", + creationSource: "web", + messageId: yield* idAllocator.allocate.message({ threadId, ordinal: 1 }), + text: "hello", + attachments: [], + }, + modelSelection, + runtimePolicy, + }) + .pipe(Effect.forkDetach); + for (let i = 0; i < 10; i += 1) { + yield* Effect.yieldNow; + } + yield* Deferred.succeed(checkGate, undefined); + yield* Fiber.join(turnFiber); + yield* Effect.yieldNow; + + assert.isTrue(Option.isSome(yield* manager.get(providerSessionId))); + assert.equal((yield* Ref.get(state)).closeCount, 0); + + const queue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId)); + assert.isDefined(queue); + yield* Queue.offer(queue!, { + type: "turn.terminal", + driver: CODEX_DRIVER, + providerThreadId: providerThread.id, + providerTurnId, + runOrdinal: 1, + status: "completed", + failure: null, + threadDisposition: "reusable", + }); + yield* TestClock.adjust("1 second"); + yield* Effect.yieldNow; + assert.isTrue(Option.isNone(yield* manager.get(providerSessionId))); + assert.equal((yield* Ref.get(state)).closeCount, 1); + }); + + yield* effect.pipe( + Effect.provide( + makeTestLayer({ + state, + idleTimeoutMs: 1000, + // Uninterruptible so the markBusy-triggered interrupt cannot land + // inside the check, mirroring an adapter that masks interruption + // while inspecting its own state. + hasPendingBackgroundWork: Effect.uninterruptible( + Effect.gen(function* () { + if (yield* Ref.getAndSet(firstCheck, false)) { + yield* Deferred.succeed(checkEntered, undefined); + yield* Deferred.await(checkGate); + } + return false; + }), + ), + }), + ), + ); + }), +); + it.effect( "ProviderSessionManagerV2 keeps active sessions alive until the provider turn terminates", () => diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 0e330180ad7..422c093490b 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -40,6 +40,7 @@ import { ProviderAdapterRegistryV2 } from "./ProviderAdapterRegistry.ts"; import { ProjectionStoreV2 } from "./ProjectionStore.ts"; const DEFAULT_IDLE_TIMEOUT_MS = 30 * 60 * 1000; +const DEFAULT_MAX_IDLE_PIN_MS = 4 * 60 * 60 * 1000; export const ProviderSessionReleaseReason = Schema.Literals([ "idle_timeout", @@ -174,6 +175,8 @@ interface LiveSessionEntry { readonly busyCount: number; readonly lastActivityAtMs: number; readonly idleFiber: Fiber.Fiber | null; + /** Set when idle release is deferred for pending background work; bounds total deferral. */ + readonly pinnedSinceMs: number | null; } type ProviderSessionEventSignal = @@ -185,6 +188,8 @@ type ProviderSessionEventSignal = export interface ProviderSessionManagerV2LayerOptions { readonly idleTimeoutMs?: number; + /** Cap on how long idle release may be deferred for pending background work. */ + readonly maxIdlePinMs?: number; /** Test replay harnesses can omit T3's MCP server from provider protocol fixtures. */ readonly configureMcp?: boolean; } @@ -252,6 +257,7 @@ export const layerWithOptions = ( const nextSubscriberId = yield* Ref.make(0); const sessionOpen = yield* makeKeyedSerialExecutor(); const idleTimeoutMs = Math.max(1, options.idleTimeoutMs ?? DEFAULT_IDLE_TIMEOUT_MS); + const maxIdlePinMs = Math.max(0, options.maxIdlePinMs ?? DEFAULT_MAX_IDLE_PIN_MS); const prepareMcpSession = (threadId: ThreadId, providerInstanceId: ProviderInstanceId) => options.configureMcp === false ? Effect.sync(() => McpProviderSession.clearMcpProviderSession(threadId)) @@ -471,6 +477,7 @@ export const layerWithOptions = ( readonly reason: ProviderSessionReleaseReason; readonly detail?: string; readonly cancelIdleFiber?: boolean; + readonly onlyIfIdleGeneration?: number; }) => Effect.acquireUseRelease( Ref.modify(sessions, (current) => { @@ -479,6 +486,12 @@ export const layerWithOptions = ( if (existing === undefined) { return [Option.none(), current] as const; } + if ( + input.onlyIfIdleGeneration !== undefined && + (existing.busyCount > 0 || existing.idleGeneration !== input.onlyIfIdleGeneration) + ) { + return [Option.none(), current] as const; + } const updated = new Map(current); updated.delete(key); return [Option.some(existing), updated] as const; @@ -532,13 +545,16 @@ export const layerWithOptions = ( ), ); + // Annotated to break the releaseIfStillIdle <-> scheduleIdleReleaseInternal + // inference cycle introduced by the pin re-arm below. const releaseIfStillIdle = (input: { readonly providerSessionId: ProviderSessionId; readonly generation: number; - }) => + }): Effect.Effect => Effect.gen(function* () { const current = yield* Ref.get(sessions); - const entry = current.get(sessionKey(input.providerSessionId)); + const key = sessionKey(input.providerSessionId); + const entry = current.get(key); if ( entry === undefined || entry.busyCount > 0 || @@ -546,10 +562,46 @@ export const layerWithOptions = ( ) { return; } + const hasPendingWork = + entry.runtime.hasPendingBackgroundWork === undefined + ? false + : yield* entry.runtime.hasPendingBackgroundWork.pipe( + Effect.catchCause(() => Effect.succeed(false)), + ); + if (hasPendingWork) { + const now = yield* Clock.currentTimeMillis; + const pinnedSinceMs = entry.pinnedSinceMs ?? now; + if (now - pinnedSinceMs < maxIdlePinMs) { + yield* Effect.logInfo("orchestration-v2.driver-session.idle-release-deferred", { + providerSessionId: input.providerSessionId, + pinnedForMs: now - pinnedSinceMs, + }); + yield* Ref.update(sessions, (latest) => { + const latestEntry = latest.get(key); + if (latestEntry === undefined || latestEntry.busyCount > 0) { + return latest; + } + const updated = new Map(latest); + updated.set(key, { ...latestEntry, pinnedSinceMs }); + return updated; + }); + yield* scheduleIdleReleaseInternal(input.providerSessionId); + return; + } + yield* Effect.logWarning("orchestration-v2.driver-session.idle-release-pin-expired", { + providerSessionId: input.providerSessionId, + pinnedForMs: now - pinnedSinceMs, + }); + } + // hasPendingBackgroundWork yields to the adapter, so the idle + // decision above can go stale; the generation guard revalidates + // busyCount and idleGeneration inside releaseEntry's atomic + // entry removal. yield* releaseEntry({ providerSessionId: input.providerSessionId, reason: "idle_timeout", cancelIdleFiber: false, + onlyIfIdleGeneration: input.generation, }).pipe( Effect.catchCause((cause) => Effect.logWarning("orchestration-v2.driver-session.idle-release-failed", { @@ -751,6 +803,7 @@ export const layerWithOptions = ( busyCount: entry.busyCount + 1, idleFiber: null, lastActivityAtMs: now, + pinnedSinceMs: null, }); return [entry.idleFiber, updated] as const; }); @@ -1154,6 +1207,7 @@ export const layerWithOptions = ( busyCount: 0, lastActivityAtMs: now, idleFiber: null, + pinnedSinceMs: null, }; yield* Ref.update(sessions, (current) => { const updated = new Map(current); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index 33a3b796a51..c2a4a3fd5a9 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -24,6 +24,8 @@ import { layer as orchestratorLayer } from "./Orchestrator.ts"; import { layer as projectionStoreLayer } from "./ProjectionStore.ts"; import { layer as projectionMaintenanceLayer } from "./ProjectionMaintenance.ts"; import { layerFromProviderInstanceRegistry as providerAdapterRegistryLayerFromProviderInstances } from "./ProviderAdapterRegistry.ts"; +import { layer as providerContinuationRequestsLayer } from "./ProviderContinuationRequests.ts"; +import { workerLive as providerContinuationWorkerLive } from "./ProviderContinuationService.ts"; import { layer as providerEventIngestorLayer } from "./ProviderEventIngestor.ts"; import { layer as providerSessionManagerLayer } from "./ProviderSessionManager.ts"; import { layer as providerRuntimeRecoveryLayer } from "./ProviderRuntimeRecoveryService.ts"; @@ -221,6 +223,11 @@ const threadLifecycleProvided = threadLifecycleServiceLayer.pipe( const scheduledTaskProvided = scheduledTaskServiceLayer.pipe( Layer.provide(Layer.mergeAll(threadLaunchProvided, threadManagementProvided)), ); +const providerContinuationWorkerProvided = providerContinuationWorkerLive.pipe( + Layer.provide( + Layer.mergeAll(providerContinuationRequestsLayer, threadManagementProvided, idAllocatorLayer), + ), +); export const OrchestrationV2LayerLive = Layer.mergeAll( orchestratorProvided, @@ -238,4 +245,5 @@ export const OrchestrationV2ProductionLayerLive = Layer.mergeAll( threadLaunchProvided, threadLifecycleProvided, scheduledTaskProvided, + providerContinuationWorkerProvided, ); diff --git a/apps/server/src/provider/Layers/ProviderOrchestrationAdapterInfrastructure.ts b/apps/server/src/provider/Layers/ProviderOrchestrationAdapterInfrastructure.ts index bf34ea30657..747654fe39f 100644 --- a/apps/server/src/provider/Layers/ProviderOrchestrationAdapterInfrastructure.ts +++ b/apps/server/src/provider/Layers/ProviderOrchestrationAdapterInfrastructure.ts @@ -13,6 +13,7 @@ import { cursorAgentSdkRunnerLiveLayer, } from "../../orchestration-v2/Adapters/CursorAgentSdk.ts"; import { IdAllocatorV2, layer as idAllocatorLayer } from "../../orchestration-v2/IdAllocator.ts"; +import { layer as providerContinuationRequestsLayer } from "../../orchestration-v2/ProviderContinuationRequests.ts"; export type ProviderOrchestrationAdapterInfrastructure = | ClaudeAgentSdkQueryRunner @@ -20,10 +21,16 @@ export type ProviderOrchestrationAdapterInfrastructure = | CursorAgentSdkRunner | IdAllocatorV2; -/** Infrastructure shared by the V2 adapters materialized inside provider instances. */ +/** + * Infrastructure shared by the V2 adapters materialized inside provider + * instances. `providerContinuationRequestsLayer` must be the same layer + * reference the orchestration runtime provides to its continuation worker so + * Effect layer memoization yields one shared queue. + */ export const ProviderOrchestrationAdapterInfrastructureLive = Layer.mergeAll( claudeAgentSdkQueryRunnerLiveLayer, codexAppServerClientFactoryFromSettingsLayer, cursorAgentSdkRunnerLiveLayer, idAllocatorLayer, + providerContinuationRequestsLayer, ); From 90e67c3b29ec42b3e4bf445fcc4f701359251736 Mon Sep 17 00:00:00 2001 From: Mike Olson Date: Mon, 6 Jul 2026 21:20:29 -0400 Subject: [PATCH 3/3] fix(orchestrator): Prevent Claude session release from hanging on idle CLI reads --- .../Adapters/ClaudeAdapterV2.test.ts | 55 +++++++++++++++++- .../Adapters/ClaudeAdapterV2.ts | 12 +++- .../ProviderSessionManager.test.ts | 58 +++++++++++++++++++ .../ProviderSessionManager.ts | 49 +++++++++++++++- 4 files changed, 169 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 2cde1fdfdd4..cb45c84cf6e 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -1,4 +1,8 @@ -import type { SDKMessage, SDKUserMessage } from "@anthropic-ai/claude-agent-sdk"; +import type { + Query as ClaudeQuery, + SDKMessage, + SDKUserMessage, +} from "@anthropic-ai/claude-agent-sdk"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { ChatAttachmentId, @@ -21,11 +25,13 @@ import { import { assert, describe, it } from "@effect/vitest"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; +import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import { attachmentRelativePath } from "../../attachmentStore.ts"; @@ -44,6 +50,7 @@ import { CLAUDE_READ_ONLY_ALLOWED_TOOLS, ClaudeProviderCapabilitiesV2, claudeMcpQueryOverrides, + claudeQueryMessages, claudeRuntimeQueryPolicyForRuntimePolicy, loggedClaudeQueryOptions, makeClaudeAdapterV2, @@ -1248,3 +1255,49 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); }); + +describe("ClaudeAdapterV2 query message stream", () => { + it.effect("closes the query when the message stream is interrupted mid-read", () => + Effect.gen(function* () { + let closed = false; + let releaseRead = () => {}; + const readStarted = Promise.withResolvers(); + async function* sdkMessages(): AsyncGenerator { + while (!closed) { + await new Promise((resolve) => { + releaseRead = resolve; + readStarted.resolve(); + }); + } + } + const generator = sdkMessages(); + const close = () => { + closed = true; + releaseRead(); + }; + const query = { + next: () => generator.next(), + return: async (value?: void) => { + close(); + return generator.return(value); + }, + throw: (error?: unknown) => generator.throw(error), + [Symbol.asyncIterator]: () => generator, + close, + } as unknown as ClaudeQuery; + + const scope = yield* Scope.make(); + yield* Stream.fromAsyncIterable(claudeQueryMessages(query), (cause) => cause).pipe( + Stream.runForEach(() => Effect.void), + Effect.forkIn(scope), + ); + yield* Effect.promise(() => readStarted.promise); + + // Iterating query[Symbol.asyncIterator]() directly deadlocks here: + // the raw generator's return() queues behind the in-flight read and + // scope close never completes. + yield* Scope.close(scope, Exit.void); + assert.isTrue(closed); + }), + ); +}); diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 8c188ac6ed7..195e017f9e6 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -320,6 +320,16 @@ function closeClaudeQuery(queryRuntime: ClaudeQuery) { }); } +// Iterate the Query itself, not query[Symbol.asyncIterator]() (the raw +// sdkMessages generator). The raw generator's return() queues behind the +// in-flight read of the next CLI message and never settles while the CLI is +// idle, deadlocking stream interruption (and with it, session scope close). +// Query.return() runs cleanup() first, which closes the transport and +// unblocks that read. +export function claudeQueryMessages(queryRuntime: ClaudeQuery): AsyncIterable { + return { [Symbol.asyncIterator]: () => queryRuntime }; +} + export interface ClaudeAgentSdkLoggedQueryOptions { readonly model: ClaudeAgentSdkQueryOptions["model"]; readonly tools: ClaudeAgentSdkQueryOptions["tools"]; @@ -513,7 +523,7 @@ export const claudeAgentSdkQueryRunnerLiveLayer: Layer.Layer< }); return { - messages: Stream.fromAsyncIterable(queryRuntime, (cause) => + messages: Stream.fromAsyncIterable(claudeQueryMessages(queryRuntime), (cause) => queryRunnerError(cause, "fromAsyncIterable"), ).pipe( Stream.tap((message) => diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 7e8623ea96c..abc70ee5b7c 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -233,6 +233,7 @@ function makeProviderAdapter( readonly providerSessionId: ProviderSessionId; }) => Effect.Effect; readonly hasPendingBackgroundWork?: Effect.Effect; + readonly hangSessionScopeClose?: boolean; } = {}, ): ProviderAdapterV2Shape { return { @@ -273,6 +274,12 @@ function makeProviderAdapter( closeCount: current.closeCount + 1, })), ); + if (options.hangSessionScopeClose === true) { + // Registered last so it runs first on scope close, wedging the + // close before the closeCount finalizer, like a provider process + // that never yields its message stream. + yield* Effect.addFinalizer(() => Effect.never); + } return { instanceId: ProviderInstanceId.make("codex"), @@ -327,6 +334,7 @@ function makeTestLayer(input: { }) => Effect.Effect; readonly failReleaseEventWrites?: boolean; readonly hasPendingBackgroundWork?: Effect.Effect; + readonly hangSessionScopeClose?: boolean; }) { const configuredEventSinkLayer = input.failReleaseEventWrites ? FailingReleaseEventSinkLayer @@ -340,6 +348,9 @@ function makeTestLayer(input: { ...(input.hasPendingBackgroundWork === undefined ? {} : { hasPendingBackgroundWork: input.hasPendingBackgroundWork }), + ...(input.hangSessionScopeClose === undefined + ? {} + : { hangSessionScopeClose: input.hangSessionScopeClose }), }), ); return Layer.mergeAll( @@ -940,6 +951,53 @@ it.effect("ProviderSessionManagerV2 releases idle sessions without sweeping all }), ); +it.effect("ProviderSessionManagerV2 persists release when session scope close hangs", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const effect = Effect.gen(function* () { + const eventSink = yield* EventSinkV2; + const idAllocator = yield* IdAllocatorV2; + const manager = yield* ProviderSessionManagerV2; + const projectionStore = yield* ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = yield* idAllocator.allocate.thread({ + fixtureName: "provider-session-manager-hung-close", + projectId: yield* idAllocator.allocate.project({ + fixtureName: "provider-session-manager-hung-close", + }), + }); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + yield* manager.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + + yield* TestClock.adjust("1 second"); + yield* Effect.yieldNow; + assert.isTrue(Option.isNone(yield* manager.get(providerSessionId))); + + yield* TestClock.adjust("30 seconds"); + yield* Effect.yieldNow; + const projection = yield* projectionStore.getThreadProjection(threadId); + assert.equal(projection.providerSessions.at(-1)?.status, "stopped"); + assert.equal((yield* Ref.get(state)).closeCount, 0); + }); + + yield* effect.pipe( + Effect.provide(makeTestLayer({ state, idleTimeoutMs: 1000, hangSessionScopeClose: true })), + ); + }), +); + it.effect("ProviderSessionManagerV2 defers idle release while background work is pending", () => Effect.gen(function* () { const state = yield* Ref.make(emptyState); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 422c093490b..48584ca7b2b 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -41,6 +41,7 @@ import { ProjectionStoreV2 } from "./ProjectionStore.ts"; const DEFAULT_IDLE_TIMEOUT_MS = 30 * 60 * 1000; const DEFAULT_MAX_IDLE_PIN_MS = 4 * 60 * 60 * 1000; +const RELEASE_SCOPE_CLOSE_TIMEOUT_MS = 30 * 1000; export const ProviderSessionReleaseReason = Schema.Literals([ "idle_timeout", @@ -512,7 +513,49 @@ export const layerWithOptions = ( input.detail ?? `Provider session released: ${input.reason}.`, ); } - const closeExit = yield* Effect.exit(Scope.close(entry.scope, Exit.void)); + // Scope close can wedge on a misbehaving adapter finalizer + // (e.g. a provider process that never yields its message + // stream). Time-box it so release still persists released + // events and leaves a diagnosable trail instead of silently + // parking the session as "ready" forever. + const closeFiber = yield* Scope.close(entry.scope, Exit.void).pipe( + Effect.exit, + Effect.forkDetach({ startImmediately: true }), + ); + const closeExit = yield* Fiber.join(closeFiber).pipe( + Effect.timeoutOption(RELEASE_SCOPE_CLOSE_TIMEOUT_MS), + ); + if (Option.isNone(closeExit)) { + yield* Effect.logWarning( + "orchestration-v2.provider-session-scope-close-timeout", + { + providerSessionId: input.providerSessionId, + reason: input.reason, + timeoutMs: RELEASE_SCOPE_CLOSE_TIMEOUT_MS, + }, + ); + yield* Fiber.join(closeFiber).pipe( + Effect.flatMap((exit) => + Exit.isFailure(exit) + ? Effect.logWarning( + "orchestration-v2.provider-session-scope-close-failed", + { + providerSessionId: input.providerSessionId, + reason: input.reason, + cause: exit.cause, + }, + ) + : Effect.logInfo( + "orchestration-v2.provider-session-scope-close-completed-late", + { + providerSessionId: input.providerSessionId, + reason: input.reason, + }, + ), + ), + Effect.forkDetach, + ); + } yield* writeReleasedSessionEvents({ entry, reason: input.reason, @@ -522,8 +565,8 @@ export const layerWithOptions = ( entry, reason: input.reason, }); - if (Exit.isFailure(closeExit)) { - return yield* Effect.failCause(closeExit.cause); + if (Option.isSome(closeExit) && Exit.isFailure(closeExit.value)) { + return yield* Effect.failCause(closeExit.value.cause); } }), }),