diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 38253cd82f1d1..e3d56367459ea 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -1414,8 +1414,12 @@ describe("ClaudeAdapterV2 resume compaction", () => { const canUseTool = openedOptions?.canUseTool; assert.isFunction(canUseTool); const longQuestion = `Choose a deployment target: ${"region ".repeat(80)}`; - const questionRequestEvent = yield* runtime.events.pipe( - Stream.filter((event) => event.type === "runtime_request.updated"), + const questionAskedEvent = yield* runtime.events.pipe( + Stream.filter( + (event) => + event.type === "runtime_request.updated" && + event.runtimeRequest.kind === "user_input", + ), Stream.runHead, Effect.forkScoped, ); @@ -1441,8 +1445,8 @@ describe("ClaudeAdapterV2 resume compaction", () => { requestId: "request-question-full-access", }, ), - ).pipe(Effect.forkScoped); - const questionEvent = yield* Fiber.join(questionRequestEvent); + ); + const questionEvent = yield* Fiber.join(questionAskedEvent); assert.isTrue(Option.isSome(questionEvent)); if ( Option.isNone(questionEvent) || @@ -1450,28 +1454,14 @@ describe("ClaudeAdapterV2 resume compaction", () => { ) { return; } - yield* runtime.respondToRuntimeRequest({ - requestId: questionEvent.value.runtimeRequest.id, - answers: { [longQuestion]: ["Production", "Staging"] }, - }); - assert.deepEqual(yield* Fiber.join(questionResult), { - behavior: "allow", - updatedInput: { - questions: [ - { - header: "Target", - question: longQuestion, - options: [ - { label: "Production", description: "Deploy to production." }, - { label: "Staging", description: "Deploy to staging." }, - ], - multiSelect: true, - }, - ], - answers: { [longQuestion]: "Production, Staging" }, - }, - toolUseID: "tool-question-full-access", - }); + const request = questionEvent.value.runtimeRequest; + assert.equal(request.status, "pending"); + assert.deepEqual(request.responseCapability, { type: "message" }); + assert.equal(questionResult?.behavior, "deny"); + assert.include( + questionResult?.behavior === "deny" ? questionResult.message : "", + `request ${request.id}`, + ); const longPlan = `# Deployment plan\n\n${"Validate every region before promotion.\n".repeat(120)}`; const proposedPlanEvent = yield* runtime.events.pipe( Stream.filter((event) => event.type === "plan.updated"), diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index b1273cee72061..ebe41c70a04a6 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -4150,6 +4150,10 @@ export function claudeUserInputQuestions( }); } +function claudeMessageQuestionResult(requestId: string): string { + return `T3 showed these questions to the user without waiting (request ${requestId}; see t3_pending_request). They stay open across turns, and the answers arrive later as one user message. If you cannot continue without them, end your turn now. Otherwise keep working.`; +} + export function claudeSdkUserInputAnswers( answers: ProviderUserInputAnswers, ): NonNullable { @@ -9436,6 +9440,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( toolName: Parameters[0], toolInput: Parameters[1], callbackOptions: Parameters[2], + /** T3's own prompts still wait for the answer inside the turn. */ + liveQuestion = false, ) { const context = yield* Ref.get(activeTurn); if (context === null) { @@ -9486,6 +9492,43 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( } } + if (toolName === "AskUserQuestion" && !liveQuestion) { + // A live native question dies when the turn ends or a subagent message lands, so it + // is answered by message instead and the agent decides whether to stop. + const artifacts = yield* buildApprovalRequestArtifacts({ + context, + nativeItemId: nativeRequestId, + nativeRequestId, + requestKind: "user_input", + questions: claudeUserInputQuestions(nativeToolInput), + }); + yield* Effect.all( + [ + emitProviderEvent({ + type: "node.updated", + driver: CLAUDE_PROVIDER, + node: artifacts.node, + }), + emitProviderEvent({ + type: "runtime_request.updated", + driver: CLAUDE_PROVIDER, + runtimeRequest: { ...artifacts.request, responseCapability: { type: "message" } }, + }), + emitProviderEvent({ + type: "turn_item.updated", + driver: CLAUDE_PROVIDER, + turnItem: artifacts.turnItem, + }), + ], + { concurrency: 1 }, + ); + return { + behavior: "deny", + message: claudeMessageQuestionResult(artifacts.request.id), + toolUseID: callbackOptions.toolUseID, + } satisfies PermissionResult; + } + if (toolName === "AskUserQuestion") { const questions = claudeUserInputQuestions(nativeToolInput); const artifacts = yield* buildApprovalRequestArtifacts({ @@ -9734,6 +9777,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( requestId: callbackOptions.requestId, toolUseID: request.toolUseID ?? callbackOptions.requestId, }, + true, ); if (result.behavior !== "allow") return { behavior: "cancelled" as const }; const answers = diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index d0edd5dacf8f6..a9b63ca9264bc 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -8794,7 +8794,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ); if ( Option.isSome(activeSession) && - activeSession.value.providerSession.capabilities.turns.supportsActiveSteering + activeSession.value.providerSession.capabilities.turns.supportsActiveSteering && + activeSession.value.providerSession.capabilities.turns.activeSteeringInterruptsTools !== + true ) { dispatchMode = { type: "steer_active", targetRunId: activeRun.id }; } diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 65c0c040e0daf..e3efd671e5f73 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -74,6 +74,7 @@ import * as ProjectStore from "./ProjectStore.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as RuntimeLayer from "./runtimeLayer.ts"; import { shellStreamItemFromThreadShell } from "./ShellStream.ts"; +import { ClaudeProviderCapabilitiesV2 } from "./Adapters/ClaudeAdapterV2.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as ThreadManagementService from "./ThreadManagementService.ts"; import * as ManagedProjectFolders from "../project/ManagedProjectFolders.ts"; @@ -907,99 +908,111 @@ it.layer(layerTest)("OrchestrationV2LayerLive", (it) => { }).pipe(Effect.provide(Layer.fresh(layerTest))), ); + const startRunningTurn = Effect.fnUntraced(function* ( + name: string, + capabilities: typeof CodexProviderCapabilitiesV2 | typeof ClaudeProviderCapabilitiesV2, + ) { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const sessions = yield* ProviderSessionManager.ProviderSessionManagerV2; + const threadId = ThreadId.make(name); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${name}-create`), + threadId, + projectId: ProjectId.make(`${name}-project`), + title: "Delivery intent", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: process.cwd(), + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${name}-first`), + threadId, + messageId: MessageId.make(`${name}-first`), + text: "Start work.", + attachments: [], + dispatchMode: { type: "start_immediately" }, + }); + const initial = yield* orchestrator.getThreadProjection(threadId); + const run = initial.runs[0]!; + const providerThread = initial.providerThreads[0]!; + const now = yield* DateTime.now; + const providerSession = { + id: providerThread.providerSessionId!, + driver, + providerInstanceId: modelSelection.instanceId, + status: "running" as const, + cwd: process.cwd(), + model: modelSelection.model, + capabilities, + createdAt: now, + updatedAt: now, + lastError: null, + }; + const providerTurn = { + id: ProviderTurnId.make(`${name}-turn`), + providerThreadId: providerThread.id, + nodeId: run.rootNodeId!, + runAttemptId: run.activeAttemptId, + nativeTurnRef: null, + ordinal: 1, + status: "running" as const, + startedAt: now, + completedAt: null, + }; + yield* eventSink.write({ + commandId: CommandId.make(`${name}-running`), + events: [ + { + id: EventId.make(`${name}-run-event`), + type: "run.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { ...run, status: "running", startedAt: now }, + }, + { + id: EventId.make(`${name}-session-event`), + type: "provider-session.attached", + threadId, + occurredAt: now, + payload: providerSession, + }, + { + id: EventId.make(`${name}-turn-event`), + type: "provider-turn.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: providerTurn, + }, + ], + }); + const sessionSpy = vi.spyOn(sessions, "get").mockReturnValue( + Effect.succeedSome({ + providerSession, + } as ProviderAdapter.ProviderAdapterV2SessionRuntime), + ); + yield* Effect.addFinalizer(() => Effect.sync(() => sessionSpy.mockRestore())); + return { threadId, run, providerThread, providerSession, providerTurn }; + }); + it.effect("resolves delivery intent against the active run and starts after it completes", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const eventSink = yield* EventSink.EventSinkV2; const outbox = yield* EffectOutbox.EffectOutboxV2; - const sessions = yield* ProviderSessionManager.ProviderSessionManagerV2; - const threadId = ThreadId.make("runtime-delivery-intent"); - yield* orchestrator.dispatch({ - type: "thread.create", - createdBy: "user", - creationSource: "web", - commandId: CommandId.make("runtime-delivery-intent-create"), - threadId, - projectId: ProjectId.make("runtime-delivery-intent-project"), - title: "Delivery intent", - modelSelection, - runtimeMode: "full-access", - interactionMode: "default", - branch: null, - worktreePath: process.cwd(), - }); - yield* orchestrator.dispatch({ - type: "message.dispatch", - createdBy: "user", - creationSource: "web", - commandId: CommandId.make("runtime-delivery-intent-first"), - threadId, - messageId: MessageId.make("runtime-delivery-intent-first"), - text: "Start work.", - attachments: [], - dispatchMode: { type: "start_immediately" }, - }); - const initial = yield* orchestrator.getThreadProjection(threadId); - const run = initial.runs[0]!; - const providerThread = initial.providerThreads[0]!; + const { threadId, run, providerThread, providerSession, providerTurn } = + yield* startRunningTurn("runtime-delivery-intent", CodexProviderCapabilitiesV2); const now = yield* DateTime.now; - const providerSession = { - id: providerThread.providerSessionId!, - driver, - providerInstanceId: modelSelection.instanceId, - status: "running" as const, - cwd: process.cwd(), - model: modelSelection.model, - capabilities: CodexProviderCapabilitiesV2, - createdAt: now, - updatedAt: now, - lastError: null, - }; - const providerTurn = { - id: ProviderTurnId.make("runtime-delivery-intent-turn"), - providerThreadId: providerThread.id, - nodeId: run.rootNodeId!, - runAttemptId: run.activeAttemptId, - nativeTurnRef: null, - ordinal: 1, - status: "running" as const, - startedAt: now, - completedAt: null, - }; - yield* eventSink.write({ - commandId: CommandId.make("runtime-delivery-intent-running"), - events: [ - { - id: EventId.make("runtime-delivery-intent-run-event"), - type: "run.updated", - threadId, - runId: run.id, - occurredAt: now, - payload: { ...run, status: "running", startedAt: now }, - }, - { - id: EventId.make("runtime-delivery-intent-session-event"), - type: "provider-session.attached", - threadId, - occurredAt: now, - payload: providerSession, - }, - { - id: EventId.make("runtime-delivery-intent-turn-event"), - type: "provider-turn.updated", - threadId, - runId: run.id, - occurredAt: now, - payload: providerTurn, - }, - ], - }); - const sessionSpy = vi.spyOn(sessions, "get").mockReturnValue( - Effect.succeedSome({ - providerSession, - } as ProviderAdapter.ProviderAdapterV2SessionRuntime), - ); - yield* Effect.addFinalizer(() => Effect.sync(() => sessionSpy.mockRestore())); const steerCommandId = CommandId.make("runtime-delivery-intent-auto"); const steerMessageId = MessageId.make("runtime-delivery-intent-auto"); @@ -1080,6 +1093,34 @@ it.layer(layerTest)("OrchestrationV2LayerLive", (it) => { }), ); + it.effect.each([ + { capabilities: CodexProviderCapabilitiesV2, effects: ["provider-turn.steer"] }, + // Claude's steer aborts running tools and pending approvals, so its answer waits for the turn. + { capabilities: ClaudeProviderCapabilitiesV2, effects: [] }, + ])( + "delivers an async answer into a running turn only when steering is safe", + ({ capabilities, effects }) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const outbox = yield* EffectOutbox.EffectOutboxV2; + const name = `runtime-async-answer-${effects.length}`; + const { threadId, run } = yield* startRunningTurn(name, capabilities); + const { requestId } = yield* seedPendingQuestion(threadId, run.id, { type: "message" }); + const commandId = CommandId.make(`${name}-answer`); + yield* orchestrator.dispatch({ + type: "runtime-request.respond", + commandId, + threadId, + requestId, + answers: { color: "Blue", sizes: ["Small"] }, + }); + assert.deepEqual( + (yield* outbox.listByCommandId(commandId)).map((effect) => effect.request.type), + effects, + ); + }), + ); + const seedPendingQuestion = Effect.fnUntraced(function* ( threadId: ThreadId, runId: RunId | null, diff --git a/docs/internals/providers.md b/docs/internals/providers.md index 42f0a70101b3a..41dc3fef94505 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -110,6 +110,13 @@ no pending RPC response to send. The Their execution nodes do not block the run. Web, desktop, and mobile use their normal question panels, and requests remain pending after a turn finishes, a provider exits, or the server restarts. +Claude's `AskUserQuestion` takes the same path from the start, because any steer, including another +agent's message, aborts a live permission callback. The +[adapter](../../apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts) records a message +request and denies the tool with its id at once, telling Claude to end its turn only if it is +blocked. The internal resume-compaction question stays live. Claude's steers also abort running +tools and approvals, so an answer to a running Claude turn queues behind it instead of steering. + `runtime-request.respond` reads the persisted request and question item, validates required answers, and commits the resolution and a user message in one transaction. Repeating the same command returns its receipt without posting the answer twice. The normal message path starts or diff --git a/docs/user/providers-claude.md b/docs/user/providers-claude.md index 206d82ce32333..7b4a926a8df22 100644 --- a/docs/user/providers-claude.md +++ b/docs/user/providers-claude.md @@ -87,6 +87,15 @@ can still be started by you. Invoke those one per message: Claude directly runs only the last named skill and may try to start earlier ones through its Skill tool, which refuses skills reserved for manual invocation. +## Questions + +When Claude asks you something, it keeps working instead of waiting. Answer in +the thread's question panel. The answer becomes a new message: it waits for +Claude's current turn to end, or starts another turn if Claude has finished. A +message you or another agent send meanwhile no longer cancels the question, and +unanswered questions survive reconnects. Dismiss a question you do not want to +answer; nothing is sent to Claude. + ## Task lists To show Claude task lists in the web and desktop apps, add