diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 54e96ffa084..a97b5b8d107 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -209,6 +209,21 @@ describe("ClaudeAdapterV2 runtime query policy", () => { installPermissionCallback: false, }); }); + + it("installs the permission callback in plan mode so ExitPlanMode can propose plans", () => { + const queryPolicy = claudeRuntimeQueryPolicyForRuntimePolicy( + ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "plan", + cwd: "/workspace", + }), + ); + + assert.deepEqual(queryPolicy, { + permissionMode: "plan", + installPermissionCallback: true, + }); + }); }); describe("ClaudeAdapterV2 native protocol logging", () => { diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 2740ceff6d2..4ada4acbc46 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -25,6 +25,7 @@ import { type ModelSelection, type OrchestrationV2ConversationMessage, type OrchestrationV2ExecutionNode, + type OrchestrationV2PlanArtifact, type OrchestrationV2ProviderCapabilities, type OrchestrationV2ProviderFailure, type OrchestrationV2ProviderSession, @@ -166,7 +167,7 @@ export const ClaudeProviderCapabilitiesV2 = { planning: { emitsPlanUpdated: false, emitsTodoList: false, - emitsProposedPlan: false, + emitsProposedPlan: true, supportsStructuredQuestions: false, planDeltasHaveItemIds: false, }, @@ -275,10 +276,25 @@ export class ClaudeAgentSdkQueryRunnerError extends Schema.TaggedErrorClass; readonly open: ( @@ -1138,11 +1154,14 @@ export function claudeRuntimeQueryPolicyForRuntimePolicy( : undefined; if (permissionMode === "plan") { + // The permission callback must be installed in plan mode: without it the + // SDK has no way to raise ExitPlanMode, so the model can never propose a + // plan and the plan interaction mode dead-ends in prose. return { permissionMode, ...(readOnlyTools === undefined ? {} : { tools: readOnlyTools }), ...(allowedTools === undefined ? {} : { allowedTools }), - installPermissionCallback: false, + installPermissionCallback: true, }; } @@ -1161,9 +1180,6 @@ export function claudeRuntimeQueryPolicyForRuntimePolicy( } function shouldInstallClaudePermissionCallback(policy: ClaudeRuntimeQueryPolicy): boolean { - if (policy.permissionMode === "plan") { - return false; - } return policy.installPermissionCallback; } @@ -1794,6 +1810,9 @@ interface ClaudeLiveQueryContext { readonly queryPolicyKey: string; readonly selectionKey: string; readonly closed: Deferred.Deferred; + readonly openedWithResumeSessionAt: boolean; + /** True once any SDK message arrived on this query. */ + receivedSdkMessage: boolean; } interface ActiveClaudeToolCall { @@ -1848,6 +1867,26 @@ export function makeClaudeAdapterV2( }); const events = yield* Queue.unbounded(); const activeTurn = yield* Ref.make(null); + /** + * Async native tasks (Task tool with isAsync) usually complete after + * their parent turn has finalized and activeTurn is null. Their turn + * contexts are retained here, keyed by native task id, so the late + * task_notification can still resolve the subagent under the original + * run — which RunExecutionService keeps draining until every child + * subagent is terminal. Entries are removed when the notification + * arrives. + */ + const pendingAsyncTaskContexts = yield* Ref.make( + new Map(), + ); + /** + * Native threads whose next open must skip the resumeSessionAt pin. + * A dirty shutdown can record a conversation head that the killed + * claude process never flushed to its transcript; resuming at that + * head fails the first query. Entries are one-shot: consumed by the + * next openQuery and re-added if the unpinned open fails again. + */ + const suppressResumeSessionAt = yield* Ref.make(new Set()); const interruptedTurns = yield* Ref.make(new Set()); const steeredTurns = yield* Ref.make(new Set()); const queryContext = yield* Ref.make(null); @@ -2665,6 +2704,16 @@ export function makeClaudeAdapterV2( ], { concurrency: 1 }, ); + yield* Ref.update(pendingAsyncTaskContexts, (current) => { + let updated: Map | null = null; + for (const [taskId, subagent] of input.context.subagentsByTaskId) { + if (subagent.task.status === "running") { + updated ??= new Map(current); + updated.set(taskId, input.context); + } + } + return updated ?? current; + }); yield* Ref.update(activeTurn, (current) => current?.providerTurnId === input.context.providerTurnId ? null : current, ); @@ -2721,6 +2770,24 @@ export function makeClaudeAdapterV2( const finalizeActiveTurnAfterQueryExit = Effect.fnUntraced(function* ( cause?: Cause.Cause, ) { + if (cause !== undefined) { + const failedQuery = yield* Ref.get(queryContext); + if ( + failedQuery !== null && + failedQuery.openedWithResumeSessionAt && + !failedQuery.receivedSdkMessage + ) { + yield* Ref.update(suppressResumeSessionAt, (current) => { + const updated = new Set(current); + updated.add(failedQuery.nativeThreadId); + return updated; + }); + yield* Effect.logWarning("orchestration-v2.claude-resume-pin-suppressed", { + providerSessionId: input.providerSessionId, + nativeThreadId: failedQuery.nativeThreadId, + }); + } + } const context = yield* Ref.get(activeTurn); if (context === null) { return; @@ -2735,7 +2802,10 @@ export function makeClaudeAdapterV2( ? {} : { failure: makeProviderFailure({ - cause: cause === undefined ? undefined : Cause.squash(cause), + cause: + cause === undefined + ? undefined + : claudeProviderFailureCause(Cause.squash(cause)), class: "transport_error", }), }), @@ -2750,7 +2820,7 @@ export function makeClaudeAdapterV2( providerSessionId: input.providerSessionId, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, - cause, + cause: Cause.pretty(cause), }); } }); @@ -2763,8 +2833,45 @@ export function makeClaudeAdapterV2( if (liveQuery?.query !== input.query) { return; } + liveQuery.receivedSdkMessage = true; const message = input.message; + if (message.type === "system" && message.subtype === "task_notification") { + // Async task completions typically land after the parent turn has + // finalized (activeTurn is null) or while an unrelated turn is + // active; resolve them against the retained owning context so the + // subagent terminalizes under its original run. + const active = yield* Ref.get(activeTurn); + const retired = (yield* Ref.get(pendingAsyncTaskContexts)).get(message.task_id); + const taskContext = + active !== null && active.subagentsByTaskId.has(message.task_id) + ? active + : (retired ?? active); + if (taskContext !== null && !taskContext.ignoredTaskIds.has(message.task_id)) { + yield* updateClaudeSubagentNode({ + context: taskContext, + taskId: message.task_id, + ...(message.tool_use_id === undefined ? {} : { toolUseId: message.tool_use_id }), + result: message.summary, + status: + message.status === "completed" + ? "completed" + : message.status === "stopped" + ? "cancelled" + : "failed", + }); + } + yield* Ref.update(pendingAsyncTaskContexts, (current) => { + if (!current.has(message.task_id)) { + return current; + } + const updated = new Map(current); + updated.delete(message.task_id); + return updated; + }); + return; + } + const context = yield* Ref.get(activeTurn); if (context === null) { return; @@ -2802,23 +2909,6 @@ export function makeClaudeAdapterV2( } } - if (message.type === "system" && message.subtype === "task_notification") { - if (!context.ignoredTaskIds.has(message.task_id)) { - yield* updateClaudeSubagentNode({ - context, - taskId: message.task_id, - ...(message.tool_use_id === undefined ? {} : { toolUseId: message.tool_use_id }), - result: message.summary, - status: - message.status === "completed" - ? "completed" - : message.status === "stopped" - ? "cancelled" - : "failed", - }); - } - } - for (const toolUse of claudeToolUseBlocksFromAssistantMessage(message)) { if (toolUse.name === "Agent") { continue; @@ -2926,6 +3016,92 @@ export function makeClaudeAdapterV2( } }); + const emitProposedPlanArtifacts = Effect.fnUntraced(function* (input: { + readonly context: ActiveClaudeTurnContext; + readonly nativeItemId: string; + readonly markdown: string; + }) { + const now = yield* DateTime.now; + const nativeItemId = `plan:${input.nativeItemId}`; + const planId = yield* idAllocator.allocate.plan({ + threadId: input.context.input.threadId, + ...(input.context.input.runId === null ? {} : { runId: input.context.input.runId }), + driver: CLAUDE_PROVIDER, + }); + const ordinal = yield* resolveItemOrdinal(input.context, nativeItemId); + const nodeId = idAllocator.derive.nodeFromProviderItem({ + driver: CLAUDE_PROVIDER, + nativeItemId, + }); + const nativeItemRef = { + driver: CLAUDE_PROVIDER, + nativeId: input.nativeItemId, + strength: "strong" as const, + }; + const plan: OrchestrationV2PlanArtifact = { + id: planId, + threadId: input.context.input.threadId, + runId: input.context.input.runId, + nodeId, + kind: "proposed_plan", + status: "active", + markdown: input.markdown, + }; + yield* emitProviderEvent({ + type: "node.updated", + 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, + kind: "plan", + status: "completed", + countsForRun: false, + providerThreadId: input.context.input.providerThread.id, + providerTurnId: input.context.providerTurnId, + nativeItemRef, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: now, + }, + }); + yield* emitProviderEvent({ + type: "plan.updated", + driver: CLAUDE_PROVIDER, + plan, + }); + yield* emitProviderEvent({ + type: "turn_item.updated", + driver: CLAUDE_PROVIDER, + turnItem: { + id: idAllocator.derive.turnItemFromProviderItem({ + driver: CLAUDE_PROVIDER, + nativeItemId, + }), + threadId: input.context.input.threadId, + runId: input.context.input.runId, + nodeId, + providerThreadId: input.context.input.providerThread.id, + providerTurnId: input.context.providerTurnId, + nativeItemRef, + parentItemId: null, + ordinal, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "proposed_plan", + planId, + markdown: input.markdown, + streaming: false, + }, + }); + }); + const canUseToolEffect = Effect.fn("ClaudeAdapterV2.canUseTool")(function* ( toolName: Parameters[0], toolInput: Parameters[1], @@ -2940,6 +3116,25 @@ export function makeClaudeAdapterV2( } satisfies PermissionResult; } + if (toolName === "ExitPlanMode" || toolName === "exit_plan_mode") { + // Plan proposals are approved through the app's plan card, not the + // SDK permission flow: capture the plan as a v2 artifact so the + // thread flips to Plan Ready, and keep the session in plan mode. + const planMarkdown = toolInput["plan"]; + yield* emitProposedPlanArtifacts({ + context, + nativeItemId: callbackOptions.toolUseID, + markdown: typeof planMarkdown === "string" ? planMarkdown : "", + }); + return { + behavior: "deny", + message: + "The proposed plan has been shared with the user for review in the app. " + + "End the turn now; do not start implementing until the user approves the plan.", + toolUseID: callbackOptions.toolUseID, + } satisfies PermissionResult; + } + const nativeRequestId = callbackOptions.toolUseID; const nativeToolInput = claudeNativeToolInputFromRecord(toolInput); if (toolName !== "Agent") { @@ -3061,7 +3256,17 @@ export function makeClaudeAdapterV2( updated.add(nativeThreadId); return [false, updated]; }); - const shouldResume = resumeSessionAt !== undefined || openedWithResume; + const pinSuppressed = yield* Ref.modify(suppressResumeSessionAt, (current) => { + if (!current.has(nativeThreadId)) { + return [false, current] as const; + } + const updated = new Set(current); + updated.delete(nativeThreadId); + return [true, updated] as const; + }); + const effectiveResumeSessionAt = pinSuppressed ? undefined : resumeSessionAt; + const shouldResume = + effectiveResumeSessionAt !== undefined || openedWithResume || pinSuppressed; const querySession = yield* queryRunner.open({ threadId: turnInput.threadId, providerSessionId: input.providerSessionId, @@ -3069,7 +3274,9 @@ export function makeClaudeAdapterV2( modelSelection: turnInput.modelSelection, nativeThreadId, resume: shouldResume, - ...(resumeSessionAt === undefined ? {} : { resumeSessionAt }), + ...(effectiveResumeSessionAt === undefined + ? {} + : { resumeSessionAt: effectiveResumeSessionAt }), cwd: turnInput.runtimePolicy.cwd, settings: adapterOptions.settings, environment: adapterOptions.environment, @@ -3089,6 +3296,8 @@ export function makeClaudeAdapterV2( queryPolicyKey, selectionKey: compiledSelection.queryIdentity, closed, + openedWithResumeSessionAt: effectiveResumeSessionAt !== undefined, + receivedSdkMessage: false, }; yield* Ref.set(queryContext, context); yield* querySession.messages.pipe( diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index a588bb25bcd..9e55693dba8 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -32,7 +32,7 @@ import { CheckpointServiceV2 } from "./CheckpointService.ts"; import { EventSinkV2 } from "./EventSink.ts"; import { layer as idAllocatorLayer } from "./IdAllocator.ts"; import type { ProviderAdapterV2Event, ProviderAdapterV2SessionRuntime } from "./ProviderAdapter.ts"; -import { ProviderEventIngestorV2 } from "./ProviderEventIngestor.ts"; +import { ProviderEventIngestorV2, ProviderEventNormalizeError } from "./ProviderEventIngestor.ts"; import { finalProviderThreadStatus, layer as runExecutionServiceLayer, @@ -459,3 +459,167 @@ it.effect("keeps ingesting owned child events after the root turn terminalizes", assert.deepEqual(yield* Ref.get(order), ["root-finalized", "child-message"]); }), ); + +it.effect("drops a poison event whose ingestion fails and keeps ingesting later events", () => + Effect.gen(function* () { + const threadId = ThreadId.make("thread:run-execution-poison"); + const runId = RunId.make("run:run-execution-poison"); + const attemptId = RunAttemptId.make("attempt:run-execution-poison"); + const providerInstanceId = ProviderInstanceId.make("codex"); + const providerSessionId = ProviderSessionId.make("session:run-execution-poison"); + const providerThreadId = ProviderThreadId.make("provider-thread:run-execution-poison"); + const rootProviderTurnId = ProviderTurnId.make("provider-turn:run-execution-poison"); + const rootNodeId = NodeId.make("node:run-execution-poison"); + const poisonMessageId = MessageId.make("message:run-execution-poison:poison"); + const goodMessageId = MessageId.make("message:run-execution-poison:good"); + const goodMessageIngested = yield* Deferred.make(); + const rootFinalized = yield* Deferred.make(); + const order = yield* Ref.make>([]); + const testLayer = runExecutionServiceLayer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(CheckpointServiceV2)({ captureBaseline: () => Effect.void }), + Layer.mock(EventSinkV2)({ + write: () => Effect.succeed([]), + writeWithEffects: (input) => + Effect.gen(function* () { + if ( + input.events.some( + (event) => event.type === "run.updated" && event.runId === runId, + ) + ) { + yield* Ref.update(order, (current) => [...current, "root-finalized"]); + yield* Deferred.succeed(rootFinalized, undefined).pipe(Effect.ignore); + } + return []; + }), + writeIfRunCurrent: () => Effect.succeed({ committed: true, storedEvents: [] }), + }), + idAllocatorLayer, + Layer.mock(ProviderEventIngestorV2)({ + ingestNormalized: (input) => + Effect.gen(function* () { + if ( + input.event.type === "message.updated" && + input.event.message.id === poisonMessageId + ) { + return yield* new ProviderEventNormalizeError({ + providerSessionId, + threadId, + providerEvent: input.event, + }); + } + if ( + input.event.type === "message.updated" && + input.event.message.id === goodMessageId + ) { + yield* Ref.update(order, (current) => [...current, "good-message"]); + yield* Deferred.succeed(goodMessageIngested, undefined).pipe(Effect.ignore); + } + return []; + }), + }), + ServerSettingsService.layerTest(), + ), + ), + ); + const events: ReadonlyArray = [ + { + type: "message.updated", + driver, + message: { + id: poisonMessageId, + threadId, + runId, + nodeId: rootNodeId, + role: "assistant", + text: "poison", + streaming: false, + }, + } as ProviderAdapterV2Event, + { + type: "message.updated", + driver, + message: { + id: goodMessageId, + threadId, + runId, + nodeId: rootNodeId, + role: "assistant", + text: "good", + streaming: false, + }, + } as ProviderAdapterV2Event, + { + type: "turn.terminal", + driver, + providerThreadId, + providerTurnId: rootProviderTurnId, + runOrdinal: 1, + status: "completed", + failure: null, + threadDisposition: "reusable", + }, + ]; + + yield* Effect.gen(function* () { + const runExecution = yield* RunExecutionServiceV2; + yield* runExecution.startRootRun({ + commandId: CommandId.make("command:run-execution-poison"), + appThread: { id: threadId } as OrchestrationV2AppThread, + providerSessionId, + session: { + events: Stream.fromIterable(events), + startTurn: () => Effect.void, + } as unknown as ProviderAdapterV2SessionRuntime, + run: { + id: runId, + threadId, + ordinal: 1, + providerInstanceId, + } as OrchestrationV2Run, + rootNode: { id: rootNodeId } as OrchestrationV2ExecutionNode, + checkpointScope: { + id: CheckpointScopeId.make("checkpoint-scope:run-execution-poison"), + } as OrchestrationV2CheckpointScope, + providerThread: { + id: providerThreadId, + driver, + } as OrchestrationV2ProviderThread, + attempt: { + id: attemptId, + providerTurnId: rootProviderTurnId, + } as OrchestrationV2RunAttempt, + attemptId, + providerTurnOrdinal: 1, + message: { + messageId: MessageId.make("message:run-execution-poison:user"), + text: "Survive a poison event.", + attachments: [], + createdBy: "user", + creationSource: "web", + }, + modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" }, + runtimePolicy: { + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + approvalPolicy: "never", + sandboxPolicy: { + type: "readOnly", + access: { type: "fullAccess" }, + networkAccess: false, + }, + }, + }); + }).pipe(Effect.provide(testLayer)); + + const observed = yield* Deferred.await(goodMessageIngested).pipe( + Effect.timeoutOption("2 seconds"), + ); + assert.isTrue(Option.isSome(observed), "good event was not ingested after the poison event"); + const finalized = yield* Deferred.await(rootFinalized).pipe(Effect.timeoutOption("2 seconds")); + assert.isTrue(finalized._tag === "Some", "root run was not finalized after the poison event"); + assert.deepEqual(yield* Ref.get(order), ["good-message", "root-finalized"]); + }), +); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index c3da3c886fd..506de32c2ed 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -681,7 +681,26 @@ export const layer: Layer.Layer< yield* finalizeRootRun(event); } yield* trackChildLifecycle(event); - }), + }).pipe( + // A single unmappable event (e.g. a late subagent notification + // that references an already-finalized run) must not tear down + // ingestion for the whole session: dropping it keeps the + // pipeline alive for every later event. Root turn.terminal + // failures still escalate so the fallback finalization in the + // outer catch keeps running. + Effect.catchCause((eventCause) => + event.type === "turn.terminal" + ? Effect.failCause(eventCause) + : Effect.logWarning( + "orchestration V2 provider event dropped after ingestion failure", + { + runId: input.run.id, + eventType: event.type, + cause: Cause.pretty(eventCause), + }, + ), + ), + ), ), Stream.takeUntilEffect(() => shouldStopProviderEventIngestion), Stream.runDrain, @@ -700,7 +719,7 @@ export const layer: Layer.Layer< Effect.flatMap((finalized) => Effect.logWarning("orchestration V2 provider event ingestion failed", { runId: input.run.id, - cause, + cause: Cause.pretty(cause), }).pipe( Effect.andThen( finalized @@ -773,7 +792,7 @@ export const layer: Layer.Layer< Effect.catchCause((cause) => Effect.logError("orchestration V2 provider turn start failed", { runId: input.run.id, - cause, + cause: Cause.pretty(cause), }).pipe( Effect.andThen(Fiber.interrupt(providerEventFiber)), Effect.andThen(Ref.get(latestProviderThread)),