diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 1add2822a01e9..7b5e61e763a72 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -11828,6 +11828,361 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + const makeIdlePermissionHarness = Effect.fnUntraced(function* ( + runtimeMode: "approval-required" | "full-access" = "approval-required", + options?: Parameters[0], + ) { + const harness = yield* makeWakeHarnessWithOptions(options); + let ordinal = 0; + let sequence = 0; + const uuid = () => `idle-permission-${++sequence}`; + const start = () => + harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: DateTime.makeUnsafe(0), + attemptId: RunAttemptId.make(`idle-permission-${++ordinal}`), + providerTurnOrdinal: ordinal, + text: "Run the auditor.", + attachments: [], + runtimePolicy: ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode, + interactionMode: "default", + cwd: "/workspace", + }), + }), + ); + const offer = (message: SDKMessage) => Queue.offer(harness.processQueues.at(-1)!, message); + const taskStarted = (toolUseId = "launch", background = true, taskId = "auditor") => + claudeSdkFrame({ + ...makeSubagentTaskStartedFrame({ taskId, toolUseId, uuid: uuid() }), + is_backgrounded: background, + }); + const settle = Effect.fnUntraced(function* () { + yield* offer(makeResultFrame({ uuid: uuid(), result: "Returned to the user." })); + return yield* Queue.take(harness.terminalReceipts); + }); + const callback = () => harness.getOpenedOptions()!.canUseTool!; + const request = ( + agentID: string | undefined, + signal = new AbortController().signal, + canUseTool = callback(), + ) => + Effect.promise(() => + canUseTool( + "Read", + { file_path: "/etc/hosts" }, + { + signal, + toolUseID: uuid(), + requestId: uuid(), + ...(agentID === undefined ? {} : { agentID }), + }, + ), + ); + const approve = Effect.fnUntraced(function* ( + agentID: string | undefined, + decision: "accept" | "decline", + providerTurnId = harness.terminalEvents().at(-1)!.providerTurnId, + ) { + const permission = yield* request(agentID).pipe(Effect.forkScoped); + const { runtimeRequest } = yield* Effect.raceFirst( + Queue.take(harness.runtimeRequestReceipts), + Fiber.join(permission).pipe( + Effect.flatMap((result) => + Effect.die(`Permission returned before approval: ${result?.behavior}`), + ), + ), + ); + assert.equal(runtimeRequest.providerTurnId, providerTurnId); + const node = harness.events.findLast( + (event) => event.type === "node.updated" && event.node.id === runtimeRequest.nodeId, + ); + assert.equal(node?.type, "node.updated"); + if (node?.type !== "node.updated") return; + assert.strictEqual(node.node.parentNodeId, node.node.rootNodeId); + assert.isFalse( + harness.events.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "dynamic_tool" && + event.turnItem.nativeItemRef?.nativeId === runtimeRequest.nativeRequestRef?.nativeId, + ), + ); + yield* harness.runtime.respondToRuntimeRequest({ requestId: runtimeRequest.id, decision }); + assert.equal( + (yield* Fiber.join(permission))?.behavior, + decision === "accept" ? "allow" : "deny", + ); + }); + return { ...harness, start, offer, taskStarted, settle, callback, request, approve }; + }); + + it.effect("asks for settled subagent approval with or without an agent ID", () => + Effect.gen(function* () { + const harness = yield* makeIdlePermissionHarness(); + yield* harness.start(); + yield* harness.offer(harness.taskStarted()); + yield* harness.settle(); + yield* harness.approve("auditor", "accept"); + yield* harness.approve(undefined, "decline"); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + + it.effect( + "limits full-access settled permissions to an eligible subagent of a completed run", + () => + Effect.gen(function* () { + const harness = yield* makeIdlePermissionHarness("full-access"); + yield* harness.start(); + yield* harness.settle(); + assert.equal((yield* harness.request(undefined))?.behavior, "deny"); + yield* harness.start(); + yield* harness.offer(harness.taskStarted()); + const launchTurn = (yield* harness.settle()).providerTurnId; + // Under bypass, the CLI forwards only requests that need a person's decision. + yield* harness.approve("auditor", "accept"); + assert.equal((yield* harness.request("another-agent"))?.behavior, "deny"); + assert.equal((yield* harness.request("auditor", AbortSignal.abort()))?.behavior, "deny"); + yield* harness.start(); + yield* harness.settle(); + yield* harness.approve("auditor", "accept", launchTurn); + assert.equal((yield* harness.request("another-agent"))?.behavior, "deny"); + assert.equal( + new Set( + harness.events.flatMap((event) => + event.type === "runtime_request.updated" ? [event.runtimeRequest.id] : [], + ), + ).size, + 2, + ); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + + it.effect("asks for a subagent of an earlier run after a wake continuation settles", () => + Effect.gen(function* () { + const harness = yield* makeIdlePermissionHarness(); + yield* harness.start(); + yield* harness.offer(harness.taskStarted("launch-auditor", true, "auditor")); + yield* harness.offer(harness.taskStarted("launch-reviewer", true, "reviewer")); + const launchTurn = (yield* harness.settle()).providerTurnId; + yield* harness.offer( + makeSubagentNotificationFrame({ + taskId: "auditor", + toolUseId: "launch-auditor", + summary: "Audit done.", + uuid: "auditor-done", + }), + ); + yield* harness.offer(wakeTurnInit); + yield* harness.awaitContinuation(1); + yield* harness.start(); + assert.notEqual((yield* harness.settle()).providerTurnId, launchTurn); + yield* harness.approve("reviewer", "accept", launchTurn); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + + it.effect("asks for a subagent of an earlier run after a later run fails", () => + Effect.gen(function* () { + const harness = yield* makeIdlePermissionHarness(); + yield* harness.start(); + yield* harness.offer(harness.taskStarted()); + const launchTurn = (yield* harness.settle()).providerTurnId; + yield* harness.start(); + yield* harness.offer( + makeResultFrame({ + uuid: "failed-turn", + result: "API Error: 500 Internal server error.", + isError: true, + apiErrorStatus: 500, + terminalReason: "api_error", + }), + ); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "failed"); + yield* harness.approve("auditor", "accept", launchTurn); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + + it.effect("closes a settled subagent approval when its subagent is stopped", () => + Effect.gen(function* () { + const harness = yield* makeIdlePermissionHarness(); + yield* harness.start(); + yield* harness.offer(harness.taskStarted()); + yield* harness.settle(); + const permission = yield* harness.request("auditor").pipe(Effect.forkScoped); + const { runtimeRequest } = yield* Queue.take(harness.runtimeRequestReceipts); + yield* harness.offer( + claudeSdkFrame({ + type: "system", + subtype: "task_notification", + task_id: "auditor", + tool_use_id: "launch", + status: "stopped", + output_file: "/tmp/auditor.output", + summary: "Stopped.", + uuid: "auditor-stopped", + session_id: WAKE_NATIVE_SESSION, + }), + ); + assert.equal((yield* Fiber.join(permission))?.behavior, "deny"); + const cancelled = (yield* Queue.take(harness.runtimeRequestReceipts)).runtimeRequest; + assert.equal(cancelled.id, runtimeRequest.id); + assert.equal(cancelled.status, "cancelled"); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + + it.effect("denies settled subagent callbacks while stopping and after process replacement", () => + Effect.gen(function* () { + const closeStarted = yield* Deferred.make(); + const finishClose = yield* Deferred.make(); + const harness = yield* makeIdlePermissionHarness("full-access", { + freshQueueOnReopen: true, + close: (messages) => + Deferred.succeed(closeStarted, undefined).pipe( + Effect.andThen(Deferred.await(finishClose)), + Effect.andThen(Queue.shutdown(messages)), + ), + }); + yield* harness.start(); + yield* harness.offer(harness.taskStarted()); + const terminal = yield* harness.settle(); + const oldCallback = harness.callback(); + const stop = yield* harness.runtime + .interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: terminal.providerTurnId, + requestRuntimeRestart: true, + }) + .pipe(Effect.forkScoped); + yield* Deferred.await(closeStarted); + assert.equal((yield* harness.request("auditor"))?.behavior, "deny"); + yield* Deferred.succeed(finishClose, undefined); + yield* Fiber.join(stop); + yield* harness.start(); + yield* harness.offer(harness.taskStarted("replacement-launch")); + yield* harness.settle(); + yield* harness.approve("auditor", "accept"); + assert.equal((yield* harness.request("auditor", undefined, oldCallback))?.behavior, "deny"); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + + it.effect("approves a completed foreground subagent resumed in the latest settled run", () => + Effect.gen(function* () { + const harness = yield* makeIdlePermissionHarness(); + yield* harness.start(); + yield* harness.offer(harness.taskStarted("launch", false)); + yield* harness.offer( + makeSubagentNotificationFrame({ + taskId: "auditor", + toolUseId: "launch", + summary: "First answer.", + uuid: "initial-answer", + }), + ); + yield* harness.settle(); + yield* harness.start(); + yield* harness.offer( + claudeSdkFrame({ + type: "assistant", + message: { + id: "resume-auditor", + type: "message", + role: "assistant", + model: "claude-sonnet-4-6", + content: [ + { + type: "tool_use", + id: "resume", + name: "SendMessage", + input: { + to: "auditor", + message: "Continue the audit.", + }, + }, + ], + }, + parent_tool_use_id: null, + uuid: "resume-message", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* harness.offer(harness.taskStarted("resume")); + yield* harness.settle(); + yield* harness.approve("auditor", "accept"); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + it.effect("answers an approval a held wake turn raises without waiting for the echo", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 7b63b7efc0b65..59eb4733a7370 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -5251,6 +5251,11 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const subagentToolCalls = new Map(); const pendingSubagentLaunchesByToolUseId = new Map(); const settledTurnByNativeThread = new Map(); + // A running subagent stays with the run that launched it, and that run keeps ingesting + // its events after a later run settles, so its approvals need that run's context. + const settledTurnsByRunId = new Map(); + // No run is left to interrupt after settle, so a settled request closes when its subagent ends. + const settledRequestStopsByTaskId = new Map>>(); const subagentStopOutcomes = new Map< string, Deferred.Deferred @@ -6964,6 +6969,11 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( return [true, new Map(current).set(input.taskId, subagent)] as const; }); if (!accepted) return; + if (input.status !== "running") { + for (const stopped of settledRequestStopsByTaskId.get(input.taskId) ?? []) { + yield* Deferred.succeed(stopped, undefined); + } + } if (input.nativeOutcome !== undefined) appliedSubagentOutcomes.add(input.nativeOutcome); if (existingSubagent !== undefined && input.status !== "running") { const openCalls = [...subagentToolCalls.values()].filter( @@ -8259,8 +8269,22 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( yield* Ref.update(activeTurn, (current) => current?.providerTurnId === input.context.providerTurnId ? null : current, ); + const runningSubagentRunIds = new Set( + [...(yield* Ref.get(sessionSubagentsByTaskId)).values()].flatMap((subagent) => + subagent.task.status === "running" ? [subagent.task.runId] : [], + ), + ); + for (const [runId, settled] of settledTurnsByRunId) { + if ( + settled.nativeThreadId === input.context.nativeThreadId && + (runId === input.context.input.runId || !runningSubagentRunIds.has(runId)) + ) { + settledTurnsByRunId.delete(runId); + } + } if (input.status === "completed") { settledTurnByNativeThread.set(input.context.nativeThreadId, input.context); + settledTurnsByRunId.set(input.context.input.runId, input.context); } else { settledTurnByNativeThread.delete(input.context.nativeThreadId); } @@ -10339,9 +10363,38 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( toolName: Parameters[0], toolInput: Parameters[1], callbackOptions: Parameters[2], - options?: { readonly startToolCall?: boolean }, + options?: { + readonly startToolCall?: boolean; + readonly owningQueryClosed?: Deferred.Deferred; + }, ) { - const context = yield* Ref.get(activeTurn); + let context = yield* Ref.get(activeTurn); + const settled = context === null; + let ownerTaskId: string | undefined; + if (settled && !callbackOptions.signal.aborted) { + const live = yield* Ref.get(queryContext); + if (live !== null && !live.stopping && live.closed === options?.owningQueryClosed) { + const owners = [...(yield* Ref.get(sessionSubagentsByTaskId))].flatMap( + ([taskId, subagent]) => { + const owner = + subagent.task.runId === null + ? undefined + : settledTurnsByRunId.get(subagent.task.runId); + return (callbackOptions.agentID === undefined || + callbackOptions.agentID === taskId) && + subagent.task.status === "running" && + !live.subagentsFromEarlierProcesses.has(subagent) && + owner?.nativeThreadId === live.nativeThreadId + ? [{ taskId, owner }] + : []; + }, + ); + const latest = settledTurnByNativeThread.get(live.nativeThreadId); + const selected = owners.find(({ owner }) => owner === latest) ?? owners[0]; + context = selected?.owner ?? null; + ownerTaskId = selected?.taskId; + } + } if (context === null) { return { behavior: "deny", @@ -10350,10 +10403,35 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( } satisfies PermissionResult; } + const ownerStopped = yield* Deferred.make(); + const untilOwnerStops = (wait: Effect.Effect, onStop: A) => + ownerTaskId === undefined + ? wait + : Effect.acquireUseRelease( + Effect.sync(() => { + const stops = settledRequestStopsByTaskId.get(ownerTaskId) ?? new Set(); + settledRequestStopsByTaskId.set(ownerTaskId, stops.add(ownerStopped)); + return stops; + }), + () => + Effect.raceFirst(wait, Deferred.await(ownerStopped).pipe(Effect.as(onStop))), + (stops) => + Effect.sync(() => { + stops.delete(ownerStopped); + if (stops.size === 0) settledRequestStopsByTaskId.delete(ownerTaskId); + }), + ); const nativeRequestId = callbackOptions.toolUseID; const nativeToolInput = claudeNativeToolInputFromRecord(toolInput); const heldForEcho = context.heldRootFrames.length > 0; - const startToolCall = toolName !== "Agent" && options?.startToolCall !== false; + const startToolCall = + toolName !== "Agent" && options?.startToolCall !== false && !settled; + // Control requests and settled-path subagent requests have no tool node to hang under. + const parentNodeId = + options?.startToolCall === false || + (settled && findToolCall(context, nativeRequestId) === undefined) + ? context.input.rootNodeId + : undefined; if (toolName === "Agent") { rememberClaudeSubagentLaunch( modelCatalog, @@ -10397,10 +10475,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( context, nativeItemId: nativeRequestId, nativeRequestId, - // No tool node exists for a control request to hang under. - ...(options?.startToolCall === false - ? { parentNodeId: context.input.rootNodeId } - : {}), + ...(parentNodeId === undefined ? {} : { parentNodeId }), requestKind: "user_input", questions, }); @@ -10433,9 +10508,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ], { concurrency: 1 }, ); - const resolvedAnswers = yield* awaitClaudeUserInputAnswers( - answers, - callbackOptions.signal, + const resolvedAnswers = yield* untilOwnerStops( + awaitClaudeUserInputAnswers(answers, callbackOptions.signal), + {}, ).pipe( Effect.ensuring( Ref.update(pendingRuntimeRequests, (current) => { @@ -10445,8 +10520,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }), ), ); - if (!(yield* Deferred.isDone(answers))) yield* cancelWithdrawnRequest(artifacts); - return callbackOptions.signal.aborted + const withdrawn = !(yield* Deferred.isDone(answers)); + if (withdrawn) yield* cancelWithdrawnRequest(artifacts); + return withdrawn ? ({ behavior: "deny", message: "User cancelled tool execution.", @@ -10537,6 +10613,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( nativeRequestId, requestKind, prompt, + ...(parentNodeId === undefined ? {} : { parentNodeId }), presentation: claudeApprovalPresentation({ permissionMode: context.queryPolicy.permissionMode, prompt, @@ -10574,9 +10651,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( { concurrency: 1 }, ); - const resolvedDecision = yield* awaitClaudeApprovalDecision( - decision, - callbackOptions.signal, + const resolvedDecision = yield* untilOwnerStops( + awaitClaudeApprovalDecision(decision, callbackOptions.signal), + { decision: "cancel" }, ).pipe( Effect.ensuring( Ref.update(pendingRuntimeRequests, (current) => { @@ -10600,9 +10677,6 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }); }); - const canUseTool: CanUseTool = (toolName, toolInput, callbackOptions) => - runPromise(canUseToolEffect(toolName, toolInput, callbackOptions)); - const onElicitationEffect = Effect.fn("ClaudeAdapterV2.onElicitation")(function* ( request: ElicitationRequest, callbackOptions: Parameters>[1], @@ -10981,6 +11055,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( turnInput.nativeThreadHasTurns ?? turnInput.providerTurnOrdinal > 1; const shouldResume = resumeSessionAt !== undefined || openedWithResume || hasPersistedProviderTurn; + const closed = yield* Deferred.make(); const toolSearchOverride = yield* resolveToolSearchOverride( turnInput.runtimePolicy.cwd, launchArgSettings, @@ -11022,7 +11097,12 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( : { allowDangerouslySkipPermissions: queryPolicy.allowDangerouslySkipPermissions, }), - canUseTool, + canUseTool: (toolName, toolInput, callbackOptions) => + runPromise( + canUseToolEffect(toolName, toolInput, callbackOptions, { + owningQueryClosed: closed, + }), + ), onUserDialog, onElicitation, supportedDialogKinds: ["resume_return"], @@ -11062,7 +11142,6 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( yield* resetBackgroundTaskStateForNativeThreadProcess(nativeThreadId, { status: "active", }); - const closed = yield* Deferred.make(); const context: ClaudeLiveQueryContext = { queryPolicy, nativeThreadId,