From 41aa3978fec776dd7891f6b38cc8298e0e4ce21a Mon Sep 17 00:00:00 2001 From: Mike Olson Date: Thu, 9 Jul 2026 00:45:53 -0400 Subject: [PATCH 1/2] feat(orchestrator): Add shared provider continuation plumbing Introduce a provider-agnostic continuation request queue, worker, and optional hasPendingBackgroundWork idle pin so adapters can surface post-settle native wake traffic as first-class runs. No adapter offers continuations yet; behavior is unchanged until Claude or Grok specialty commits wire attach mode. --- .../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 +- 7 files changed, 437 insertions(+), 4 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/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 c489f02704f5529dd75973d331bbeb3b17439e2c Mon Sep 17 00:00:00 2001 From: Mike Olson Date: Fri, 10 Jul 2026 13:14:13 -0400 Subject: [PATCH 2/2] fix(orchestrator): Restore Claude session continuity for resume, wake, and idle release Resume persisted Claude sessions after provider session recycle. Surface 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; buffer null-activeTurn wake messages, request one internal continuation run per wake, and drain the buffer in attach mode without re-prompting the CLI, reporting pending background work so idle release stays pinned until the wake is ingested. Prevent session release from hanging on idle CLI reads. Require claude-agent-sdk 0.3.205 and align the replay testkit with it: derive requestId from toolUseID for replayed permission requests and fail the replay when canUseTool returns null. Regenerate the native-stack patch header so current aube parses it. --- apps/server/package.json | 2 +- .../Adapters/ClaudeAdapterV2.test.ts | 1327 ++++++++++++++++- .../Adapters/ClaudeAdapterV2.testkit.ts | 13 + .../Adapters/ClaudeAdapterV2.ts | 421 +++++- .../ProviderSessionManager.test.ts | 58 + .../ProviderSessionManager.ts | 49 +- ...act-navigation%2Fnative-stack@7.17.6.patch | 2 + pnpm-lock.yaml | 76 +- 8 files changed, 1880 insertions(+), 68 deletions(-) diff --git a/apps/server/package.json b/apps/server/package.json index 0ce49d35421..6fc162f168d 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -25,7 +25,7 @@ "test": "vp test run" }, "dependencies": { - "@anthropic-ai/claude-agent-sdk": "^0.3.170", + "@anthropic-ai/claude-agent-sdk": "^0.3.205", "@connectrpc/connect": "1.7.0", "@connectrpc/connect-node": "1.7.0", "@cursor/sdk": "1.0.22", diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 54e96ffa084..63f5739a9f4 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 { 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,10 +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"; @@ -32,8 +39,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, @@ -41,6 +50,7 @@ import { CLAUDE_READ_ONLY_ALLOWED_TOOLS, ClaudeProviderCapabilitiesV2, claudeMcpQueryOverrides, + claudeQueryMessages, claudeRuntimeQueryPolicyForRuntimePolicy, loggedClaudeQueryOptions, makeClaudeAdapterV2, @@ -101,19 +111,22 @@ function makeClaudeTestTurnInput(input: { readonly attemptId: RunAttemptId; 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), 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, 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, @@ -729,3 +742,1309 @@ 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); + }), + ); +}); + +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))), + ), + ); + + it.effect("wakes and hydrates a subagent that completes after the root turn settled", () => + Effect.scoped( + Effect.gen(function* () { + const SUBAGENT_TASK_ID = "task-wake-subagent"; + const SUBAGENT_TOOL_USE_ID = "toolu-wake-subagent"; + const SUBAGENT_SUMMARY = "SUB_SETTLE_DONE"; + const subagentTaskStarted = claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: SUBAGENT_TASK_ID, + tool_use_id: SUBAGENT_TOOL_USE_ID, + description: "Sleep then echo done token", + subagent_type: "general-purpose", + task_type: "local_agent", + prompt: "Run the shell command, then return exactly SUB_SETTLE_DONE.", + uuid: "00000000-0000-4000-8000-000000000201", + session_id: WAKE_NATIVE_SESSION, + }); + const subagentNotification = claudeSdkFrame({ + type: "system", + subtype: "task_notification", + task_id: SUBAGENT_TASK_ID, + tool_use_id: SUBAGENT_TOOL_USE_ID, + status: "completed", + output_file: "/tmp/task-wake-subagent.output", + summary: SUBAGENT_SUMMARY, + uuid: "00000000-0000-4000-8000-000000000202", + session_id: WAKE_NATIVE_SESSION, + }); + // The SDK resolves a background Agent tool_use immediately with an + // async-launch ACK; it must not terminalize the subagent. + const subagentAsyncAck = claudeSdkFrame({ + type: "user", + message: { + role: "user", + content: [ + { + type: "tool_result", + tool_use_id: SUBAGENT_TOOL_USE_ID, + content: [{ type: "text", text: "Async agent launched successfully." }], + }, + ], + }, + parent_tool_use_id: null, + uuid: "00000000-0000-4000-8000-000000000205", + session_id: WAKE_NATIVE_SESSION, + tool_use_result: { + isAsync: true, + status: "async_launched", + agentId: SUBAGENT_TASK_ID, + prompt: "Run the shell command, then return exactly SUB_SETTLE_DONE.", + }, + }); + + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + const subagentEvents = () => + harness.events.filter( + (event): event is Extract => + event.type === "subagent.updated", + ); + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-6a"), + text: "Spawn a background subagent and stop.", + attachments: [], + }), + ); + yield* Queue.offer(harness.sdkMessages, subagentTaskStarted); + yield* awaitUntil(() => subagentEvents().length >= 1, "subagent node created"); + assert.equal(subagentEvents()[0]?.subagent.status, "running"); + yield* Queue.offer(harness.sdkMessages, subagentAsyncAck); + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000203", + result: "Spawned the subagent in the background.", + }), + ); + yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + assert.equal(harness.terminalEvents()[0]?.status, "completed"); + // The ACK tool_result did not terminalize the row, and a still-running + // subagent pins idle release like a background task. + assert.equal(subagentEvents().at(-1)?.subagent.status, "running"); + assert.isTrue(yield* harness.hasPendingBackgroundWork); + assert.lengthOf(harness.continuationRequests, 0); + + yield* Queue.offer(harness.sdkMessages, subagentNotification); + yield* awaitUntil(() => harness.continuationRequests.length === 1, "continuation request"); + assert.equal(harness.continuationRequests[0]?.threadId, harness.threadId); + assert.equal(harness.continuationRequests[0]?.detail, SUBAGENT_SUMMARY); + + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000204", + result: "The subagent finished with SUB_SETTLE_DONE.", + }), + ); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-6b"), + 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 replayed notification hydrates the original subagent node with + // its terminal status and result, and keeps the original run + // attribution instead of re-parenting to the continuation run. + const finalSubagent = subagentEvents().at(-1)?.subagent; + assert.equal(finalSubagent?.status, "completed"); + assert.equal(finalSubagent?.result, SUBAGENT_SUMMARY); + assert.equal(finalSubagent?.runId, subagentEvents()[0]?.subagent.runId); + const subagentNodeEvents = harness.events.filter( + (event): event is Extract => + event.type === "node.updated" && + event.node.kind === "subagent" && + event.node.nativeItemRef?.nativeId === SUBAGENT_TASK_ID, + ); + const finalSubagentNode = subagentNodeEvents.at(-1)?.node; + assert.equal(finalSubagentNode?.status, "completed"); + assert.equal(finalSubagentNode?.runId, subagentNodeEvents[0]?.node.runId); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + }).pipe(Effect.provide(Layer.merge(idAllocatorLayer, NodeServices.layer))), + ), + ); + + it.effect("releases the idle pin when a post-settle subagent stops without completing", () => + Effect.scoped( + Effect.gen(function* () { + const SUBAGENT_TASK_ID = "task-wake-subagent-stopped"; + const subagentTaskStarted = claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: SUBAGENT_TASK_ID, + tool_use_id: "toolu-wake-subagent-stopped", + description: "Long-running research task", + subagent_type: "general-purpose", + task_type: "local_agent", + prompt: "Investigate the flaky test.", + uuid: "00000000-0000-4000-8000-000000000301", + session_id: WAKE_NATIVE_SESSION, + }); + const subagentStoppedNotification = claudeSdkFrame({ + type: "system", + subtype: "task_notification", + task_id: SUBAGENT_TASK_ID, + tool_use_id: "toolu-wake-subagent-stopped", + status: "stopped", + output_file: "/tmp/task-wake-subagent-stopped.output", + summary: "Agent was stopped before finishing.", + uuid: "00000000-0000-4000-8000-000000000302", + session_id: WAKE_NATIVE_SESSION, + }); + + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + const subagentEvents = () => + harness.events.filter( + (event): event is Extract => + event.type === "subagent.updated", + ); + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-7a"), + text: "Spawn a background subagent and stop.", + attachments: [], + }), + ); + yield* Queue.offer(harness.sdkMessages, subagentTaskStarted); + yield* awaitUntil(() => subagentEvents().length >= 1, "subagent node created"); + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000303", + result: "Spawned the subagent in the background.", + }), + ); + yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + assert.isTrue(yield* harness.hasPendingBackgroundWork); + + yield* Queue.offer(harness.sdkMessages, subagentStoppedNotification); + yield* awaitUntil(() => harness.continuationRequests.length === 1, "continuation request"); + + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000304", + result: "The subagent was stopped.", + }), + ); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-7b"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal"); + + assert.equal(subagentEvents().at(-1)?.subagent.status, "cancelled"); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + }).pipe(Effect.provide(Layer.merge(idAllocatorLayer, NodeServices.layer))), + ), + ); + + it.effect("re-opens a resumed subagent and hydrates its second result", () => + Effect.scoped( + Effect.gen(function* () { + const SUBAGENT_TASK_ID = "task-resume-subagent"; + const SUBAGENT_TOOL_USE_ID = "toolu-resume-subagent"; + const RESUME_TOOL_USE_ID = "toolu-resume-sendmessage"; + const FIRST_SUMMARY = "Timer armed. Waiting for it to complete."; + const SECOND_SUMMARY = "RESUME_DONE"; + const subagentTaskStarted = claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: SUBAGENT_TASK_ID, + tool_use_id: SUBAGENT_TOOL_USE_ID, + description: "Sleep then echo done token", + subagent_type: "general-purpose", + task_type: "local_agent", + prompt: "Run the shell command, then return exactly RESUME_DONE.", + uuid: "00000000-0000-4000-8000-000000000401", + session_id: WAKE_NATIVE_SESSION, + }); + const firstNotification = claudeSdkFrame({ + type: "system", + subtype: "task_notification", + task_id: SUBAGENT_TASK_ID, + tool_use_id: SUBAGENT_TOOL_USE_ID, + status: "completed", + output_file: "/tmp/task-resume-subagent.output", + summary: FIRST_SUMMARY, + uuid: "00000000-0000-4000-8000-000000000402", + session_id: WAKE_NATIVE_SESSION, + }); + // SendMessage to a completed subagent resumes it: the CLI re-emits + // task_started with the same task id but the SendMessage call's + // tool_use_id, not the original Agent launch's. + const resumeTaskStarted = claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: SUBAGENT_TASK_ID, + tool_use_id: RESUME_TOOL_USE_ID, + description: "Sleep then echo done token", + subagent_type: "general-purpose", + task_type: "local_agent", + prompt: "Run the shell command, then return exactly RESUME_DONE.", + uuid: "00000000-0000-4000-8000-000000000405", + session_id: WAKE_NATIVE_SESSION, + }); + const secondNotification = claudeSdkFrame({ + type: "system", + subtype: "task_notification", + task_id: SUBAGENT_TASK_ID, + tool_use_id: SUBAGENT_TOOL_USE_ID, + status: "completed", + output_file: "/tmp/task-resume-subagent.output", + summary: SECOND_SUMMARY, + uuid: "00000000-0000-4000-8000-000000000407", + session_id: WAKE_NATIVE_SESSION, + }); + + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + const subagentEvents = () => + harness.events.filter( + (event): event is Extract => + event.type === "subagent.updated", + ); + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-8a"), + text: "Spawn a background subagent and stop.", + attachments: [], + }), + ); + yield* Queue.offer(harness.sdkMessages, subagentTaskStarted); + yield* awaitUntil(() => subagentEvents().length >= 1, "subagent node created"); + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000403", + result: "Spawned the subagent in the background.", + }), + ); + yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + + yield* Queue.offer(harness.sdkMessages, firstNotification); + yield* awaitUntil( + () => harness.continuationRequests.length === 1, + "first continuation request", + ); + assert.equal(harness.continuationRequests[0]?.detail, FIRST_SUMMARY); + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000404", + result: "The subagent finished early.", + }), + ); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-8b"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal"); + assert.equal(subagentEvents().at(-1)?.subagent.status, "completed"); + assert.equal(subagentEvents().at(-1)?.subagent.result, FIRST_SUMMARY); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + + // A user turn nudges the completed subagent via SendMessage; the + // resume task_started re-opens the row across turn contexts (the new + // turn's maps are empty, so this exercises the session registry). + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-8c"), + text: "Nudge the subagent to finish.", + attachments: [], + providerTurnOrdinal: 3, + }), + ); + yield* awaitUntil(() => harness.offeredMessages.length === 2, "nudge prompt offered"); + // The resume rides on a SendMessage tool call: the CLI re-emits + // task_started with the SendMessage tool_use_id, and that tool call's + // result is a delivery ACK which must not terminalize the subagent. + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + type: "assistant", + message: { + role: "assistant", + content: [ + { + type: "tool_use", + id: RESUME_TOOL_USE_ID, + name: "SendMessage", + input: { agent_id: SUBAGENT_TASK_ID, message: "Continue and return the token." }, + }, + ], + }, + parent_tool_use_id: null, + uuid: "00000000-0000-4000-8000-000000000411", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* Queue.offer(harness.sdkMessages, resumeTaskStarted); + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + type: "user", + message: { + role: "user", + content: [ + { + type: "tool_result", + tool_use_id: RESUME_TOOL_USE_ID, + content: [ + { + type: "text", + text: '{"success":true,"message":"Message sent to agent; it will resume."}', + }, + ], + }, + ], + }, + parent_tool_use_id: null, + uuid: "00000000-0000-4000-8000-000000000412", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* awaitUntil( + () => subagentEvents().at(-1)?.subagent.status === "running", + "subagent re-opened", + ); + const reopened = subagentEvents().at(-1)?.subagent; + assert.isNull(reopened?.result); + // The reopen re-attributes the subagent to the resuming run: + // RunExecutionService routes parent-thread events by runId, and the + // launch run's ingestion fiber stops once its child subagents + // terminalize, so only the resuming run's fiber can persist the + // resumed lifecycle. + assert.equal(reopened?.runId, "run-attempt-claude-wake-8c"); + assert.notEqual(reopened?.runId, subagentEvents()[0]?.subagent.runId); + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000406", + result: "Nudged the subagent.", + }), + ); + yield* awaitUntil(() => harness.terminalEvents().length === 3, "nudge turn terminal"); + // The re-opened subagent pins idle release again. + assert.isTrue(yield* harness.hasPendingBackgroundWork); + + // The resumed run's notification is wake evidence again and carries + // its summary as the continuation detail. + yield* Queue.offer(harness.sdkMessages, secondNotification); + yield* awaitUntil( + () => harness.continuationRequests.length === 2, + "second continuation request", + ); + assert.equal(harness.continuationRequests[1]?.detail, SECOND_SUMMARY); + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000408", + result: "The subagent finished with RESUME_DONE.", + }), + ); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-8d"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 4, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* awaitUntil( + () => harness.terminalEvents().length === 4, + "second continuation terminal", + ); + + const finalSubagent = subagentEvents().at(-1)?.subagent; + assert.equal(finalSubagent?.status, "completed"); + assert.equal(finalSubagent?.result, SECOND_SUMMARY); + // The completion keeps the resuming run's attribution. + assert.equal(finalSubagent?.runId, "run-attempt-claude-wake-8c"); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + + // Only task_started may re-open a terminal subagent: a late + // task_progress must not flip the row back to running or re-pin idle. + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-8e"), + text: "Anything new?", + attachments: [], + providerTurnOrdinal: 5, + }), + ); + yield* awaitUntil(() => harness.offeredMessages.length === 3, "final prompt offered"); + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + type: "system", + subtype: "task_progress", + task_id: SUBAGENT_TASK_ID, + tool_use_id: SUBAGENT_TOOL_USE_ID, + description: "Stale progress line", + uuid: "00000000-0000-4000-8000-000000000409", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000410", + result: "Nothing new.", + }), + ); + yield* awaitUntil(() => harness.terminalEvents().length === 5, "final turn terminal"); + assert.equal(subagentEvents().at(-1)?.subagent.status, "completed"); + assert.equal(subagentEvents().at(-1)?.subagent.result, SECOND_SUMMARY); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + }).pipe(Effect.provide(Layer.merge(idAllocatorLayer, NodeServices.layer))), + ), + ); + + it.effect("re-opens a resumed subagent whose task_started races past settle", () => + Effect.scoped( + Effect.gen(function* () { + const SUBAGENT_TASK_ID = "task-resume-postsettle"; + const SUBAGENT_TOOL_USE_ID = "toolu-resume-postsettle"; + const RESUME_TOOL_USE_ID = "toolu-resume-postsettle-sendmessage"; + const FIRST_SUMMARY = "Answered early."; + const SECOND_SUMMARY = "RESUME_SETTLE_DONE"; + const subagentTaskStarted = claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: SUBAGENT_TASK_ID, + tool_use_id: SUBAGENT_TOOL_USE_ID, + description: "Sleep then echo done token", + subagent_type: "general-purpose", + task_type: "local_agent", + prompt: "Run the shell command, then return exactly RESUME_SETTLE_DONE.", + uuid: "00000000-0000-4000-8000-000000000501", + session_id: WAKE_NATIVE_SESSION, + }); + const firstNotification = claudeSdkFrame({ + type: "system", + subtype: "task_notification", + task_id: SUBAGENT_TASK_ID, + tool_use_id: SUBAGENT_TOOL_USE_ID, + status: "completed", + output_file: "/tmp/task-resume-postsettle.output", + summary: FIRST_SUMMARY, + uuid: "00000000-0000-4000-8000-000000000502", + session_id: WAKE_NATIVE_SESSION, + }); + const resumeTaskStarted = claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: SUBAGENT_TASK_ID, + tool_use_id: RESUME_TOOL_USE_ID, + description: "Sleep then echo done token", + subagent_type: "general-purpose", + task_type: "local_agent", + prompt: "Run the shell command, then return exactly RESUME_SETTLE_DONE.", + uuid: "00000000-0000-4000-8000-000000000505", + session_id: WAKE_NATIVE_SESSION, + }); + const secondNotification = claudeSdkFrame({ + type: "system", + subtype: "task_notification", + task_id: SUBAGENT_TASK_ID, + tool_use_id: SUBAGENT_TOOL_USE_ID, + status: "completed", + output_file: "/tmp/task-resume-postsettle.output", + summary: SECOND_SUMMARY, + uuid: "00000000-0000-4000-8000-000000000506", + session_id: WAKE_NATIVE_SESSION, + }); + + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + const subagentEvents = () => + harness.events.filter( + (event): event is Extract => + event.type === "subagent.updated", + ); + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-9a"), + text: "Spawn a background subagent and stop.", + attachments: [], + }), + ); + yield* Queue.offer(harness.sdkMessages, subagentTaskStarted); + yield* awaitUntil(() => subagentEvents().length >= 1, "subagent node created"); + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000503", + result: "Spawned the subagent in the background.", + }), + ); + yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + + yield* Queue.offer(harness.sdkMessages, firstNotification); + yield* awaitUntil( + () => harness.continuationRequests.length === 1, + "first continuation request", + ); + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000504", + result: "The subagent answered early.", + }), + ); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-9b"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal"); + assert.equal(subagentEvents().at(-1)?.subagent.status, "completed"); + assert.equal(subagentEvents().at(-1)?.subagent.result, FIRST_SUMMARY); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + + // The resume task_started races past settle: no turn is active, so it + // must re-open the session registry entry (pinning idle again) and + // buffer for replay. Its notification then counts as wake evidence + // and carries the new summary as the continuation detail. The resume + // rides on a SendMessage tool call whose frames race past settle too; + // on drain replay the SendMessage tool_result is a delivery ACK and + // must not terminalize the re-opened subagent. + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + type: "assistant", + message: { + role: "assistant", + content: [ + { + type: "tool_use", + id: RESUME_TOOL_USE_ID, + name: "SendMessage", + input: { agent_id: SUBAGENT_TASK_ID, message: "Continue and return the token." }, + }, + ], + }, + parent_tool_use_id: null, + uuid: "00000000-0000-4000-8000-000000000508", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* Queue.offer(harness.sdkMessages, resumeTaskStarted); + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + type: "user", + message: { + role: "user", + content: [ + { + type: "tool_result", + tool_use_id: RESUME_TOOL_USE_ID, + content: [ + { + type: "text", + text: '{"success":true,"message":"Message sent to agent; it will resume."}', + }, + ], + }, + ], + }, + parent_tool_use_id: null, + uuid: "00000000-0000-4000-8000-000000000509", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* Queue.offer(harness.sdkMessages, secondNotification); + yield* awaitUntil( + () => harness.continuationRequests.length === 2, + "second continuation request", + ); + assert.equal(harness.continuationRequests[1]?.detail, SECOND_SUMMARY); + assert.isTrue(yield* harness.hasPendingBackgroundWork); + + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000507", + result: "The subagent finished with RESUME_SETTLE_DONE.", + }), + ); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-9c"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 3, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* awaitUntil( + () => harness.terminalEvents().length === 3, + "resume continuation terminal", + ); + + // The drained replay re-opens the row (running, stale result cleared) + // before the second notification terminalizes it again. + const statuses = subagentEvents().map((event) => event.subagent.status); + const firstCompleted = statuses.indexOf("completed"); + const reopenedIndex = statuses.lastIndexOf("running"); + assert.isAbove(reopenedIndex, firstCompleted); + assert.isNull(subagentEvents()[reopenedIndex]?.subagent.result); + // The drain-replayed reopen re-attributes the subagent to the + // continuation run performing the replay, so that run's ingestion + // fiber routes the resumed lifecycle and lingers past settle until + // the resumed task completes. + assert.equal(subagentEvents()[reopenedIndex]?.subagent.runId, "run-attempt-claude-wake-9c"); + // The execution node re-opens too, even though the registry entry was + // already pre-opened by the wake buffer before the drain replay. + const nodeStatuses = harness.events + .filter( + (event): event is Extract => + event.type === "node.updated" && + event.node.kind === "subagent" && + event.node.nativeItemRef?.nativeId === SUBAGENT_TASK_ID, + ) + .map((event) => event.node.status); + assert.isAbove(nodeStatuses.lastIndexOf("running"), nodeStatuses.indexOf("completed")); + const finalSubagent = subagentEvents().at(-1)?.subagent; + assert.equal(finalSubagent?.status, "completed"); + assert.equal(finalSubagent?.result, SECOND_SUMMARY); + // The completion keeps the resuming run's attribution. + assert.equal(finalSubagent?.runId, "run-attempt-claude-wake-9c"); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + }).pipe(Effect.provide(Layer.merge(idAllocatorLayer, NodeServices.layer))), + ), + ); +}); + +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.testkit.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.testkit.ts index b7e5067d72f..0e1c28a85a2 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.testkit.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.testkit.ts @@ -346,6 +346,9 @@ function permissionRequestOptionsFromFrame( ...(options.description === undefined ? {} : { description: options.description }), toolUseID: options.toolUseID, ...(options.agentID === undefined ? {} : { agentID: options.agentID }), + // Replay transcripts predate the SDK requiring requestId on canUseTool + // options; the adapter only reads toolUseID, so a derived id is safe. + requestId: options.toolUseID, }; } @@ -463,6 +466,16 @@ function makeReplayQueryRunner(transcript: ClaudeAgentSdkReplayTranscript): Clau request.input, permissionRequestOptionsFromFrame(request), ); + if (result === null) { + const error = new ClaudeReplayUnexpectedOutboundError({ + scenario: transcript.scenario, + cursor, + expectedType: "permission.response", + actual: null, + }); + failure = error; + throw error; + } assertNextOutboundFrame(makeClaudePermissionResponseFrame(result)); continue; } diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 2740ceff6d2..91c027a3794 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, @@ -315,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"]; @@ -508,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) => @@ -1406,6 +1421,19 @@ function claudeSubagentResultText(output: ClaudeNativeToolOutput): string { return claudeNativeToolOutputText(output); } +function isClaudeSubagentAsyncLaunchAck(output: ClaudeNativeToolOutput): boolean { + const value = claudeNativeToolOutputValue(output); + if (typeof value === "object" && value !== null) { + if ("isAsync" in value && value.isAsync === true) { + return true; + } + if ("status" in value && value.status === "async_launched") { + return true; + } + } + return claudeSubagentResultText(output).startsWith("Async agent launched successfully."); +} + function webSearchPatternsFromClaudeTool(input: { readonly toolInput: ClaudeNativeToolInput; readonly output: ClaudeNativeToolOutput; @@ -1823,12 +1851,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 +1892,31 @@ 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()); + // Subagent registry that survives turn settle: a background subagent + // (Agent with run_in_background) can complete after the root turn + // ended, and its task_notification must both count as wake evidence + // and hydrate the original subagent node instead of being dropped. + const sessionSubagentsByTaskId = yield* Ref.make(new Map()); + 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); @@ -2060,17 +2120,45 @@ export function makeClaudeAdapterV2( OrchestrationV2ExecutionNode["status"], "running" | "completed" | "failed" | "cancelled" >; + readonly reopen?: boolean; }) { + // The session registry lets a wake-replay turn (fresh context maps) + // hydrate a subagent that was created by an earlier, settled turn. const existingSubagent = input.context.subagentsByTaskId.get(input.taskId) ?? (input.toolUseId === undefined ? undefined - : input.context.subagentsByToolUseId.get(input.toolUseId)); + : input.context.subagentsByToolUseId.get(input.toolUseId)) ?? + (yield* Ref.get(sessionSubagentsByTaskId)).get(input.taskId); if (existingSubagent === undefined && input.status !== "running") { return; } + // Status is monotone with one exception: task_started for a known + // task id is an authoritative CLI lifecycle event (SendMessage to a + // completed subagent resumes it and re-emits task_started with the + // same id), so it may re-open a terminal entry. A late or + // out-of-order task_progress must not. + const isReopen = + input.reopen === true && + existingSubagent !== undefined && + existingSubagent.task.status !== "running" && + input.status === "running"; + if ( + existingSubagent !== undefined && + existingSubagent.task.status !== "running" && + input.status === "running" && + !isReopen + ) { + return; + } const lifecycleChanged = - existingSubagent === undefined || existingSubagent.task.status !== input.status; + existingSubagent === undefined || + existingSubagent.task.status !== input.status || + // A drain-replayed resume task_started finds the registry entry + // already pre-opened to running by bufferWakeMessage while the + // projection node still holds the old terminal status; re-emit + // the node lifecycle for authoritative task_started updates. + (input.reopen === true && input.status === "running"); const now = yield* DateTime.now; const nativeItemId = `task:${input.taskId}`; @@ -2098,8 +2186,19 @@ export function makeClaudeAdapterV2( const turnItemOrdinal = existingSubagent?.turnItemOrdinal ?? (yield* resolveItemOrdinal(input.context, `${nativeItemId}:subagent`)); + // A resumed subagent's previous final answer and progress no longer + // represent its outcome; the next task_progress/task_notification + // carry the new ones. + const priorTask = + existingSubagent === undefined + ? undefined + : isReopen + ? (({ progress: _staleProgress, ...rest }) => ({ ...rest, result: null }))( + existingSubagent.task, + ) + : existingSubagent.task; const task = { - ...(existingSubagent?.task ?? { + ...(priorTask ?? { id: nodeId, threadId: input.context.input.threadId, runId: input.context.input.runId, @@ -2122,6 +2221,18 @@ export function makeClaudeAdapterV2( startedAt: now, }), status: input.status, + // A reopen replayed under a continuation run re-attributes the + // subagent to that run. RunExecutionService routes parent-thread + // events by runId, and only the resuming run's ingestion fiber is + // guaranteed alive (the launch run's fiber stops once its child + // subagents terminalize); attribution also enrolls the subagent + // in the resuming run's active-child tracking so its fiber + // outlives settle until the resumed task completes. + ...(input.reopen === true && + input.status === "running" && + existingSubagent !== undefined + ? { runId: input.context.input.runId } + : {}), ...(input.prompt === undefined ? {} : { prompt: input.prompt }), ...(input.title === undefined ? {} : { title: input.title }), ...(input.progress === undefined ? {} : { progress: input.progress }), @@ -2149,6 +2260,25 @@ export function makeClaudeAdapterV2( if (input.toolUseId !== undefined) { input.context.subagentsByToolUseId.set(input.toolUseId, subagent); } + // The same terminal protection, applied atomically: a concurrent + // fiber (live stream vs continuation drain) may have terminalized + // the registry entry after this update's lookup read it. A resume + // re-open (task_started) bypasses it only when the registered entry + // is still the terminal generation the lookup resolved; if a + // concurrent fiber installed a newer terminal entry meanwhile, the + // re-open must not clobber its result. + yield* Ref.update(sessionSubagentsByTaskId, (current) => { + const registered = current.get(input.taskId); + if ( + registered !== undefined && + registered.task.status !== "running" && + input.status === "running" && + !(isReopen && registered === existingSubagent) + ) { + return current; + } + return new Map(current).set(input.taskId, subagent); + }); if (existingSubagent === undefined) { const childThread = makeSubagentChildThread({ @@ -2181,10 +2311,13 @@ export function makeClaudeAdapterV2( driver: CLAUDE_PROVIDER, node: { id: nodeId, - threadId: input.context.input.threadId, - runId: input.context.input.runId, - parentNodeId: input.context.input.rootNodeId, - rootNodeId: input.context.input.rootNodeId, + // Parenting stays with the launch run's root node even on + // wake-replay; runId follows task.runId, which a reopen + // re-attributes to the resuming run (see task construction). + threadId: task.threadId, + runId: task.runId, + parentNodeId: task.parentNodeId, + rootNodeId: task.parentNodeId, kind: "subagent", status: input.status, countsForRun: false, @@ -2755,6 +2888,138 @@ 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 tasks count as wake evidence: a + // pending local_bash background task, or a session-registered + // subagent that is still running (Agent with run_in_background + // settling after the root turn). A stray notification for an + // unknown task is dropped as before instead of triggering a + // spurious continuation. + const isPendingTaskNotification = + isNotification && (yield* Ref.get(pendingBackgroundTaskIds)).has(message.task_id); + const isPendingSubagentNotification = + isNotification && + !isPendingTaskNotification && + (yield* Ref.get(sessionSubagentsByTaskId)).get(message.task_id)?.task.status === + "running"; + // A task_started for a session-registered subagent that races past + // settle is a resume (SendMessage to a completed subagent re-emits + // task_started with the same task id). Re-open the registry entry + // so the resumed run pins idle and its eventual notification counts + // as wake evidence again, and buffer the frame so the continuation + // drain re-opens the projection row; it does not itself offer a + // continuation. + const isKnownSubagentTaskStarted = + message.type === "system" && + message.subtype === "task_started" && + (yield* Ref.get(sessionSubagentsByTaskId)).has(message.task_id); + if ( + isKnownSubagentTaskStarted && + message.type === "system" && + message.subtype === "task_started" + ) { + const now = yield* DateTime.now; + yield* Ref.update(sessionSubagentsByTaskId, (current) => { + const registered = current.get(message.task_id); + if (registered === undefined || registered.task.status === "running") { + return current; + } + const { progress: _staleProgress, ...priorTask } = registered.task; + return new Map(current).set(message.task_id, { + ...registered, + task: { + ...priorTask, + status: "running", + result: null, + completedAt: null, + updatedAt: now, + }, + }); + }); + } + const isWakeEvidence = + isPendingTaskNotification || + isPendingSubagentNotification || + isKnownSubagentTaskStarted || + 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 && + !isPendingSubagentNotification && + 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 +3032,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 +3043,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, @@ -2785,13 +3054,21 @@ export function makeClaudeAdapterV2( ...(message.prompt === undefined ? {} : { prompt: message.prompt }), title: message.description, status: "running", + reopen: true, }); } } 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 +3080,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, @@ -2834,7 +3114,18 @@ export function makeClaudeAdapterV2( for (const { toolResult, output } of claudeToolResultEntriesFromMessage(message)) { const subagent = context.subagentsByToolUseId.get(toolResult.tool_use_id); - if (subagent !== undefined) { + // A resume task_started reuses the resuming tool call's + // tool_use_id (e.g. SendMessage), whose tool_result only + // acknowledges delivery. Only the Agent launch's tool_result may + // terminalize the subagent, and Agent tool_uses never enter + // toolCalls (they project as subagent rows instead). + if (subagent !== undefined && !context.toolCalls.has(toolResult.tool_use_id)) { + // A background Agent launch resolves its tool_use immediately + // with an async-launch ACK while the task keeps running; only + // the eventual task_notification terminalizes the subagent. + if (isClaudeSubagentAsyncLaunchAck(output)) { + continue; + } const result = claudeSubagentResultText(output); yield* updateClaudeSubagentNode({ context, @@ -3061,7 +3352,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, @@ -3129,6 +3427,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, @@ -3147,15 +3453,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({ @@ -3167,7 +3482,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( @@ -3338,6 +3694,23 @@ 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; + } + for (const subagent of (yield* Ref.get(sessionSubagentsByTaskId)).values()) { + if (subagent.task.status === "running") { + 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; @@ -3610,6 +3983,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({ @@ -3620,6 +3994,7 @@ export const ClaudeAdapterV2Driver: ProviderAdapterDriver< fileSystem, idAllocator, queryRunner, + continuationRequests, }); }, (effect, input) => @@ -3643,6 +4018,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, @@ -3652,6 +4028,7 @@ const makeDefaultClaudeAdapterV2 = Effect.fn("ClaudeAdapterV2.layer")(function* fileSystem, idAllocator, queryRunner, + continuationRequests, }); }); 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); } }), }), diff --git a/patches/@react-navigation%2Fnative-stack@7.17.6.patch b/patches/@react-navigation%2Fnative-stack@7.17.6.patch index cd2b86fb76c..1ec4d978529 100644 --- a/patches/@react-navigation%2Fnative-stack@7.17.6.patch +++ b/patches/@react-navigation%2Fnative-stack@7.17.6.patch @@ -1,3 +1,5 @@ +diff --git a/lib/module/views/useHeaderConfigProps.js b/lib/module/views/useHeaderConfigProps.js +index 0b75c70b4e0d233ee3b5faaf9cfbc40d4f8ed494..eb174e3fde91a7783f132b3fb16b01175ec19220 100644 --- a/lib/module/views/useHeaderConfigProps.js +++ b/lib/module/views/useHeaderConfigProps.js @@ -19,6 +19,12 @@ diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 17be82f7e4c..c4ef97bf201 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -73,7 +73,7 @@ patchedDependencies: '@legendapp/list@3.2.0': 45e4cbcbeeca1b628b8480869374d7cbc77e3fbfa97ee0107a0d72c3c4a5d7f3 '@pierre/diffs@1.3.0-beta.5': 7cb6da88544119adda056b2f46f43956f99326227732da0b345081e285a6c53a '@react-native-menu/menu@2.0.0': 5ea3ae4bf1d9baf5443b65c269bb09621c27a68d556f713778f37b1e8d46aaae - '@react-navigation/native-stack@7.17.6': 2d7fec7933e324ccf082b1ae0df3907c35d16d48c117d628e27d9c3921c26bca + '@react-navigation/native-stack@7.17.6': c7fc101b78d434904425e5a24c22fb0042298dec6f807250486e784f3c717273 effect@4.0.0-beta.78: c502bc684210b707dfceb87d8fe6ad6843395af6e19cfc02cd65854898bde2c5 expo-modules-jsi@56.0.10: 9170f8074ae4e35a0a086e756c8f815794fd3abe51eac67ca3ba02804225ec1f react-native-gesture-handler@2.31.2: 808eb26f9e57cf4945efd3985af4d9c764da6f91f4c9764433cc868602bbf4d3 @@ -419,8 +419,8 @@ importers: apps/server: dependencies: '@anthropic-ai/claude-agent-sdk': - specifier: ^0.3.170 - version: 0.3.170(@anthropic-ai/sdk@0.93.0(zod@4.4.3))(@modelcontextprotocol/sdk@1.29.0(zod@4.4.3))(zod@4.4.3) + specifier: ^0.3.205 + version: 0.3.205(@anthropic-ai/sdk@0.93.0(zod@4.4.3))(@modelcontextprotocol/sdk@1.29.0(zod@4.4.3))(zod@4.4.3) '@connectrpc/connect': specifier: 1.7.0 version: 1.7.0(@bufbuild/protobuf@1.10.0) @@ -917,52 +917,52 @@ packages: '@alchemy.run/node-utils@0.0.4': resolution: {integrity: sha512-TiIhPXCTCi3tk0zmdYJJ14CNSesSfsJxXdIOP0HTSItQ1mZWLocrF7qCuEWKyW/IEFzp6kaiOf19aIA/mbCp1g==} - '@anthropic-ai/claude-agent-sdk-darwin-arm64@0.3.170': - resolution: {integrity: sha512-rwfgArIa5WI0QPNqFsRBgvtSI0mrtpynUm0oK6+l6/KX4hcgnYGEzciZR1bOeD9/7sSZlTdIgt+T9alKeZmXcg==} + '@anthropic-ai/claude-agent-sdk-darwin-arm64@0.3.205': + resolution: {integrity: sha512-lrfJ4eVtzfPkCpbSkBOGSMQCBbvmW6nbPzgHE4IwMN3scZlpuFMUFqh2aaJa/X2SAcWD9H2S0t2WWvSRgM7BjA==} cpu: [arm64] os: [darwin] - '@anthropic-ai/claude-agent-sdk-darwin-x64@0.3.170': - resolution: {integrity: sha512-0e58h8UQMtsQxLGIv9r4foxfBFWKZ7NeDtoplLhuD7EwQonehomw1sBXCch77t/IfUS+q5vQ5zv+fOGmap5nLQ==} + '@anthropic-ai/claude-agent-sdk-darwin-x64@0.3.205': + resolution: {integrity: sha512-G6ETPmL5mNzJ2DFsWxG3jmsmrXgZX1N2ZCJvxaGUUpjTsKZJ4Tup1cWYvcd/m7o5fYZmx9REmgzTwsAIc1fdPQ==} cpu: [x64] os: [darwin] - '@anthropic-ai/claude-agent-sdk-linux-arm64-musl@0.3.170': - resolution: {integrity: sha512-SRYfQcsXlOq+CD/FqkQBTSHbaD++w73GnnO+NUV9adLYrca3kfetRwWT1iguY1cNS0l34dCR3rlzCPq78vg1Jg==} + '@anthropic-ai/claude-agent-sdk-linux-arm64-musl@0.3.205': + resolution: {integrity: sha512-91fgdG4aTnQ29sKOcUqgH4+tKCW2ut6PWGRSYmXNDbROasJm1rAlPdzC5brdu/e4c0CDSNV6TWyE5JCjaS/jlQ==} cpu: [arm64] os: [linux] libc: [musl] - '@anthropic-ai/claude-agent-sdk-linux-arm64@0.3.170': - resolution: {integrity: sha512-gLbaFqcGppFJQd4DLNV4IXoeahejT/p2/M8bSSvRDbla9GOsBr1AxV5XLRyBn1e7xFGozZIAIQr3+1chp7NJgQ==} + '@anthropic-ai/claude-agent-sdk-linux-arm64@0.3.205': + resolution: {integrity: sha512-CXzySK3PV3EizCRPXnxPqeaAtgrBFDnMFOVpMe36oC3U16yDb1b1tAJGqZi/7uFrVvAiaXvnSFxhUWnDDSaO+A==} cpu: [arm64] os: [linux] libc: [glibc] - '@anthropic-ai/claude-agent-sdk-linux-x64-musl@0.3.170': - resolution: {integrity: sha512-m4+I0qBEk7cxRKS+pL+eoWXbXTFOAo83fQ0tQvap4z/mDMm06IWJtEPoYTaMBwsp32GJWLkHWKbZSBCHZnp2DQ==} + '@anthropic-ai/claude-agent-sdk-linux-x64-musl@0.3.205': + resolution: {integrity: sha512-vvsb7GlnA8CTSVvvTkrXjcSeRKqxSM7p/tU3Od9ICAZeWHglptekEyzLEApzLuLbI5ewfFF/F0q3NwOBbo18dg==} cpu: [x64] os: [linux] libc: [musl] - '@anthropic-ai/claude-agent-sdk-linux-x64@0.3.170': - resolution: {integrity: sha512-Xl/m7TaSC3T5IDBdHrZQ9fCQYyDmPELN34CL+MoyPIf7uSmuZnjE9fUOqDh2Rv26JxWssi1M6X+BBvVuKd6Cpg==} + '@anthropic-ai/claude-agent-sdk-linux-x64@0.3.205': + resolution: {integrity: sha512-siS+1iNqBSlGFZZvJY6+mhzZ/6/ec/TbX9GMuwmTF0E6fxGhIIp797jJxR1q8r6FAq7d39mEoRNhC0Ffo60uNQ==} cpu: [x64] os: [linux] libc: [glibc] - '@anthropic-ai/claude-agent-sdk-win32-arm64@0.3.170': - resolution: {integrity: sha512-IG+8isJNNJKbnnhO7m+PGhfVCg+XoQ/MDxGde5eigFI0WsEfitjuWSWwx82bT9ghxI1aa6qNvI+UPgPcZuo5Fg==} + '@anthropic-ai/claude-agent-sdk-win32-arm64@0.3.205': + resolution: {integrity: sha512-SpP5zF68weFez/6pKrGzq/UVAJDMDNphWqmkLfOpWTDBL5xy6XlIZw5Bl4EXoVnfi2VLFkwuffNeFe+9SdX7kw==} cpu: [arm64] os: [win32] - '@anthropic-ai/claude-agent-sdk-win32-x64@0.3.170': - resolution: {integrity: sha512-7cuqSKbHVItPGVwRbd3A0BEJwcNtc7Fhoh6qHN4C6yrmjSrvdYYx3MLvq/VI768/RoG7mAMDxb+j7WfEfoP9BA==} + '@anthropic-ai/claude-agent-sdk-win32-x64@0.3.205': + resolution: {integrity: sha512-kg2kkXyeSoFLruO3Ic2IruLxzBR0xCUtmlJHdWi3SYW7JhAKNJg4fcrdJsWcardmEw23Y2UDGDJbRyxqSVx6wg==} cpu: [x64] os: [win32] - '@anthropic-ai/claude-agent-sdk@0.3.170': - resolution: {integrity: sha512-pAvhfk+iTodXZ6RF18Kz7BEUWFjL7EcR3tKuhUNdPpE1NAYCR3mSHGbafi72JsrNwKEDIs7FU31z3fqhwy8QzA==} + '@anthropic-ai/claude-agent-sdk@0.3.205': + resolution: {integrity: sha512-ft6iBw9kXudsusiXNpeybIPBJ07Z3tqp1ROSg5cEJqgA+9i+JJj2sRfQth+QD+lyenbbAU8yPieLxIimvfBhtw==} engines: {node: '>=18.0.0'} peerDependencies: '@anthropic-ai/sdk': '>=0.93.0' @@ -10405,44 +10405,44 @@ snapshots: '@alchemy.run/node-utils@0.0.4': {} - '@anthropic-ai/claude-agent-sdk-darwin-arm64@0.3.170': + '@anthropic-ai/claude-agent-sdk-darwin-arm64@0.3.205': optional: true - '@anthropic-ai/claude-agent-sdk-darwin-x64@0.3.170': + '@anthropic-ai/claude-agent-sdk-darwin-x64@0.3.205': optional: true - '@anthropic-ai/claude-agent-sdk-linux-arm64-musl@0.3.170': + '@anthropic-ai/claude-agent-sdk-linux-arm64-musl@0.3.205': optional: true - '@anthropic-ai/claude-agent-sdk-linux-arm64@0.3.170': + '@anthropic-ai/claude-agent-sdk-linux-arm64@0.3.205': optional: true - '@anthropic-ai/claude-agent-sdk-linux-x64-musl@0.3.170': + '@anthropic-ai/claude-agent-sdk-linux-x64-musl@0.3.205': optional: true - '@anthropic-ai/claude-agent-sdk-linux-x64@0.3.170': + '@anthropic-ai/claude-agent-sdk-linux-x64@0.3.205': optional: true - '@anthropic-ai/claude-agent-sdk-win32-arm64@0.3.170': + '@anthropic-ai/claude-agent-sdk-win32-arm64@0.3.205': optional: true - '@anthropic-ai/claude-agent-sdk-win32-x64@0.3.170': + '@anthropic-ai/claude-agent-sdk-win32-x64@0.3.205': optional: true - '@anthropic-ai/claude-agent-sdk@0.3.170(@anthropic-ai/sdk@0.93.0(zod@4.4.3))(@modelcontextprotocol/sdk@1.29.0(zod@4.4.3))(zod@4.4.3)': + '@anthropic-ai/claude-agent-sdk@0.3.205(@anthropic-ai/sdk@0.93.0(zod@4.4.3))(@modelcontextprotocol/sdk@1.29.0(zod@4.4.3))(zod@4.4.3)': dependencies: '@anthropic-ai/sdk': 0.93.0(zod@4.4.3) '@modelcontextprotocol/sdk': 1.29.0(zod@4.4.3) zod: 4.4.3 optionalDependencies: - '@anthropic-ai/claude-agent-sdk-darwin-arm64': 0.3.170 - '@anthropic-ai/claude-agent-sdk-darwin-x64': 0.3.170 - '@anthropic-ai/claude-agent-sdk-linux-arm64': 0.3.170 - '@anthropic-ai/claude-agent-sdk-linux-arm64-musl': 0.3.170 - '@anthropic-ai/claude-agent-sdk-linux-x64': 0.3.170 - '@anthropic-ai/claude-agent-sdk-linux-x64-musl': 0.3.170 - '@anthropic-ai/claude-agent-sdk-win32-arm64': 0.3.170 - '@anthropic-ai/claude-agent-sdk-win32-x64': 0.3.170 + '@anthropic-ai/claude-agent-sdk-darwin-arm64': 0.3.205 + '@anthropic-ai/claude-agent-sdk-darwin-x64': 0.3.205 + '@anthropic-ai/claude-agent-sdk-linux-arm64': 0.3.205 + '@anthropic-ai/claude-agent-sdk-linux-arm64-musl': 0.3.205 + '@anthropic-ai/claude-agent-sdk-linux-x64': 0.3.205 + '@anthropic-ai/claude-agent-sdk-linux-x64-musl': 0.3.205 + '@anthropic-ai/claude-agent-sdk-win32-arm64': 0.3.205 + '@anthropic-ai/claude-agent-sdk-win32-x64': 0.3.205 '@anthropic-ai/sdk@0.93.0(zod@4.4.3)': dependencies: