// @effect-diagnostics nodeBuiltinImport:off import { type ChatAttachment, type ModelSelection, type OrchestrationV2ConversationMessage, type OrchestrationV2ExecutionNode, type OrchestrationV2PlanArtifact, type OrchestrationV2PlanStep, type OrchestrationV2ProviderCapabilities, type OrchestrationV2ProviderFailure, type OrchestrationV2ProviderRetry, type OrchestrationV2ProviderSession, type OrchestrationV2ProviderThread, type OrchestrationV2ProviderThreadNativeMetadata, type OrchestrationV2ProviderTurn, type OrchestrationV2RuntimeRequest, type OrchestrationV2Subagent, type OrchestrationV2TurnItem, type OrchestrationV2UserInputQuestion, type OrchestrationV2WebSearchResult, type ProviderApprovalDecision, type ProviderApprovalOption, type ProviderInstanceId, type ProviderInteractionMode, type ProviderDriverKind, type ProviderRequestKind, type ProviderThreadId, type ProviderUserInputAnswers, type RuntimeRequestId, type ThreadTokenUsageSnapshot, type ThreadId, } from "@t3tools/contracts"; import { AgentScope, AgentScopeThreadId } from "@t3tools/shared/AgentScope"; import { HTML_RENDER_TOOL_NAME } from "@t3tools/shared/htmlRender"; import { modelSelectionsEqual } from "@t3tools/shared/model"; import { type SelfInvocation, selfInvocationArgs } from "@t3tools/shared/nodeRuntime"; import { resolveT3McpToolId } from "@t3tools/shared/t3McpToolPresentation"; import { compactDynamicToolOutput, htmlRenderFromToolItem } from "@t3tools/shared/toolOutput"; import { FILE_HEADERS_ONLY, formatPatch, structuredPatch } from "diff"; import * as Cause from "effect/Cause"; import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import type * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; import * as Fiber from "effect/Fiber"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Result from "effect/Result"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as EffectAcpErrors from "effect-acp/errors"; import type * as EffectAcpProtocol from "effect-acp/protocol"; import type * as EffectAcpSchema from "effect-acp/compat"; import { formatReadToolLabel, formatSearchToolLabel } from "@t3tools/shared/toolActivity"; import * as ProviderHost from "@t3tools/provider-core/server/ProviderHost"; import { makeAcpMcpOverAcpBridge, type AcpMcpOverAcpBridge } from "./mcpOverAcpBridge.ts"; import { mcpToolPresentation } from "@t3tools/provider-core/server/mcpToolPresentation"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; import * as McpProviderSession from "@t3tools/provider-core/server/mcpSession"; import { applyAcpAgentTerminalUpdate, acpContentBlockDisplayText, embeddedTerminalIdsFromSessionUpdate, extractMcpToolCallIdentity, findModelConfigOption, mergeToolCallState, parsePermissionRequest, parseSessionUpdateEvent, toolCallVisibleOutputChanged, withTrackedPermissionToolCall, type AcpPlanUpdate, type AcpAgentTerminalState, type AcpSessionModeState, type AcpToolCallState, } from "./runtimeModel.ts"; import { acpClientExecuteDisposition, acpMcpToolApprovalElicitationDisposition, acpPermissionDisposition, type AcpPermissionDisposition, makeAcpClientPolicyGrants, unknownRecord, } from "./clientPolicy.ts"; import { makeAcpClientTerminals, resolveEmbeddedTerminalContent, type AcpClientTerminals, } from "./clientTerminals.ts"; import { ACP_EMPTY_OPTION_VALUE, ACP_SESSION_MODE_OPTION_ID } from "./sessionConfig.ts"; import * as AcpSessionRuntime from "./AcpSessionRuntime.ts"; import { t3AcpPromptWithInstructions, type T3AcpInstructionState, } from "@t3tools/provider-core/server/orchestrationInstructions"; import { buildRuntimeInstructions } from "@t3tools/provider-core/server/runtimeInstructions"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import type * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; import { type BackgroundWork, type BackgroundWorkReport, backgroundWorkNotification, } from "@t3tools/provider-core/server/notification"; import { makeProviderFailure, makeProviderRetryTurnItem, } from "@t3tools/provider-core/server/failure"; import { acpSelectionTransition } from "@t3tools/provider-core/server/selectionTransition"; import { isProviderNativeImageAttachment, providerMessageTextWithAttachmentPaths, } from "@t3tools/provider-core/server/attachmentPrompt"; import { makeSubagentChildThread, makeSubagentConversationArtifacts, subagentThreadTitle, } from "@t3tools/provider-core/server/subagentProjection"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; export const ACP_PROTOCOL = "acp.ndjson-jsonrpc" as const; // Grok summaries can trail hydrated tools by over two seconds; shorter waits split the wake. const ACP_DEFERRED_FINALIZE_DEBOUNCE: Duration.Input = "3000 millis"; /** * Emit first chunks and segment-close events immediately; coalesce later chunks into full-text snapshots. * This avoids quadratic output; scheduled flushes can run late under host load. */ const ACP_TEXT_FLUSH_INTERVAL: Duration.Input = "50 millis"; export interface AcpAdapterV2RuntimeInput { readonly cwd: string; readonly runtimePolicy: ProviderAdapter.ProviderAdapterV2RuntimePolicy; readonly mcpServers: ReadonlyArray; readonly acpMcpServers?: ReadonlyArray; readonly processEnvironment?: NodeJS.ProcessEnv; readonly resumeSessionId?: string; readonly interruptPromptOnCancel?: boolean; readonly clientCapabilities: EffectAcpSchema.InitializeRequest["clientCapabilities"]; readonly clientInfo: AcpSessionRuntime.AcpSessionRuntimeOptions["clientInfo"]; readonly requestLogger?: NonNullable; readonly protocolLogging: NonNullable< AcpSessionRuntime.AcpSessionRuntimeOptions["protocolLogging"] >; readonly onTermination: NonNullable; readonly onOutgoingResponseFailure?: AcpSessionRuntime.AcpSessionRuntimeOptions["onOutgoingResponseFailure"]; readonly onOutgoingResponse?: AcpSessionRuntime.AcpSessionRuntimeOptions["onOutgoingResponse"]; } export type AcpAdapterV2NativeLogging = Pick< AcpSessionRuntime.AcpSessionRuntimeOptions, "requestLogger" | "protocolLogging" >; export interface AcpAdapterV2UserInputRequest { readonly nativeItemId: string; readonly nativeRequestId: string; readonly questions: ReadonlyArray; } export interface AcpAdapterV2ExtensionContext { readonly runtime: AcpSessionRuntime.AcpSessionRuntime["Service"]; readonly reportProviderRetry: (input: { readonly sessionId: string; readonly failure: OrchestrationV2ProviderFailure; }) => Effect.Effect; readonly finishSubagent: (notice: { readonly sessionId: string; readonly childSessionId: string; readonly status: "completed" | "failed" | "cancelled"; readonly result: string | null; }) => Effect.Effect; readonly applyBackgroundTaskMutation: (mutation: { readonly sessionId: string; readonly taskId: string; readonly status: "running" | "completed" | "failed"; readonly output?: string; readonly report?: BackgroundWork; }) => Effect.Effect; readonly requestUserInput: ( input: AcpAdapterV2UserInputRequest, requestContext: EffectAcpProtocol.AcpRequestContext, ) => Effect.Effect< { readonly acknowledgeNativeResponse: Effect.Effect; readonly answers: ProviderUserInputAnswers | null; }, EffectAcpErrors.AcpError >; readonly captureProposedPlan: (input: { readonly planMarkdown: string }) => Effect.Effect; readonly lastProposedPlanMarkdown: Effect.Effect; } /** * Raised when an agent does not accept the session's model selection: the * session's model option does not offer it, or the agent answers the write * with invalid params. `makeProviderFailure` shows its message to the user; * the agent's reply stays on `cause`. */ class AcpModelSelectionError extends Schema.TaggedError()( "AcpModelSelectionError", { model: Schema.String, cause: Schema.Defect(), }, ) { override get message(): string { return `The agent did not accept the model "${this.model}". Choose another model for this provider, then retry.`; } } export interface AcpAdapterV2Flavor { readonly buildPrompt?: (input: { readonly text: string; readonly attachments: ReadonlyArray; readonly promptCapabilities: EffectAcpSchema.AgentCapabilities["promptCapabilities"]; }) => Effect.Effect, EffectAcpErrors.AcpError>; readonly promptFailure?: (cause: unknown) => OrchestrationV2ProviderFailure; /** * Runs before attachment paths, T3 instructions and native slash-command detection, * so a rewrite can turn the user's text into a native command. */ readonly rewriteUserText?: ( text: string, turnInput: ProviderAdapter.ProviderAdapterV2TurnInput, ) => Effect.Effect; readonly driver: ProviderDriverKind; readonly capabilities: OrchestrationV2ProviderCapabilities; readonly clientCapabilitiesMeta?: NonNullable< EffectAcpSchema.InitializeRequest["clientCapabilities"] >["_meta"]; readonly normalizeSessionUpdate?: ( notification: EffectAcpSchema.SessionNotification, ) => EffectAcpSchema.SessionNotification; readonly onAvailableCommandsUpdate?: ( commands: ReadonlyArray, ) => Effect.Effect; readonly onSessionConfigurationUpdate?: ( configOptions: ReadonlyArray, modeState: AcpSessionModeState | undefined, modelState: EffectAcpSchema.SessionModelState | null | undefined, ) => Effect.Effect; readonly onUrlElicitation?: (input: { readonly elicitationId: string; readonly url: string; readonly message: string; }) => Effect.Effect; readonly extractElicitationForm?: (request: EffectAcpSchema.CreateElicitationRequest) => | { readonly questions: ReadonlyArray; readonly respond: ( answers: ProviderUserInputAnswers, ) => EffectAcpSchema.CreateElicitationResponse; } | undefined; readonly withRuntimeStartup?: (effect: Effect.Effect) => Effect.Effect; readonly makeRuntime: ( input: AcpAdapterV2RuntimeInput, ) => Effect.Effect< AcpSessionRuntime.AcpSessionRuntime["Service"], EffectAcpErrors.AcpError, ChildProcessSpawner.ChildProcessSpawner | Crypto.Crypto | Scope.Scope >; readonly customModels?: ReadonlyArray; readonly resolveModelId?: (selection: ModelSelection) => string | undefined; readonly getModelContextWindow?: ( selection: ModelSelection, session: AcpSessionRuntime.AcpSessionRuntimeStartResult, ) => number | undefined; readonly applyModelSelection?: (input: { readonly runtime: AcpSessionRuntime.AcpSessionRuntime["Service"]; readonly startResult: AcpSessionRuntime.AcpSessionRuntimeStartResult; readonly modelSelection: ModelSelection; }) => Effect.Effect; readonly sessionModeForPolicy?: ( policy: ProviderAdapter.ProviderAdapterV2RuntimePolicy, ) => string | undefined; readonly clientFileSystem?: { readonly readTextFile: ( request: EffectAcpSchema.ReadTextFileRequest, cwd: string | null, ) => Effect.Effect; readonly writeTextFile?: ( request: EffectAcpSchema.WriteTextFileRequest, cwd: string | null, ) => Effect.Effect; }; readonly extractPermissionQuestion?: (request: EffectAcpSchema.RequestPermissionRequest) => | { readonly question: OrchestrationV2UserInputQuestion; readonly respond: ( answers: ProviderUserInputAnswers, ) => EffectAcpSchema.RequestPermissionResponse | undefined; } | undefined; /** Grok Auto asks only about classifier refusals; T3 policy must not auto-approve them. */ readonly permissionDisposition?: ( policy: ProviderAdapter.ProviderAdapterV2RuntimePolicy, request: EffectAcpSchema.RequestPermissionRequest, ) => AcpPermissionDisposition; readonly approvalOptions?: ( request: EffectAcpSchema.RequestPermissionRequest, ) => ReadonlyArray; readonly onSessionEvent?: ( event: AcpSessionRuntime.AcpSessionRuntimeEvent, ) => Effect.Effect; readonly subagentsIdleOnTurnCompletion?: boolean; readonly supportsCompaction?: boolean; readonly runtimeHarness?: string; readonly registerExtensions?: ( context: AcpAdapterV2ExtensionContext, ) => Effect.Effect; readonly extractSubagentSessionEvent?: ( notification: EffectAcpSchema.SessionNotification, ) => AcpSubagentSessionEvent | undefined; readonly extractSubagentUpdate?: ( toolCall: AcpToolCallState, ) => AcpAdapterV2SubagentUpdate | undefined; readonly isProvisionalSubagentTool?: (toolCall: AcpToolCallState) => boolean; readonly normalizeToolCall?: (toolCall: AcpToolCallState) => AcpToolCallState; readonly extractProposedPlanMarkdown?: (toolCall: AcpToolCallState) => string | undefined; readonly extractBackgroundTaskId?: (toolCall: AcpToolCallState) => string | undefined; /** Distinguishes commands from monitors when lifecycle notifications omit their kind. */ readonly isBackgroundCommandTool?: (toolCall: AcpToolCallState) => boolean; readonly extractBackgroundToolMutation?: (text: string) => ReadonlyArray<{ readonly taskId: string; readonly status: "running" | "completed" | "failed"; readonly appendOutput: string; }>; readonly extractSubagentEndNotice?: (text: string) => | { readonly childSessionId: string; readonly status: "completed" | "failed"; } | undefined; readonly extractBackgroundTaskCompletion?: (toolCall: AcpToolCallState) => ReadonlyArray<{ readonly taskId: string; readonly status: "running" | "completed" | "failed"; readonly appendOutput: string; }>; readonly isPersistentBackgroundTool?: (toolCall: AcpToolCallState) => boolean; readonly isProviderWakeNotification?: ( notification: EffectAcpSchema.SessionNotification, ) => boolean; readonly deferFinalizeForBackgroundWork?: boolean; readonly assertComplete?: Effect.Effect; /** Interrupt the local prompt fiber before `session/cancel` (Grok wedged prompts). */ readonly interruptPromptOnCancel?: boolean; /** Grok can retain task_already_running until process exit. */ readonly restartRuntimeAfterInterrupt?: boolean; readonly restartRuntimeOnEveryInterrupt?: boolean; readonly terminateRuntimeProcessGroupOnInterrupt?: boolean; readonly preserveRuntimeOnSettledInterrupt?: boolean; readonly enablePostSettleContinuation?: boolean; /** Settle on native idle after the accepted user echo, only on negotiated ACP v2. */ readonly followAgentWorkState?: boolean; /** * Grok CLI currently accepts and vision-processes image blocks while still * reporting the capability as false; without this override, screenshot turns * fail before `session/prompt`. */ readonly supportsImagePrompts?: boolean; } function acpSupportsImagePrompts(input: { readonly flavorSupportsImagePrompts?: boolean | undefined; readonly negotiatedImage?: boolean | undefined; }): boolean { return input.flavorSupportsImagePrompts === true || input.negotiatedImage === true; } export function acpScopedNativeId(instanceId: ProviderInstanceId, nativeId: string): string { return `provider-instance:${encodeURIComponent(instanceId)}:${nativeId}`; } function acpProviderItemNativeId(input: { readonly instanceId: ProviderInstanceId; readonly itemIdentityVersion: 2 | undefined; readonly nativeId: string; }): string { return input.itemIdentityVersion === 2 ? acpScopedNativeId(input.instanceId, input.nativeId) : input.nativeId; } export type AcpSubagentSessionEvent = | { readonly type: "state"; readonly childSessionId: string; readonly parentSessionId: string; readonly status: "running" | "completed" | "failed" | "cancelled"; readonly result: string | null; } | { readonly type: "delegation"; readonly childSessionId: string; readonly parentSessionId: string | null; readonly delegationId: string; readonly prompt: string; }; export interface AcpAdapterV2SubagentUpdate { readonly nativeTaskId: string; readonly prompt: string; readonly title: string | null; readonly model: string | null; readonly status: | "pending" | "running" | "idle" | "completed" | "failed" | "interrupted" | "cancelled"; readonly childSessionId: string | null; readonly result: string | null; readonly suppressNormalTool?: boolean; } export interface AcpAdapterV2Options { readonly instanceId: ProviderInstanceId; readonly flavor: AcpAdapterV2Flavor; readonly selfInvocation: SelfInvocation; readonly clientTerminals?: { readonly environment?: NodeJS.ProcessEnv; readonly shellCommands?: boolean; }; readonly nativeLogging?: (threadId: ThreadId) => AcpAdapterV2NativeLogging; readonly continuationRequests?: { readonly offer: ( request: ProviderContinuationRequests.ProviderContinuationRequest, ) => Effect.Effect; }; readonly testHooks?: { readonly onDeferredFinalizeScheduled?: (debounce: Duration.Input) => Effect.Effect; readonly afterPromptSettledWithBackgroundWork?: () => Effect.Effect; readonly onNativeResponseLifecycle?: (event: { readonly generation?: number; readonly requestId?: string; readonly type: | "admission_rejected" | "failed" | "late_noop" | "registered" | "removed" | "timer_exited" | "timer_started" | "watcher_exited" | "watcher_started"; }) => Effect.Effect; }; } export const AcpProviderCapabilitiesV2 = { sessions: { supportsMultipleProviderThreadsPerSession: false, supportsModelSwitchInSession: false, supportsProviderSwitchingViaHandoff: true, supportsRuntimeModeSwitchInSession: false, pendingRequestsSurviveRestart: false, }, threads: { canCreateEmptyThread: true, canReadThreadSnapshot: false, canRollbackThread: true, canForkThread: false, canForkFromTurn: false, canForkFromSubagentThread: false, exposesNativeThreadId: true, }, turns: { exposesNativeTurnId: false, emitsTurnStarted: true, emitsTurnCompleted: true, supportsInterrupt: true, supportsActiveSteering: false, supportsSteeringByInterruptRestart: true, supportsQueuedMessages: true, terminalStatusQuality: "strong", }, streaming: { streamsAssistantText: true, streamsReasoning: true, streamsToolOutput: true, streamsPlanText: false, emitsMessageCompleted: true, }, tools: { exposesToolItemIds: true, emitsToolStarted: true, emitsToolCompleted: true, emitsToolOutput: true, supportsMcpTools: false, supportsDynamicToolCallbacks: false, }, approvals: { supportsCommandApproval: true, supportsFileReadApproval: true, supportsFileChangeApproval: true, supportsApplyPatchApproval: false, approvalsHaveNativeRequestIds: false, approvalCallbacksAreLiveOnly: true, approvalsCanOriginateFromSubagents: false, }, planning: { emitsPlanUpdated: true, emitsTodoList: true, emitsProposedPlan: false, supportsStructuredQuestions: true, planDeltasHaveItemIds: false, }, subagents: { supportsSubagents: false, exposesSubagentThreadIds: false, emitsSubagentLifecycle: false, canWaitForSubagents: false, canCloseSubagents: false, canForkSubagentThread: false, }, context: { acceptsSystemContext: false, acceptsDeveloperContext: false, acceptsSyntheticUserContext: true, canGenerateSummaries: true, canConsumeHandoffSummaries: true, supportsDeltaHandoff: true, supportsFullThreadHandoff: true, maxRecommendedHandoffChars: null, }, checkpointing: { appCanCheckpointFilesystem: true, supportsNestedCheckpointScopes: true, // ACP lacks truncation; rollback restores T3 checkpoints and starts a fresh agent session. providerCanRollbackConversation: true, providerRollbackReturnsSnapshot: true, providerCanReadConversationSnapshot: false, }, identity: { nativeThreadIds: "strong", nativeTurnIds: "weak", nativeItemIds: "weak", nativeRequestIds: "weak", }, runtimePolicy: { enforcement: "client-boundary", }, } satisfies OrchestrationV2ProviderCapabilities; function negotiatedCapabilities( base: OrchestrationV2ProviderCapabilities, started: AcpSessionRuntime.AcpSessionRuntimeStartResult, ): OrchestrationV2ProviderCapabilities { const agent = started.initializeResult.agentCapabilities ?? {}; const session = agent.sessionCapabilities; const setup = started.sessionSetupResult; const canSwitchModel = findModelConfigOption(setup.configOptions) !== undefined || setup.models != null; const canLoad = agent.loadSession === true; const canFork = session?.fork != null; return { ...base, sessions: { ...base.sessions, supportsModelSwitchInSession: canSwitchModel, }, threads: { ...base.threads, canReadThreadSnapshot: canLoad, canForkThread: canFork, canForkFromTurn: false, }, tools: { ...base.tools, supportsMcpTools: true, }, checkpointing: { ...base.checkpointing, providerCanReadConversationSnapshot: canLoad, }, }; } interface AcpMcpContext { readonly servers: ReadonlyArray; readonly acpServers: ReadonlyArray; readonly processEnvironment?: NodeJS.ProcessEnv; readonly endpoint?: string; readonly authorization?: string; } function acpMcpContext( session: McpProviderSession.McpProviderSessionConfig | undefined, self: SelfInvocation, ): AcpMcpContext { if (session === undefined) { return { servers: [], acpServers: [] }; } // codex-acp 1.2.0 and pi-acp drop injected HTTP servers; bridge through stdio instead. return { servers: [ { name: "t3-code", command: self.command, args: [...selfInvocationArgs(self, ["acp-mcp-bridge"])], env: [ { name: "ELECTRON_RUN_AS_NODE", value: "1" }, { name: "T3_ACP_MCP_ENDPOINT", value: session.endpoint }, { name: "T3_ACP_MCP_AUTHORIZATION", value: session.authorizationHeader }, ], }, ], acpServers: [{ type: "acp", name: "t3-code", serverId: "t3-code" }], endpoint: session.endpoint, authorization: session.authorizationHeader, processEnvironment: { T3_ACP_MCP_ENDPOINT: session.endpoint, T3_ACP_MCP_AUTHORIZATION: session.authorizationHeader, T3_ACP_MCP_NODE: self.command, ...(self.entrypoint === undefined ? {} : { T3_ACP_MCP_ENTRYPOINT: self.entrypoint }), }, }; } function acpMcpActivation(context: AcpMcpContext) { return { mcpServers: context.servers, acpMcpServers: context.acpServers }; } function nativeThreadId( driver: ProviderDriverKind, thread: OrchestrationV2ProviderThread, ): Effect.Effect { const id = thread.nativeThreadRef?.nativeId; if (id === null || id === undefined || id.trim().length === 0) { return Effect.fail( new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: `Provider thread ${thread.id} is missing its ACP session id`, }), ); } return Effect.succeed(id); } function makeProviderThread(input: { readonly driver: ProviderDriverKind; readonly providerInstanceId: ProviderInstanceId; readonly idAllocator: IdAllocator.IdAllocatorV2["Service"]; readonly appThreadId: OrchestrationV2ProviderThread["appThreadId"]; readonly providerSessionId: OrchestrationV2ProviderThread["providerSessionId"]; readonly nativeThreadId: string; readonly ownerNodeId?: OrchestrationV2ProviderThread["ownerNodeId"]; readonly forkedFrom?: OrchestrationV2ProviderThread["forkedFrom"]; readonly itemIdentityVersion?: 2; readonly now: DateTime.Utc; }): OrchestrationV2ProviderThread { return { id: input.idAllocator.derive.providerThread({ driver: input.driver, providerInstanceId: input.providerInstanceId, nativeThreadId: input.nativeThreadId, }), driver: input.driver, providerInstanceId: input.providerInstanceId, providerSessionId: input.providerSessionId, appThreadId: input.appThreadId, ownerNodeId: input.ownerNodeId ?? null, nativeThreadRef: { driver: input.driver, nativeId: input.nativeThreadId, strength: "strong", }, nativeConversationHeadRef: null, status: "idle", firstRunOrdinal: null, lastRunOrdinal: null, handoffIds: [], forkedFrom: input.forkedFrom ?? null, contextUsage: null, nativeMetadata: input.itemIdentityVersion === undefined ? null : { itemIdentityVersion: input.itemIdentityVersion }, createdAt: input.now, updatedAt: input.now, }; } function acpMcpToolCallOutput(rawOutput: unknown): unknown { const record = unknownRecord(rawOutput); if (record === undefined) return rawOutput; if (!("result" in record) && !("error" in record)) return rawOutput; const result = unknownRecord(record.result); const resultOutput = result === undefined ? undefined : (result.structuredContent ?? result.content ?? undefined); const errorMessage = unknownRecord(record.error)?.message; if (typeof errorMessage !== "string") { return resultOutput ?? rawOutput; } return resultOutput === undefined ? { error: errorMessage } : { error: errorMessage, result: resultOutput }; } function nonEmptyText(value: unknown, fallback: string): string { return typeof value === "string" && value.trim().length > 0 ? value.trim() : fallback; } function decodeByteText(value: unknown): string | undefined { if (!Array.isArray(value) || value.length === 0) return undefined; if (!value.every((entry) => typeof entry === "number" && Number.isInteger(entry))) { return undefined; } try { const text = new TextDecoder().decode(Uint8Array.from(value as number[])); return text.length > 0 ? text : undefined; } catch { return undefined; } } function textFromUnknown(value: unknown): string | undefined { if (typeof value === "string") { return value; } const fromBytes = decodeByteText(value); if (fromBytes !== undefined) { return fromBytes; } if (Array.isArray(value)) { const parts = value.flatMap((entry) => { const text = textFromUnknown(entry); return text === undefined || text.length === 0 ? [] : [text]; }); return parts.length === 0 ? undefined : parts.join("\n"); } const record = unknownRecord(value); if (record === undefined) { return undefined; } for (const key of [ "output_for_prompt", "stdout", "stderr", "output", "combinedOutput", "combined_output", "content", "text", "message", ]) { const direct = record[key]; if (typeof direct === "string" && direct.length > 0) { return direct; } const decoded = decodeByteText(direct); if (decoded !== undefined) { return decoded; } const text = textFromUnknown(direct); if (text !== undefined && text.length > 0) { return text; } } const result = unknownRecord(record.Result) ?? unknownRecord(record.result); if (result !== undefined) { return textFromUnknown(result); } return undefined; } function commandExitCode(value: unknown): number | undefined { const record = unknownRecord(value); for (const key of ["exitCode", "exit_code", "code"]) { const candidate = record?.[key]; if (typeof candidate === "number" && Number.isInteger(candidate)) { return candidate; } } return undefined; } /** * Mid-stream Grok Bash re-reports carry exit_code 0 while still in progress; * interrupted tools must not retain that stale success code. */ function acpProjectedCommandExitCode( status: "pending" | "running" | "waiting" | "completed" | "failed" | "interrupted", rawOutput: unknown, ): number | undefined { if (status !== "completed" && status !== "failed") { return undefined; } return commandExitCode(rawOutput); } function structuredFileChanges(toolCall: AcpToolCallState) { const content = toolCall.data.content; if (!Array.isArray(content)) return []; return content.flatMap((entry) => { const diff = unknownRecord(entry); if (diff?.type !== "diff" || !Array.isArray(diff.changes)) return []; return diff.changes.flatMap((candidate) => { const change = unknownRecord(candidate); const operation = typeof change?.operation === "string" ? change.operation.trim() : ""; const path = typeof change?.path === "string" ? change.path.trim() : ""; if (operation.length === 0 || path.length === 0) return []; const oldPath = typeof change?.oldPath === "string" ? change.oldPath.trim() : ""; const fileType = typeof change?.fileType === "string" ? change.fileType.trim() : ""; const mimeType = typeof change?.mimeType === "string" ? change.mimeType.trim() : ""; return [ { operation, path, ...(oldPath.length === 0 ? {} : { oldPath }), ...(fileType.length === 0 ? {} : { fileType }), ...(mimeType.length === 0 ? {} : { mimeType }), }, ]; }); }); } // Bound diff search to protect the event loop; empty-sided diffs need no search. const ACP_V1_DIFF_MAX_EDITS = 1_000; export function acpToolCallDiffPatch(content: unknown): string | undefined { if (!Array.isArray(content)) return undefined; const diffs = content.flatMap((entry) => { const diff = unknownRecord(entry); return diff?.type === "diff" ? [diff] : []; }); for (const diff of diffs) { const patch = unknownRecord(diff.patch); if (typeof patch?.text === "string") { const change = Array.isArray(diff.changes) && diff.changes.length === 1 ? unknownRecord(diff.changes[0]) : undefined; if ( patch.format === "git_patch" && patch.text.trimStart().startsWith("@@ ") && typeof change?.path === "string" ) { const oldPath = typeof change.oldPath === "string" ? change.oldPath : change.path; const before = change.operation === "add" ? "/dev/null" : JSON.stringify(`a/${oldPath}`); const after = change.operation === "delete" ? "/dev/null" : JSON.stringify(`b/${change.path}`); return `diff --git ${JSON.stringify(`a/${oldPath}`)} ${JSON.stringify(`b/${change.path}`)}\n--- ${before}\n+++ ${after}\n${patch.text.trimStart()}`; } return patch.text; } } const v1Patches = diffs.flatMap((diff) => { if (typeof diff.path !== "string" || typeof diff.newText !== "string") return []; const oldText = typeof diff.oldText === "string" ? diff.oldText : undefined; const patch = structuredPatch( oldText === undefined ? "/dev/null" : diff.path, diff.path, oldText ?? "", diff.newText, undefined, undefined, { context: 3, maxEditLength: oldText && diff.newText ? ACP_V1_DIFF_MAX_EDITS : Number.POSITIVE_INFINITY, }, ); return patch === undefined || patch.hunks.length === 0 ? [] : [patch]; }); return v1Patches.length === 0 ? undefined : formatPatch(v1Patches, FILE_HEADERS_ONLY); } function pathFromToolCall(toolCall: AcpToolCallState): string | undefined { const locations = toolCall.data.locations; if (Array.isArray(locations)) { for (const location of locations) { const path = unknownRecord(location)?.path; if (typeof path === "string" && path.trim().length > 0) { return path.trim(); } } } const rawInput = unknownRecord(toolCall.data.rawInput); for (const key of ["path", "filePath", "file_path", "url", "query", "pattern"]) { const candidate = rawInput?.[key]; if (typeof candidate === "string" && candidate.trim().length > 0) { return candidate.trim(); } } return undefined; } function acpBackendWebSearch( rawInput: Record | undefined, rawOutput: Record | undefined, ): | { readonly query: string | undefined; readonly results: OrchestrationV2WebSearchResult[] } | undefined { const variant = typeof rawInput?.variant === "string" ? rawInput.variant.toLowerCase() : ""; const action = unknownRecord(rawOutput?.action); if (variant !== "xsearch" && variant !== "websearch" && action?.type !== "search") { return undefined; } let args: Record | undefined; if (typeof rawOutput?.input === "string") { try { args = unknownRecord(JSON.parse(rawOutput.input)); } catch { args = undefined; } } const argsText = Object.entries(args ?? {}) .filter(([, value]) => typeof value === "string" || typeof value === "number") .map(([key, value]) => `${key}: ${value}`) .join(", "); const query = [action?.query, args?.query, argsText] .find((value): value is string => typeof value === "string" && value.trim().length > 0) ?.trim(); const urls = new Set(); for (const source of Array.isArray(action?.sources) ? action.sources : []) { const url = unknownRecord(source)?.url; if (typeof url === "string" && url.trim().length > 0) urls.add(url.trim()); } return { query, results: [...urls].map((url) => ({ url })) }; } function providerRequestKind(kind: string | "unknown"): ProviderRequestKind { switch (kind) { case "execute": return "command"; case "read": case "search": case "fetch": return "file-read"; case "edit": case "delete": case "move": return "file-change"; default: return "command"; } } function toolStatus( status: AcpToolCallState["status"], ): "pending" | "running" | "waiting" | "completed" | "failed" { switch (status) { case "completed": return "completed"; case "failed": return "failed"; case "pending": return "pending"; case "requiresAction": return "waiting"; default: return "running"; } } type ProjectedToolStatus = ReturnType | "interrupted"; function nodeStatus(status: ProjectedToolStatus): OrchestrationV2ExecutionNode["status"] { return status === "pending" ? "running" : status; } function completedAtForStatus(status: ProjectedToolStatus, now: DateTime.Utc): DateTime.Utc | null { return status === "completed" || status === "failed" || status === "interrupted" ? now : null; } function selectPermissionOptionId( request: EffectAcpSchema.RequestPermissionRequest, decision: Exclude, ): string | undefined { const kind = decision === "acceptForSession" ? "allow_always" : decision === "accept" ? "allow_once" : "reject_once"; return request.options.find((option) => option.kind === kind)?.optionId.trim() || undefined; } // Prefer allow-once; allow-always can persist beyond the session. function selectAutoApprovedPermissionOption( request: EffectAcpSchema.RequestPermissionRequest, ): string | undefined { return ( selectPermissionOptionId(request, "accept") ?? selectPermissionOptionId(request, "acceptForSession") ); } function elicitationContent( answers: ProviderUserInputAnswers, allowedKeys: ReadonlySet, ): Record { const content: Record = {}; for (const [key, value] of Object.entries(answers)) { if (!allowedKeys.has(key)) continue; if (typeof value === "string" || typeof value === "number" || typeof value === "boolean") { content[key] = value; } else if (Array.isArray(value)) { content[key] = value.filter((entry): entry is string => typeof entry === "string"); } } return content; } interface ActiveTextSegment { readonly nativeItemId: string; readonly startedAt: DateTime.Utc; sourceMessageId: string | null; text: string; // The running node and first text have been emitted. emitted: boolean; // Text changed since the last emission; a delayed flush is scheduled. dirty: boolean; } interface ActiveTextStream { current: ActiveTextSegment | null; nextSegment: number; } interface AcpNativeBuildConfiguration { readonly modeId?: string; readonly configOptions: ReadonlyArray<{ readonly id: string; readonly value: string }>; } interface ActiveAcpTurn { readonly input: ProviderAdapter.ProviderAdapterV2TurnInput; readonly providerTurnId: OrchestrationV2ProviderTurn["id"]; readonly nativeThreadId: string; readonly nativeTurnId: string; readonly startedAt: DateTime.Utc; readonly completed: Deferred.Deferred; readonly itemOrdinals: Map; readonly user: ActiveTextStream; readonly assistant: ActiveTextStream; readonly reasoning: ActiveTextStream; providerRetry?: | { readonly failure: OrchestrationV2ProviderFailure; readonly retry: OrchestrationV2ProviderRetry; readonly startedAt: DateTime.Utc; readonly itemOrdinal: number; } | undefined; contextUsage: ThreadTokenUsageSnapshot | null; nativeMetadata: OrchestrationV2ProviderThreadNativeMetadata | null; readonly tools: Map; /** Tools the agent failed after Stop; later updates keep them interrupted. */ readonly stoppedToolCallIds: Set; readonly toolUpdatesSkipped: Map; readonly toolStartedAt: Map; readonly subagents: Map; readonly subagentsBySessionId: Map; readonly pendingSubagentNotifications: Map>; readonly toolCallIdsByBackgroundTaskId: Map; /** Command ACKs register null; terminal results wait here for prompt settlement. */ readonly backgroundCommands: Map< string, { readonly status: "completed" | "failed"; readonly output?: string } | null >; readonly persistentBackgroundTaskIds: Set; /** Monitor end notices lack output; wait for hydration or the safety timeout. */ readonly awaitingBackgroundHydration: Set; readonly pendingInjectedReport: Set; /** Reports can precede monitor end notices; consume their completion marker without re-arming. */ earlyInjectedReportObserved: boolean; readonly plans: Map< string, { readonly id: OrchestrationV2PlanArtifact["id"]; readonly startedAt: DateTime.Utc; latest: OrchestrationV2PlanArtifact | null; } >; interrupted: boolean; finalized: boolean; finalizedStatus: "completed" | "interrupted" | "failed" | "cancelled" | null; promptSettled: boolean; promptSettledStatus: "completed" | "interrupted" | "failed" | "cancelled" | null; workAccepted: boolean; readonly promptWireSettled: Deferred.Deferred; backgroundFinalizeGeneration: number; } type AcpRuntimeTeardownState = | { readonly _tag: "Idle" } | { readonly _tag: "InProgress"; readonly completed: Deferred.Deferred; } | { readonly _tag: "Failed"; readonly error: ProviderAdapter.ProviderAdapterProtocolError }; function acpRootSessionUpdateIngestsOutput( notification: EffectAcpSchema.SessionNotification, ): boolean { const update = notification.update; switch (update.sessionUpdate) { case "agent_message_chunk": case "agent_thought_chunk": return (acpContentBlockDisplayText(update.content)?.length ?? 0) > 0; case "agent_message": case "agent_thought": return (update.content ?? []).some( (content) => (acpContentBlockDisplayText(content)?.length ?? 0) > 0, ); case "tool_call": case "tool_call_update": case "plan": case "plan_update": case "plan_removed": case "compaction_update": case "compaction_summary_chunk": return parseSessionUpdateEvent(notification).events.some( (event) => event._tag === "ToolCallUpdated" || event._tag === "PlanUpdated", ); default: return false; } } export function acpPostSettleWakeEvidence( notification: EffectAcpSchema.SessionNotification, flavor: Pick = {}, ): boolean { if (!acpRootSessionUpdateIngestsOutput(notification)) return false; const update = notification.update; if ( (update.sessionUpdate === "user_message_chunk" || update.sessionUpdate === "agent_message_chunk") && update.content.type === "text" ) { const text = update.content.text; if ((flavor.extractBackgroundToolMutation?.(text) ?? []).length > 0) return false; if (update.sessionUpdate === "agent_message_chunk" && text.trim().length === 0) return false; if (/ = {}, ): boolean { if (!acpPostSettleWakeEvidence(notification, flavor)) { return false; } const update = notification.update; if (update.sessionUpdate === "agent_message_chunk") { return (acpContentBlockDisplayText(update.content)?.trim().length ?? 0) > 0; } if (update.sessionUpdate === "tool_call" || update.sessionUpdate === "tool_call_update") { return parseSessionUpdateEvent(notification).events.some((event) => { if (event._tag !== "ToolCallUpdated") return false; // Grok monitor start ACKs say completed while the task is still running. const toolCall = flavor.normalizeToolCall?.(event.toolCall) ?? event.toolCall; return toolCall.status === "completed" || toolCall.status === "failed"; }); } return false; } function acpPostSettleWakeShouldBuffer( notification: EffectAcpSchema.SessionNotification, backgroundWorkRunning: boolean, ): boolean { if (!backgroundWorkRunning) return true; const update = notification.update; return ( update.sessionUpdate !== "agent_message_chunk" && update.sessionUpdate !== "agent_thought_chunk" ); } /** * An app-owned wake (a delegated child finishing) is injected by the * orchestrator, not typed by the user. It reports on a sibling child and says * nothing about this session's own pending wake frames, so it must not discard * them the way a real user turn does. */ function acpIsAppOwnedWakeTurn(message: { readonly createdBy: string; readonly creationSource: string; }): boolean { return message.createdBy === "agent" && message.creationSource === "server"; } function acpIsProviderContinuationMessage(message: { readonly createdBy: string; readonly creationSource: string; }): boolean { return message.createdBy === "agent" && message.creationSource === "provider"; } function acpCarryoverTerminalShouldClearContinuation(input: { readonly continuationOffered: boolean; readonly wakeBufferLength: number; }): boolean { return !input.continuationOffered && input.wakeBufferLength === 0; } function acpTurnStartShouldPreserveContinuation(input: { readonly continuationRequested: boolean; readonly isContinuationTurn: boolean; readonly wakeBufferLength: number; }): boolean { return !input.isContinuationTurn && input.continuationRequested && input.wakeBufferLength > 0; } function acpPostSettleMonitorPromptShouldSuppress( mutation: | { readonly taskId: string; readonly status: "running" | "completed" | "failed"; } | undefined, ): boolean { return mutation?.status === "running"; } export function acpCompletedTurnShouldTerminalizeTool( tool: AcpToolCallState, flavor: Pick, ): boolean { const status = toolStatus(tool.status); if (status !== "pending" && status !== "running") return false; if (flavor.extractBackgroundTaskId?.(tool) !== undefined) return false; return flavor.extractSubagentUpdate?.(tool) === undefined; } interface ActiveAcpSubagent { task: OrchestrationV2Subagent; readonly childThreadId: ThreadId; readonly childRootNodeId: OrchestrationV2ExecutionNode["id"]; readonly turnItemId: OrchestrationV2TurnItem["id"]; turnItemOrdinal: number; providerTurnId: OrchestrationV2ProviderTurn["id"]; readonly parentProviderThreadId: ProviderThreadId; childSessionId: string | null; assistantText: string; readonly assistantMessages: Map; readonly childItemOrdinals: Map; nextChildOrdinal: number; terminalStatusProjected: boolean; readonly toolState: Pick; readonly toolSessionIds: Map; } interface AcpNativeChildSession { parentSessionId: string | null; status: "running" | "completed" | "failed" | "cancelled"; result: string | null; delegationId: string | null; prompt: string; subagent: ActiveAcpSubagent | undefined; owner: ActiveAcpTurn | undefined; readonly pending: Array; } function acpSubagentStatusIsTerminal(status: OrchestrationV2Subagent["status"]): boolean { return ( status === "completed" || status === "failed" || status === "interrupted" || status === "cancelled" || status === "idle" ); } function acpSubagentStatusBlocksTurnSettlement(status: OrchestrationV2Subagent["status"]): boolean { return status === "running" || status === "pending"; } function acpSubagentHasPendingBackgroundWork(subagent: ActiveAcpSubagent): boolean { return ( acpSubagentStatusBlocksTurnSettlement(subagent.task.status) || !subagent.terminalStatusProjected ); } type AcpCarryoverSubagents = { readonly sessionId: string; /** The root's subscriber keeps ingesting, so late child terminals can still project. */ readonly projectsLateTerminals: boolean; readonly subagents: ReadonlyArray; }; type PendingRuntimeRequest = { readonly generation: number; readonly nativeResponseAcknowledgement: Deferred.Deferred; readonly transportRequestId: string; readonly requestId: RuntimeRequestId; readonly runtimeRequest: OrchestrationV2RuntimeRequest; readonly node: OrchestrationV2ExecutionNode; readonly turnItem: OrchestrationV2TurnItem; } & ( | { readonly type: "approval"; readonly decision: Deferred.Deferred; } | { readonly type: "user_input"; readonly answers: Deferred.Deferred; } ); interface SnapshotMessageState { readonly order: Array; readonly messages: Map; loadingRole: "user" | "assistant" | "thought" | null; loadingMessageId: string | null; loadingIndex: number; } const TOOL_UPDATE_PERSIST_EVERY = 10; // Agents can resend whole calls while streaming arguments. function shouldPersistToolUpdate( context: Pick, key: string, previous: AcpToolCallState | undefined, next: AcpToolCallState, reportedStatus: AcpToolCallState["status"], ): boolean { const skipped = context.toolUpdatesSkipped.get(key) ?? 0; const persist = reportedStatus === "completed" || reportedStatus === "failed" || previous === undefined || previous.status !== next.status || previous.title !== next.title || toolCallVisibleOutputChanged(previous, next) || skipped + 1 >= TOOL_UPDATE_PERSIST_EVERY; context.toolUpdatesSkipped.set(key, persist ? 0 : skipped + 1); return persist; } export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( options: AcpAdapterV2Options, ) { const crypto = yield* Crypto.Crypto; const fileSystem = yield* FileSystem.FileSystem; const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const agentScope = yield* AgentScope; const idAllocator = yield* IdAllocator.IdAllocatorV2; const host = yield* ProviderHost.ProviderHost; const mcpSessions = yield* McpProviderSessions.McpProviderSessions; const { flavor, selfInvocation: self } = options; const readMcpContext = (threadId: ThreadId | null) => threadId === null ? Effect.succeed(acpMcpContext(undefined, self)) : mcpSessions.read(threadId).pipe(Effect.map((session) => acpMcpContext(session, self))); const driver = flavor.driver; const continuationRequests = options.continuationRequests; const postSettleContinuationEnabled = flavor.enablePostSettleContinuation === true && continuationRequests !== undefined; return ProviderAdapter.ProviderAdapterV2.of({ instanceId: options.instanceId, driver, getCapabilities: Effect.fn("AcpAdapter.getCapabilities")(function () { return Effect.succeed(flavor.capabilities); }), planSelectionTransition: Effect.fn("AcpAdapter.planSelectionTransition")(function (input) { return Effect.succeed(acpSelectionTransition(input)); }), openSession: Effect.fn("AcpAdapterV2.openSession")( function* (input: ProviderAdapter.ProviderAdapterV2OpenSessionInput) { const sessionScope = yield* Effect.scope; let itemIdentityVersion: 2 | undefined = input.initialProviderItemIdentityVersion ?? (input.initialNativeThreadId === undefined ? 2 : undefined); const providerNativeId = (nativeId: string) => acpProviderItemNativeId({ instanceId: options.instanceId, itemIdentityVersion, nativeId, }); const providerNodeId = (nativeItemId: string) => idAllocator.derive.nodeFromProviderItem({ driver, nativeItemId: providerNativeId(nativeItemId), }); const providerMessageId = (nativeItemId: string) => idAllocator.derive.messageFromProviderItem({ driver, nativeItemId: providerNativeId(nativeItemId), }); const providerTurnItemId = (nativeItemId: string) => idAllocator.derive.turnItemFromProviderItem({ driver, nativeItemId: providerNativeId(nativeItemId), }); const deriveProviderTurnId = (nativeTurnId: string) => idAllocator.derive.providerTurn({ driver, nativeTurnId: providerNativeId(nativeTurnId), }); const useProviderThreadIdentity = (thread: OrchestrationV2ProviderThread): void => { itemIdentityVersion = thread.nativeMetadata?.itemIdentityVersion; }; const terminalEnvironmentBySessionId = new Map(); interface PendingTerminalEnvironment { readonly environment: NodeJS.ProcessEnv | undefined; readonly claimUnknownSession: boolean; readonly sessionId: string | null; } let pendingTerminalEnvironment: PendingTerminalEnvironment | null = { environment: (yield* readMcpContext(input.threadId)).processEnvironment, claimUnknownSession: input.initialNativeThreadId === undefined, sessionId: input.initialNativeThreadId ?? null, }; const prepareTerminalEnvironment = ( mcpContext: AcpMcpContext, sessionId?: string, ): void => { pendingTerminalEnvironment = { environment: mcpContext.processEnvironment, claimUnknownSession: false, sessionId: sessionId ?? null, }; }; const prepareClaimableTerminalEnvironment = (mcpContext: AcpMcpContext): void => { pendingTerminalEnvironment = { environment: mcpContext.processEnvironment, claimUnknownSession: true, sessionId: null, }; }; const rememberTerminalEnvironment = ( sessionId: string, mcpContext: AcpMcpContext, ): void => { const environment = mcpContext.processEnvironment; pendingTerminalEnvironment = null; if (environment === undefined) { terminalEnvironmentBySessionId.delete(sessionId); } else { terminalEnvironmentBySessionId.set(sessionId, environment); } }; const clientTerminals: AcpClientTerminals | undefined = options.clientTerminals === undefined ? undefined : yield* makeAcpClientTerminals({ defaultCwd: input.runtimePolicy.cwd ?? process.cwd(), environment: options.clientTerminals.environment, shellCommands: options.clientTerminals.shellCommands, environmentForSession: (sessionId) => { const remembered = terminalEnvironmentBySessionId.get(sessionId); if (remembered !== undefined) return remembered; if (pendingTerminalEnvironment === null) return undefined; if ( pendingTerminalEnvironment.sessionId === null && pendingTerminalEnvironment.claimUnknownSession ) { pendingTerminalEnvironment = { ...pendingTerminalEnvironment, sessionId, }; } return pendingTerminalEnvironment.sessionId === sessionId ? pendingTerminalEnvironment.environment : undefined; }, }).pipe( Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, childProcessSpawner), ); if (clientTerminals !== undefined) { yield* Scope.addFinalizer(sessionScope, clientTerminals.disposeAll); } const sessionScopedId = (sessionId: string, nativeId: string): string => JSON.stringify([sessionId, nativeId]); const embeddedTerminalsByToolCallId = new Map< string, { readonly sessionId: string; readonly terminalIds: ReadonlyArray; readonly toolCallId: string; } >(); const toolCallIdsByAgentTerminalId = new Map>(); const agentTerminalsById = new Map(); const rememberEmbeddedTerminals = (input: { readonly sessionId: string; readonly toolCallId: string; readonly terminalIds: ReadonlyArray; }): void => { const toolCallKey = sessionScopedId(input.sessionId, input.toolCallId); for (const terminalId of embeddedTerminalsByToolCallId.get(toolCallKey)?.terminalIds ?? []) { const terminalKey = sessionScopedId(input.sessionId, terminalId); const toolCallIds = toolCallIdsByAgentTerminalId.get(terminalKey); toolCallIds?.delete(input.toolCallId); if (toolCallIds?.size === 0) toolCallIdsByAgentTerminalId.delete(terminalKey); } embeddedTerminalsByToolCallId.delete(toolCallKey); embeddedTerminalsByToolCallId.set(toolCallKey, { sessionId: input.sessionId, terminalIds: input.terminalIds, toolCallId: input.toolCallId, }); for (const terminalId of input.terminalIds) { const terminalKey = sessionScopedId(input.sessionId, terminalId); const toolCallIds = toolCallIdsByAgentTerminalId.get(terminalKey) ?? new Set(); toolCallIds.add(input.toolCallId); toolCallIdsByAgentTerminalId.set(terminalKey, toolCallIds); } for (const oldest of embeddedTerminalsByToolCallId.keys()) { if (embeddedTerminalsByToolCallId.size <= 256) break; const remembered = embeddedTerminalsByToolCallId.get(oldest); if (remembered === undefined) continue; for (const terminalId of remembered.terminalIds) { const terminalKey = sessionScopedId(remembered.sessionId, terminalId); const toolCallIds = toolCallIdsByAgentTerminalId.get(terminalKey); toolCallIds?.delete(remembered.toolCallId); if (toolCallIds?.size === 0) toolCallIdsByAgentTerminalId.delete(terminalKey); } embeddedTerminalsByToolCallId.delete(oldest); } }; const embeddedTerminalCommands = ( sessionId: string, toolCallId: string, ): ReadonlyArray => ( embeddedTerminalsByToolCallId.get(sessionScopedId(sessionId, toolCallId)) ?.terminalIds ?? [] ).flatMap((terminalId) => { const command = clientTerminals?.readCommandLine(terminalId) ?? agentTerminalsById.get(sessionScopedId(sessionId, terminalId))?.command; return command === undefined ? [] : [command]; }); // Client terminals run with server privileges; apply turn policy and prior user approvals. const clientPolicyGrants = makeAcpClientPolicyGrants(); let latestRuntimePolicy: ProviderAdapter.ProviderAdapterV2RuntimePolicy = input.runtimePolicy; const events = yield* Queue.unbounded(); const activeTurn = yield* Ref.make(null); let pendingStopCloseTurn: ActiveAcpTurn | null = null; let followsAgentWorkState = false; let agentWork: Extract< EffectAcpSchema.SessionUpdate, { sessionUpdate: "state_update" } > | null = null; const agentRequests = new Set(); const nativeChildren = new Map(); const extractSubagentUpdate = (tool: AcpToolCallState) => flavor.extractSubagentSessionEvent !== undefined && !followsAgentWorkState ? undefined : flavor.extractSubagentUpdate?.(tool); let turnAttached = yield* Deferred.make(); const activeSessionId = yield* Ref.make(null); const contextUsageBySessionId = yield* Ref.make( new Map(), ); const nativeMetadataBySessionId = yield* Ref.make( new Map(), ); const providerThreadByNativeSessionId = yield* Ref.make( new Map(), ); const nativeBuildConfigurationBySessionId = new Map(); const initialSessionActivationFailure = yield* Ref.make<{ readonly sessionId: string; readonly error: EffectAcpErrors.AcpError; } | null>(null); const activeSessionSetup = yield* Ref.make(null); const setActiveSessionSetup = Effect.fnUntraced(function* ( setup: AcpSessionRuntime.AcpSessionRuntimeStartResult, ) { followsAgentWorkState = flavor.followAgentWorkState === true && continuationRequests !== undefined && setup.initializeResult.protocolVersion === 2; agentWork = null; agentRequests.clear(); nativeChildren.clear(); yield* Ref.set(activeSessionSetup, setup); }); const activeSelection = yield* Ref.make(null); const activeInteractionMode = yield* Ref.make(null); const promptInstructionStates = yield* Ref.make(new Map()); // Unique markers keep delayed prompt completions from clearing later compactions. const compactionMarkers = yield* Ref.make(new Map()); const runtimeRestartRequired = yield* Ref.make(false); const runtimeTeardownState = yield* Ref.make({ _tag: "Idle" }); const runtimeCallbackGeneration = yield* Ref.make(0); const runtimeCallbackGenerationCounter = yield* Ref.make(0); const allocateRuntimeCallbackGeneration = Ref.updateAndGet( runtimeCallbackGenerationCounter, (generation) => generation + 1, ); const advanceRuntimeCallbackGeneration = allocateRuntimeCallbackGeneration.pipe( Effect.tap((generation) => Ref.set(runtimeCallbackGeneration, generation)), ); const runtimeCallbackPermit = yield* Semaphore.make(1); const runtimeTransitionPermit = yield* Semaphore.make(1); const nativeResponseAcknowledgements = yield* Ref.make( new Map< string, { readonly acknowledgement: Deferred.Deferred; readonly generation: number; } >(), ); const pendingRuntimeRequests = yield* Ref.make(new Map()); const emitNativeResponseLifecycle = options.testHooks?.onNativeResponseLifecycle ?? (() => Effect.void); const nextElicitationOrdinal = yield* Ref.make(0); const providerTurns = yield* Ref.make(new Map()); const snapshot = yield* Ref.make({ order: [], messages: new Map(), loadingRole: null, loadingMessageId: null, loadingIndex: 0, }); const awaitRuntimeTeardown = Effect.fnUntraced(function* () { const state = yield* Ref.get(runtimeTeardownState); if (state._tag === "InProgress") { yield* Deferred.await(state.completed); } else if (state._tag === "Failed") { return yield* state.error; } }); const runRuntimeCallbackAtGeneration = ( generation: number, effect: Effect.Effect, ) => runtimeCallbackPermit.withPermit( Effect.gen(function* () { if ((yield* Ref.get(runtimeTeardownState))._tag !== "Idle") { return Option.none(); } if ((yield* Ref.get(runtimeCallbackGeneration)) !== generation) { return Option.none(); } return Option.some(yield* effect); }), ); const registerNativeResponseAcknowledgement = /* Provider events include streaming deltas; avoid a span for each update. */ Effect.fnUntraced(function* ( generation: number, transportRequestId: string, acknowledgement: Deferred.Deferred, ) { yield* Ref.update(nativeResponseAcknowledgements, (current) => { const updated = new Map(current); updated.set(transportRequestId, { acknowledgement, generation }); return updated; }); yield* emitNativeResponseLifecycle({ type: "registered", generation, requestId: transportRequestId, }); if (!(yield* Deferred.isDone(acknowledgement))) return; yield* Ref.update(nativeResponseAcknowledgements, (current) => { if (current.get(transportRequestId)?.acknowledgement !== acknowledgement) { return current; } const updated = new Map(current); updated.delete(transportRequestId); return updated; }); }); const acknowledgeNativeResponse = (generation: number, transportRequestId: string) => runRuntimeCallbackAtGeneration(generation, Effect.void).pipe( Effect.flatMap((registered) => Option.isSome(registered) ? Effect.void : emitNativeResponseLifecycle({ type: "admission_rejected", generation, requestId: transportRequestId, }).pipe( Effect.andThen( new EffectAcpErrors.AcpTransportError({ detail: "The ACP runtime closed before its response could be admitted", cause: "Native response registration rejected during teardown", }), ), ), ), ); const awaitNativeResponseAcknowledgements = Effect.fnUntraced(function* ( acknowledgements: ReadonlyArray< readonly [string | undefined, Deferred.Deferred] >, ) { if (acknowledgements.length === 0) return true; const completed = yield* Deferred.make<"settled" | "timeout">(); yield* emitNativeResponseLifecycle({ type: "watcher_started" }); const acknowledgementFiber = yield* Effect.forEach( acknowledgements, ([, acknowledgement]) => Deferred.await(acknowledgement).pipe(Effect.exit), { concurrency: "unbounded", discard: true }, ).pipe( Effect.andThen(Deferred.succeed(completed, "settled")), Effect.interruptible, Effect.ensuring(emitNativeResponseLifecycle({ type: "watcher_exited" })), Effect.forkDetach, ); yield* emitNativeResponseLifecycle({ type: "timer_started" }); const timerFiber = yield* Effect.sleep("2 seconds").pipe( Effect.andThen(Deferred.succeed(completed, "timeout")), Effect.interruptible, Effect.ensuring(emitNativeResponseLifecycle({ type: "timer_exited" })), Effect.forkDetach, ); const outcome = yield* Deferred.await(completed); yield* Fiber.interrupt(acknowledgementFiber); yield* Fiber.interrupt(timerFiber); if (outcome === "settled") return true; yield* Ref.update(nativeResponseAcknowledgements, (current) => { const updated = new Map(current); for (const [requestId, acknowledgement] of acknowledgements) { if ( requestId !== undefined && updated.get(requestId)?.acknowledgement === acknowledgement ) { updated.delete(requestId); } } return updated; }); yield* Effect.forEach( acknowledgements, ([requestId]) => emitNativeResponseLifecycle({ type: "removed", ...(requestId === undefined ? {} : { requestId }), }), { concurrency: "unbounded", discard: true }, ); const timeoutError = new EffectAcpErrors.AcpTransportError({ detail: "Timed out waiting for an admitted ACP response to reach the transport queue", cause: "Native response acknowledgement timed out", }); yield* Effect.forEach( acknowledgements, ([requestId, acknowledgement]) => Deferred.fail(acknowledgement, timeoutError).pipe( Effect.andThen( emitNativeResponseLifecycle({ type: "failed", ...(requestId === undefined ? {} : { requestId }), }), ), ), { concurrency: "unbounded", discard: true }, ); return false; }); const awaitAdmittedNativeResponses = Ref.get(nativeResponseAcknowledgements).pipe( Effect.flatMap((current) => awaitNativeResponseAcknowledgements( [...current.entries()].map( ([requestId, entry]) => [requestId, entry.acknowledgement] as const, ), ), ), ); const quarantineNativeTransportAtGeneration = Effect.fnUntraced(function* ( generation: number, ) { const quarantined = yield* Ref.modify(nativeResponseAcknowledgements, (current) => { const updated = new Map(current); const acknowledgements: Array> = []; for (const [requestId, entry] of updated) { if (entry.generation !== generation) continue; updated.delete(requestId); acknowledgements.push(entry.acknowledgement); } return [acknowledgements, updated] as const; }); const error = new EffectAcpErrors.AcpTransportError({ detail: "The ACP runtime was replaced before its response reached the transport queue", cause: "ACP runtime transport was quarantined during teardown", }); yield* Effect.forEach( quarantined, (acknowledgement) => Deferred.fail(acknowledgement, error), { concurrency: "unbounded", discard: true }, ); }); const closeNativeTransport = runtimeCallbackPermit.withPermit( Effect.gen(function* () { yield* advanceRuntimeCallbackGeneration; const acknowledgements = yield* Ref.getAndSet( nativeResponseAcknowledgements, new Map(), ); const error = new EffectAcpErrors.AcpTransportError({ detail: "The ACP session closed before its admitted response reached the transport", cause: "ACP session transport closed", }); yield* Effect.forEach( acknowledgements, ([requestId, entry]) => Deferred.fail(entry.acknowledgement, error).pipe( Effect.andThen( emitNativeResponseLifecycle({ type: "removed", generation: entry.generation, requestId, }), ), Effect.andThen( emitNativeResponseLifecycle({ type: "failed", generation: entry.generation, requestId, }), ), ), { concurrency: "unbounded", discard: true }, ); return acknowledgements.size > 0; }), ); const lastTurnRoute = yield* Ref.make<{ readonly threadId: ThreadId; readonly providerThreadId: ProviderThreadId; } | null>(null); const wakeBuffer = yield* Ref.make>([]); const noWakeReports = { reports: new Map(), offered: new Set(), }; const wakeReports = yield* Ref.make<{ readonly reports: ReadonlyMap; readonly offered: ReadonlySet; }>(noWakeReports); const recordWakeReport = Effect.fnUntraced(function* ( key: string, report: BackgroundWorkReport, ) { const context = yield* Ref.get(activeTurn); if (context !== null && !context.promptSettled && !context.finalized) return; yield* Ref.update(wakeReports, ({ reports, offered }) => ({ reports: new Map(reports).set(key, report), offered, })); }); const continuationRequested = yield* Ref.make(false); const continuationGeneration = yield* Ref.make(0); const continuationPermit = yield* Semaphore.make(1); const continuationClosed = yield* Ref.make(false); // Quarantine stopped-run events so they cannot wake or attach to later runs. const stoppedRunQuarantine = yield* Ref.make(false); // Native subagents can outlive interrupts; carry lineage forward to finish their original turn items. const carryoverSubagents = yield* Ref.make(null); const handledBackgroundTaskIdsInActiveTurn = yield* Ref.make>( new Set(), ); const midTurnUnreportedCompletedTaskIds = yield* Ref.make>(new Set()); // Grok re-prompts on stale monitor notices; suppress output until a genuine terminal or app turn. const suppressPostSettleMonitorPrompt = yield* Ref.make(false); const runningBackgroundTaskIds = yield* Ref.make>(new Set()); // Tombstones prevent mid-stream exit_code 0 and stale monitor events from resurrecting ended tasks. const endedBackgroundTaskIds = yield* Ref.make>(new Set()); const endedBackgroundTaskIdLimit = 128; const setBackgroundTaskRunning = /* Provider events include streaming deltas; avoid a span for each update. */ Effect.fnUntraced(function* (taskId: string, running: boolean) { if (running && (yield* Ref.get(endedBackgroundTaskIds)).has(taskId)) { return; } yield* Ref.update(runningBackgroundTaskIds, (current) => { if (current.has(taskId) === running) return current; const next = new Set(current); if (running) { next.add(taskId); } else { next.delete(taskId); } return next; }); }); const markBackgroundTaskEnded = (taskId: string) => Ref.update(endedBackgroundTaskIds, (current) => { if (current.has(taskId)) return current; const next = new Set(current).add(taskId); for (const oldest of next) { if (next.size <= endedBackgroundTaskIdLimit) break; next.delete(oldest); } return next; }).pipe(Effect.andThen(setBackgroundTaskRunning(taskId, false))); const applyBackgroundTaskMutationRunning = (mutation: { readonly taskId: string; readonly status: "running" | "completed" | "failed"; }) => mutation.status === "running" ? setBackgroundTaskRunning(mutation.taskId, true) : markBackgroundTaskEnded(mutation.taskId); const trackRunningBackgroundTools = /* Provider events include streaming deltas; avoid a span for each update. */ Effect.fnUntraced(function* ( notification: EffectAcpSchema.SessionNotification, ): Effect.fn.Return { if (flavor.extractBackgroundTaskId === undefined) return; for (const event of parseSessionUpdateEvent(notification).events) { if (event._tag !== "ToolCallUpdated") continue; const toolCall = flavor.normalizeToolCall?.(event.toolCall) ?? event.toolCall; const taskId = flavor.extractBackgroundTaskId(toolCall); if (taskId === undefined) continue; const status = toolStatus(toolCall.status); yield* setBackgroundTaskRunning(taskId, status === "pending" || status === "running"); } }); const emitProviderEvent = (event: ProviderAdapter.ProviderAdapterV2Event) => Queue.offer(events, event).pipe(Effect.asVoid); let scheduleDeferredFinalize: (context: ActiveAcpTurn) => Effect.Effect = () => Effect.void; const nativeLogging = options.nativeLogging?.(input.threadId); const handleRuntimeTerminationAtGeneration = (runtimeGeneration: number) => runRuntimeCallbackAtGeneration( runtimeGeneration, Effect.gen(function* () { yield* Ref.set(runtimeRestartRequired, true); if (!followsAgentWorkState) return; agentWork = { sessionUpdate: "state_update", state: "idle", stopReason: "error" }; for (const child of nativeChildren.values()) { if (child.status !== "running") continue; child.status = "failed"; child.result = null; } for (const child of nativeChildren.values()) { if ( child.subagent === undefined || acpSubagentStatusIsTerminal(child.subagent.task.status) ) continue; child.result = null; yield* publishNativeChildStatus(child, "failed"); } agentRequests.clear(); yield* Ref.set(wakeBuffer, []); yield* Ref.set(continuationRequested, false); yield* cancelPendingRuntimeRequests(); const context = yield* Ref.get(activeTurn); if (context !== null) yield* finalizeTurn( context, "failed", makeProviderFailure({ class: "transport_error" }), ); }), ).pipe(Effect.asVoid); const makeRuntimeInput = ( runtimeGeneration: number, mcpContext: AcpMcpContext, resumeSessionId?: string, onTermination: AcpAdapterV2RuntimeInput["onTermination"] = () => handleRuntimeTerminationAtGeneration(runtimeGeneration), ): AcpAdapterV2RuntimeInput => { return { cwd: input.runtimePolicy.cwd ?? process.cwd(), runtimePolicy: input.runtimePolicy, mcpServers: mcpContext.servers, acpMcpServers: mcpContext.acpServers, ...(mcpContext.processEnvironment === undefined ? {} : { processEnvironment: mcpContext.processEnvironment }), ...(resumeSessionId === undefined ? {} : { resumeSessionId }), interruptPromptOnCancel: flavor.interruptPromptOnCancel ?? false, clientCapabilities: { fs: { readTextFile: flavor.clientFileSystem !== undefined, writeTextFile: flavor.clientFileSystem?.writeTextFile !== undefined, }, terminal: clientTerminals !== undefined, elicitation: { form: {}, ...(flavor.onUrlElicitation ? { url: {} } : {}) }, ...(flavor.clientCapabilitiesMeta ? { _meta: flavor.clientCapabilitiesMeta } : {}), }, clientInfo: { name: "t3-code", version: "0.0.0" }, onTermination, onOutgoingResponseFailure: (requestId, error) => Ref.modify(nativeResponseAcknowledgements, (current) => { const entry = current.get(requestId); if (entry === undefined || entry.generation !== runtimeGeneration) { return [ emitNativeResponseLifecycle({ type: "late_noop", generation: runtimeGeneration, requestId, }), current, ] as const; } const updated = new Map(current); updated.delete(requestId); return [ Deferred.fail(entry.acknowledgement, error).pipe( Effect.andThen( emitNativeResponseLifecycle({ type: "removed", generation: runtimeGeneration, requestId, }), ), Effect.asVoid, ), updated, ] as const; }).pipe(Effect.flatten), onOutgoingResponse: (requestId) => Ref.modify(nativeResponseAcknowledgements, (current) => { const entry = current.get(requestId); if (entry === undefined || entry.generation !== runtimeGeneration) { return [ emitNativeResponseLifecycle({ type: "late_noop", generation: runtimeGeneration, requestId, }), current, ] as const; } const updated = new Map(current); updated.delete(requestId); return [ Deferred.succeed(entry.acknowledgement, undefined).pipe( Effect.andThen( emitNativeResponseLifecycle({ type: "removed", generation: runtimeGeneration, requestId, }), ), Effect.asVoid, ), updated, ] as const; }).pipe(Effect.flatten), ...(nativeLogging?.requestLogger === undefined ? {} : { requestLogger: nativeLogging.requestLogger }), protocolLogging: nativeLogging?.protocolLogging ?? { logIncoming: true, logOutgoing: true, logger: () => Effect.void, }, }; }; let runtimeScope: Scope.Closeable | undefined; let runtime!: AcpSessionRuntime.AcpSessionRuntime["Service"]; let runtimeMcpBridge: AcpMcpOverAcpBridge | undefined; yield* Effect.addFinalizer(() => runtimeScope === undefined ? Effect.void : Scope.close(runtimeScope, Exit.void).pipe(Effect.ignore), ); const resolveItemOrdinal = (context: ActiveAcpTurn, nativeItemId: string) => Effect.sync(() => { const existing = context.itemOrdinals.get(nativeItemId); if (existing !== undefined) return existing; const ordinal = context.input.providerTurnOrdinal * 100 + context.itemOrdinals.size + 1; context.itemOrdinals.set(nativeItemId, ordinal); return ordinal; }); const resolveSubagentChildOrdinal = (subagent: ActiveAcpSubagent, nativeItemId: string) => { const existing = subagent.childItemOrdinals.get(nativeItemId); if (existing !== undefined) return existing; const ordinal = subagent.nextChildOrdinal++; subagent.childItemOrdinals.set(nativeItemId, ordinal); return ordinal; }; const emitProviderRetry = Effect.fnUntraced(function* ( context: ActiveAcpTurn, status: "running" | "completed" | "interrupted" | "cancelled", ) { const state = context.providerRetry; if (state === undefined) return; yield* emitProviderEvent({ type: "turn_item.updated", driver, turnItem: makeProviderRetryTurnItem({ driver, threadId: context.input.threadId, runId: context.input.runId, nodeId: context.input.rootNodeId, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, ...state, status, updatedAt: yield* DateTime.now, }), }); if (status !== "running") context.providerRetry = undefined; }); const reportProviderRetry = Effect.fnUntraced(function* (notice: { readonly sessionId: string; readonly failure: OrchestrationV2ProviderFailure; }) { const context = yield* Ref.get(activeTurn); if ( context === null || context.finalized || context.interrupted || context.nativeThreadId !== notice.sessionId || (yield* Ref.get(stoppedRunQuarantine)) ) return; const previous = context.providerRetry; context.providerRetry = { failure: notice.failure, retry: { attempt: (previous?.retry.attempt ?? 0) + 1, maxAttempts: null, retryDelayMs: null, }, startedAt: previous?.startedAt ?? (yield* DateTime.now), itemOrdinal: previous?.itemOrdinal ?? (yield* resolveItemOrdinal(context, `terminal-failure:${context.providerTurnId}`)), }; yield* emitProviderRetry(context, "running"); }); const lastCapturedProposedPlan = yield* Ref.make<{ readonly nativeTurnId: string; readonly markdown: string; readonly planId: OrchestrationV2PlanArtifact["id"]; } | null>(null); const captureProposedPlan = Effect.fnUntraced(function* (input: { readonly planMarkdown: string; }) { const context = yield* Ref.get(activeTurn); if (context === null) return; const markdown = input.planMarkdown.trim(); if (markdown.length === 0) return; const previous = yield* Ref.get(lastCapturedProposedPlan); if ( previous !== null && previous.nativeTurnId === context.nativeTurnId && previous.markdown === markdown ) { return; } const nativeItemId = `${context.nativeTurnId}:proposed-plan`; const planId = previous !== null && previous.nativeTurnId === context.nativeTurnId ? previous.planId : yield* idAllocator.allocate .plan({ threadId: context.input.threadId, runId: context.input.runId, driver, }) .pipe(Effect.orDie); yield* Ref.set(lastCapturedProposedPlan, { nativeTurnId: context.nativeTurnId, markdown, planId, }); const now = yield* DateTime.now; const nodeId = idAllocator.derive.nodeFromProviderItem({ driver, nativeItemId }); const turnItemId = idAllocator.derive.turnItemFromProviderItem({ driver, nativeItemId }); const ordinal = yield* resolveItemOrdinal(context, nativeItemId); const nativeItemRef = { driver, nativeId: nativeItemId, strength: "weak" as const }; yield* emitProviderEvent({ type: "node.updated", driver, node: { id: nodeId, threadId: context.input.threadId, runId: context.input.runId, parentNodeId: context.input.rootNodeId, rootNodeId: context.input.rootNodeId, kind: "plan", status: "completed", countsForRun: false, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, nativeItemRef, runtimeRequestId: null, checkpointScopeId: null, startedAt: context.startedAt, completedAt: now, }, }); yield* emitProviderEvent({ type: "plan.updated", driver, plan: { id: planId, threadId: context.input.threadId, runId: context.input.runId, nodeId, kind: "proposed_plan", status: "completed", markdown, }, }); yield* emitProviderEvent({ type: "turn_item.updated", driver, turnItem: { id: turnItemId, threadId: context.input.threadId, runId: context.input.runId, nodeId, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, nativeItemRef, parentItemId: null, ordinal, status: "completed", title: null, startedAt: context.startedAt, completedAt: now, updatedAt: now, type: "proposed_plan", planId, markdown, streaming: false, }, }); }); const lastProposedPlanMarkdown = Effect.gen(function* () { const context = yield* Ref.get(activeTurn); const previous = yield* Ref.get(lastCapturedProposedPlan); return context !== null && previous?.nativeTurnId === context.nativeTurnId ? previous.markdown : undefined; }); const rememberSnapshotMessage = (message: OrchestrationV2ConversationMessage) => Ref.update(snapshot, (current) => { const key = String(message.id); const exists = current.messages.has(key); const messages = new Map(current.messages); messages.set(key, message); return { ...current, order: exists ? current.order : [...current.order, key], messages, }; }); const textStreamFor = ( context: ActiveAcpTurn, kind: "user" | "assistant" | "reasoning", ): ActiveTextStream => kind === "user" ? context.user : kind === "assistant" ? context.assistant : context.reasoning; const textEmissionPermit = yield* Semaphore.make(1); /** * Emits the stream's current segment. A `flushOf` call is a delayed * flush: it emits only while that segment is still current and has * unsent text. The permit keeps a flush from landing after the * segment's completed snapshot. */ const emitTextSegment = Effect.fnUntraced(function* ( context: ActiveAcpTurn, kind: "user" | "assistant" | "reasoning", completed: boolean, flushOf?: ActiveTextSegment, ) { const stream = textStreamFor(context, kind); const segment = stream.current; if (segment === null) return; if (flushOf !== undefined && (segment !== flushOf || !segment.dirty)) return; // The node carries no text, so it changes only when the segment // opens and when it completes. const emitsNode = completed || !segment.emitted; segment.emitted = true; segment.dirty = false; // Read the text in the same step that clears `dirty`. Chunks keep // appending while this emission yields; each one marks the segment // dirty again and reaches the next flush or the close. const text = segment.text; const now = yield* DateTime.now; const ordinal = yield* resolveItemOrdinal(context, segment.nativeItemId); const nodeId = providerNodeId(segment.nativeItemId); const turnItemId = providerTurnItemId(segment.nativeItemId); const nativeItemRef = { driver, nativeId: segment.nativeItemId, strength: "weak" as const, }; if (kind !== "user" && emitsNode) { yield* emitProviderEvent({ type: "node.updated", driver, node: { id: nodeId, threadId: context.input.threadId, runId: context.input.runId, parentNodeId: context.input.rootNodeId, rootNodeId: context.input.rootNodeId, kind: kind === "assistant" ? "assistant_message" : "reasoning", status: completed ? "completed" : "running", countsForRun: false, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, nativeItemRef, runtimeRequestId: null, checkpointScopeId: null, startedAt: segment.startedAt, completedAt: completed ? now : null, }, }); } if (kind !== "reasoning") { const messageId = providerMessageId(segment.nativeItemId); const messageNodeId = kind === "user" ? context.input.rootNodeId : nodeId; const message: OrchestrationV2ConversationMessage = { createdBy: kind === "user" ? "user" : "agent", creationSource: "provider", id: messageId, threadId: context.input.threadId, runId: context.input.runId, nodeId: messageNodeId, role: kind, text, attachments: [], streaming: !completed, createdAt: segment.startedAt, updatedAt: now, }; yield* emitProviderEvent({ type: "message.updated", driver, message }); yield* emitProviderEvent( kind === "user" ? { type: "turn_item.updated", driver, turnItem: { createdBy: "user", creationSource: "provider", id: turnItemId, threadId: context.input.threadId, runId: context.input.runId, nodeId: messageNodeId, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, nativeItemRef, parentItemId: null, ordinal, status: completed ? "completed" : "running", title: null, startedAt: segment.startedAt, completedAt: completed ? now : null, updatedAt: now, type: "user_message", messageId, inputIntent: "turn_start", text, attachments: [], }, } : { type: "turn_item.updated", driver, turnItem: { id: turnItemId, threadId: context.input.threadId, runId: context.input.runId, nodeId, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, nativeItemRef, parentItemId: null, ordinal, status: completed ? "completed" : "running", title: null, startedAt: segment.startedAt, completedAt: completed ? now : null, updatedAt: now, type: "assistant_message", messageId, text, streaming: !completed, }, }, ); if (completed) yield* rememberSnapshotMessage(message); return; } yield* emitProviderEvent({ type: "turn_item.updated", driver, turnItem: { id: turnItemId, threadId: context.input.threadId, runId: context.input.runId, nodeId, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, nativeItemRef, parentItemId: null, ordinal, status: completed ? "completed" : "running", title: null, startedAt: segment.startedAt, completedAt: completed ? now : null, updatedAt: now, type: "reasoning", text, streaming: !completed, }, }); }, textEmissionPermit.withPermit); /** * Publishes a text change. A segment's first text emits at once so * live output starts immediately; later changes coalesce into one * snapshot per ACP_TEXT_FLUSH_INTERVAL. Closing the segment emits its * final text and makes a pending flush a no-op. */ const publishTextSegment = Effect.fnUntraced(function* ( context: ActiveAcpTurn, kind: "user" | "assistant" | "reasoning", ) { const segment = textStreamFor(context, kind).current; if (segment === null) return; if (!segment.emitted) return yield* emitTextSegment(context, kind, false); if (segment.dirty) return; segment.dirty = true; yield* emitTextSegment(context, kind, false, segment).pipe( Effect.delay(ACP_TEXT_FLUSH_INTERVAL), Effect.forkIn(sessionScope), ); }); const closeTextStream = Effect.fnUntraced(function* ( context: ActiveAcpTurn, kind: "user" | "assistant" | "reasoning", ) { const stream = textStreamFor(context, kind); if (stream.current === null) return; yield* emitTextSegment(context, kind, true); stream.current = null; }); const closeTextStreams = Effect.fnUntraced(function* (context: ActiveAcpTurn) { yield* closeTextStream(context, "user"); yield* closeTextStream(context, "reasoning"); yield* closeTextStream(context, "assistant"); }); const appendText = Effect.fnUntraced(function* ( context: ActiveAcpTurn, kind: "user" | "assistant" | "reasoning", text: string, messageId?: string | null, ) { if (text.length === 0) return; for (const other of ["user", "reasoning", "assistant"] as const) { if (other !== kind) yield* closeTextStream(context, other); } const stream = textStreamFor(context, kind); const sourceMessageId = messageId?.trim() || null; if ( stream.current !== null && sourceMessageId !== null && stream.current.sourceMessageId !== null && stream.current.sourceMessageId !== sourceMessageId ) { yield* closeTextStream(context, kind); } if (stream.current === null) { const now = yield* DateTime.now; stream.current = { nativeItemId: sourceMessageId === null ? `${context.nativeTurnId}:${kind}:${stream.nextSegment}` : `${context.nativeTurnId}:${kind}:message:${sourceMessageId}`, startedAt: now, sourceMessageId, text: "", emitted: false, dirty: false, }; stream.nextSegment += 1; } else if (stream.current.sourceMessageId === null && sourceMessageId !== null) { // Agents may omit the first messageId; retain projected identity until the first later id. stream.current.sourceMessageId = sourceMessageId; } stream.current.text += text; yield* publishTextSegment(context, kind); }); const replaceText = Effect.fnUntraced(function* ( context: ActiveAcpTurn, kind: "user" | "assistant" | "reasoning", text: string, messageId: string, ) { for (const other of ["user", "reasoning", "assistant"] as const) { if (other !== kind) yield* closeTextStream(context, other); } const stream = textStreamFor(context, kind); if (stream.current?.sourceMessageId !== messageId) { yield* closeTextStream(context, kind); const now = yield* DateTime.now; stream.current = { nativeItemId: `${context.nativeTurnId}:${kind}:message:${messageId}`, startedAt: now, sourceMessageId: messageId, text, emitted: false, dirty: false, }; stream.nextSegment += 1; } else { stream.current.text = text; } yield* publishTextSegment(context, kind); }); const emitSubagentAssistant = Effect.fnUntraced(function* ( subagent: ActiveAcpSubagent, text: string, mode: "append" | "replace" = "append", messageId?: string | null, ) { if (text.length === 0 && mode === "append") return; const nativeItemId = `${subagent.task.nativeTaskRef?.nativeId ?? subagent.task.id}:message:${messageId ?? "result"}`; const previous = subagent.assistantMessages.get(nativeItemId) ?? ""; const messageText = mode === "replace" ? text : `${previous}${text}`; subagent.assistantMessages.set(nativeItemId, messageText); subagent.assistantText = messageText; const now = yield* DateTime.now; const ordinal = resolveSubagentChildOrdinal(subagent, nativeItemId); const artifacts = makeSubagentConversationArtifacts({ messageId: providerMessageId(nativeItemId), turnItemId: providerTurnItemId(nativeItemId), threadId: subagent.childThreadId, rootNodeId: subagent.childRootNodeId, providerThreadId: subagent.task.providerThreadId, providerTurnId: null, nativeItemRef: { driver, nativeId: nativeItemId, strength: "weak" }, role: "assistant", text: subagent.assistantText, ordinal, now, }); yield* emitProviderEvent({ type: "message.updated", driver, message: artifacts.message }); yield* emitProviderEvent({ type: "turn_item.updated", driver, turnItem: artifacts.turnItem, }); }); const projectSubagentNotification = Effect.fnUntraced(function* ( subagent: ActiveAcpSubagent, notification: EffectAcpSchema.SessionNotification, ) { const update = notification.update; if (update.sessionUpdate === "agent_message_chunk") { const text = acpContentBlockDisplayText(update.content); if (text !== undefined) { yield* emitSubagentAssistant(subagent, text, "append", update.messageId); } } else if (update.sessionUpdate === "agent_message") { if (update.content === undefined) return; const text = (update.content ?? []) .flatMap((content) => { const display = acpContentBlockDisplayText(content); return display === undefined ? [] : [display]; }) .join("\n"); yield* emitSubagentAssistant(subagent, text, "replace", update.messageId); } }); const projectSubagentTool = Effect.fnUntraced(function* ( subagent: ActiveAcpSubagent, sessionId: string, toolCall: AcpToolCallState, projectedStatus?: ProjectedToolStatus, ) { const nativeTaskId = subagent.task.nativeTaskRef?.nativeId ?? String(subagent.task.id); const key = sessionId === subagent.childSessionId ? `${nativeTaskId}:tool:${toolCall.toolCallId}` : `${nativeTaskId}:session:${sessionId}:tool:${toolCall.toolCallId}`; const context = subagent.toolState; const rootSessionId = yield* Ref.get(activeSessionId); subagent.toolSessionIds.set(key, sessionId); const previous = context.tools.get(key); const merged = mergeToolCallState(previous, toolCall); context.tools.set(key, merged); if ( projectedStatus === undefined && !shouldPersistToolUpdate(context, key, previous, merged, merged.status) ) return; const mcpIdentity = extractMcpToolCallIdentity(merged, { embeddedTerminalCommands: [ ...embeddedTerminalCommands(sessionId, toolCall.toolCallId), ...(rootSessionId === null ? [] : embeddedTerminalCommands(rootSessionId, toolCall.toolCallId)), ], }); const now = yield* DateTime.now; const status = projectedStatus ?? toolStatus(merged.status); const startedAt = context.toolStartedAt.get(key) ?? now; context.toolStartedAt.set(key, startedAt); const ordinal = resolveSubagentChildOrdinal(subagent, key); yield* emitProviderEvent({ type: "turn_item.updated", driver, turnItem: { id: providerTurnItemId(key), threadId: subagent.childThreadId, runId: null, nodeId: subagent.childRootNodeId, providerThreadId: subagent.task.providerThreadId, providerTurnId: null, nativeItemRef: { driver, nativeId: key, strength: "strong" }, parentItemId: null, ordinal, status, title: merged.title ?? merged.kind ?? "Tool", startedAt, completedAt: completedAtForStatus(status, now), updatedAt: now, type: "dynamic_tool", ...(mcpIdentity === undefined ? {} : mcpToolPresentation({ serverName: mcpIdentity.server, toolName: mcpIdentity.tool, source: unknownRecord( ( unknownRecord(unknownRecord(merged.data.rawOutput)?.result) ?? unknownRecord(merged.data.rawOutput) )?._meta, )?.source, })), toolName: mcpIdentity === undefined ? (merged.title ?? merged.kind ?? "Tool") : `${mcpIdentity.server}.${mcpIdentity.tool}`, input: merged.data.rawInput ?? null, output: merged.data.rawOutput ?? merged.data.content ?? null, }, }); }); const terminalizeSubagentTools = Effect.fnUntraced(function* (subagent: ActiveAcpSubagent) { for (const [key, tool] of subagent.toolState.tools) { if ( toolStatus(tool.status) !== "running" && toolStatus(tool.status) !== "pending" && toolStatus(tool.status) !== "waiting" ) continue; const sessionId = subagent.toolSessionIds.get(key); if (sessionId !== undefined) yield* projectSubagentTool( subagent, sessionId, { ...tool, status: "failed" }, "interrupted", ); } }); const emitSubagent = Effect.fnUntraced(function* ( context: ActiveAcpTurn, update: AcpAdapterV2SubagentUpdate, ) { // Hydration uses a new toolCallId but references the original child by task id. const existing = context.subagents.get(update.nativeTaskId) ?? (update.childSessionId !== null ? context.subagentsBySessionId.get(update.childSessionId) : undefined); const updateIsTerminal = acpSubagentStatusIsTerminal(update.status); if ( existing !== undefined && acpSubagentStatusIsTerminal(existing.task.status) && !updateIsTerminal ) { return; } if ( existing !== undefined && existing.task.status === update.status && updateIsTerminal && existing.terminalStatusProjected ) { return; } // Empty-prompt hydration must match existing lineage; only spawn ACKs can create it. if (existing === undefined) { const isHydrationOnly = update.prompt === "" && update.title === null; if (isHydrationOnly || update.suppressNormalTool === false) { return; } } const now = yield* DateTime.now; const nativeTaskId = existing?.task.nativeTaskRef?.nativeId ?? update.nativeTaskId; const nativeItemRef = { driver, nativeId: nativeTaskId, strength: "strong" as const, }; const nodeId = existing?.task.id ?? providerNodeId(nativeTaskId); const childThreadId = existing?.childThreadId ?? idAllocator.derive.threadFromProviderThread({ driver, providerInstanceId: context.input.modelSelection.instanceId, nativeThreadId: `${context.nativeThreadId}:task:${nativeTaskId}`, }); const childRootNodeId = existing?.childRootNodeId ?? providerNodeId(`${nativeTaskId}:child-root`); const turnItemId = existing?.turnItemId ?? providerTurnItemId(nativeTaskId); const turnItemOrdinal = existing?.turnItemOrdinal ?? (yield* resolveItemOrdinal(context, nativeTaskId)); const taskStatus = update.status; const task: OrchestrationV2Subagent = { ...(existing?.task ?? { id: nodeId, threadId: context.input.threadId, runId: context.input.runId, parentNodeId: context.input.rootNodeId, origin: "provider_native" as const, createdBy: "agent" as const, driver, providerInstanceId: context.input.modelSelection.instanceId, providerThreadId: null, childThreadId, nativeTaskRef: nativeItemRef, prompt: update.prompt, title: update.title, result: null, startedAt: now, }), model: update.model?.trim() || existing?.task.model || null, status: taskStatus, result: update.result ?? existing?.task.result ?? null, completedAt: acpSubagentStatusIsTerminal(taskStatus) ? now : null, updatedAt: now, }; const subagent: ActiveAcpSubagent = existing ?? { task, childThreadId, childRootNodeId, turnItemId, turnItemOrdinal, providerTurnId: context.providerTurnId, parentProviderThreadId: context.input.providerThread.id, childSessionId: null, assistantText: "", assistantMessages: new Map(), childItemOrdinals: new Map(), nextChildOrdinal: 101, terminalStatusProjected: false, toolState: { tools: new Map(), toolUpdatesSkipped: new Map(), toolStartedAt: new Map(), }, toolSessionIds: new Map(), }; subagent.task = task; context.subagents.set(nativeTaskId, subagent); if (existing === undefined) { yield* emitProviderEvent({ type: "app_thread.created", driver, appThread: makeSubagentChildThread({ parentThread: context.input.appThread, childThreadId, parentNodeId: nodeId, activeProviderThreadId: null, providerInstanceId: context.input.modelSelection.instanceId, modelSelection: { ...context.input.modelSelection, model: task.model ?? context.input.modelSelection.model, }, title: subagentThreadTitle({ parentTitle: context.input.appThread.title, title: update.title, prompt: update.prompt, ordinal: context.subagents.size, }), now, createdBy: "agent", creationSource: "provider", }), }); const promptNativeItemId = `${nativeTaskId}:prompt`; const promptArtifacts = makeSubagentConversationArtifacts({ senderThreadId: context.input.threadId, messageId: providerMessageId(promptNativeItemId), turnItemId: providerTurnItemId(promptNativeItemId), threadId: childThreadId, rootNodeId: childRootNodeId, providerThreadId: null, providerTurnId: null, nativeItemRef: { driver, nativeId: promptNativeItemId, strength: "weak" }, role: "user", text: update.prompt, ordinal: 100, now, }); yield* emitProviderEvent({ type: "message.updated", driver, message: promptArtifacts.message, }); yield* emitProviderEvent({ type: "turn_item.updated", driver, turnItem: promptArtifacts.turnItem, }); } const childSessionId = update.childSessionId; if (childSessionId !== null && subagent.childSessionId === null) { subagent.childSessionId = childSessionId; context.subagentsBySessionId.set(childSessionId, subagent); const providerThread = makeProviderThread({ driver, providerInstanceId: context.input.modelSelection.instanceId, idAllocator, appThreadId: childThreadId, providerSessionId: input.providerSessionId, nativeThreadId: childSessionId, ...(itemIdentityVersion === undefined ? {} : { itemIdentityVersion }), forkedFrom: { providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, }, now, }); subagent.task = { ...subagent.task, providerThreadId: providerThread.id }; yield* Ref.update(providerThreadByNativeSessionId, (current) => new Map(current).set(childSessionId, providerThread), ); yield* emitProviderEvent({ type: "provider_thread.updated", driver, providerThread: { ...providerThread, status: "idle" }, }); const buffered = context.pendingSubagentNotifications.get(childSessionId) ?? []; context.pendingSubagentNotifications.delete(childSessionId); yield* Effect.forEach( buffered, (notification) => projectSubagentNotification(subagent, notification), { concurrency: 1, discard: true }, ); } if ( taskStatus !== "running" && subagent.assistantText.length === 0 && update.result !== null ) { yield* emitSubagentAssistant(subagent, update.result); } const result = update.result ?? subagent.task.result ?? (subagent.assistantText || null); subagent.task = { ...subagent.task, status: taskStatus, result, completedAt: acpSubagentStatusIsTerminal(taskStatus) ? now : null, updatedAt: now, }; const providerThreadId = subagent.task.providerThreadId; yield* emitProviderEvent({ type: "node.updated", driver, node: { id: nodeId, threadId: context.input.threadId, runId: subagent.task.runId, parentNodeId: subagent.task.parentNodeId, rootNodeId: subagent.task.parentNodeId, kind: "subagent", status: taskStatus, countsForRun: false, providerThreadId: context.input.providerThread.id, providerTurnId: subagent.providerTurnId, nativeItemRef, runtimeRequestId: null, checkpointScopeId: null, startedAt: subagent.task.startedAt, completedAt: subagent.task.completedAt, }, }); yield* emitProviderEvent({ type: "node.updated", driver, node: { id: childRootNodeId, threadId: childThreadId, runId: null, parentNodeId: null, rootNodeId: childRootNodeId, kind: "root_turn", status: taskStatus, countsForRun: false, providerThreadId, providerTurnId: null, nativeItemRef, runtimeRequestId: null, checkpointScopeId: null, startedAt: subagent.task.startedAt, completedAt: subagent.task.completedAt, }, }); yield* emitProviderEvent({ type: "subagent.updated", driver, subagent: subagent.task }); yield* emitProviderEvent({ type: "turn_item.updated", driver, turnItem: { id: turnItemId, threadId: context.input.threadId, runId: subagent.task.runId, nodeId, providerThreadId: context.input.providerThread.id, providerTurnId: subagent.providerTurnId, nativeItemRef, parentItemId: null, ordinal: turnItemOrdinal, status: taskStatus, title: subagent.task.title, startedAt: subagent.task.startedAt, completedAt: subagent.task.completedAt, updatedAt: now, type: "subagent", subagentId: subagent.task.id, origin: "provider_native", driver, providerInstanceId: context.input.modelSelection.instanceId, childThreadId, prompt: subagent.task.prompt, result, }, }); if (acpSubagentStatusIsTerminal(taskStatus)) { subagent.terminalStatusProjected = true; } }); const nativeChild = (sessionId: string) => { let child = nativeChildren.get(sessionId); if (child === undefined) { child = { parentSessionId: null, status: "running", result: null, delegationId: null, prompt: "", subagent: undefined, owner: undefined, pending: [], }; nativeChildren.set(sessionId, child); } return child; }; const setNativeChildState = ( child: AcpNativeChildSession, status: AcpNativeChildSession["status"], result: string | null, ) => { if (child.status !== "running" && status === "running" && child.subagent !== undefined) { // A new execution replaces its result; live ownership transfers keep their progress. child.subagent.task = { ...child.subagent.task, result: null }; child.subagent.assistantText = ""; child.subagent.assistantMessages.clear(); } child.status = status; child.result = result; }; const nativeChildOwner = (sessionId: string): AcpNativeChildSession | undefined => { const visited = new Set(); let child = nativeChildren.get(sessionId); while (child !== undefined && !visited.has(sessionId)) { if (child.subagent !== undefined) return child; visited.add(sessionId); if (child.parentSessionId === null) return; sessionId = child.parentSessionId; child = nativeChildren.get(sessionId); } }; const hasWorkingDescendant = (sessionId: string) => { for (const [id, child] of nativeChildren) { if (id === sessionId || child.status !== "running") continue; const visited = new Set(); let parent = child.parentSessionId; while (parent !== null && !visited.has(parent)) { if (parent === sessionId) return true; visited.add(parent); parent = nativeChildren.get(parent)?.parentSessionId ?? null; } } return false; }; const publishNativeChildStatus = Effect.fnUntraced(function* ( child: AcpNativeChildSession, status: AcpAdapterV2SubagentUpdate["status"], ) { const { subagent, owner } = child; if (subagent === undefined || owner === undefined) return; if (acpSubagentStatusIsTerminal(status)) yield* terminalizeSubagentTools(subagent); yield* emitSubagent(owner, { nativeTaskId: subagent.task.nativeTaskRef?.nativeId ?? subagent.task.id, childSessionId: subagent.childSessionId, prompt: subagent.task.prompt, title: subagent.task.title, model: subagent.task.model, status, result: child.result, }); }); const syncNativeChildren = Effect.fnUntraced(function* ( context: ActiveAcpTurn, ): Effect.fn.Return { const rootSessionId = yield* Ref.get(activeSessionId); for (const [sessionId, child] of nativeChildren) { if (child.parentSessionId !== rootSessionId) continue; if (child.subagent === undefined && child.delegationId !== null) { const subagent = context.subagents.get(child.delegationId); if (subagent !== undefined) { child.subagent = subagent; child.owner = context; yield* emitSubagent(context, { nativeTaskId: child.delegationId, childSessionId: sessionId, prompt: subagent.task.prompt, title: subagent.task.title, model: subagent.task.model, status: "running", result: child.result, }); } } const subagent = child.subagent; if (subagent === undefined) continue; const newDelegation = child.delegationId !== null && child.delegationId !== subagent.task.nativeTaskRef?.nativeId && !subagent.childItemOrdinals.has(`${sessionId}:prompt:${child.delegationId}`); if (newDelegation && child.status !== "running") setNativeChildState(child, "running", null); const hasNativeWork = child.status === "running" || hasWorkingDescendant(sessionId); if ( (hasNativeWork && acpSubagentStatusIsTerminal(subagent.task.status)) || (child.owner !== context && (child.owner?.finalizedStatus !== "completed" || newDelegation) && (hasNativeWork || child.pending.length > 0)) ) { // The ownership event precedes child tools so the old run can detach first. child.owner = context; subagent.task = { ...subagent.task, runId: context.input.runId, parentNodeId: context.input.rootNodeId, status: "running", completedAt: null, }; subagent.providerTurnId = context.providerTurnId; subagent.turnItemOrdinal = yield* resolveItemOrdinal( context, subagent.task.nativeTaskRef?.nativeId ?? subagent.task.id, ); subagent.terminalStatusProjected = false; context.subagents.set( subagent.task.nativeTaskRef?.nativeId ?? subagent.task.id, subagent, ); context.subagentsBySessionId.set(sessionId, subagent); yield* publishNativeChildStatus(child, "running"); for (const [key, tool] of subagent.toolState.tools) { const toolSessionId = subagent.toolSessionIds.get(key); if ( toolSessionId === undefined || (toolStatus(tool.status) !== "running" && toolStatus(tool.status) !== "pending") ) continue; subagent.toolState.toolUpdatesSkipped.delete(key); yield* projectSubagentTool(subagent, toolSessionId, tool, toolStatus(tool.status)); } } if ( child.delegationId !== null && child.delegationId !== subagent.task.nativeTaskRef?.nativeId ) { const nativeItemId = `${sessionId}:prompt:${child.delegationId}`; if (!subagent.childItemOrdinals.has(nativeItemId)) { const now = yield* DateTime.now; const artifacts = makeSubagentConversationArtifacts({ senderThreadId: subagent.task.threadId, messageId: providerMessageId(nativeItemId), turnItemId: providerTurnItemId(nativeItemId), threadId: subagent.childThreadId, rootNodeId: subagent.childRootNodeId, providerThreadId: subagent.task.providerThreadId, providerTurnId: null, nativeItemRef: { driver, nativeId: nativeItemId, strength: "weak" }, role: "user", text: child.prompt, ordinal: resolveSubagentChildOrdinal(subagent, nativeItemId), now, }); yield* emitProviderEvent({ type: "message.updated", driver, message: artifacts.message, }); yield* emitProviderEvent({ type: "turn_item.updated", driver, turnItem: artifacts.turnItem, }); } } } for (const [sessionId, child] of nativeChildren) { const owner = nativeChildOwner(sessionId); if (owner?.owner !== context && owner?.owner?.finalizedStatus !== "completed") continue; for (const notification of child.pending.splice(0)) yield* projectNativeChildFrame(notification); if ( child.subagent?.childSessionId != null && child.status !== "running" && !hasWorkingDescendant(child.subagent.childSessionId) ) { yield* publishNativeChildStatus(child, child.status); } } }); const projectNativeChildFrame = Effect.fnUntraced(function* ( notification: EffectAcpSchema.SessionNotification, ): Effect.fn.Return { const child = nativeChild(notification.sessionId); const owner = nativeChildOwner(notification.sessionId); const subagent = owner?.subagent; if ( subagent === undefined || owner?.owner === undefined || (owner.owner.finalized && owner.owner.finalizedStatus !== "completed") || (acpSubagentStatusIsTerminal(subagent.task.status) && child.status === "running") ) { if (child.pending.length < 256) child.pending.push(notification); if (subagent !== undefined) { const root = yield* Ref.get(activeSessionId); if (root !== null) yield* offerContinuationRun(root); } return; } const update = notification.update; if (update.sessionUpdate === "tool_call" || update.sessionUpdate === "tool_call_update") { for (const event of parseSessionUpdateEvent(notification).events) { if (event._tag === "ToolCallUpdated") { yield* projectSubagentTool( subagent, notification.sessionId, flavor.normalizeToolCall?.(event.toolCall) ?? event.toolCall, ); } } } else { yield* projectSubagentNotification(subagent, notification); } }); const followNativeChild = Effect.fnUntraced(function* ( notification: EffectAcpSchema.SessionNotification, context: ActiveAcpTurn | null, ): Effect.fn.Return { if (!followsAgentWorkState || flavor.extractSubagentSessionEvent === undefined) return false; const root = yield* Ref.get(activeSessionId); const event = flavor.extractSubagentSessionEvent(notification); if (event !== undefined) { if (yield* Ref.get(stoppedRunQuarantine)) return true; const child = nativeChild(event.childSessionId); child.parentSessionId = event.parentSessionId ?? child.parentSessionId; if (event.type === "state") { setNativeChildState(child, event.status, event.result); } else { child.delegationId = event.delegationId; child.prompt = event.prompt; } if (context !== null) yield* syncNativeChildren(context); const owner = nativeChildOwner(event.childSessionId); if (owner?.subagent !== undefined && owner.owner !== undefined) { const sessionId = owner.subagent.childSessionId; if ( owner.status !== "running" && sessionId !== null && !hasWorkingDescendant(sessionId) ) { yield* publishNativeChildStatus(owner, owner.status); } else if (acpSubagentStatusIsTerminal(owner.subagent.task.status)) { if (context !== null) yield* syncNativeChildren(context); else if (root !== null) yield* offerContinuationRun(root); } } return true; } if (notification.sessionId === root) return false; if (yield* Ref.get(stoppedRunQuarantine)) return true; const update = notification.update; if (update.sessionUpdate === "state_update") { const child = nativeChild(notification.sessionId); setNativeChildState( child, update.state !== "idle" ? "running" : update.stopReason === "error" ? "failed" : update.stopReason === "cancelled" ? "cancelled" : "completed", update.state === "idle" ? (update.error?.message ?? null) : null, ); } if (context !== null) yield* syncNativeChildren(context); yield* projectNativeChildFrame(notification); const owner = nativeChildOwner(notification.sessionId); if ( owner?.subagent?.childSessionId != null && owner.status !== "running" && !hasWorkingDescendant(owner.subagent.childSessionId) ) { yield* publishNativeChildStatus(owner, owner.status); } return true; }); const toolOutputText = (toolCall: AcpToolCallState): string => { if (typeof toolCall.data.rawOutput === "string") { return toolCall.data.rawOutput; } if ( typeof (toolCall.data.rawOutput as { text?: unknown } | undefined)?.text === "string" ) { return String((toolCall.data.rawOutput as { text: string }).text); } return textFromUnknown(toolCall.data.rawOutput) ?? ""; }; const setToolOutputText = (toolCall: AcpToolCallState, text: string): AcpToolCallState => ({ ...toolCall, data: { ...toolCall.data, rawOutput: { type: "Text", text, }, }, }); const appendToolOutputText = ( toolCall: AcpToolCallState, appendOutput: string, ): AcpToolCallState => { if (appendOutput.length === 0) return toolCall; return setToolOutputText(toolCall, `${toolOutputText(toolCall)}${appendOutput}`); }; const isMonitorEndNoticeText = (text: string): boolean => /Monitor\s+["']?[0-9a-f-]{8,}/i.test(text) && /ended/i.test(text); const hasDeferredBackgroundWork = (context: ActiveAcpTurn): boolean => { if (context.awaitingBackgroundHydration.size > 0) return true; if (context.pendingInjectedReport.size > 0) return true; for (const [taskId, toolCallId] of context.toolCallIdsByBackgroundTaskId) { if (context.persistentBackgroundTaskIds.has(taskId)) continue; const tool = context.tools.get(toolCallId); if (tool === undefined) continue; const status = toolStatus(tool.status); if (status === "pending" || status === "running") return true; } for (const subagent of context.subagents.values()) { if (acpSubagentStatusBlocksTurnSettlement(subagent.task.status)) { return true; } } return false; }; const hasSurvivingBackgroundWork = (context: ActiveAcpTurn): boolean => [...context.toolCallIdsByBackgroundTaskId.values()].some((toolCallId) => { const tool = context.tools.get(toolCallId); const status = tool === undefined ? undefined : toolStatus(tool.status); return status === "pending" || status === "running"; }) || [...context.subagents.values()].some((subagent) => acpSubagentStatusBlocksTurnSettlement(subagent.task.status), ); const rearmDeferredFinalize = /* Provider events include streaming deltas; avoid a span for each update. */ Effect.fnUntraced(function* (context: ActiveAcpTurn) { if (!flavor.deferFinalizeForBackgroundWork) return; if (!context.promptSettled || context.finalized) return; if (hasDeferredBackgroundWork(context)) return; yield* scheduleDeferredFinalize(context); }); // let breaks circular inference from monitor hydration re-entry. let emitTool: ( context: ActiveAcpTurn, incoming: AcpToolCallState, projectedStatus?: ProjectedToolStatus, ) => Effect.Effect = () => Effect.void; const markAwaitingBackgroundHydration = /* Provider events include streaming deltas; avoid a span for each update. */ Effect.fnUntraced(function* (context: ActiveAcpTurn, taskId: string) { if (context.awaitingBackgroundHydration.has(taskId)) return; context.awaitingBackgroundHydration.add(taskId); yield* Effect.gen(function* () { yield* Effect.sleep("60000 millis"); if (context.finalized || !context.awaitingBackgroundHydration.has(taskId)) return; context.awaitingBackgroundHydration.delete(taskId); const toolCallId = context.toolCallIdsByBackgroundTaskId.get(taskId); const tool = toolCallId !== undefined ? context.tools.get(toolCallId) : undefined; if (tool !== undefined) { const status = toolStatus(tool.status); if (status === "pending" || status === "running") { yield* emitTool(context, { ...tool, status: "completed" }); } } yield* rearmDeferredFinalize(context); }).pipe(Effect.forkIn(sessionScope), Effect.asVoid); }); const markPendingInjectedReport = /* Provider events include streaming deltas; avoid a span for each update. */ Effect.fnUntraced(function* (context: ActiveAcpTurn, taskId: string) { if (!context.promptSettled) return; if (context.pendingInjectedReport.has(taskId)) return; context.pendingInjectedReport.add(taskId); yield* Effect.gen(function* () { yield* Effect.sleep("25000 millis"); if (context.finalized || !context.pendingInjectedReport.has(taskId)) return; context.pendingInjectedReport.delete(taskId); yield* rearmDeferredFinalize(context); }).pipe(Effect.forkIn(sessionScope), Effect.asVoid); }); const clearMidTurnUnreportedTaskIds = (taskIds: ReadonlyArray) => Ref.update(midTurnUnreportedCompletedTaskIds, (current) => { let next: Set | null = null; for (const taskId of taskIds) { if (!current.has(taskId)) continue; if (next === null) next = new Set(current); next.delete(taskId); } return next ?? current; }); emitTool = Effect.fnUntraced(function* ( context: ActiveAcpTurn, incoming: AcpToolCallState, projectedStatus?: ProjectedToolStatus, ) { if (context.assistant.current?.sourceMessageId == null) yield* closeTextStream(context, "assistant"); if (context.reasoning.current?.sourceMessageId == null) yield* closeTextStream(context, "reasoning"); yield* closeTextStream(context, "user"); const previous = context.tools.get(incoming.toolCallId); const merged = mergeToolCallState(previous, incoming); const normalized = flavor.normalizeToolCall?.(merged) ?? merged; const backgroundTaskId = flavor.extractBackgroundTaskId?.(normalized); const backgroundCommandTaskId = flavor.isBackgroundCommandTool?.(normalized) === true ? backgroundTaskId : previous?.backgroundCommandTaskId; if ( backgroundCommandTaskId !== undefined && !context.backgroundCommands.has(backgroundCommandTaskId) ) { context.backgroundCommands.set(backgroundCommandTaskId, null); } const completion = context.promptSettled && backgroundCommandTaskId !== undefined ? (context.backgroundCommands.get(backgroundCommandTaskId) ?? undefined) : undefined; // Preserve hydrated output; only a replayed start ACK needs its output restored. const toolCall = completion === undefined ? normalized : { ...(backgroundTaskId === undefined || completion.output === undefined ? normalized : setToolOutputText(normalized, completion.output)), status: completion.status, }; context.tools.set( toolCall.toolCallId, backgroundCommandTaskId === undefined ? toolCall : { ...toolCall, backgroundCommandTaskId }, ); // fx 0.0.15 fails the tools a Stop cut off; they read as interrupted, like ones left open. const previousStatus = previous === undefined ? undefined : toolStatus(previous.status); if ( context.interrupted && toolCall.status === "failed" && (previousStatus === "pending" || previousStatus === "running" || previousStatus === "waiting") ) { context.stoppedToolCallIds.add(toolCall.toolCallId); } const stoppedStatus = projectedStatus ?? (toolCall.status === "failed" && context.stoppedToolCallIds.has(toolCall.toolCallId) ? "interrupted" : undefined); if ( followsAgentWorkState && projectedStatus === undefined && flavor.isProvisionalSubagentTool?.(toolCall) ) { yield* rearmDeferredFinalize(context); return; } if (backgroundTaskId !== undefined) { context.toolCallIdsByBackgroundTaskId.set(backgroundTaskId, toolCall.toolCallId); if (flavor.isPersistentBackgroundTool?.(toolCall) === true) { context.persistentBackgroundTaskIds.add(backgroundTaskId); } const backgroundStatus = projectedStatus ?? toolStatus(toolCall.status); yield* setBackgroundTaskRunning( backgroundTaskId, backgroundStatus === "pending" || backgroundStatus === "running", ); if ( !context.promptSettled && backgroundStatus !== "pending" && backgroundStatus !== "running" ) { yield* Ref.update(handledBackgroundTaskIdsInActiveTurn, (current) => new Set(current).add(backgroundTaskId), ); yield* Ref.update(midTurnUnreportedCompletedTaskIds, (current) => { if (!current.has(backgroundTaskId)) return current; const next = new Set(current); next.delete(backgroundTaskId); return next; }); } } const backgroundCompletions = projectedStatus === undefined ? (flavor.extractBackgroundTaskCompletion?.(toolCall) ?? []) : []; let hydratedRegisteredMonitor = false; for (const backgroundCompletion of backgroundCompletions) { yield* applyBackgroundTaskMutationRunning(backgroundCompletion); if (backgroundCompletion.status !== "running") { yield* Ref.update(handledBackgroundTaskIdsInActiveTurn, (current) => new Set(current).add(backgroundCompletion.taskId), ); yield* Ref.update(midTurnUnreportedCompletedTaskIds, (current) => { if (!current.has(backgroundCompletion.taskId)) return current; const next = new Set(current); next.delete(backgroundCompletion.taskId); return next; }); } if (backgroundCompletion.status !== "running") { context.awaitingBackgroundHydration.delete(backgroundCompletion.taskId); } const targetToolCallId = context.toolCallIdsByBackgroundTaskId.get( backgroundCompletion.taskId, ); if (targetToolCallId !== undefined) { hydratedRegisteredMonitor = true; } const target = targetToolCallId !== undefined ? context.tools.get(targetToolCallId) : undefined; if (target !== undefined && target.toolCallId !== toolCall.toolCallId) { const nextStatus = backgroundCompletion.status === "running" ? ("inProgress" as const) : backgroundCompletion.status === "failed" ? ("failed" as const) : ("completed" as const); const hydrated = backgroundCompletion.appendOutput.length > 0 ? isMonitorEndNoticeText(toolOutputText(target)) || toolOutputText(target).trim().length === 0 ? setToolOutputText( { ...target, status: nextStatus }, backgroundCompletion.appendOutput, ) : appendToolOutputText( { ...target, status: nextStatus }, backgroundCompletion.appendOutput, ) : { ...target, status: nextStatus }; // BackgroundTaskStarted normalization can reopen this row, so keep the // completion for settlement when no task_completed recorded one. if ( backgroundCompletion.status !== "running" && context.backgroundCommands.get(backgroundCompletion.taskId) === null ) { context.backgroundCommands.set(backgroundCompletion.taskId, { status: backgroundCompletion.status, output: toolOutputText(hydrated), }); } yield* emitTool(context, hydrated); } } // Monitor TaskOutput resembles subagent hydration; do not invent a subagent. const subagentUpdate = hydratedRegisteredMonitor ? undefined : extractSubagentUpdate(toolCall); if (subagentUpdate !== undefined) { yield* emitSubagent( context, stoppedStatus === "interrupted" && subagentUpdate.status === "failed" ? { ...subagentUpdate, status: "interrupted" } : subagentUpdate, ); if (followsAgentWorkState) yield* syncNativeChildren(context); if (subagentUpdate.suppressNormalTool !== false) { yield* rearmDeferredFinalize(context); return; } } if ( stoppedStatus === undefined && !shouldPersistToolUpdate( context, toolCall.toolCallId, previous, toolCall, merged.status, ) ) { yield* rearmDeferredFinalize(context); return; } const status = stoppedStatus ?? toolStatus(toolCall.status); const now = yield* DateTime.now; const nativeItemId = `${context.nativeThreadId}:tool:${toolCall.toolCallId}`; const ordinal = yield* resolveItemOrdinal(context, nativeItemId); const nodeId = providerNodeId(nativeItemId); const turnItemId = providerTurnItemId(nativeItemId); const nativeItemRef = { driver, nativeId: toolCall.toolCallId, strength: "strong" as const, }; const startedAt = context.toolStartedAt.get(toolCall.toolCallId) ?? now; context.toolStartedAt.set(toolCall.toolCallId, startedAt); const completedAt = completedAtForStatus(status, now); const title = toolCall.title ?? null; yield* emitProviderEvent({ type: "node.updated", driver, node: { id: nodeId, threadId: context.input.threadId, runId: context.input.runId, parentNodeId: context.input.rootNodeId, rootNodeId: context.input.rootNodeId, kind: "tool_call", status: nodeStatus(status), countsForRun: true, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, nativeItemRef, runtimeRequestId: null, checkpointScopeId: null, startedAt, completedAt, }, }); const base = { id: turnItemId, threadId: context.input.threadId, runId: context.input.runId, nodeId, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, nativeItemRef, parentItemId: null, ordinal, status, title, startedAt, completedAt, updatedAt: now, } as const; const rawInput = toolCall.data.rawInput; const rawOutput = toolCall.data.rawOutput ?? toolCall.data.content; const changes = structuredFileChanges(toolCall); const path = changes[0]?.path ?? pathFromToolCall(toolCall); const diffText = acpToolCallDiffPatch(toolCall.data.content) ?? textFromUnknown(rawOutput); const rawInputRecord = unknownRecord(rawInput); const inputVariant = typeof rawInputRecord?.variant === "string" ? rawInputRecord.variant.trim().toLowerCase() : ""; const rawOutputRecord = unknownRecord(rawOutput); const outputCommand = typeof rawOutputRecord?.command === "string" && rawOutputRecord.command.trim().length > 0 ? rawOutputRecord.command.trim() : undefined; const monitorCommand = (typeof rawInputRecord?.command === "string" && rawInputRecord.command.trim().length > 0 ? rawInputRecord.command.trim() : undefined) ?? outputCommand; const outputIsBashResult = typeof rawOutputRecord?.type === "string" && rawOutputRecord.type.trim().toLowerCase() === "bash" && commandExitCode(rawOutput) !== undefined; const projectAsCommandExecution = inputVariant === "monitor" || outputIsBashResult; const mcpIdentity = extractMcpToolCallIdentity(toolCall, { embeddedTerminalCommands: embeddedTerminalCommands( context.nativeThreadId, toolCall.toolCallId, ), }); let turnItem: OrchestrationV2TurnItem; if (toolCall.toolCallId.startsWith("acp-compaction:")) { const summary = textFromUnknown(rawOutput); turnItem = { ...base, type: "compaction", driver, ...(summary === undefined ? {} : { summary }), ...(context.contextUsage?.usedTokens === undefined ? {} : { beforeTokenCount: context.contextUsage.usedTokens }), }; if (status === "completed") { context.contextUsage = null; yield* Ref.update(contextUsageBySessionId, (current) => { const updated = new Map(current); updated.delete(context.nativeThreadId); return updated; }); yield* Ref.update(compactionMarkers, (current) => { const updated = new Map(current); updated.set(context.nativeThreadId, Symbol()); return updated; }); } } else if (mcpIdentity !== undefined) { const toolName = `${mcpIdentity.server}.${mcpIdentity.tool}`; const input = mcpIdentity.input ?? unknownRecord(rawInputRecord?.arguments) ?? rawInputRecord ?? {}; let output = rawOutput === undefined ? undefined : acpMcpToolCallOutput(rawOutput); const isHtmlRender = resolveT3McpToolId(toolName) === HTML_RENDER_TOOL_NAME; const isError = compactDynamicToolOutput(output)?.isError === true || compactDynamicToolOutput(rawOutputRecord?.result)?.isError === true; if ( isHtmlRender && (status === "failed" || (status === "completed" && isError)) && previous?.status !== "failed" && previous?.status !== "completed" ) { // A failed call may have published before the provider lost its result. yield* mcpSessions.takeHtmlRender( context.input.threadId, input, htmlRenderFromToolItem({ toolName, output }), ); } if (status === "completed" && isHtmlRender && !isError) { const reportedPage = htmlRenderFromToolItem({ toolName, output }); // Antigravity replaces MCP results with summaries, including on repeated completions. const previousPage = htmlRenderFromToolItem({ toolName, output: acpMcpToolCallOutput(previous?.data.rawOutput ?? previous?.data.content), }); const page = reportedPage ? yield* mcpSessions.takeHtmlRender(context.input.threadId, input, reportedPage) : (previousPage ?? (yield* mcpSessions.takeHtmlRender(context.input.threadId, input))); if (reportedPage === undefined && page !== undefined) { output = { htmlRender: page }; context.tools.set(toolCall.toolCallId, { ...toolCall, data: { ...toolCall.data, rawOutput: output }, }); } } turnItem = { ...base, title: null, ...mcpToolPresentation({ serverName: mcpIdentity.server, toolName: mcpIdentity.tool, source: unknownRecord( (unknownRecord(rawOutputRecord?.result) ?? rawOutputRecord)?._meta, )?.source, }), type: "dynamic_tool", toolName, input, ...(output === undefined ? {} : { output }), }; yield* emitProviderEvent({ type: "turn_item.updated", driver, turnItem }); yield* rearmDeferredFinalize(context); return; } else if (changes.length > 0) { turnItem = { ...base, type: "file_change", fileName: changes[0]!.path, changes, ...(diffText === undefined ? {} : { diffStr: diffText }), }; } else { switch (toolCall.kind) { case "read": turnItem = { ...base, title: path ? formatReadToolLabel(path) : (title ?? "Read file"), type: "dynamic_tool", toolName: "Read", input: path === undefined || ["path", "filePath", "file_path"].some((key) => rawInputRecord?.[key] === path) ? (rawInputRecord ?? {}) : { ...rawInputRecord, path }, ...(rawOutput === undefined ? {} : { output: rawOutput }), }; break; case "search": { const backendSearch = acpBackendWebSearch(rawInputRecord, rawOutputRecord); if (backendSearch !== undefined) { const label = nonEmptyText(toolCall.data.title, title ?? "Web search").replace( /:\s*$/u, "", ); turnItem = { ...base, title: backendSearch.query === undefined ? label : `${label}: ${backendSearch.query}`, type: "web_search", ...(backendSearch.query === undefined ? {} : { patterns: [backendSearch.query] }), ...(backendSearch.results.length === 0 ? {} : { results: backendSearch.results }), }; break; } turnItem = { ...base, title: formatSearchToolLabel({ rawInput: rawInputRecord, input: rawInputRecord, ...(path === undefined ? {} : { pattern: path }), }) ?? title, type: "file_search", ...(path === undefined ? {} : { pattern: path }), ...(path === undefined ? {} : { results: [ { fileName: path, ...(textFromUnknown(rawOutput) === undefined ? {} : { preview: textFromUnknown(rawOutput) }), }, ], }), }; break; } case "execute": { const exitCode = acpProjectedCommandExitCode(status, rawOutput); turnItem = { ...base, type: "command_execution", input: toolCall.command ?? monitorCommand ?? toolCall.title ?? "Command", ...(textFromUnknown(rawOutput) === undefined ? {} : { output: textFromUnknown(rawOutput) }), ...(exitCode === undefined ? {} : { exitCode }), }; break; } case "edit": case "delete": case "move": turnItem = { ...base, type: "file_change", fileName: path ?? toolCall.title ?? "File change", ...(diffText === undefined ? {} : { diffStr: diffText }), }; break; case "fetch": { // Grok rawOutput.Content is unreadable by textFromUnknown; bounded content blocks duplicate it. const snippet = textFromUnknown(toolCall.data.content) ?? textFromUnknown(rawOutput); turnItem = { ...base, type: "web_search", ...(path === undefined ? {} : { patterns: [path] }), ...(path === undefined ? {} : { results: [ { url: path, ...(snippet === undefined ? {} : { snippet }), }, ], }), }; break; } default: if (projectAsCommandExecution) { const exitCode = acpProjectedCommandExitCode(status, rawOutput); turnItem = { ...base, type: "command_execution", input: toolCall.command ?? monitorCommand ?? toolCall.title ?? (inputVariant === "monitor" ? "Monitor" : "Command"), ...(textFromUnknown(rawOutput) === undefined ? {} : { output: textFromUnknown(rawOutput) }), ...(exitCode === undefined ? {} : { exitCode }), ...(inputVariant === "monitor" ? { monitor: true } : {}), }; } else { turnItem = { ...base, type: "dynamic_tool", toolName: toolCall.title ?? toolCall.kind ?? null, input: rawInput ?? {}, ...(rawOutput === undefined ? {} : { output: rawOutput }), }; } } } yield* emitProviderEvent({ type: "turn_item.updated", driver, turnItem }); yield* rearmDeferredFinalize(context); }); /** * Notices have no id; each is its own item, numbered by the turn's item count. * A notice is not conversation content, so an open message keeps streaming across it. */ const emitNotice = Effect.fnUntraced(function* ( context: ActiveAcpTurn, notice: { readonly title: string; readonly description?: string }, ) { const nativeItemId = `${context.nativeTurnId}:notice:${context.itemOrdinals.size}`; const now = yield* DateTime.now; yield* emitProviderEvent({ type: "turn_item.updated", driver, turnItem: { id: providerTurnItemId(nativeItemId), threadId: context.input.threadId, runId: context.input.runId, nodeId: context.input.rootNodeId, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, nativeItemRef: { driver, nativeId: nativeItemId, strength: "weak" }, parentItemId: null, ordinal: yield* resolveItemOrdinal(context, nativeItemId), type: "system_notice", status: "completed", title: notice.title, message: notice.description === undefined ? notice.title : `${notice.title}\n\n${notice.description}`, startedAt: now, completedAt: now, updatedAt: now, }, }); }); const emitPlan = Effect.fnUntraced(function* ( context: ActiveAcpTurn, update: AcpPlanUpdate, ) { yield* closeTextStreams(context); const existing = context.plans.get(update.nativePlanId); if (update.kind === "removed" && existing === undefined) return; const nativeItemId = `${context.nativeTurnId}:plan:${encodeURIComponent(update.nativePlanId)}`; const ordinal = yield* resolveItemOrdinal(context, nativeItemId); const now = yield* DateTime.now; const nodeId = providerNodeId(nativeItemId); const turnItemId = providerTurnItemId(nativeItemId); const planState = existing ?? { id: yield* idAllocator.allocate.plan({ threadId: context.input.threadId, runId: context.input.runId, driver, }), startedAt: now, latest: null, }; const planId = planState.id; const nativeItemRef = { driver, nativeId: nativeItemId, strength: "weak" as const }; const base = { id: planId, threadId: context.input.threadId, runId: context.input.runId, nodeId, } as const; let plan: OrchestrationV2PlanArtifact; if (update.kind === "removed") { if (planState.latest === null) return; plan = { ...planState.latest, status: "superseded" }; } else if (update.kind === "items") { const steps: ReadonlyArray = update.plan.map( (step, index) => ({ id: `acp-step-${index + 1}`, text: nonEmptyText(step.step, `Step ${index + 1}`), status: step.status === "inProgress" ? "running" : step.status === "completed" ? "completed" : "pending", }), ); const completed = steps.length > 0 && steps.every((step) => step.status === "completed"); plan = { ...base, status: completed ? "completed" : "active", kind: "todo_list", steps, ...(update.explanation == null ? {} : { explanation: update.explanation }), }; } else { const markdown = update.kind === "markdown" ? update.markdown : update.kind === "file" ? `Plan file: ${update.uri}` : `[Unsupported ACP plan content: ${update.contentType}]`; plan = { ...base, status: "active", kind: "proposed_plan", markdown, }; } planState.latest = plan; context.plans.set(update.nativePlanId, planState); const completed = plan.status === "completed" || plan.status === "superseded"; const nodeKind = plan.kind === "todo_list" ? "todo_list" : "plan"; yield* emitProviderEvent({ type: "node.updated", driver, node: { id: nodeId, threadId: context.input.threadId, runId: context.input.runId, parentNodeId: context.input.rootNodeId, rootNodeId: context.input.rootNodeId, kind: nodeKind, status: completed ? "completed" : "running", countsForRun: false, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, nativeItemRef, runtimeRequestId: null, checkpointScopeId: null, startedAt: planState.startedAt, completedAt: completed ? now : null, }, }); yield* emitProviderEvent({ type: "plan.updated", driver, plan }); yield* emitProviderEvent({ type: "turn_item.updated", driver, turnItem: { id: turnItemId, threadId: context.input.threadId, runId: context.input.runId, nodeId, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, nativeItemRef, parentItemId: null, ordinal, status: completed ? "completed" : "running", title: null, startedAt: planState.startedAt, completedAt: completed ? now : null, updatedAt: now, ...(plan.kind === "todo_list" ? { type: "todo_list" as const, planId, steps: plan.steps, ...(plan.explanation === undefined ? {} : { explanation: plan.explanation }), } : { type: "proposed_plan" as const, planId, markdown: plan.markdown, streaming: !completed, }), }, }); }); const appendLoadedHistory = /* Provider events include streaming deltas; avoid a span for each update. */ Effect.fnUntraced(function* ( notification: EffectAcpSchema.SessionNotification, role: "user" | "assistant" | "thought", text: string, replace = false, ) { if (text.length === 0 && !replace) return; const now = yield* DateTime.now; yield* Ref.update(snapshot, (current) => { const update = notification.update; const sourceMessageId = (update.sessionUpdate === "user_message_chunk" || update.sessionUpdate === "agent_message_chunk" || update.sessionUpdate === "agent_thought_chunk" || update.sessionUpdate === "user_message" || update.sessionUpdate === "agent_message" || update.sessionUpdate === "agent_thought") && update.messageId ? update.messageId : null; const startsNew = current.loadingRole !== role || (sourceMessageId !== null && current.loadingMessageId !== sourceMessageId); const loadingIndex = startsNew ? current.loadingIndex + 1 : current.loadingIndex; const nativeItemId = sourceMessageId === null ? `${notification.sessionId}:history:${role}:${loadingIndex}` : `${notification.sessionId}:history:${role}:message:${sourceMessageId}`; const messageId = providerMessageId(nativeItemId); const key = String(messageId); const previous = current.messages.get(key); const messages = new Map(current.messages); messages.set(key, { createdBy: previous?.createdBy ?? (role === "user" ? "user" : "agent"), creationSource: previous?.creationSource ?? "provider", id: messageId, threadId: input.threadId, runId: null, nodeId: null, role: role === "user" ? "user" : "assistant", text: replace ? text : `${previous?.text ?? ""}${text}`, attachments: [], streaming: false, createdAt: previous?.createdAt ?? now, updatedAt: now, }); return { order: current.order.includes(key) ? current.order : [...current.order, key], messages, loadingRole: role, loadingMessageId: sourceMessageId ?? (startsNew ? null : current.loadingMessageId), loadingIndex, }; }); }); const offerContinuationRun = Effect.fnUntraced(function* (_sessionId: string) { if (continuationRequests === undefined) { return false; } const pending = yield* continuationPermit.withPermit( Effect.gen(function* () { if (yield* Ref.get(continuationClosed)) return Option.none(); if (yield* Ref.get(stoppedRunQuarantine)) return Option.none(); if (yield* Ref.get(continuationRequested)) return Option.none(); const route = yield* Ref.get(lastTurnRoute); if (route === null) return Option.none(); yield* Ref.set(continuationRequested, true); const generation = yield* Ref.updateAndGet( continuationGeneration, (value) => value + 1, ); const reports = yield* Ref.modify(wakeReports, ({ reports }) => [ reports, { reports, offered: new Set(reports.keys()) }, ]); return Option.some({ route, generation, reports }); }), ); if (Option.isNone(pending)) return false; const { route, generation, reports } = pending.value; const notification = backgroundWorkNotification([...reports.values()]); yield* Effect.logInfo("orchestration-v2.acp-wake-turn-detected", { driver, providerSessionId: input.providerSessionId, threadId: route.threadId, providerThreadId: route.providerThreadId, }); yield* continuationRequests.offer({ threadId: route.threadId, providerThreadId: route.providerThreadId, driver, detail: null, ...(notification === null ? {} : { notification }), clearIfCurrent: () => continuationPermit.withPermit( Effect.gen(function* () { if ((yield* Ref.get(continuationGeneration)) === generation) { yield* Ref.set(continuationRequested, false); } }), ), dispatchIfCurrent: (effect) => continuationPermit.withPermit( Effect.gen(function* () { const clearIfOwner = Effect.gen(function* () { if ((yield* Ref.get(continuationGeneration)) === generation) { yield* Ref.set(continuationRequested, false); } }); if (yield* Ref.get(stoppedRunQuarantine)) { yield* clearIfOwner; return Option.none(); } if ((yield* Ref.get(continuationGeneration)) !== generation) { return Option.none(); } if (!(yield* Ref.get(continuationRequested))) return Option.none(); // Retain the dispatch flag until continuation starts so late frames cannot enqueue duplicates. const exit = yield* Effect.exit(effect); if (Exit.isFailure(exit)) { yield* clearIfOwner; return yield* Effect.failCause(exit.cause); } return Option.some(exit.value); }), ), }); return true; }); const applyLateBackgroundMutation = Effect.fnUntraced(function* ( sessionId: string, mutation: { readonly taskId: string; readonly status: "running" | "completed" | "failed"; }, ) { const taskAlreadyEnded = (yield* Ref.get(endedBackgroundTaskIds)).has(mutation.taskId); yield* applyBackgroundTaskMutationRunning(mutation); if (mutation.status === "running") { if (taskAlreadyEnded && acpPostSettleMonitorPromptShouldSuppress(mutation)) { yield* Ref.set(suppressPostSettleMonitorPrompt, true); } return; } // Only a new terminal may un-suppress output; re-delivery must not revive monitor ACK chatter. if (!taskAlreadyEnded) { yield* Ref.set(suppressPostSettleMonitorPrompt, false); } const activeContext = yield* Ref.get(activeTurn); if (activeContext !== null && !activeContext.finalized) { const handled = yield* Ref.get(handledBackgroundTaskIdsInActiveTurn); if ( !activeContext.promptSettled && !handled.has(mutation.taskId) && activeContext.toolCallIdsByBackgroundTaskId.has(mutation.taskId) ) { yield* Ref.update(midTurnUnreportedCompletedTaskIds, (current) => new Set(current).add(mutation.taskId), ); } return; } if ( postSettleContinuationEnabled && (yield* Ref.get(activeSessionId)) === sessionId && (yield* Ref.get(runningBackgroundTaskIds)).size === 0 && ((yield* Ref.get(wakeBuffer)).length > 0 || (yield* Ref.get(midTurnUnreportedCompletedTaskIds)).size > 0) ) { yield* offerContinuationRun(sessionId); } }); // Structured task ends are authoritative, but emitting before prompt // settlement marks the result handled and suppresses its continuation. const finishRegisteredBackgroundTool = Effect.fnUntraced(function* (mutation: { readonly taskId: string; readonly status: "completed" | "failed"; readonly output?: string; }) { const context = yield* Ref.get(activeTurn); if (context === null || context.finalized || !context.promptSettled) return; const toolCallId = context.toolCallIdsByBackgroundTaskId.get(mutation.taskId); const tool = toolCallId === undefined ? undefined : context.tools.get(toolCallId); if (tool === undefined) return; const status = toolStatus(tool.status); if ( status !== "pending" && status !== "running" && !context.backgroundCommands.has(mutation.taskId) ) { return; } context.awaitingBackgroundHydration.delete(mutation.taskId); const finished = { ...tool, status: mutation.status }; yield* emitTool( context, mutation.output === undefined ? finished : setToolOutputText(finished, mutation.output), mutation.status, ); yield* rearmDeferredFinalize(context); }); const bufferPostSettleWake = Effect.fnUntraced(function* ( notification: EffectAcpSchema.SessionNotification, ) { if (!postSettleContinuationEnabled || continuationRequests === undefined) { return { buffered: false, offerContinuation: false, stopProcessing: false, }; } if (yield* Ref.get(stoppedRunQuarantine)) { return { buffered: false, offerContinuation: false, stopProcessing: true, }; } const rootSessionId = yield* Ref.get(activeSessionId); if (rootSessionId === null || notification.sessionId !== rootSessionId) { return { buffered: false, offerContinuation: false, stopProcessing: false, }; } if (!acpPostSettleWakeEvidence(notification, flavor)) { return { buffered: false, offerContinuation: false, stopProcessing: false, }; } const update = notification.update; let alreadyHandledToolUpdate = false; if (update.sessionUpdate === "tool_call" || update.sessionUpdate === "tool_call_update") { yield* trackRunningBackgroundTools(notification); // Long-timeout get_command_or_subagent_output consumption ends with TaskOutput alone. for (const event of parseSessionUpdateEvent(notification).events) { if (event._tag !== "ToolCallUpdated") continue; const toolCall = flavor.normalizeToolCall?.(event.toolCall) ?? event.toolCall; const proposedPlanMarkdown = flavor.extractProposedPlanMarkdown?.(toolCall); if (proposedPlanMarkdown !== undefined && proposedPlanMarkdown.length > 0) { yield* captureProposedPlan({ planMarkdown: proposedPlanMarkdown }); } const toolTaskId = flavor.extractBackgroundTaskId?.(toolCall); if ( toolTaskId !== undefined && (yield* Ref.get(handledBackgroundTaskIdsInActiveTurn)).has(toolTaskId) ) { alreadyHandledToolUpdate = true; } if (flavor.extractBackgroundTaskCompletion !== undefined) { for (const completion of flavor.extractBackgroundTaskCompletion(toolCall)) { alreadyHandledToolUpdate = alreadyHandledToolUpdate || (yield* Ref.get(handledBackgroundTaskIdsInActiveTurn)).has(completion.taskId); yield* applyBackgroundTaskMutationRunning(completion); } } } } const backgroundWorkRunning = (yield* Ref.get(runningBackgroundTaskIds)).size > 0; const handledInTurnCount = (yield* Ref.get(handledBackgroundTaskIdsInActiveTurn)).size; const isInTurnHandledAgentChatter = handledInTurnCount > 0 && (update.sessionUpdate === "agent_message_chunk" || update.sessionUpdate === "agent_thought_chunk"); // Grok re-prompts for each post-settle monitor event; retaining them would replay the whole burst. let buffered = false; if ( !alreadyHandledToolUpdate && !isInTurnHandledAgentChatter && acpPostSettleWakeShouldBuffer(notification, backgroundWorkRunning) ) { yield* Ref.update(wakeBuffer, (current) => [...current, notification]); buffered = true; } if (!acpPostSettleContinuationOfferEvidence(notification, flavor)) { return { buffered, offerContinuation: false, stopProcessing: true, }; } if (backgroundWorkRunning) { return { buffered, offerContinuation: false, stopProcessing: true, }; } if (alreadyHandledToolUpdate) { if (!(yield* Ref.get(continuationRequested))) { yield* Ref.set(wakeBuffer, []); } return { buffered, offerContinuation: false, stopProcessing: true, }; } // Preserve wake frames for other tasks until their completion offers. if (isInTurnHandledAgentChatter) { return { buffered, offerContinuation: false, stopProcessing: true, }; } return { buffered, offerContinuation: true, stopProcessing: true, }; }); let applyFinalizedActiveTurnSubagentTerminal: ( context: ActiveAcpTurn, notification: EffectAcpSchema.SessionNotification, ) => Effect.Effect = () => Effect.succeed(false); const projectAgentTerminalUpdate = Effect.fnUntraced(function* ( notification: EffectAcpSchema.SessionNotification, context: ActiveAcpTurn | null, ) { const update = notification.update; if ( update.sessionUpdate !== "terminal_update" && update.sessionUpdate !== "terminal_output_chunk" ) { return false; } const terminalKey = sessionScopedId(notification.sessionId, update.terminalId); const next = applyAcpAgentTerminalUpdate(agentTerminalsById.get(terminalKey), update); agentTerminalsById.set(terminalKey, next); if (context === null || notification.sessionId !== context.nativeThreadId) return true; const status = next.exitStatus === undefined ? ("inProgress" as const) : next.exitStatus.exitCode === 0 ? ("completed" as const) : ("failed" as const); const embeddedToolCallIds = toolCallIdsByAgentTerminalId.get(terminalKey); const projectedToolCallIds = embeddedToolCallIds === undefined || embeddedToolCallIds.size === 0 ? [`acp-agent-terminal:${update.terminalId}`] : [...embeddedToolCallIds]; for (const toolCallId of projectedToolCallIds) { const existing = context.tools.get(toolCallId); const seeded: AcpToolCallState = existing ?? ({ toolCallId, kind: "execute", title: next.command ?? "Terminal", status, ...(next.command === undefined ? {} : { command: next.command }), data: { toolCallId }, } satisfies AcpToolCallState); yield* emitTool( context, setToolOutputText( { ...seeded, status, ...(next.command === undefined ? {} : { command: next.command }), }, next.output, ), ); } return true; }); const idleErrorFailure = (error: EffectAcpSchema.PromptResponse["error"]) => flavor.promptFailure?.( new EffectAcpErrors.AcpRequestError({ code: error?.code ?? -32603, errorMessage: error?.message ?? "Provider turn failed.", ...(error?.data === undefined ? {} : { data: error.data }), }), ) ?? makeProviderFailure({ message: error?.message, code: error == null ? null : String(error.code), class: "provider_error", }); const settleIdleWork = Effect.fnUntraced(function* ( context: ActiveAcpTurn, ): Effect.fn.Return { if ( !followsAgentWorkState || context.finalized || (!context.workAccepted && !context.interrupted) || agentWork?.state !== "idle" || agentRequests.size > 0 ) return; const work = agentWork; // Native close sends cancelled idle before acknowledging the user's Stop. if (work.stopReason === "cancelled" && pendingStopCloseTurn === context) { context.interrupted = true; } const failure = work.stopReason === "error" ? idleErrorFailure(work.error) : undefined; yield* finalizeTurn( context, failure !== undefined ? "failed" : work.stopReason === "cancelled" ? "cancelled" : "completed", failure, ); }); const awaitRequestTurn = Effect.fnUntraced(function* (generation: number) { if (!followsAgentWorkState) return true; while (true) { const waiting = yield* runRuntimeCallbackAtGeneration( generation, Effect.gen(function* () { if (yield* Ref.get(stoppedRunQuarantine)) return false; if ((yield* Ref.get(activeTurn)) !== null) return null; const sessionId = yield* Ref.get(activeSessionId); if (sessionId !== null) yield* offerContinuationRun(sessionId); return turnAttached; }), ); if (Option.isNone(waiting)) return yield* Effect.never; if (waiting.value === false) return false; if (waiting.value === null) return true; if (!(yield* Deferred.await(waiting.value))) return false; } }); const holdingAgentRequest = (generation: number) => (effect: Effect.Effect) => Effect.suspend(() => !followsAgentWorkState ? effect : Effect.acquireUseRelease( runRuntimeCallbackAtGeneration( generation, Effect.sync(() => { const token = {}; agentRequests.add(token); return token; }), ), (token) => (Option.isNone(token) ? Effect.never : effect), (token) => runRuntimeCallbackAtGeneration( generation, Effect.gen(function* () { if (Option.isSome(token)) agentRequests.delete(token.value); const context = yield* Ref.get(activeTurn); if (context !== null) yield* settleIdleWork(context); }), ).pipe(Effect.asVoid), ), ); const handleSessionUpdate = Effect.fnUntraced(function* ( notification: EffectAcpSchema.SessionNotification, ) { const context = yield* Ref.get(activeTurn); const update = notification.update; if ( followsAgentWorkState && flavor.extractSubagentSessionEvent !== undefined && notification.sessionId !== (yield* Ref.get(activeSessionId)) && (yield* Ref.get(stoppedRunQuarantine)) ) return; if (yield* projectAgentTerminalUpdate(notification, context)) return; if ( update.sessionUpdate === "usage_update" || update.sessionUpdate === "session_info_update" || (update.sessionUpdate === "state_update" && update.state === "idle" && update.usage != null) ) { const stateEvent = parseSessionUpdateEvent(notification).events.find( (event) => event._tag === "UsageUpdated" || event._tag === "SessionInfoUpdated", ); if (stateEvent === undefined) return; if (stateEvent._tag === "UsageUpdated") { yield* Ref.update(contextUsageBySessionId, (current) => new Map(current).set(notification.sessionId, stateEvent.usage), ); if (context?.nativeThreadId === notification.sessionId) { context.contextUsage = stateEvent.usage; } } else { const metadata = yield* Ref.modify(nativeMetadataBySessionId, (current) => { const merged = { ...current.get(notification.sessionId), ...stateEvent.metadata, }; return [merged, new Map(current).set(notification.sessionId, merged)] as const; }); if (context?.nativeThreadId === notification.sessionId) { context.nativeMetadata = metadata; } } const knownProviderThread = (yield* Ref.get(providerThreadByNativeSessionId)).get( notification.sessionId, ); if (knownProviderThread !== undefined) { const now = yield* DateTime.now; const providerThread: OrchestrationV2ProviderThread = { ...knownProviderThread, contextUsage: (yield* Ref.get(contextUsageBySessionId)).get(notification.sessionId) ?? knownProviderThread.contextUsage ?? null, nativeMetadata: (yield* Ref.get(nativeMetadataBySessionId)).get(notification.sessionId) ?? knownProviderThread.nativeMetadata ?? null, updatedAt: now, }; yield* Ref.update(providerThreadByNativeSessionId, (current) => new Map(current).set(notification.sessionId, providerThread), ); yield* emitProviderEvent({ type: "provider_thread.updated", driver, providerThread, }); } if (update.sessionUpdate !== "state_update") return; } if (yield* followNativeChild(notification, context)) return; if ( followsAgentWorkState && !(yield* Ref.get(stoppedRunQuarantine)) && notification.sessionId === (yield* Ref.get(activeSessionId)) ) { if (update.sessionUpdate === "state_update") agentWork = update; if ( update.sessionUpdate === "user_message" || update.sessionUpdate === "user_message_chunk" ) { agentWork = { sessionUpdate: "state_update", state: "running" }; if (context !== null) { context.workAccepted = true; if (!context.finalized) return; } } if ( context !== null && update.sessionUpdate === "state_update" && update.state === "idle" ) { yield* settleIdleWork(context); } else if ( context === null && !(yield* Ref.get(stoppedRunQuarantine)) && (acpRootSessionUpdateIngestsOutput(notification) || update.sessionUpdate === "user_message" || update.sessionUpdate === "user_message_chunk" || (update.sessionUpdate === "state_update" && update.state === "running")) ) { yield* Ref.update(wakeBuffer, (current) => [...current, notification]); yield* offerContinuationRun(notification.sessionId); return; } } if ( context?.finalized === true && (yield* applyFinalizedActiveTurnSubagentTerminal(context, notification)) ) { return; } if ( context !== null && context.finalized && (yield* Ref.get(handledBackgroundTaskIdsInActiveTurn)).size > 0 && (update.sessionUpdate === "agent_message_chunk" || update.sessionUpdate === "agent_thought_chunk") ) { yield* Ref.set(suppressPostSettleMonitorPrompt, true); return; } if ( context !== null && (yield* Ref.get(suppressPostSettleMonitorPrompt)) && (update.sessionUpdate === "agent_message_chunk" || update.sessionUpdate === "agent_thought_chunk") ) { return; } if ( context?.finalized === true && update.sessionUpdate === "user_message_chunk" && update.content.type === "text" ) { const mutations = flavor.extractBackgroundToolMutation?.(update.content.text) ?? []; for (const mutation of mutations) { yield* applyLateBackgroundMutation(notification.sessionId, mutation); } return; } if (context === null) { if (yield* Ref.get(stoppedRunQuarantine)) { return; } const bufferOutcome = yield* bufferPostSettleWake(notification); // Completed roots retain subscribers and can project terminals now; interrupted roots may lack runId ownership. const carryover = yield* Ref.get(carryoverSubagents); const rootTerminalCanStillProject = carryover !== null && carryover.sessionId === (yield* Ref.get(activeSessionId)) && carryover.projectsLateTerminals; const projectCarryover = rootTerminalCanStillProject; let carryoverTerminalized = false; if ( flavor.extractSubagentUpdate !== undefined && (update.sessionUpdate === "tool_call" || update.sessionUpdate === "tool_call_update") ) { for (const event of parseSessionUpdateEvent(notification).events) { if (event._tag !== "ToolCallUpdated") continue; const toolCall = flavor.normalizeToolCall?.(event.toolCall) ?? event.toolCall; const subagentUpdate = extractSubagentUpdate(toolCall); if (subagentUpdate === undefined) continue; if (!acpSubagentStatusIsTerminal(subagentUpdate.status)) { continue; } if ( yield* updateCarryoverSubagentStatus( subagentUpdate.nativeTaskId, subagentUpdate.status, subagentUpdate.result, { project: projectCarryover }, ) ) { carryoverTerminalized = true; } if ( subagentUpdate.childSessionId !== null && (yield* updateCarryoverSubagentStatus( subagentUpdate.childSessionId, subagentUpdate.status, subagentUpdate.result, { project: projectCarryover }, )) ) { carryoverTerminalized = true; } } } if ( update.sessionUpdate === "user_message_chunk" && update.content.type === "text" && flavor.extractSubagentEndNotice !== undefined ) { const notice = flavor.extractSubagentEndNotice(update.content.text); if ( notice !== undefined && (yield* updateCarryoverSubagentStatus( notice.childSessionId, notice.status, undefined, { project: projectCarryover, }, )) ) { carryoverTerminalized = true; } } const continuationOffered = bufferOutcome.offerContinuation ? yield* offerContinuationRun(notification.sessionId) : false; const wakeOutcome = { ...bufferOutcome, continuationOffered }; if (carryoverTerminalized && !wakeOutcome.buffered) { if ( acpCarryoverTerminalShouldClearContinuation({ continuationOffered: wakeOutcome.continuationOffered, wakeBufferLength: (yield* Ref.get(wakeBuffer)).length, }) ) { yield* Ref.set(continuationRequested, false); } return; } if (wakeOutcome.stopProcessing) { return; } if ( update.sessionUpdate === "user_message_chunk" || update.sessionUpdate === "agent_message_chunk" || update.sessionUpdate === "agent_thought_chunk" ) { const text = acpContentBlockDisplayText(update.content); if (text === undefined) return; const lateBackgroundMutations = update.content.type === "text" ? (flavor.extractBackgroundToolMutation?.(text) ?? []) : []; for (const lateBackgroundMutation of lateBackgroundMutations) { yield* applyLateBackgroundMutation(notification.sessionId, lateBackgroundMutation); } const lateMonitorChatter = / 0 && lateBackgroundMutations.length === 0 && !lateMonitorChatter ) { yield* appendLoadedHistory( notification, update.sessionUpdate === "user_message_chunk" ? "user" : update.sessionUpdate === "agent_thought_chunk" ? "thought" : "assistant", text, ); } } else if ( update.sessionUpdate === "user_message" || update.sessionUpdate === "agent_message" || update.sessionUpdate === "agent_thought" ) { if (update.content === undefined) return; const text = (update.content ?? []) .flatMap((content) => { const display = acpContentBlockDisplayText(content); return display === undefined ? [] : [display]; }) .join("\n"); yield* appendLoadedHistory( notification, update.sessionUpdate === "user_message" ? "user" : update.sessionUpdate === "agent_thought" ? "thought" : "assistant", text, true, ); } else if ( update.sessionUpdate === "tool_call" || update.sessionUpdate === "tool_call_update" || update.sessionUpdate === "plan" || update.sessionUpdate === "plan_update" || update.sessionUpdate === "plan_removed" ) { yield* Ref.update(snapshot, (current) => ({ ...current, loadingRole: null, loadingMessageId: null, })); } return; } if (context.finalized) return; const rootSessionId = yield* Ref.get(activeSessionId); if (notification.sessionId !== rootSessionId) { if (context.finalized) return; if (flavor.extractSubagentUpdate === undefined) return; const subagent = context.subagentsBySessionId.get(notification.sessionId); if ( update.sessionUpdate === "tool_call" || update.sessionUpdate === "tool_call_update" ) { if (subagent === undefined) return; const nativeTaskId = subagent.task.nativeTaskRef?.nativeId ?? String(subagent.task.id); for (const event of parseSessionUpdateEvent(notification).events) { if (event._tag !== "ToolCallUpdated") continue; const toolCall = flavor.normalizeToolCall?.(event.toolCall) ?? event.toolCall; const subagentUpdate = extractSubagentUpdate(toolCall); if (subagentUpdate !== undefined) { if ( subagentUpdate.nativeTaskId === nativeTaskId || subagentUpdate.childSessionId === notification.sessionId ) { yield* emitSubagent(context, subagentUpdate); } continue; } yield* projectSubagentTool(subagent, notification.sessionId, toolCall); } return; } const isDisplayableAssistantUpdate = (update.sessionUpdate === "agent_message_chunk" && acpContentBlockDisplayText(update.content) !== undefined) || (update.sessionUpdate === "agent_message" && update.content !== undefined); if (!isDisplayableAssistantUpdate) { return; } if (subagent !== undefined) { yield* projectSubagentNotification(subagent, notification); return; } const buffered = context.pendingSubagentNotifications.get(notification.sessionId) ?? []; buffered.push(notification); context.pendingSubagentNotifications.set(notification.sessionId, buffered); return; } // The activeSessionId yield may let idle/prompt settlement finalize this context. if (context.finalized) return; if ( context.promptSettled && !acpIsProviderContinuationMessage(context.input.message) && flavor.isProviderWakeNotification?.(notification) === true ) { if (hasDeferredBackgroundWork(context)) { yield* Ref.update(wakeBuffer, (current) => [...current, notification]); return; } yield* finalizeTurn(context, context.promptSettledStatus ?? "completed"); const wake = yield* bufferPostSettleWake(notification); if (wake.offerContinuation) { yield* offerContinuationRun(notification.sessionId); } return; } if ( update.sessionUpdate !== "tool_call_update" && update.sessionUpdate !== "plan" && update.sessionUpdate !== "plan_update" && update.sessionUpdate !== "plan_removed" && acpRootSessionUpdateIngestsOutput(notification) ) { yield* emitProviderRetry(context, "completed"); } switch (update.sessionUpdate) { case "state_update": { const toolCallId = `${context.nativeTurnId}:requires-action`; const existing = context.tools.get(toolCallId); if (update.state === "requires_action") { yield* emitTool(context, { toolCallId, kind: "other", title: "Action required", status: "requiresAction", data: { state: "requires_action" }, }); } else if (existing?.status === "requiresAction") { yield* emitTool(context, { ...existing, status: "completed", data: { ...existing.data, state: update.state }, }); } break; } case "agent_message_chunk": { const text = acpContentBlockDisplayText(update.content); if (text !== undefined) { const startsNewAssistantSegment = context.assistant.current === null; if (context.pendingInjectedReport.size > 0) { const reportedTaskIds = [...context.pendingInjectedReport]; context.pendingInjectedReport.clear(); yield* clearMidTurnUnreportedTaskIds(reportedTaskIds); } else if ( startsNewAssistantSegment && context.promptSettled && (yield* Ref.get(midTurnUnreportedCompletedTaskIds)).size > 0 ) { context.earlyInjectedReportObserved = true; } yield* appendText(context, "assistant", text, update.messageId); } break; } case "agent_thought_chunk": { const text = acpContentBlockDisplayText(update.content); if (text !== undefined) { yield* appendText(context, "reasoning", text, update.messageId); } break; } case "user_message": case "agent_message": case "agent_thought": { if (update.content === undefined) break; const text = (update.content ?? []) .flatMap((content) => { const display = acpContentBlockDisplayText(content); return display === undefined ? [] : [display]; }) .join("\n"); yield* replaceText( context, update.sessionUpdate === "user_message" ? "user" : update.sessionUpdate === "agent_message" ? "assistant" : "reasoning", text, update.messageId, ); break; } case "tool_call_content_chunk": { const text = update.content.type === "content" ? acpContentBlockDisplayText(update.content.content) : update.content.type === "diff" ? "changes" in update.content ? update.content.patch?.text : update.content.newText : undefined; if (text !== undefined) { const previous = context.tools.get(update.toolCallId) ?? { toolCallId: update.toolCallId, status: "inProgress" as const, data: { toolCallId: update.toolCallId }, }; yield* emitTool(context, appendToolOutputText(previous, text)); } break; } case "compaction_summary_chunk": { const text = acpContentBlockDisplayText(update.content); if (text !== undefined) { const toolCallId = `acp-compaction:${update.compactionId}`; const previous = context.tools.get(toolCallId) ?? { toolCallId, kind: "think", title: "Compact context", status: "inProgress" as const, data: { toolCallId }, }; yield* emitTool(context, appendToolOutputText(previous, text)); } break; } case "user_message_chunk": { if (update.content.type === "text" && flavor.extractBackgroundToolMutation) { const mutations = flavor.extractBackgroundToolMutation(update.content.text); const terminalTaskIds = mutations .filter((mutation) => mutation.status !== "running") .map((mutation) => mutation.taskId); const reportArrivedBeforeNotice = context.earlyInjectedReportObserved && terminalTaskIds.length > 0; if (reportArrivedBeforeNotice) { context.earlyInjectedReportObserved = false; yield* clearMidTurnUnreportedTaskIds(terminalTaskIds); } for (const mutation of mutations) { const toolCallId = context.toolCallIdsByBackgroundTaskId.get(mutation.taskId); const previous = toolCallId !== undefined ? context.tools.get(toolCallId) : undefined; if (previous !== undefined) { let nextStatus = mutation.status === "running" ? ("inProgress" as const) : mutation.status === "failed" ? ("failed" as const) : ("completed" as const); if (nextStatus !== "inProgress") { yield* markAwaitingBackgroundHydration(context, mutation.taskId); nextStatus = "inProgress"; } yield* emitTool( context, appendToolOutputText( { ...previous, status: nextStatus }, mutation.appendOutput, ), ); } yield* applyBackgroundTaskMutationRunning(mutation); if (mutation.status !== "running" && !reportArrivedBeforeNotice) { yield* markPendingInjectedReport(context, mutation.taskId); } } } if (update.content.type === "text" && flavor.extractSubagentEndNotice) { const notice = flavor.extractSubagentEndNotice(update.content.text); const subagent = notice !== undefined ? context.subagentsBySessionId.get(notice.childSessionId) : undefined; if ( notice !== undefined && subagent !== undefined && (subagent.task.status === "running" || subagent.task.status === "pending") ) { // Agents may answer without hydrating output; finish the subagent so deferred finalize can close. yield* emitSubagent(context, { nativeTaskId: subagent.task.nativeTaskRef?.nativeId ?? notice.childSessionId, prompt: subagent.task.prompt, title: subagent.task.title, model: subagent.task.model, status: notice.status, childSessionId: notice.childSessionId, result: null, suppressNormalTool: true, }); } } break; } default: { const parsed = parseSessionUpdateEvent(notification); for (const event of parsed.events) { if (event._tag === "ToolCallUpdated") { yield* emitTool(context, event.toolCall); } else if (event._tag === "PlanUpdated") { yield* emitPlan(context, event.payload); } else if (event._tag === "NoticeReceived") { yield* emitNotice(context, event); } else if (event._tag === "UnknownUpdate") { yield* emitTool(context, { toolCallId: `${context.nativeTurnId}:unsupported:${event.updateType}`, kind: "other", title: `Unsupported ACP update: ${event.updateType}`, status: "completed", data: { updateType: event.updateType }, }); } } } } yield* rearmDeferredFinalize(context); }); const activeContext = Effect.gen(function* () { const context = yield* Ref.get(activeTurn); if (context === null) { return yield* new EffectAcpErrors.AcpTransportError({ detail: "ACP agent requested input without an active turn", cause: "No active ACP turn", }); } return context; }); const beginApprovalRequest = Effect.fnUntraced(function* ( context: ActiveAcpTurn, params: EffectAcpSchema.RequestPermissionRequest, generation: number, transportRequestId: string, ) { yield* closeTextStreams(context); const parsed = parsePermissionRequest(params); const approvalOptions = flavor.approvalOptions?.(params); const nativeRequestId = params.toolCall.toolCallId; const requestId = yield* idAllocator.allocate.runtimeRequest({ driver, providerTurnId: context.providerTurnId, nativeRequestId, }); const decision = yield* Deferred.make(); const nativeResponseAcknowledgement = yield* Deferred.make< void, EffectAcpErrors.AcpError >(); yield* registerNativeResponseAcknowledgement( generation, transportRequestId, nativeResponseAcknowledgement, ); const now = yield* DateTime.now; const nodeId = idAllocator.derive.approvalNode({ requestId }); const requestKind = providerRequestKind(parsed.kind); const nativeItemRef = { driver, nativeId: nativeRequestId, strength: "weak" as const }; const ordinal = yield* resolveItemOrdinal( context, `${context.nativeTurnId}:approval:${nativeRequestId}`, ); const runtimeRequest: OrchestrationV2RuntimeRequest = { id: requestId, nodeId, providerTurnId: context.providerTurnId, nativeRequestRef: nativeItemRef, kind: requestKind, status: "pending", responseCapability: { type: "live", providerSessionId: input.providerSessionId, }, createdAt: now, resolvedAt: null, }; const node: OrchestrationV2ExecutionNode = { id: nodeId, threadId: context.input.threadId, runId: context.input.runId, parentNodeId: context.input.rootNodeId, rootNodeId: context.input.rootNodeId, kind: "approval_request", status: "waiting", countsForRun: false, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, nativeItemRef, runtimeRequestId: requestId, checkpointScopeId: null, startedAt: now, completedAt: null, }; const turnItem: OrchestrationV2TurnItem = { id: idAllocator.derive.approvalTurnItem({ requestId }), threadId: context.input.threadId, runId: context.input.runId, nodeId, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, nativeItemRef, parentItemId: null, ordinal, status: "waiting", title: null, startedAt: now, completedAt: null, updatedAt: now, type: "approval_request", requestId, requestKind, ...(parsed.detail === undefined ? {} : { prompt: parsed.detail }), ...(approvalOptions === undefined ? {} : { options: approvalOptions }), }; yield* Ref.update(pendingRuntimeRequests, (current) => { const updated = new Map(current); updated.set(String(requestId), { type: "approval", generation, nativeResponseAcknowledgement, requestId, transportRequestId, decision, runtimeRequest, node, turnItem, }); return updated; }); yield* emitProviderEvent({ type: "node.updated", driver, node, }); yield* emitProviderEvent({ type: "runtime_request.updated", driver, threadId: context.input.threadId, runtimeRequest, }); yield* emitProviderEvent({ type: "turn_item.updated", driver, turnItem, }); return { context, decision, nativeResponseAcknowledgement, requestId, transportRequestId, } as const; }); const beginUserInputRequest = Effect.fnUntraced(function* ( request: AcpAdapterV2UserInputRequest, generation: number, transportRequestId: string, ) { const context = yield* activeContext; yield* closeTextStreams(context); const requestId = yield* idAllocator.allocate.runtimeRequest({ driver, providerTurnId: context.providerTurnId, nativeRequestId: request.nativeRequestId, }); const answers = yield* Deferred.make(); const nativeResponseAcknowledgement = yield* Deferred.make< void, EffectAcpErrors.AcpError >(); yield* registerNativeResponseAcknowledgement( generation, transportRequestId, nativeResponseAcknowledgement, ); const now = yield* DateTime.now; const nodeId = providerNodeId(request.nativeItemId); const turnItemId = providerTurnItemId(request.nativeItemId); const nativeItemRef = { driver, nativeId: request.nativeItemId, strength: "weak" as const, }; const ordinal = yield* resolveItemOrdinal(context, request.nativeItemId); const runtimeRequest: OrchestrationV2RuntimeRequest = { id: requestId, nodeId, providerTurnId: context.providerTurnId, nativeRequestRef: { driver, nativeId: request.nativeRequestId, strength: "weak", }, kind: "user_input", status: "pending", responseCapability: { type: "live", providerSessionId: input.providerSessionId, }, createdAt: now, resolvedAt: null, }; const node: OrchestrationV2ExecutionNode = { id: nodeId, threadId: context.input.threadId, runId: context.input.runId, parentNodeId: context.input.rootNodeId, rootNodeId: context.input.rootNodeId, kind: "user_input_request", status: "waiting", countsForRun: false, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, nativeItemRef, runtimeRequestId: requestId, checkpointScopeId: null, startedAt: now, completedAt: null, }; const turnItem: OrchestrationV2TurnItem = { id: turnItemId, threadId: context.input.threadId, runId: context.input.runId, nodeId, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, nativeItemRef, parentItemId: null, ordinal, status: "waiting", title: null, startedAt: now, completedAt: null, updatedAt: now, type: "user_input_request", requestId, questions: [...request.questions], }; yield* Ref.update(pendingRuntimeRequests, (current) => { const updated = new Map(current); updated.set(String(requestId), { type: "user_input", generation, nativeResponseAcknowledgement, requestId, transportRequestId, answers, runtimeRequest, node, turnItem, }); return updated; }); yield* emitProviderEvent({ type: "node.updated", driver, node, }); yield* emitProviderEvent({ type: "runtime_request.updated", driver, threadId: context.input.threadId, runtimeRequest, }); yield* emitProviderEvent({ type: "turn_item.updated", driver, turnItem, }); return { answers, context, nativeResponseAcknowledgement, requestId, transportRequestId, } as const; }); const requestUserInputWithAdmission = Effect.fn( "AcpAdapterV2.requestUserInputWithAdmission", )( function* ( generation: number, request: Effect.Effect, transportRequestId: string, ) { if (!(yield* awaitRequestTurn(generation))) { return { answers: null, acknowledgeNativeResponse: Effect.void }; } return yield* runRuntimeCallbackAtGeneration( generation, request.pipe( Effect.flatMap((value) => beginUserInputRequest(value, generation, transportRequestId), ), ), ).pipe( Effect.flatMap((pending) => { if (Option.isNone(pending)) return Effect.never; const { answers, requestId, transportRequestId } = pending.value; return Deferred.await(answers).pipe( Effect.flatMap((result) => runRuntimeCallbackAtGeneration(generation, Effect.succeed(result)).pipe( Effect.flatMap((checked) => Option.isSome(checked) ? Effect.succeed({ acknowledgeNativeResponse: acknowledgeNativeResponse( generation, transportRequestId, ), answers: checked.value, }) : Effect.never, ), ), ), Effect.ensuring( runRuntimeCallbackAtGeneration( generation, Ref.update(pendingRuntimeRequests, (current) => { const updated = new Map(current); updated.delete(String(requestId)); return updated; }), ).pipe(Effect.asVoid), ), ); }), ); }, Effect.mapError( (cause) => new EffectAcpErrors.AcpTransportError({ detail: "Failed to handle ACP user input request", cause, }), ), ); const cancelPendingRuntimeRequests = Effect.fnUntraced(function* () { const waiting = turnAttached; turnAttached = yield* Deferred.make(); yield* Deferred.succeed(waiting, false); const requests = yield* Ref.modify(pendingRuntimeRequests, (current) => [ [...current.values()], new Map(), ]); if (requests.length === 0) return; const now = yield* DateTime.now; yield* Effect.forEach( requests, (request) => Effect.gen(function* () { const cancelled = yield* request.type === "approval" ? Deferred.succeed(request.decision, "cancel") : Deferred.succeed(request.answers, null); if (!cancelled) return; yield* emitProviderEvent({ type: "runtime_request.updated", driver, threadId: request.node.threadId, runtimeRequest: { ...request.runtimeRequest, status: "cancelled", resolvedAt: now, }, }); yield* emitProviderEvent({ type: "node.updated", driver, node: { ...request.node, status: "cancelled", completedAt: now, }, }); yield* emitProviderEvent({ type: "turn_item.updated", driver, turnItem: { ...request.turnItem, status: "cancelled", completedAt: now, updatedAt: now, }, }); }), { concurrency: 1, discard: true }, ); }); const mutateCarryoverSubagentStatus = Effect.fnUntraced(function* ( subagent: ActiveAcpSubagent, status: OrchestrationV2Subagent["status"], resultOverride?: string | null, ) { const now = yield* DateTime.now; const result = resultOverride !== undefined && resultOverride !== null && resultOverride.length > 0 ? resultOverride : (subagent.task.result ?? (subagent.assistantText || null)); const completedAt = acpSubagentStatusIsTerminal(status) ? now : null; subagent.task = { ...subagent.task, status, result, completedAt, updatedAt: now, }; }); const projectCarryoverSubagentStatus = Effect.fnUntraced(function* ( subagent: ActiveAcpSubagent, status: OrchestrationV2Subagent["status"], resultOverride?: string | null, ) { yield* mutateCarryoverSubagentStatus(subagent, status, resultOverride); const now = subagent.task.updatedAt; const nativeTaskId = subagent.task.nativeTaskRef?.nativeId ?? subagent.task.id; const nativeItemRef = { driver, nativeId: nativeTaskId, strength: "strong" as const, }; const parentProviderThreadId = subagent.parentProviderThreadId; const result = subagent.task.result; const completedAt = subagent.task.completedAt; yield* emitProviderEvent({ type: "node.updated", driver, node: { id: subagent.task.id, threadId: subagent.task.threadId, runId: subagent.task.runId, parentNodeId: subagent.task.parentNodeId, rootNodeId: subagent.task.parentNodeId, kind: "subagent", status, countsForRun: false, providerThreadId: parentProviderThreadId, providerTurnId: subagent.providerTurnId, nativeItemRef, runtimeRequestId: null, checkpointScopeId: null, startedAt: subagent.task.startedAt, completedAt, }, }); yield* emitProviderEvent({ type: "node.updated", driver, node: { id: subagent.childRootNodeId, threadId: subagent.childThreadId, runId: null, parentNodeId: null, rootNodeId: subagent.childRootNodeId, kind: "root_turn", status, countsForRun: false, providerThreadId: subagent.task.providerThreadId, providerTurnId: null, nativeItemRef, runtimeRequestId: null, checkpointScopeId: null, startedAt: subagent.task.startedAt, completedAt, }, }); yield* emitProviderEvent({ type: "subagent.updated", driver, subagent: subagent.task, }); yield* emitProviderEvent({ type: "turn_item.updated", driver, turnItem: { id: subagent.turnItemId, threadId: subagent.task.threadId, runId: subagent.task.runId, nodeId: subagent.task.id, providerThreadId: parentProviderThreadId, providerTurnId: subagent.providerTurnId, nativeItemRef, parentItemId: null, ordinal: subagent.turnItemOrdinal, status, title: subagent.task.title, startedAt: subagent.task.startedAt, completedAt, updatedAt: now, type: "subagent", subagentId: subagent.task.id, origin: "provider_native", driver, providerInstanceId: subagent.task.providerInstanceId, childThreadId: subagent.childThreadId, prompt: subagent.task.prompt, result, }, }); if (acpSubagentStatusIsTerminal(status)) { subagent.terminalStatusProjected = true; } }); const updateCarryoverSubagentStatus = Effect.fnUntraced(function* ( nativeIdOrChildSessionId: string, status: OrchestrationV2Subagent["status"], result?: string | null, options?: { readonly project?: boolean }, ) { const carryover = yield* Ref.get(carryoverSubagents); if (carryover === null) return false; const match = carryover.subagents.find((subagent) => { const nativeId = subagent.task.nativeTaskRef?.nativeId ?? String(subagent.task.id); return ( nativeId === nativeIdOrChildSessionId || subagent.childSessionId === nativeIdOrChildSessionId ); }); if (match === undefined) return false; const shouldProject = options?.project !== false; if (match.task.status === status) { if (!shouldProject || match.terminalStatusProjected) { return true; } yield* projectCarryoverSubagentStatus(match, status, result); return true; } if (match.task.status !== "running" && match.task.status !== "pending") { return false; } if (!shouldProject) { yield* mutateCarryoverSubagentStatus(match, status, result); } else { yield* projectCarryoverSubagentStatus(match, status, result); } return true; }); const finishSubagentFromNotice = Effect.fnUntraced(function* (notice: { readonly childSessionId: string; readonly status: "completed" | "failed" | "cancelled"; readonly result: string | null; }) { const context = yield* Ref.get(activeTurn); const subagent = context === null ? undefined : context.subagentsBySessionId.get(notice.childSessionId); const known = subagent ?? (yield* Ref.get(carryoverSubagents))?.subagents.find( (candidate) => candidate.childSessionId === notice.childSessionId || candidate.task.nativeTaskRef?.nativeId === notice.childSessionId, ); if (known !== undefined && acpSubagentStatusBlocksTurnSettlement(known.task.status)) { yield* recordWakeReport(notice.childSessionId, { kind: "subagent", label: known.task.title ?? known.task.prompt, outcome: notice.status, childThreadId: known.childThreadId, }); } if (context !== null && subagent !== undefined && !context.finalized) { if (!acpSubagentStatusBlocksTurnSettlement(subagent.task.status)) return; yield* emitSubagent(context, { nativeTaskId: subagent.task.nativeTaskRef?.nativeId ?? notice.childSessionId, prompt: subagent.task.prompt, title: subagent.task.title, model: subagent.task.model, status: notice.status, childSessionId: notice.childSessionId, result: notice.result, suppressNormalTool: true, }); yield* rearmDeferredFinalize(context); return; } const carryover = yield* Ref.get(carryoverSubagents); yield* updateCarryoverSubagentStatus( notice.childSessionId, notice.status, notice.result, { project: carryover !== null && carryover.sessionId === (yield* Ref.get(activeSessionId)) && carryover.projectsLateTerminals, }, ); }); applyFinalizedActiveTurnSubagentTerminal = Effect.fnUntraced(function* ( context: ActiveAcpTurn, notification: EffectAcpSchema.SessionNotification, ) { if (flavor.extractSubagentUpdate === undefined) return false; const applyTerminal = Effect.fnUntraced(function* ( subagent: ActiveAcpSubagent, update: AcpAdapterV2SubagentUpdate, ) { if (!acpSubagentStatusIsTerminal(update.status)) return false; if (acpSubagentStatusIsTerminal(subagent.task.status)) { if (subagent.terminalStatusProjected || context.finalizedStatus !== "completed") { return true; } } if (context.finalizedStatus === "completed") { yield* emitSubagent(context, update); } else { yield* mutateCarryoverSubagentStatus(subagent, update.status, update.result); } return true; }); const sessionUpdate = notification.update; if ( sessionUpdate.sessionUpdate === "tool_call" || sessionUpdate.sessionUpdate === "tool_call_update" ) { for (const event of parseSessionUpdateEvent(notification).events) { if (event._tag !== "ToolCallUpdated") continue; const toolCall = flavor.normalizeToolCall?.(event.toolCall) ?? event.toolCall; const subagentUpdate = extractSubagentUpdate(toolCall); if ( subagentUpdate === undefined || !acpSubagentStatusIsTerminal(subagentUpdate.status) ) { continue; } const subagent = context.subagents.get(subagentUpdate.nativeTaskId) ?? (subagentUpdate.childSessionId === null ? undefined : context.subagentsBySessionId.get(subagentUpdate.childSessionId)) ?? context.subagentsBySessionId.get(notification.sessionId); if (subagent === undefined) continue; const nativeTaskId = subagent.task.nativeTaskRef?.nativeId ?? String(subagent.task.id); if ( subagentUpdate.nativeTaskId !== nativeTaskId && (subagentUpdate.childSessionId === null || (subagentUpdate.childSessionId !== subagent.childSessionId && subagentUpdate.childSessionId !== notification.sessionId)) ) { continue; } if (yield* applyTerminal(subagent, subagentUpdate)) return true; } } if ( sessionUpdate.sessionUpdate === "user_message_chunk" && sessionUpdate.content.type === "text" && flavor.extractSubagentEndNotice !== undefined ) { const notice = flavor.extractSubagentEndNotice(sessionUpdate.content.text); const subagent = notice === undefined ? undefined : context.subagentsBySessionId.get(notice.childSessionId); if (notice !== undefined && subagent !== undefined) { return yield* applyTerminal(subagent, { nativeTaskId: subagent.task.nativeTaskRef?.nativeId ?? String(subagent.task.id), prompt: subagent.task.prompt, title: subagent.task.title, model: subagent.task.model, status: notice.status, childSessionId: notice.childSessionId, result: null, suppressNormalTool: true, }); } } return false; }); const projectDeferredCarryoverTerminals = Effect.fnUntraced(function* ( subagents: ReadonlyArray, ) { for (const subagent of subagents) { if ( subagent.task.status === "running" || subagent.task.status === "pending" || subagent.terminalStatusProjected ) { continue; } yield* projectCarryoverSubagentStatus( subagent, subagent.task.status, subagent.task.result, ); } }); const terminalizeCarryoverSubagents = Effect.fnUntraced(function* ( carryover: AcpCarryoverSubagents | null, ) { if (carryover === null) return; for (const subagent of carryover.subagents) { yield* terminalizeSubagentTools(subagent); if (acpSubagentStatusIsTerminal(subagent.task.status)) { if (!subagent.terminalStatusProjected) { yield* projectCarryoverSubagentStatus( subagent, subagent.task.status, subagent.task.result, ); } continue; } yield* projectCarryoverSubagentStatus(subagent, "interrupted"); } }); const projectAcpRuntimeSessionUpdateEffect = /* Provider events include streaming deltas; avoid a span for each update. */ Effect.fnUntraced( function* (rawNotification: EffectAcpSchema.SessionNotification) { const embedded = embeddedTerminalIdsFromSessionUpdate(rawNotification); if (embedded !== undefined) { rememberEmbeddedTerminals({ ...embedded, sessionId: rawNotification.sessionId, }); } const notification = clientTerminals === undefined ? rawNotification : resolveEmbeddedTerminalContent( rawNotification, clientTerminals.readOutputSnapshot, ); if (notification.update.sessionUpdate === "available_commands_update") { yield* ( flavor.onAvailableCommandsUpdate?.(notification.update.availableCommands) ?? Effect.void ); } if ( notification.update.sessionUpdate === "config_option_update" || notification.update.sessionUpdate === "current_mode_update" ) { yield* ( flavor.onSessionConfigurationUpdate?.( yield* runtime.getConfigOptions, yield* runtime.getModeState, (yield* Ref.get(activeSessionSetup))?.sessionSetupResult.models, ) ?? Effect.void ); } yield* handleSessionUpdate( flavor.normalizeSessionUpdate?.(notification) ?? notification, ); }, Effect.mapError( (cause) => new EffectAcpErrors.AcpTransportError({ detail: "Failed to project an ACP session update", cause, }), ), ); const projectAcpRuntimeSessionUpdate = ( handlerGeneration: number, rawNotification: EffectAcpSchema.SessionNotification, ) => runRuntimeCallbackAtGeneration( handlerGeneration, projectAcpRuntimeSessionUpdateEffect(rawNotification), ).pipe(Effect.asVoid); // Post-settle background work retains the latest turn policy. const clientPolicyContext = Effect.map(Ref.get(activeTurn), (context) => ({ policy: context?.input.runtimePolicy ?? latestRuntimePolicy, turnKey: context === null ? null : String(context.providerTurnId), })); const denyClientRequest = (operation: string, disposition: "ask" | "deny") => Effect.logWarning("ACP client policy denied a client-mediated operation", { operation, disposition, }).pipe( Effect.andThen( Effect.fail( EffectAcpErrors.AcpRequestError.internalError( disposition === "ask" ? `The active T3 runtime policy requires approval for ${operation}. Request permission with session/request_permission before retrying.` : `The active T3 runtime policy does not allow ${operation}.`, ), ), ), ); const guardClientTerminalCreate = clientPolicyContext.pipe( Effect.flatMap(({ policy, turnKey }) => { const disposition = acpClientExecuteDisposition(policy); if ( disposition === "allow" || (disposition === "ask" && clientPolicyGrants.allowsExecute(turnKey)) ) { return Effect.void; } return denyClientRequest("terminal/create", disposition); }), ); const wireAcpRuntimeTerminalHandlers = Effect.fnUntraced(function* ( targetRuntime: AcpSessionRuntime.AcpSessionRuntime["Service"], ) { if (clientTerminals === undefined) return; yield* targetRuntime.handleCreateTerminal((request) => guardClientTerminalCreate.pipe(Effect.andThen(clientTerminals.create(request))), ); yield* targetRuntime.handleTerminalOutput(clientTerminals.output); yield* targetRuntime.handleTerminalWaitForExit(clientTerminals.waitForExit); yield* targetRuntime.handleTerminalKill(clientTerminals.kill); yield* targetRuntime.handleTerminalRelease(clientTerminals.release); }); const wireAcpRuntimeMcpHandlers = Effect.fnUntraced(function* ( targetRuntime: AcpSessionRuntime.AcpSessionRuntime["Service"], mcpBridge: AcpMcpOverAcpBridge | undefined, ) { if (mcpBridge === undefined) return; const mapMcpBridgeError = Effect.mapError( (error: Error) => new EffectAcpErrors.AcpRequestError({ code: -32603, errorMessage: error.message, cause: error, }), ); yield* targetRuntime.handleMcpConnect((request) => mcpBridge.connect(request).pipe(mapMcpBridgeError), ); yield* targetRuntime.handleMcpMessage((request) => mcpBridge.message(request).pipe(mapMcpBridgeError), ); yield* targetRuntime.handleMcpNotification((request) => mcpBridge.notification(request).pipe(mapMcpBridgeError), ); yield* targetRuntime.handleMcpDisconnect((request) => mcpBridge.disconnect(request).pipe(mapMcpBridgeError), ); }); const wireAcpRuntimeHandlers = Effect.fnUntraced(function* ( targetRuntime: AcpSessionRuntime.AcpSessionRuntime["Service"], handlerGeneration: number, handlerOptions: { readonly sessionUpdates?: boolean; readonly terminals?: boolean; readonly mcp?: boolean; } = {}, ) { yield* targetRuntime.getEvents().pipe( Stream.runForEach((event) => { if (event._tag === "EventStreamBarrier") { return Deferred.succeed(event.acknowledge, undefined).pipe(Effect.asVoid); } return runRuntimeCallbackAtGeneration( handlerGeneration, flavor.onSessionEvent?.(event) ?? Effect.void, ).pipe(Effect.asVoid); }), Effect.forkIn(runtimeScope ?? sessionScope), ); const requestUserInput = ( request: AcpAdapterV2UserInputRequest, requestContext: EffectAcpProtocol.AcpRequestContext, ) => requestUserInputWithAdmission( handlerGeneration, Effect.succeed(request), requestContext.requestId, ); const clientFileSystem = flavor.clientFileSystem; if (clientFileSystem !== undefined) { yield* targetRuntime.handleReadTextFile((request) => clientPolicyContext.pipe( Effect.flatMap(({ policy }) => clientFileSystem.readTextFile(request, policy.cwd)), ), ); const writeTextFile = clientFileSystem.writeTextFile; if (writeTextFile !== undefined) { yield* targetRuntime.handleWriteTextFile((request) => clientPolicyContext.pipe( Effect.flatMap(({ policy }) => writeTextFile(request, policy.cwd)), ), ); } } if (handlerOptions.mcp !== false) { yield* wireAcpRuntimeMcpHandlers(targetRuntime, runtimeMcpBridge); } if (handlerOptions.terminals !== false) { yield* wireAcpRuntimeTerminalHandlers(targetRuntime); } if (handlerOptions.sessionUpdates !== false) { yield* targetRuntime.handleSessionUpdate((rawNotification) => projectAcpRuntimeSessionUpdate(handlerGeneration, rawNotification), ); } yield* targetRuntime.handleRequestPermission((nativeParams, requestContext) => Effect.gen(function* () { const trackedTurn = yield* Ref.get(activeTurn); const params = trackedTurn !== null && nativeParams.sessionId === (yield* Ref.get(activeSessionId)) ? withTrackedPermissionToolCall( nativeParams, trackedTurn.tools.get(nativeParams.toolCall.toolCallId), ) : nativeParams; const transportRequestId = requestContext.requestId; const permissionQuestion = flavor.extractPermissionQuestion?.(params); if (permissionQuestion !== undefined) { const userInput = yield* requestUserInputWithAdmission( handlerGeneration, Effect.succeed({ nativeItemId: `${params.sessionId}:question:${params.toolCall.toolCallId}`, nativeMethod: "session/request_permission", nativeRequestId: params.toolCall.toolCallId, nativeSessionId: params.sessionId, questions: [permissionQuestion.question], }), transportRequestId, ); const response = userInput.answers === null ? undefined : permissionQuestion.respond(userInput.answers); yield* userInput.acknowledgeNativeResponse; return response ?? ({ outcome: { outcome: "cancelled" } } as const); } const dispositionFor = (policy: ProviderAdapter.ProviderAdapterV2RuntimePolicy) => (flavor.permissionDisposition ?? acpPermissionDisposition)(policy, params); if (followsAgentWorkState && dispositionFor(latestRuntimePolicy) === "ask") { if (!(yield* awaitRequestTurn(handlerGeneration))) return { outcome: { outcome: "cancelled" } } as const; } const admitted = yield* runRuntimeCallbackAtGeneration( handlerGeneration, Effect.gen(function* () { const context = followsAgentWorkState ? yield* Ref.get(activeTurn) : yield* activeContext; const disposition = dispositionFor( context?.input.runtimePolicy ?? latestRuntimePolicy, ); if (disposition === "allow") { const optionId = selectAutoApprovedPermissionOption(params); return { _tag: "Immediate" as const, response: optionId === undefined ? ({ outcome: { outcome: "cancelled" } } as const) : ({ outcome: { outcome: "selected", optionId } } as const), }; } if (disposition === "deny") { const optionId = selectPermissionOptionId(params, "decline"); return { _tag: "Immediate" as const, response: optionId === undefined ? ({ outcome: { outcome: "cancelled" } } as const) : ({ outcome: { outcome: "selected", optionId } } as const), }; } return { _tag: "Pending" as const, pending: yield* beginApprovalRequest( yield* activeContext, params, handlerGeneration, transportRequestId, ), }; }), ); if (Option.isNone(admitted)) { return yield* Effect.never; } if (admitted.value._tag === "Immediate") { const response = admitted.value.response; const checked = yield* runRuntimeCallbackAtGeneration( handlerGeneration, Effect.gen(function* () { const nativeResponseAcknowledgement = yield* Deferred.make< void, EffectAcpErrors.AcpError >(); yield* registerNativeResponseAcknowledgement( handlerGeneration, transportRequestId, nativeResponseAcknowledgement, ); return response; }), ); return Option.isSome(checked) ? checked.value : yield* Effect.never; } const { context, decision: pendingDecision, requestId, transportRequestId: pendingTransportRequestId, } = admitted.value.pending; const parsedPermission = parsePermissionRequest(params); const decision = yield* Deferred.await(pendingDecision).pipe( Effect.ensuring( runRuntimeCallbackAtGeneration( handlerGeneration, Ref.update(pendingRuntimeRequests, (current) => { const updated = new Map(current); updated.delete(String(requestId)); return updated; }), ).pipe(Effect.asVoid), ), ); if ( parsedPermission.kind !== "unknown" && (decision === "accept" || decision === "acceptForSession") ) { clientPolicyGrants.recordApproval({ kind: providerRequestKind(parsedPermission.kind), scope: decision === "acceptForSession" ? "session" : "turn", turnKey: String(context.providerTurnId), }); } const response = (() => { if (decision === "cancel") { return { outcome: { outcome: "cancelled" } } as const; } const optionId = selectPermissionOptionId(params, decision); return optionId === undefined ? ({ outcome: { outcome: "cancelled" } } as const) : ({ outcome: { outcome: "selected", optionId } } as const); })(); const checked = yield* runRuntimeCallbackAtGeneration( handlerGeneration, Effect.succeed(response), ); if (Option.isNone(checked)) return yield* Effect.never; yield* acknowledgeNativeResponse(handlerGeneration, pendingTransportRequestId); return checked.value; }).pipe( Effect.mapError( (cause) => new EffectAcpErrors.AcpTransportError({ detail: "Failed to handle an ACP permission request", cause, }), ), holdingAgentRequest(handlerGeneration), ), ); yield* targetRuntime.handleElicitation((params, requestContext) => Effect.gen(function* () { const transportRequestId = requestContext.requestId; if ( params.mode === "form" && (unknownRecord(params._meta)?.codex_approval_kind === "mcp_tool_call" || transportRequestId.startsWith("mcp_tool_call_approval_")) ) { const mcpApprovalDisposition = yield* runRuntimeCallbackAtGeneration( handlerGeneration, Effect.gen(function* () { const context = yield* activeContext; const disposition = acpMcpToolApprovalElicitationDisposition( context.input.runtimePolicy, params, transportRequestId, ); if (disposition === undefined || disposition === "ask") { return disposition; } const nativeResponseAcknowledgement = yield* Deferred.make< void, EffectAcpErrors.AcpError >(); yield* registerNativeResponseAcknowledgement( handlerGeneration, transportRequestId, nativeResponseAcknowledgement, ); return disposition; }), ); if (Option.isNone(mcpApprovalDisposition)) return yield* Effect.never; if (mcpApprovalDisposition.value === "allow") { return { action: "accept", content: {} } as const; } } if ( params.mode === "url" && "url" in params && typeof params.url === "string" && "elicitationId" in params && typeof params.elicitationId === "string" ) { const { elicitationId, url } = params; const admitted = yield* runRuntimeCallbackAtGeneration( handlerGeneration, Effect.gen(function* () { const nativeResponseAcknowledgement = yield* Deferred.make< void, EffectAcpErrors.AcpError >(); yield* registerNativeResponseAcknowledgement( handlerGeneration, transportRequestId, nativeResponseAcknowledgement, ); const accepted = yield* ( flavor.onUrlElicitation?.({ elicitationId, url, message: params.message, }) ?? Effect.succeed(false) ); return accepted ? ({ action: "accept" } as const) : ({ action: "decline" } as const); }), ); if (Option.isNone(admitted)) return yield* Effect.never; return admitted.value; } if (params.mode !== "form" || !("requestedSchema" in params)) { return { action: "decline" } as const; } const requestedSchema = unknownRecord(params.requestedSchema); const properties = unknownRecord(requestedSchema?.properties) ?? {}; const nativeForm = flavor.extractElicitationForm?.(params); const elicitationScopeId = "sessionId" in params ? params.sessionId : `request:${params.requestId}`; const questions = nativeForm?.questions ?? Object.entries(properties).map( ([id, property], index): OrchestrationV2UserInputQuestion => { const record = unknownRecord(property); const enumValues = Array.isArray(record?.enum) ? record.enum.filter((value): value is string => typeof value === "string") : []; const options = enumValues.length > 0 ? enumValues.map((value) => ({ label: value, description: value })) : record?.type === "boolean" ? [ { label: "true", description: "Yes" }, { label: "false", description: "No" }, ] : []; return { id, header: nonEmptyText(record?.title, `Question ${index + 1}`), question: nonEmptyText(record?.description, params.message), options, }; }, ); const userInput = yield* requestUserInputWithAdmission( handlerGeneration, Effect.gen(function* () { const ordinal = yield* Ref.getAndUpdate( nextElicitationOrdinal, (current) => current + 1, ); const nativeRequestId = `${elicitationScopeId}:elicitation:${ordinal}`; return { nativeItemId: nativeRequestId, nativeRequestId, questions, }; }), transportRequestId, ); const response = userInput.answers === null ? ({ action: "cancel" } as const) : (nativeForm?.respond(userInput.answers) ?? ({ action: "accept", content: elicitationContent( userInput.answers, new Set(Object.keys(properties)), ), } as const)); yield* userInput.acknowledgeNativeResponse; return response; }).pipe(holdingAgentRequest(handlerGeneration)), ); if (flavor.registerExtensions !== undefined) { yield* flavor.registerExtensions({ runtime: targetRuntime, reportProviderRetry: (notice) => runRuntimeCallbackAtGeneration(handlerGeneration, reportProviderRetry(notice)).pipe( Effect.asVoid, ), requestUserInput, captureProposedPlan, lastProposedPlanMarkdown, finishSubagent: (notice) => runRuntimeCallbackAtGeneration( handlerGeneration, Effect.gen(function* () { if (yield* Ref.get(stoppedRunQuarantine)) return; if ((yield* Ref.get(activeSessionId)) !== notice.sessionId) return; yield* finishSubagentFromNotice(notice); }), ).pipe(Effect.asVoid), applyBackgroundTaskMutation: (incoming) => runRuntimeCallbackAtGeneration( handlerGeneration, Effect.gen(function* () { if (yield* Ref.get(stoppedRunQuarantine)) return; if ((yield* Ref.get(activeSessionId)) !== incoming.sessionId) return; const context = yield* Ref.get(activeTurn); let mutation = incoming; if ( incoming.status !== "running" && context !== null && !context.finalized && (incoming.report?.kind === "command" || context.backgroundCommands.has(incoming.taskId)) ) { const previous = context.backgroundCommands.get(incoming.taskId); const toolCallId = context.toolCallIdsByBackgroundTaskId.get(incoming.taskId); const tool = toolCallId === undefined ? undefined : context.tools.get(toolCallId); const output = incoming.output ?? previous?.output ?? (tool === undefined ? undefined : toolOutputText(tool)); // Missing exit codes decode as success; they cannot undo a known failure. const completion = { status: previous?.status === "failed" ? "failed" : incoming.status, ...(output === undefined ? {} : { output }), }; context.backgroundCommands.set(incoming.taskId, completion); mutation = { ...incoming, ...completion }; } if ( mutation.status !== "running" && mutation.report !== undefined && !(yield* Ref.get(handledBackgroundTaskIdsInActiveTurn)).has(mutation.taskId) ) { yield* recordWakeReport(mutation.taskId, { ...mutation.report, outcome: mutation.status, }); } yield* applyLateBackgroundMutation(mutation.sessionId, mutation); if (mutation.status !== "running") { yield* finishRegisteredBackgroundTool({ taskId: mutation.taskId, status: mutation.status, ...(mutation.output === undefined ? {} : { output: mutation.output }), }); } }), ).pipe(Effect.asVoid), }); } }); const makeRuntimeMcpBridge = Effect.fnUntraced(function* ( mcpContext: AcpMcpContext, scope: Scope.Scope, ) { if (mcpContext.endpoint === undefined || mcpContext.authorization === undefined) { return undefined; } const mcpBridge = yield* makeAcpMcpOverAcpBridge({ endpoint: mcpContext.endpoint, authorization: mcpContext.authorization, allocateConnectionId: crypto.randomUUIDv4().pipe(Effect.orDie), }); yield* Scope.addFinalizer(scope, mcpBridge.dispose); return mcpBridge; }); const spawnAcpRuntime = Effect.fnUntraced(function* ( threadId: ThreadId | null, resumeSessionId?: string, ) { if (runtimeScope !== undefined) { yield* Scope.close(runtimeScope, Exit.void); } runtimeScope = yield* Scope.make(); const mcpContext = yield* readMcpContext(threadId); runtimeMcpBridge = yield* makeRuntimeMcpBridge(mcpContext, runtimeScope); const runtimeGeneration = yield* Ref.get(runtimeCallbackGeneration); runtime = yield* flavor .makeRuntime(makeRuntimeInput(runtimeGeneration, mcpContext, resumeSessionId)) .pipe( Effect.provideService(Scope.Scope, runtimeScope), Effect.provideService(Crypto.Crypto, crypto), Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, childProcessSpawner), Effect.provideService(AgentScope, agentScope), Effect.provideService(AgentScopeThreadId, threadId ?? undefined), ); }); const startAcpRuntime = Effect.fnUntraced(function* ( threadId: ThreadId | null, resumeSessionId?: string, ) { const startup = Effect.gen(function* () { yield* spawnAcpRuntime(threadId, resumeSessionId); yield* wireAcpRuntimeHandlers(runtime, yield* Ref.get(runtimeCallbackGeneration)); return yield* runtime.start(); }); return yield* flavor.withRuntimeStartup?.(startup) ?? startup; }); const restartAcpRuntime = Effect.fnUntraced(function* (threadId: ThreadId | null) { yield* spawnAcpRuntime(threadId); yield* wireAcpRuntimeHandlers(runtime, yield* Ref.get(runtimeCallbackGeneration)); }); const startReplacementAcpRuntime = Effect.fnUntraced(function* ( threadId: ThreadId | null, commitSessionState: ( replacement: AcpSessionRuntime.AcpSessionRuntimeStartResult, ) => Effect.Effect, ) { const previousScope = runtimeScope; const previousGeneration = yield* Ref.get(runtimeCallbackGeneration); yield* runtimeCallbackPermit.withPermit(awaitAdmittedNativeResponses); // Keep live callbacks active until replacement startup succeeds. const replacementGeneration = yield* allocateRuntimeCallbackGeneration; const replacementScope = yield* Scope.make(); type CandidateLifecycle = | { readonly _tag: "Starting" } | { readonly _tag: "Terminated"; readonly error: EffectAcpErrors.AcpError } | { readonly _tag: "Committed" }; const candidateLifecycle = yield* Ref.make({ _tag: "Starting" }); const handleCandidateTermination: AcpAdapterV2RuntimeInput["onTermination"] = (error) => Ref.modify( candidateLifecycle, (current): readonly [Effect.Effect, CandidateLifecycle] => { if (current._tag === "Committed") { return [handleRuntimeTerminationAtGeneration(replacementGeneration), current]; } return [ Effect.void, current._tag === "Terminated" ? current : ({ _tag: "Terminated", error } as const), ]; }, ).pipe(Effect.flatten); const stagedSessionUpdates: Array = []; let handleCandidateSessionUpdate: ( notification: EffectAcpSchema.SessionNotification, ) => Effect.Effect = (notification) => Effect.sync(() => { stagedSessionUpdates.push(notification); }); let replacementMcpBridge: AcpMcpOverAcpBridge | undefined; const startup = Effect.gen(function* () { const mcpContext = yield* readMcpContext(threadId); replacementMcpBridge = yield* makeRuntimeMcpBridge(mcpContext, replacementScope); const replacementRuntime = yield* flavor .makeRuntime( makeRuntimeInput( replacementGeneration, mcpContext, undefined, handleCandidateTermination, ), ) .pipe( Effect.provideService(Scope.Scope, replacementScope), Effect.provideService(Crypto.Crypto, crypto), Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, childProcessSpawner), Effect.provideService(AgentScope, agentScope), Effect.provideService(AgentScopeThreadId, threadId ?? undefined), ); yield* replacementRuntime.handleSessionUpdate((notification) => Effect.suspend(() => handleCandidateSessionUpdate(notification)), ); yield* wireAcpRuntimeMcpHandlers(replacementRuntime, replacementMcpBridge); const started = yield* replacementRuntime.start(); return { replacementRuntime, started }; }); const replacementExit = yield* Effect.exit( (flavor.withRuntimeStartup?.(startup) ?? startup).pipe( Effect.onInterrupt(() => Scope.close(replacementScope, Exit.void).pipe(Effect.ignore), ), ), ); if (Exit.isFailure(replacementExit)) { yield* runtimeCallbackPermit.withPermit( quarantineNativeTransportAtGeneration(replacementGeneration), ); yield* Scope.close(replacementScope, Exit.void).pipe(Effect.ignore); return yield* Effect.failCause(replacementExit.cause); } yield* runtimeCallbackPermit .withPermit( Effect.uninterruptible( Effect.gen(function* () { yield* awaitAdmittedNativeResponses; yield* wireAcpRuntimeHandlers( replacementExit.value.replacementRuntime, replacementGeneration, { sessionUpdates: false, terminals: false, mcp: false }, ); const lifecycle = yield* Ref.modify( candidateLifecycle, (current): readonly [CandidateLifecycle, CandidateLifecycle] => current._tag === "Terminated" ? [current, current] : [{ _tag: "Committed" }, { _tag: "Committed" }], ); if (lifecycle._tag === "Terminated") { return yield* lifecycle.error; } yield* quarantineNativeTransportAtGeneration(previousGeneration); runtime = replacementExit.value.replacementRuntime; runtimeScope = replacementScope; runtimeMcpBridge = replacementMcpBridge; yield* Ref.set(runtimeCallbackGeneration, replacementGeneration); prepareTerminalEnvironment( yield* readMcpContext(threadId), replacementExit.value.started.sessionId, ); yield* wireAcpRuntimeTerminalHandlers(replacementExit.value.replacementRuntime); yield* commitSessionState(replacementExit.value.started); while (true) { const buffered = stagedSessionUpdates.splice(0, stagedSessionUpdates.length); if (buffered.length === 0) { handleCandidateSessionUpdate = (notification) => projectAcpRuntimeSessionUpdate(replacementGeneration, notification); break; } yield* Effect.forEach( buffered, (notification) => projectAcpRuntimeSessionUpdateEffect(notification).pipe( Effect.catchCause((cause) => Effect.logError("failed to replay staged ACP session update", { driver, cause, }), ), ), { discard: true }, ); } yield* cancelPendingRuntimeRequests(); if (previousScope !== undefined) { yield* Scope.close(previousScope, Exit.void).pipe( Effect.catchCause((cause) => Effect.logError("failed to close replaced ACP runtime scope", { driver, cause, }), ), ); } }), ), ) .pipe( Effect.onExit((exit) => Exit.isFailure(exit) ? Ref.get(candidateLifecycle).pipe( Effect.flatMap((lifecycle) => lifecycle._tag === "Committed" ? Effect.void : Scope.close(replacementScope, Exit.void).pipe(Effect.ignore), ), ) : Effect.void, ), ); return replacementExit.value.started; }); const commitReplacementSession = Effect.fn("AcpAdapterV2.commitReplacementSession")( function* ( threadId: ThreadId | null, candidate: AcpSessionRuntime.AcpSessionRuntimeStartResult, ) { rememberTerminalEnvironment(candidate.sessionId, yield* readMcpContext(threadId)); yield* Ref.set(runtimeRestartRequired, false); yield* Ref.set(activeSessionId, candidate.sessionId); yield* setActiveSessionSetup(candidate); yield* Ref.set(activeSelection, null); yield* Ref.set(activeInteractionMode, null); yield* Ref.set(promptInstructionStates, new Map()); yield* Ref.set(compactionMarkers, new Map()); yield* Ref.set(providerTurns, new Map()); yield* Ref.set(snapshot, { order: [], messages: new Map(), loadingRole: null, loadingMessageId: null, loadingIndex: 0, }); yield* continuationPermit.withPermit( Effect.gen(function* () { yield* Ref.update(continuationGeneration, (value) => value + 1); yield* Ref.set(stoppedRunQuarantine, false); yield* Ref.set(wakeBuffer, []); yield* Ref.set(wakeReports, noWakeReports); yield* Ref.set(continuationRequested, false); yield* Ref.set(runningBackgroundTaskIds, new Set()); yield* Ref.set(endedBackgroundTaskIds, new Set()); yield* Ref.set(midTurnUnreportedCompletedTaskIds, new Set()); yield* Ref.set(handledBackgroundTaskIdsInActiveTurn, new Set()); yield* Ref.set(carryoverSubagents, null); yield* Ref.set(suppressPostSettleMonitorPrompt, false); yield* Ref.set(lastTurnRoute, null); }), ); }, ); const initialStart = yield* Effect.result( startAcpRuntime(input.threadId, input.initialNativeThreadId), ); const started = Result.isSuccess(initialStart) ? initialStart.success : yield* Effect.gen(function* () { const failedMethod = "method" in initialStart.failure ? initialStart.failure.method : undefined; if ( input.initialNativeThreadId === undefined || (failedMethod !== "session/load" && failedMethod !== "session/resume") ) { return yield* initialStart.failure; } yield* Ref.set(initialSessionActivationFailure, { sessionId: input.initialNativeThreadId, error: initialStart.failure, }); itemIdentityVersion = 2; yield* Ref.set(runtimeRestartRequired, false); prepareClaimableTerminalEnvironment(yield* readMcpContext(input.threadId)); return yield* startAcpRuntime(input.threadId); }); yield* Ref.set(activeSessionId, started.sessionId); yield* setActiveSessionSetup(started); rememberTerminalEnvironment(started.sessionId, yield* readMcpContext(input.threadId)); const capabilities = negotiatedCapabilities(flavor.capabilities, started); const canLoadSession = started.initializeResult.agentCapabilities?.loadSession === true; const canResumeSession = started.initializeResult.agentCapabilities?.sessionCapabilities?.resume != null; const supportsImagePrompts = acpSupportsImagePrompts({ flavorSupportsImagePrompts: flavor.supportsImagePrompts, negotiatedImage: started.initializeResult.agentCapabilities?.promptCapabilities?.image === true, }); const activateSession = Effect.fnUntraced(function* ( sessionId: string, threadId: ThreadId | null, ) { const initialFailure = yield* Ref.modify(initialSessionActivationFailure, (failure) => failure?.sessionId === sessionId ? [failure.error, null] : [undefined, failure], ); if (initialFailure !== undefined) { return yield* initialFailure; } const mcpContext = yield* readMcpContext(threadId); const activationOptions = acpMcpActivation(mcpContext); prepareTerminalEnvironment(mcpContext, sessionId); const activated = canLoadSession ? yield* runtime.loadSession(sessionId, activationOptions) : canResumeSession ? yield* runtime.resumeSession(sessionId, activationOptions) : yield* new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: `ACP driver cannot load or resume session ${sessionId}`, }); rememberTerminalEnvironment(activated.sessionId, yield* readMcpContext(threadId)); return activated; }); const configureSession = Effect.fnUntraced(function* ( startResult: AcpSessionRuntime.AcpSessionRuntimeStartResult, modelSelection: ModelSelection, runtimePolicy: ProviderAdapter.ProviderAdapterV2RuntimePolicy, ) { yield* Ref.set(activeSelection, null); const requestedModel = flavor.resolveModelId?.(modelSelection) ?? modelSelection.model; // Invalid params is the agent refusing the value, and the runtime // raises the same code when the session's model option does not offer // it. Other failures, such as sign-in or a lost connection, keep their // own error. const rejectedModel = (error: EffectAcpErrors.AcpError) => error._tag === "AcpRequestError" && error.code === -32602 ? new AcpModelSelectionError({ model: modelSelection.model, cause: error }) : error; let appliedModel: string | undefined; if (flavor.applyModelSelection !== undefined) { appliedModel = yield* flavor .applyModelSelection({ runtime, startResult, modelSelection, }) .pipe(Effect.mapError(rejectedModel)); } else if (requestedModel.length > 0) { // Stored catalog choices can expire; explicitly configured custom IDs // are validated by the agent instead. const isCustomModel = flavor.customModels?.includes(requestedModel) === true; const modelOption = findModelConfigOption(yield* runtime.getConfigOptions); const modelState = startResult.sessionSetupResult.models; if (modelOption?.type === "select") { if ( isCustomModel || modelOption.options.some((entry) => "value" in entry ? entry.value === requestedModel : entry.options.some((choice) => choice.value === requestedModel), ) ) { yield* runtime .setConfigOption(modelOption.id, requestedModel) .pipe(Effect.mapError(rejectedModel)); appliedModel = requestedModel; } } else if ( modelState != null && (isCustomModel || modelState.availableModels.some((model) => model.modelId === requestedModel)) ) { if (modelState.currentModelId !== requestedModel) { yield* runtime.setSessionModel(requestedModel).pipe(Effect.mapError(rejectedModel)); } appliedModel = requestedModel; } } if (appliedModel !== undefined) { const applied = appliedModel; yield* Ref.update(activeSessionSetup, (setup) => { if (setup === null) { return setup; } const models = setup.sessionSetupResult.models; if (models == null || models.currentModelId === applied) { return setup; } return { ...setup, sessionSetupResult: { ...setup.sessionSetupResult, models: { ...models, currentModelId: applied }, }, }; }); } const optionSelections = (modelSelection.options ?? []).map((selection) => selection.value === ACP_EMPTY_OPTION_VALUE ? { ...selection, value: "" } : selection, ); const configOptions = yield* runtime.getConfigOptions; const availableConfigIds = new Set(configOptions.map((option) => option.id)); const hasNativeConfigWithSyntheticModeId = availableConfigIds.has( ACP_SESSION_MODE_OPTION_ID, ); const modeSelection = hasNativeConfigWithSyntheticModeId ? undefined : optionSelections.find((selection) => selection.id === ACP_SESSION_MODE_OPTION_ID); const configSelections = hasNativeConfigWithSyntheticModeId ? optionSelections : optionSelections.filter((selection) => selection.id !== ACP_SESSION_MODE_OPTION_ID); // Probe descriptors union models; skip options absent from the live session and use agent defaults. const unsupportedConfigIds = configSelections .map((selection) => selection.id) .filter((id) => !availableConfigIds.has(id)); if (unsupportedConfigIds.length > 0) { yield* Effect.logWarning( "ACP session does not expose requested configuration option(s)", { driver, sessionId: startResult.sessionId, optionIds: unsupportedConfigIds, }, ); } for (const selection of configSelections) { if (!availableConfigIds.has(selection.id)) continue; // Advertised values union models; skip per-model rejections instead of failing session open. const option = configOptions.find((candidate) => candidate.id === selection.id); if ( option !== undefined && option.type === "select" && typeof selection.value === "string" ) { const advertisedValues = option.options.flatMap((entry) => "value" in entry ? [entry.value] : entry.options.map((choice) => choice.value), ); if (!advertisedValues.includes(selection.value)) continue; } yield* runtime.setConfigOption(selection.id, selection.value).pipe( Effect.catchTags({ AcpRequestError: (error) => Effect.logWarning("ACP session rejected a configuration option value", { optionId: selection.id, value: selection.value, detail: error.message, }), }), ); } const policyMode = flavor.sessionModeForPolicy?.(runtimePolicy); if (policyMode !== undefined) { yield* runtime.setMode(policyMode); } const modeState = yield* runtime.getModeState; if ( modeSelection !== undefined && typeof modeSelection.value === "string" && modeState?.availableModes.some((mode) => mode.id === modeSelection.value) === true && modeState.currentModeId !== modeSelection.value ) { yield* runtime.setMode(modeSelection.value); } const effectiveModeState = yield* runtime.getModeState; const effectiveConfigOptions = yield* runtime.getConfigOptions; const planSensitiveOptions = effectiveConfigOptions.filter( ( option, ): option is Extract< EffectAcpSchema.SessionConfigOption, { readonly type: "select" } > => option.type === "select" && (option.category === "mode" || option.category === "collaboration_mode"), ); if (runtimePolicy.interactionMode === "plan") { if (!nativeBuildConfigurationBySessionId.has(startResult.sessionId)) { nativeBuildConfigurationBySessionId.set(startResult.sessionId, { ...(effectiveModeState === undefined ? {} : { modeId: effectiveModeState.currentModeId }), configOptions: planSensitiveOptions.map((option) => ({ id: option.id, value: option.currentValue, })), }); } const planMode = effectiveModeState?.availableModes.find( (mode) => mode.id === "plan" || mode.id === "architect", ); if (planMode !== undefined && effectiveModeState?.currentModeId !== planMode.id) { yield* runtime.setMode(planMode.id); } for (const option of planSensitiveOptions) { const choices = option.options.flatMap((entry) => "value" in entry ? [entry.value] : entry.options.map((choice) => choice.value), ); const requested = choices.find( (choice) => choice === "plan" || choice === "architect", ); if (requested !== undefined && option.currentValue !== requested) { yield* runtime.setConfigOption(option.id, requested); } } } else { const nativeBuild = nativeBuildConfigurationBySessionId.get(startResult.sessionId); if (nativeBuild !== undefined) { if ( nativeBuild.modeId !== undefined && effectiveModeState?.currentModeId !== nativeBuild.modeId && effectiveModeState?.availableModes.some( (mode) => mode.id === nativeBuild.modeId, ) === true ) { yield* runtime.setMode(nativeBuild.modeId); } for (const saved of nativeBuild.configOptions) { const option = effectiveConfigOptions.find( (candidate) => candidate.type === "select" && candidate.id === saved.id, ); if (option === undefined || option.type !== "select") continue; const choices = option.options.flatMap((entry) => "value" in entry ? [entry.value] : entry.options.map((choice) => choice.value), ); if (option.currentValue !== saved.value && choices.includes(saved.value)) { yield* runtime.setConfigOption(option.id, saved.value); } } nativeBuildConfigurationBySessionId.delete(startResult.sessionId); } } yield* ( flavor.onSessionConfigurationUpdate?.( yield* runtime.getConfigOptions, yield* runtime.getModeState, (yield* Ref.get(activeSessionSetup))?.sessionSetupResult.models, ) ?? Effect.void ); }); yield* configureSession(started, input.modelSelection, input.runtimePolicy); yield* Ref.set(activeSelection, input.modelSelection); yield* Ref.set(activeInteractionMode, input.runtimePolicy.interactionMode); const createdAt = yield* DateTime.now; const providerSession: OrchestrationV2ProviderSession = { id: input.providerSessionId, driver, providerInstanceId: options.instanceId, status: "ready", cwd: input.runtimePolicy.cwd ?? process.cwd(), model: input.modelSelection.model, capabilities, createdAt, updatedAt: createdAt, lastError: null, }; const providerTurnPayload = ( context: ActiveAcpTurn, status: OrchestrationV2ProviderTurn["status"], completedAt: DateTime.Utc | null, ): OrchestrationV2ProviderTurn => ({ id: context.providerTurnId, providerThreadId: context.input.providerThread.id, nodeId: context.input.rootNodeId, runAttemptId: context.input.attemptId, nativeTurnRef: { driver, nativeId: context.nativeTurnId, strength: "weak", }, ordinal: context.input.providerTurnOrdinal, status, startedAt: context.startedAt, completedAt, }); const terminalizeOpenRunOwnedItems = Effect.fnUntraced(function* ( context: ActiveAcpTurn, options: { readonly terminalizeSubagents: boolean }, ) { for (const tool of context.tools.values()) { const status = toolStatus(tool.status); if (status === "pending" || status === "running" || status === "waiting") { yield* emitTool(context, tool, "interrupted"); } } if (!options.terminalizeSubagents) return; for (const subagent of context.subagents.values()) { yield* terminalizeSubagentTools(subagent); if (subagent.task.status !== "running" && subagent.task.status !== "pending") { continue; } yield* emitSubagent(context, { nativeTaskId: subagent.task.nativeTaskRef?.nativeId ?? subagent.task.id, prompt: subagent.task.prompt, title: subagent.task.title, model: subagent.task.model, status: "interrupted", childSessionId: subagent.childSessionId, result: subagent.task.result, suppressNormalTool: true, }); } }); const terminalizeOpenForegroundTools = Effect.fnUntraced(function* ( context: ActiveAcpTurn, status: "completed" | "failed" = "completed", ) { for (const tool of context.tools.values()) { if (!acpCompletedTurnShouldTerminalizeTool(tool, { ...flavor, extractSubagentUpdate })) continue; yield* emitTool(context, tool, status); } }); const quarantineStoppedRun = Effect.fnUntraced(function* () { yield* continuationPermit.withPermit( Effect.gen(function* () { yield* Ref.update(continuationGeneration, (value) => value + 1); yield* Ref.set(stoppedRunQuarantine, true); yield* Ref.set(wakeBuffer, []); yield* Ref.set(wakeReports, noWakeReports); yield* Ref.set(continuationRequested, false); yield* Ref.set(runningBackgroundTaskIds, new Set()); yield* Ref.set(midTurnUnreportedCompletedTaskIds, new Set()); yield* Ref.set(carryoverSubagents, null); yield* Ref.set(lastTurnRoute, null); }), ); }); const finalizeTurn = Effect.fnUntraced(function* ( context: ActiveAcpTurn, status: "completed" | "interrupted" | "failed" | "cancelled", failure?: OrchestrationV2ProviderFailure, options?: { readonly childTurnsContinue?: boolean }, ): Effect.fn.Return { if (context.finalized) return; const settledStatus = context.interrupted ? "interrupted" : status; context.finalizedStatus = settledStatus; context.finalized = true; const directStopQuarantine = yield* Ref.get(stoppedRunQuarantine); if (flavor.subagentsIdleOnTurnCompletion === true) { for (const subagent of context.subagents.values()) { if (!acpSubagentStatusBlocksTurnSettlement(subagent.task.status)) continue; yield* emitSubagent(context, { nativeTaskId: subagent.task.nativeTaskRef?.nativeId ?? subagent.task.id, prompt: subagent.task.prompt, title: subagent.task.title, model: subagent.task.model, status: settledStatus === "completed" ? "idle" : settledStatus, childSessionId: subagent.childSessionId, result: null, }); } } if (settledStatus === "completed") { yield* terminalizeOpenForegroundTools(context); } else if (settledStatus === "interrupted") { yield* terminalizeOpenRunOwnedItems(context, { terminalizeSubagents: directStopQuarantine, }); } yield* closeTextStreams(context); if (settledStatus !== "failed") { yield* emitProviderRetry(context, settledStatus); } const now = yield* DateTime.now; if ( flavor.supportsCompaction === true && context.input.message.text.trim() === "/compact" && context.input.message.attachments.length === 0 && settledStatus === "completed" ) { const nativeItemId = `${context.nativeTurnId}:compaction`; yield* Ref.update(compactionMarkers, (current) => { const updated = new Map(current); updated.set(context.nativeThreadId, Symbol()); return updated; }); yield* emitProviderEvent({ type: "turn_item.updated", driver, turnItem: { id: idAllocator.derive.turnItemFromProviderItem({ driver, nativeItemId }), threadId: context.input.threadId, runId: context.input.runId, nodeId: context.input.rootNodeId, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, nativeItemRef: { driver, nativeId: nativeItemId, strength: "weak" }, parentItemId: null, ordinal: yield* resolveItemOrdinal(context, nativeItemId), type: "compaction", driver, status: "completed", title: "Context compacted", startedAt: context.startedAt, completedAt: now, updatedAt: now, }, }); } const turn = providerTurnPayload(context, settledStatus, now); yield* Ref.update(providerTurns, (current) => { const updated = new Map(current); updated.set(String(turn.id), turn); return updated; }); yield* emitProviderEvent({ type: "provider_turn.updated", driver, threadId: context.input.threadId, providerTurn: turn, }); const updatedProviderThread: OrchestrationV2ProviderThread = { ...context.input.providerThread, providerSessionId: input.providerSessionId, status: "active", lastRunOrdinal: context.input.runOrdinal, firstRunOrdinal: context.input.providerThread.firstRunOrdinal ?? context.input.runOrdinal, contextUsage: context.contextUsage, nativeMetadata: context.nativeMetadata, updatedAt: now, }; yield* Ref.update(providerThreadByNativeSessionId, (current) => new Map(current).set(context.nativeThreadId, updatedProviderThread), ); yield* emitProviderEvent({ type: "provider_thread.updated", driver, providerThread: updatedProviderThread, }); const rootSessionId = yield* Ref.get(activeSessionId); const childTurnsContinue = settledStatus === "failed" && (options?.childTurnsContinue === true || (followsAgentWorkState && rootSessionId !== null && hasWorkingDescendant(rootSessionId))); yield* emitProviderEvent( settledStatus === "failed" ? { type: "turn.terminal", driver, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, runOrdinal: context.input.runOrdinal, failureItemOrdinal: yield* resolveItemOrdinal( context, `terminal-failure:${context.providerTurnId}`, ), status: settledStatus, ...(childTurnsContinue ? { childTurnsContinue: true } : {}), failure: failure ?? makeProviderFailure({ class: "provider_error" }), ...(context.providerRetry === undefined ? {} : { retry: context.providerRetry.retry, retryStartedAt: context.providerRetry.startedAt, }), threadDisposition: "reusable", } : { type: "turn.terminal", driver, providerThreadId: context.input.providerThread.id, providerTurnId: context.providerTurnId, runOrdinal: context.input.runOrdinal, status: settledStatus, failure: null, threadDisposition: "reusable", }, ); const subagentsRequiringCarryover = [...context.subagents.values()].filter( acpSubagentHasPendingBackgroundWork, ); if (subagentsRequiringCarryover.length > 0 && !directStopQuarantine) { const sessionId = yield* Ref.get(activeSessionId); if (sessionId !== null) { yield* Ref.set(carryoverSubagents, { sessionId, projectsLateTerminals: settledStatus === "completed" || childTurnsContinue, subagents: subagentsRequiringCarryover, }); } } yield* Ref.set(activeTurn, null); if ( postSettleContinuationEnabled && settledStatus === "completed" && !directStopQuarantine && (yield* Ref.get(runningBackgroundTaskIds)).size === 0 && ((yield* Ref.get(wakeBuffer)).length > 0 || (yield* Ref.get(midTurnUnreportedCompletedTaskIds)).size > 0) ) { const sessionId = yield* Ref.get(activeSessionId); if (sessionId !== null) { yield* offerContinuationRun(sessionId); } } if ( settledStatus !== "completed" || (yield* Ref.get(runningBackgroundTaskIds)).size === 0 ) { yield* Ref.set(midTurnUnreportedCompletedTaskIds, new Set()); } yield* Deferred.succeed(context.completed, undefined).pipe(Effect.ignore); }); scheduleDeferredFinalize = Effect.fn("scheduleDeferredFinalize")(function* (context) { if (!flavor.deferFinalizeForBackgroundWork) return; if (!context.promptSettled || context.finalized || context.interrupted) return; if (hasDeferredBackgroundWork(context)) return; context.backgroundFinalizeGeneration += 1; const generation = context.backgroundFinalizeGeneration; yield* Effect.gen(function* () { yield* Effect.sleep(ACP_DEFERRED_FINALIZE_DEBOUNCE); if ( context.finalized || context.interrupted || context.backgroundFinalizeGeneration !== generation ) { return; } if (hasDeferredBackgroundWork(context)) return; const status = context.promptSettledStatus ?? "completed"; yield* finalizeTurn(context, status); }).pipe(Effect.forkIn(sessionScope), Effect.asVoid); yield* ( options.testHooks?.onDeferredFinalizeScheduled?.(ACP_DEFERRED_FINALIZE_DEBOUNCE) ?? Effect.void ); }); const resolvePromptParts = Effect.fnUntraced(function* ( turnInput: ProviderAdapter.ProviderAdapterV2TurnInput, sessionId: string, ) { const prompt: Array = []; const mcpSession = yield* mcpSessions.read(turnInput.threadId); const instructionState = { interactionMode: turnInput.runtimePolicy.interactionMode, hasT3Mcp: acpMcpContext(mcpSession, self).servers.length > 0, hasDeviceTools: mcpSession?.capabilities?.has("device") === true, } satisfies T3AcpInstructionState; const previousInstructionState = (yield* Ref.get(promptInstructionStates)).get(sessionId); const compactionMarker = (yield* Ref.get(compactionMarkers)).get(sessionId); const userText = flavor.rewriteUserText === undefined ? turnInput.message.text : yield* flavor .rewriteUserText(turnInput.message.text, turnInput) .pipe( Effect.provideService( ChildProcessSpawner.ChildProcessSpawner, childProcessSpawner, ), ); const messageText = yield* providerMessageTextWithAttachmentPaths({ text: userText, attachments: turnInput.message.attachments, resolveAttachmentPath: host.resolveAttachmentPath, resolveExtractedTextPath: host.resolveExtractedTextPath, }); const isNativeSlashCommand = userText.trimStart().startsWith("/"); const text = t3AcpPromptWithInstructions({ prompt: messageText, state: instructionState, ...(previousInstructionState === undefined || compactionMarker !== undefined ? {} : { previousState: previousInstructionState }), }); if (text.length > 0) { prompt.push({ type: "text", text }); } const imageAttachments = turnInput.message.attachments.filter( isProviderNativeImageAttachment, ); if (imageAttachments.length > 0 && !supportsImagePrompts) { return yield* new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: "ACP driver did not negotiate image prompt support", }); } if (flavor.buildPrompt) { const setup = yield* Ref.get(activeSessionSetup); const blocks = yield* flavor .buildPrompt({ text, attachments: turnInput.message.attachments, promptCapabilities: setup?.initializeResult.agentCapabilities?.promptCapabilities, }) .pipe( Effect.mapError( (cause) => new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: "Failed to build ACP attachment prompt", payload: cause, }), ), ); prompt.splice(0, prompt.length, ...blocks); } for (const attachment of flavor.buildPrompt ? [] : imageAttachments) { const path = host.resolveAttachmentPath(attachment as ChatAttachment); if (path === null) { return yield* new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: `Invalid attachment id '${attachment.id}'`, }); } const bytes = yield* fileSystem.readFile(path).pipe( Effect.mapError( (cause) => new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: `Failed to read attachment '${attachment.id}'`, reason: "attachment_unreadable", payload: cause, }), ), ); prompt.push({ type: "image", data: Buffer.from(bytes).toString("base64"), mimeType: attachment.mimeType, }); } if (prompt.length === 0) { return yield* new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: "ACP turn requires non-empty text or attachments", }); } if (!isNativeSlashCommand) { prompt.push({ type: "text", text: buildRuntimeInstructions({ harness: flavor.runtimeHarness ?? driver, model: turnInput.modelSelection.model, }), }); } return { prompt, instructionState: text === messageText ? undefined : instructionState, compactionMarker, }; }); const restartRuntimeAfterTeardownIfRequired = Effect.fnUntraced(function* ( threadId: ThreadId | null, ) { const restartRequired = yield* Ref.get(runtimeRestartRequired); if (!restartRequired) return false; yield* restartAcpRuntime(threadId); yield* Ref.set(runtimeRestartRequired, false); yield* Ref.set(activeSessionId, null); yield* Ref.set(activeSessionSetup, null); yield* Ref.set(activeSelection, null); yield* Ref.set(activeInteractionMode, null); yield* Ref.set(snapshot, { order: [], messages: new Map(), loadingRole: null, loadingMessageId: null, loadingIndex: 0, }); return true; }); const startTurnUnlocked = Effect.fn("AcpAdapterV2.startTurn")( function* (turnInput: ProviderAdapter.ProviderAdapterV2TurnInput) { yield* awaitRuntimeTeardown(); const existing = yield* Ref.get(activeTurn); if (existing !== null) { return yield* new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: `ACP provider turn ${existing.providerTurnId} is still active`, reason: "previous_turn_active", }); } useProviderThreadIdentity(turnInput.providerThread); // load/resume can invoke client methods; install the incoming policy before activation. latestRuntimePolicy = turnInput.runtimePolicy; const requestedSessionId = yield* nativeThreadId(driver, turnInput.providerThread); const restartAfterInterrupt = yield* restartRuntimeAfterTeardownIfRequired( turnInput.threadId, ); const needsSessionActivation = (yield* Ref.get(activeSessionId)) !== requestedSessionId || restartAfterInterrupt; if (needsSessionActivation) { const activated = yield* activateSession(requestedSessionId, turnInput.threadId); yield* Ref.set(activeSessionId, activated.sessionId); yield* setActiveSessionSetup(activated); yield* configureSession(activated, turnInput.modelSelection, turnInput.runtimePolicy); yield* Ref.set(activeSelection, turnInput.modelSelection); yield* Ref.set(activeInteractionMode, turnInput.runtimePolicy.interactionMode); } else { const configuredSelection = yield* Ref.get(activeSelection); const configuredInteractionMode = yield* Ref.get(activeInteractionMode); if ( configuredSelection === null || !modelSelectionsEqual(configuredSelection, turnInput.modelSelection) || configuredInteractionMode !== turnInput.runtimePolicy.interactionMode ) { const currentSessionSetup = yield* Ref.get(activeSessionSetup); if (currentSessionSetup === null) { return yield* new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: `ACP session ${requestedSessionId} has no active setup metadata`, }); } yield* configureSession( currentSessionSetup, turnInput.modelSelection, turnInput.runtimePolicy, ); yield* Ref.set(activeSelection, turnInput.modelSelection); yield* Ref.set(activeInteractionMode, turnInput.runtimePolicy.interactionMode); } } yield* Ref.set(lastTurnRoute, { threadId: turnInput.threadId, providerThreadId: turnInput.providerThread.id, }); yield* Ref.set(suppressPostSettleMonitorPrompt, false); yield* Ref.set(handledBackgroundTaskIdsInActiveTurn, new Set()); // Continuations consume existing wake traffic without re-prompting ACP. const isContinuationTurn = (postSettleContinuationEnabled || followsAgentWorkState) && acpIsProviderContinuationMessage(turnInput.message); const isAppOwnedWakeTurn = acpIsAppOwnedWakeTurn(turnInput.message); if (!isAppOwnedWakeTurn) { yield* Ref.set(midTurnUnreportedCompletedTaskIds, new Set()); } const continuationWasRequested = yield* continuationPermit.withPermit( Effect.gen(function* () { const wasRequested = yield* Ref.get(continuationRequested); const preserveBufferedContinuation = acpTurnStartShouldPreserveContinuation({ continuationRequested: wasRequested, isContinuationTurn, wakeBufferLength: (yield* Ref.get(wakeBuffer)).length, }); if (!isContinuationTurn && !isAppOwnedWakeTurn && !preserveBufferedContinuation) { yield* Ref.set(wakeBuffer, []); yield* Ref.set(wakeReports, noWakeReports); } if (preserveBufferedContinuation) return wasRequested; yield* Ref.update(continuationGeneration, (value) => value + 1); yield* Ref.set(continuationRequested, false); return wasRequested; }), ); const promptParts = isContinuationTurn ? null : yield* resolvePromptParts(turnInput, requestedSessionId); const startedAt = yield* DateTime.now; const nativeTurnId = `${requestedSessionId}:turn:${turnInput.providerTurnOrdinal}`; const providerTurnId = deriveProviderTurnId(nativeTurnId); const completed = yield* Deferred.make(); const promptWireSettled = yield* Deferred.make(); const rememberedContextUsage = (yield* Ref.get(contextUsageBySessionId)).get( requestedSessionId, ); const rememberedNativeMetadata = (yield* Ref.get(nativeMetadataBySessionId)).get( requestedSessionId, ); const initialNativeMetadata = rememberedNativeMetadata === undefined ? (turnInput.providerThread.nativeMetadata ?? null) : { ...turnInput.providerThread.nativeMetadata, ...rememberedNativeMetadata, }; if (initialNativeMetadata !== null) { yield* Ref.update(nativeMetadataBySessionId, (current) => new Map(current).set(requestedSessionId, initialNativeMetadata), ); } const context: ActiveAcpTurn = { input: turnInput, providerTurnId, nativeThreadId: requestedSessionId, nativeTurnId, startedAt, completed, itemOrdinals: new Map(), user: { current: null, nextSegment: 0 }, assistant: { current: null, nextSegment: 0 }, reasoning: { current: null, nextSegment: 0 }, contextUsage: rememberedContextUsage ?? turnInput.providerThread.contextUsage ?? null, nativeMetadata: initialNativeMetadata, tools: new Map(), stoppedToolCallIds: new Set(), toolUpdatesSkipped: new Map(), toolStartedAt: new Map(), subagents: new Map(), subagentsBySessionId: new Map(), pendingSubagentNotifications: new Map(), toolCallIdsByBackgroundTaskId: new Map(), backgroundCommands: new Map(), persistentBackgroundTaskIds: new Set(), awaitingBackgroundHydration: new Set(), pendingInjectedReport: new Set(), earlyInjectedReportObserved: false, plans: new Map(), interrupted: false, finalized: false, finalizedStatus: null, promptSettled: false, promptSettledStatus: null, workAccepted: false, promptWireSettled, backgroundFinalizeGeneration: 0, }; const carryover = yield* Ref.getAndSet(carryoverSubagents, null); let rehydratedCarryoverSubagents: ReadonlyArray = []; if (carryover !== null && carryover.sessionId === requestedSessionId) { rehydratedCarryoverSubagents = carryover.subagents; for (const subagent of carryover.subagents) { const nativeId = subagent.task.nativeTaskRef?.nativeId ?? null; if (nativeId !== null) { context.subagents.set(nativeId, subagent); } if (subagent.childSessionId !== null) { context.subagentsBySessionId.set(subagent.childSessionId, subagent); } } } yield* Ref.set(activeTurn, context); yield* Ref.set(stoppedRunQuarantine, false); const runningTurn = providerTurnPayload(context, "running", null); yield* Ref.update(providerTurns, (current) => { const updated = new Map(current); updated.set(String(runningTurn.id), runningTurn); return updated; }); yield* emitProviderEvent({ type: "provider_turn.updated", driver, threadId: turnInput.threadId, providerTurn: runningTurn, }); const activeProviderThread: OrchestrationV2ProviderThread = { ...turnInput.providerThread, providerSessionId: input.providerSessionId, status: "active", contextUsage: context.contextUsage, nativeMetadata: context.nativeMetadata, updatedAt: startedAt, }; yield* Ref.update(providerThreadByNativeSessionId, (current) => new Map(current).set(requestedSessionId, activeProviderThread), ); yield* emitProviderEvent({ type: "provider_thread.updated", driver, providerThread: activeProviderThread, }); yield* rememberSnapshotMessage({ createdBy: turnInput.message.createdBy, creationSource: turnInput.message.creationSource, ...(turnInput.message.scheduledTaskId === undefined ? {} : { scheduledTaskId: turnInput.message.scheduledTaskId }), ...(turnInput.message.senderThreadId === undefined ? {} : { senderThreadId: turnInput.message.senderThreadId }), id: turnInput.message.messageId, threadId: turnInput.threadId, runId: turnInput.runId, nodeId: turnInput.rootNodeId, role: "user", text: turnInput.message.text, attachments: [...turnInput.message.attachments], streaming: false, createdAt: startedAt, updatedAt: startedAt, }); if (followsAgentWorkState) yield* runtimeCallbackPermit.withPermit(syncNativeChildren(context)); const attached = turnAttached; turnAttached = yield* Deferred.make(); yield* Deferred.succeed(attached, true); // Every attach ends the deferred-terminal contract, but only the // queued provider continuation owns wake traffic. A user turn may // start before that continuation and must leave its buffer intact. yield* projectDeferredCarryoverTerminals(rehydratedCarryoverSubagents); // Retain projected terminals during wake drain so monotonic guards reject replayed spawns. if (isContinuationTurn) { yield* Ref.set(continuationRequested, false); yield* Ref.update(wakeReports, ({ reports, offered }) => ({ reports: new Map([...reports].filter(([key]) => !offered.has(key))), offered: noWakeReports.offered, })); const takeWakeBuffer = Ref.modify(wakeBuffer, (current) => { const next: Array = []; return [ current.filter((notification) => notification.sessionId === requestedSessionId), next, ] as const; }); if (followsAgentWorkState) { yield* runtimeCallbackPermit.withPermit( Effect.gen(function* () { const liveWork = agentWork; yield* Effect.forEach(yield* takeWakeBuffer, handleSessionUpdate, { discard: true, }); agentWork = liveWork; context.workAccepted = true; yield* settleIdleWork(context); }), ); return; } const drainedWakeCount = yield* takeWakeBuffer.pipe( Effect.tap((drained) => Effect.forEach(drained, handleSessionUpdate, { concurrency: 1, discard: true, }), ), Effect.map((drained) => drained.length), ); context.promptSettled = true; context.promptSettledStatus = "completed"; if (drainedWakeCount === 0) { if (continuationWasRequested && flavor.deferFinalizeForBackgroundWork === true) { if (hasDeferredBackgroundWork(context)) { yield* rearmDeferredFinalize(context); } else { yield* scheduleDeferredFinalize(context); } } else { yield* finalizeTurn(context, "completed"); } return; } if (!context.finalized) { if (hasDeferredBackgroundWork(context)) { yield* rearmDeferredFinalize(context); } else { yield* scheduleDeferredFinalize(context); } } return; } const promptGeneration = yield* Ref.get(runtimeCallbackGeneration); yield* runtime.prompt({ prompt: promptParts!.prompt }).pipe( Effect.tap(() => Ref.update(promptInstructionStates, (current) => { if (promptParts?.instructionState === undefined) return current; const updated = new Map(current); updated.set(requestedSessionId, promptParts.instructionState); return updated; }), ), Effect.tap(() => Ref.update(compactionMarkers, (current) => { if (promptParts?.instructionState === undefined) return current; const observedMarker = promptParts.compactionMarker; if ( observedMarker === undefined || current.get(requestedSessionId) !== observedMarker ) return current; const updated = new Map(current); updated.delete(requestedSessionId); return updated; }), ), // Wire settlement precedes the completion callback's permit request so // settled-soft classification can observe the native return even when // the completion fiber has not yet set promptSettled under the permit. Effect.tap(() => Deferred.succeed(context.promptWireSettled, undefined).pipe(Effect.asVoid), ), Effect.flatMap((result) => runRuntimeCallbackAtGeneration( promptGeneration, Effect.gen(function* () { if (context.finalized || followsAgentWorkState) return; // ACP v2 reports a failed prompt as an idle `error`, not an RPC error. const idleError = result.stopReason === "error" && !context.interrupted; const status = idleError ? "failed" : result.stopReason === "cancelled" ? context.interrupted ? "interrupted" : "cancelled" : "completed"; if (flavor.deferFinalizeForBackgroundWork === true && !context.interrupted) { context.promptSettled = true; context.promptSettledStatus = status; for (const [taskId, completion] of context.backgroundCommands) { if (completion === null) continue; yield* finishRegisteredBackgroundTool({ taskId, ...completion }); } } if (idleError) { yield* terminalizeOpenForegroundTools(context, "failed"); // Unlike an RPC failure, the agent survives an idle error, so its // background tools and subagents keep reporting to this run. yield* finalizeTurn(context, "failed", idleErrorFailure(result.error), { childTurnsContinue: hasSurvivingBackgroundWork(context), }); return; } if ( flavor.deferFinalizeForBackgroundWork === true && !context.interrupted && hasDeferredBackgroundWork(context) ) { yield* closeTextStreams(context); yield* ( options.testHooks?.afterPromptSettledWithBackgroundWork?.() ?? Effect.void ); return; } yield* finalizeTurn(context, status); }), ).pipe(Effect.asVoid), ), Effect.catchCause((cause) => runRuntimeCallbackAtGeneration( promptGeneration, Effect.gen(function* () { if (context.finalized) return; const error = Cause.findError(cause); // Native close rejects prompts cancelled before insertion with -32800. const interrupted = context.interrupted || (pendingStopCloseTurn === context && Result.isSuccess(error) && error.success._tag === "AcpRequestError" && error.success.code === -32800); yield* finalizeTurn( context, interrupted ? "interrupted" : "failed", flavor.promptFailure?.(Cause.squash(cause)) ?? makeProviderFailure({ cause: Cause.squash(cause), class: "provider_error", }), ).pipe( Effect.andThen( Effect.logWarning("orchestration-v2.acp-prompt-failed", { driver, providerSessionId: input.providerSessionId, providerThreadId: turnInput.providerThread.id, providerTurnId, cause, }), ), ); }), ).pipe(Effect.asVoid), ), Effect.forkIn(sessionScope), ); }, (effect, turnInput) => effect.pipe( Effect.mapError( (cause) => new ProviderAdapter.ProviderAdapterTurnStartError({ driver, threadId: turnInput.threadId, providerThreadId: turnInput.providerThread.id, runId: turnInput.runId, cause, }), ), ), ); const startTurn = Effect.fn("AcpAdapterV2.startTurn.transition")(function* ( turnInput: ProviderAdapter.ProviderAdapterV2TurnInput, ) { return yield* runtimeTransitionPermit.withPermit(startTurnUnlocked(turnInput)); }); yield* Effect.addFinalizer(() => Effect.gen(function* () { yield* continuationPermit.withPermit( Effect.gen(function* () { yield* Ref.set(continuationClosed, true); yield* Ref.update(continuationGeneration, (value) => value + 1); }), ); const requests = [...(yield* Ref.get(pendingRuntimeRequests)).values()]; yield* Effect.forEach( requests, (request) => request.type === "approval" ? Deferred.succeed(request.decision, "cancel").pipe(Effect.ignore) : Deferred.succeed(request.answers, null).pipe(Effect.ignore), { discard: true }, ); const closingError = new EffectAcpErrors.AcpTransportError({ detail: "The ACP session closed before its admitted response reached the transport", cause: "ACP session transport closed", }); yield* Effect.forEach( requests, (request) => Deferred.fail(request.nativeResponseAcknowledgement, closingError), { discard: true }, ); const transportHadOutstandingResponses = yield* closeNativeTransport; const sessionCapabilities = started.initializeResult.agentCapabilities?.sessionCapabilities; if (sessionCapabilities?.close != null) { yield* runtimeTransitionPermit.withPermitsIfAvailable(1)( Effect.gen(function* () { const teardownState = yield* Ref.get(runtimeTeardownState); const restartRequired = yield* Ref.get(runtimeRestartRequired); if ( teardownState._tag === "Idle" && !restartRequired && !transportHadOutstandingResponses ) { yield* runtime.closeSession().pipe(Effect.ignore); } }), ); } if (flavor.assertComplete !== undefined) { yield* flavor.assertComplete.pipe(Effect.orDie); } if (runtimeScope !== undefined) { yield* Scope.close(runtimeScope, Exit.void).pipe(Effect.ignore); } }), ); const sessionRuntime: ProviderAdapter.ProviderAdapterV2SessionRuntime = { instanceId: options.instanceId, driver, providerSessionId: input.providerSessionId, providerSession, events: Stream.fromEffectRepeat(Queue.take(events)), ...(flavor.getModelContextWindow === undefined ? {} : { getModelContextWindow: (selection) => { const session = Ref.getUnsafe(activeSessionSetup); return session === null ? undefined : flavor.getModelContextWindow?.(selection, session); }, }), ...(postSettleContinuationEnabled || followsAgentWorkState ? { hasPendingBackgroundWork: Effect.gen(function* () { const rootSessionId = yield* Ref.get(activeSessionId); if ( followsAgentWorkState && rootSessionId !== null && hasWorkingDescendant(rootSessionId) ) return true; if ( followsAgentWorkState && (agentWork?.state === "running" || agentWork?.state === "requires_action" || agentRequests.size > 0) ) return true; if ((yield* Ref.get(wakeBuffer)).length > 0) return true; if (yield* Ref.get(continuationRequested)) return true; if ((yield* Ref.get(runningBackgroundTaskIds)).size > 0) return true; const active = yield* Ref.get(activeTurn); if ( active !== null && [...active.subagents.values()].some(acpSubagentHasPendingBackgroundWork) ) { return true; } const carryover = yield* Ref.get(carryoverSubagents); if ( carryover !== null && carryover.subagents.some(acpSubagentHasPendingBackgroundWork) ) { return true; } return false; }), } : {}), ensureThread: Effect.fn("AcpAdapterV2.ensureThread")( function* (threadInput: ProviderAdapter.ProviderAdapterV2EnsureThreadInput) { return yield* runtimeTransitionPermit.withPermit( Effect.gen(function* () { if (threadInput.forceNewNativeThread) { const currentTurn = yield* Ref.get(activeTurn); const pendingWork = yield* ( sessionRuntime.hasPendingBackgroundWork ?? Effect.succeed(false) ); if (currentTurn !== null || pendingWork) return yield* new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: "Cannot replace ACP session while a turn or background work is active", }); yield* awaitRuntimeTeardown(); const previousItemIdentityVersion = itemIdentityVersion; itemIdentityVersion = 2; prepareTerminalEnvironment(yield* readMcpContext(threadInput.threadId)); yield* startReplacementAcpRuntime(threadInput.threadId, (candidate) => commitReplacementSession(threadInput.threadId, candidate), ).pipe( Effect.tapError(() => Effect.sync(() => { itemIdentityVersion = previousItemIdentityVersion; }), ), ); } const now = yield* DateTime.now; const sessionId = yield* Ref.get(activeSessionId); if (sessionId === null) { return yield* new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: "ACP runtime did not produce a session id", }); } const created = makeProviderThread({ driver, providerInstanceId: options.instanceId, idAllocator, appThreadId: threadInput.threadId, providerSessionId: input.providerSessionId, nativeThreadId: sessionId, ...(itemIdentityVersion === undefined ? {} : { itemIdentityVersion }), now, }); const providerThread = threadInput.forceNewNativeThread && threadInput.existingProviderThread ? { ...created, id: threadInput.existingProviderThread.id } : created; yield* Ref.update(providerThreadByNativeSessionId, (current) => new Map(current).set(sessionId, providerThread), ); return providerThread; }), ); }, (effect, threadInput) => effect.pipe( Effect.mapError( (cause) => new ProviderAdapter.ProviderAdapterEnsureThreadError({ driver, threadId: threadInput.threadId, cause, }), ), ), ), resumeThread: Effect.fn("AcpAdapterV2.resumeThread")( function* (threadInput: { readonly providerThread: OrchestrationV2ProviderThread; readonly modelSelection?: ModelSelection; readonly runtimePolicy?: ProviderAdapter.ProviderAdapterV2RuntimePolicy; }) { return yield* runtimeTransitionPermit.withPermit( Effect.gen(function* () { yield* awaitRuntimeTeardown(); const sessionId = yield* nativeThreadId(driver, threadInput.providerThread); const previousItemIdentityVersion = itemIdentityVersion; useProviderThreadIdentity(threadInput.providerThread); const restorePreviousItemIdentity = Effect.sync(() => { itemIdentityVersion = previousItemIdentityVersion; }); const restartAfterInterrupt = yield* restartRuntimeAfterTeardownIfRequired( threadInput.providerThread.appThreadId, ).pipe(Effect.tapError(() => restorePreviousItemIdentity)); if ((yield* Ref.get(activeSessionId)) !== sessionId || restartAfterInterrupt) { yield* Ref.set(snapshot, { order: [], messages: new Map(), loadingRole: null, loadingMessageId: null, loadingIndex: 0, }); const activated = yield* activateSession( sessionId, threadInput.providerThread.appThreadId, ).pipe(Effect.tapError(() => restorePreviousItemIdentity)); yield* Ref.set(activeSessionId, activated.sessionId); yield* setActiveSessionSetup(activated); const nextSelection = threadInput.modelSelection ?? input.modelSelection; const nextRuntimePolicy = threadInput.runtimePolicy ?? input.runtimePolicy; yield* configureSession(activated, nextSelection, nextRuntimePolicy); yield* Ref.set(activeSelection, nextSelection); yield* Ref.set(activeInteractionMode, nextRuntimePolicy.interactionMode); } const now = yield* DateTime.now; return { ...threadInput.providerThread, providerSessionId: input.providerSessionId, status: "idle" as const, updatedAt: now, }; }), ); }, (effect, threadInput) => effect.pipe( Effect.mapError( (cause) => new ProviderAdapter.ProviderAdapterResumeThreadError({ driver, providerSessionId: input.providerSessionId, providerThreadId: threadInput.providerThread.id, cause, }), ), ), ), startTurn, ...(flavor.supportsCompaction === true ? { compactThread: (turnInput: ProviderAdapter.ProviderAdapterV2TurnInput) => startTurn({ ...turnInput, message: { ...turnInput.message, text: "/compact" }, }), } : {}), steerTurn: (turnInput) => Effect.fail( new ProviderAdapter.ProviderAdapterSteerRunUnsupportedError({ driver, providerThreadId: turnInput.providerThread.id, }), ), interruptTurn: Effect.fn("AcpAdapterV2.interruptTurn")( function* (turnInput: ProviderAdapter.ProviderAdapterV2InterruptInput) { return yield* Effect.uninterruptibleMask((restore) => Effect.gen(function* () { yield* restore(runtimeTransitionPermit.take(1)); return yield* Effect.gen(function* () { const transition = yield* Effect.gen(function* () { const interruptContext = yield* Ref.get(activeTurn); if ( followsAgentWorkState && turnInput.requestRuntimeRestart === true && (interruptContext === null || interruptContext.providerTurnId === turnInput.providerTurnId) ) { // ACP cancel leaves native children alive; close confirms session-wide Stop. pendingStopCloseTurn = interruptContext; const closed = yield* runtime.closeSession().pipe( Effect.timeoutOption("10 seconds"), Effect.mapError( (cause) => new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: "The agent refused to close its session; background work may still run", cause, }), ), ); if (Option.isNone(closed)) { return yield* new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: "The agent did not confirm session close; background work may still run", }); } yield* runtimeCallbackPermit.withPermit( Effect.gen(function* () { // Callbacks can settle the turn while close is awaiting confirmation. const context = yield* Ref.get(activeTurn); const carryover = yield* Ref.get(carryoverSubagents); yield* Ref.set(stoppedRunQuarantine, true); yield* Ref.set(runtimeRestartRequired, true); agentWork = null; agentRequests.clear(); nativeChildren.clear(); yield* quarantineStoppedRun(); if (context !== null) context.interrupted = true; yield* cancelPendingRuntimeRequests(); yield* terminalizeCarryoverSubagents(carryover); if (context !== null) { yield* finalizeTurn(context, "interrupted"); } }), ); return undefined; } const softSteerCandidate = flavor.preserveRuntimeOnSettledInterrupt === true && turnInput.requestRuntimeRestart !== true && interruptContext?.providerTurnId === turnInput.providerTurnId; // Native return can precede its callback; drain responses under the permit and check wire settlement before cancel. let settledSoftInterrupt = false; if (softSteerCandidate && interruptContext !== null) { const promptSettledUnderPermit = yield* runtimeCallbackPermit.withPermit( awaitAdmittedNativeResponses.pipe( Effect.andThen( Effect.sync(() => interruptContext.promptSettled === true), ), ), ); const wireSettled = yield* Deferred.isDone( interruptContext.promptWireSettled, ); settledSoftInterrupt = promptSettledUnderPermit || wireSettled; } const restartRuntime = !settledSoftInterrupt && (turnInput.requestRuntimeRestart === true || flavor.restartRuntimeOnEveryInterrupt === true); const hardRestart = restartRuntime && flavor.terminateRuntimeProcessGroupOnInterrupt === true; if (hardRestart) { const teardownState = yield* Ref.get(runtimeTeardownState); if (teardownState._tag === "Failed") { return yield* teardownState.error; } if (teardownState._tag === "InProgress") { yield* Effect.logInfo( "ACP interrupt awaiting in-progress hard teardown", { driver, providerTurnId: turnInput.providerTurnId, }, ); yield* Deferred.await(teardownState.completed); return undefined; } } const context = interruptContext; if (context?.providerTurnId !== turnInput.providerTurnId) { // Contain orphan runtimes on Stop; session-wide teardown must not touch another active turn. const containOrphanRuntime = context === null && hardRestart && turnInput.requestRuntimeRestart === true; yield* Effect.logWarning( containOrphanRuntime ? "ACP Stop raced a soft interrupt that cleared the turn; containing the orphan runtime" : "ACP interrupt raced transport teardown or a prior Stop; treating as already interrupted", { driver, requestedProviderTurnId: turnInput.providerTurnId, activeProviderTurnId: context?.providerTurnId ?? null, hardRestart, requestRuntimeRestart: turnInput.requestRuntimeRestart === true, teardownState: (yield* Ref.get(runtimeTeardownState))._tag, }, ); if (containOrphanRuntime) { const teardownBarrier = yield* Deferred.make< void, ProviderAdapter.ProviderAdapterProtocolError >(); yield* runtimeCallbackPermit.withPermit( Effect.gen(function* () { yield* awaitAdmittedNativeResponses; const stoppedGeneration = yield* Ref.get(runtimeCallbackGeneration); yield* quarantineNativeTransportAtGeneration(stoppedGeneration); yield* Ref.set(runtimeTeardownState, { _tag: "InProgress", completed: teardownBarrier, }); yield* advanceRuntimeCallbackGeneration; }), ); const orphanCarryover = yield* Ref.getAndSet(carryoverSubagents, null); yield* quarantineStoppedRun(); yield* cancelPendingRuntimeRequests(); yield* terminalizeCarryoverSubagents(orphanCarryover); if (runtime.terminateProcessGroup === undefined) { const error = new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: "ACP runtime does not expose its required process-group teardown; the session is poisoned", }); yield* Ref.set(runtimeTeardownState, { _tag: "Failed", error }); yield* Deferred.fail(teardownBarrier, error).pipe(Effect.ignore); return yield* error; } const teardownExit = yield* runtime.terminateProcessGroup.pipe( Effect.exit, ); if (Exit.isFailure(teardownExit)) { const error = new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: "ACP orphan runtime process-group teardown failed; the session is poisoned", payload: Cause.squash(teardownExit.cause), }); yield* Ref.set(runtimeTeardownState, { _tag: "Failed", error }); yield* Deferred.fail(teardownBarrier, error).pipe(Effect.ignore); return yield* error; } yield* Ref.set(runtimeRestartRequired, true); yield* Ref.set(runtimeTeardownState, { _tag: "Idle" }); yield* Deferred.succeed(teardownBarrier, undefined).pipe(Effect.ignore); } return undefined; } const teardownBarrier = hardRestart ? yield* Deferred.make() : null; if (teardownBarrier !== null) { yield* runtimeCallbackPermit.withPermit( Effect.gen(function* () { yield* awaitAdmittedNativeResponses; const stoppedGeneration = yield* Ref.get(runtimeCallbackGeneration); yield* quarantineNativeTransportAtGeneration(stoppedGeneration); yield* Ref.set(runtimeTeardownState, { _tag: "InProgress", completed: teardownBarrier, }); yield* advanceRuntimeCallbackGeneration; }), ); } return { context, restartRuntime, hardRestart, teardownBarrier, settledSoftInterrupt, }; }).pipe( Effect.ensuring( Effect.sync(() => { pendingStopCloseTurn = null; }), ), ); if (transition === undefined) return; const { context, restartRuntime, hardRestart, teardownBarrier, settledSoftInterrupt, } = transition; const poisonTeardown = Effect.fnUntraced(function* ( detail: string, payload?: unknown, ) { const error = new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail, ...(payload === undefined ? {} : { payload }), }); yield* Ref.set(runtimeTeardownState, { _tag: "Failed", error }); yield* Deferred.fail(teardownBarrier!, error).pipe(Effect.ignore); return error; }); const runTransition = Effect.gen(function* () { context.interrupted = true; if (hardRestart || turnInput.requestRuntimeRestart === true) { yield* quarantineStoppedRun(); } yield* cancelPendingRuntimeRequests(); // Finalize before killing so projection catches up and concurrent interrupts see activeTurn cleared. if ( (hardRestart || context.promptSettled || settledSoftInterrupt) && !context.finalized ) { yield* finalizeTurn(context, "interrupted"); } if (hardRestart) { if (runtime.terminateProcessGroup === undefined) { return yield* poisonTeardown( "ACP runtime does not expose its required process-group teardown; the session is poisoned", ).pipe(Effect.flatMap(Effect.fail)); } const teardownExit = yield* runtime.terminateProcessGroup!.pipe( Effect.exit, ); if (Exit.isFailure(teardownExit)) { return yield* poisonTeardown( "ACP runtime process-group teardown failed; the session is poisoned", Cause.squash(teardownExit.cause), ).pipe(Effect.flatMap(Effect.fail)); } yield* Ref.set(runtimeRestartRequired, true); yield* Ref.set(runtimeTeardownState, { _tag: "Idle" }); yield* Deferred.succeed(teardownBarrier!, undefined).pipe(Effect.ignore); } else { // Settled soft interrupts have no prompt to cancel; cancel could kill background subagents. if (!settledSoftInterrupt) { yield* runtime.cancel; } if (restartRuntime && flavor.restartRuntimeAfterInterrupt === true) { yield* Ref.set(runtimeRestartRequired, true); } } const stopped = yield* Deferred.await(context.completed).pipe( Effect.timeoutOption("10 seconds"), ); if (Option.isNone(stopped)) { if (!context.finalized) { yield* finalizeTurn(context, "interrupted"); } return yield* new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: `ACP provider turn ${turnInput.providerTurnId} did not acknowledge cancellation before the interrupt timeout`, }); } }); if (!hardRestart) { return yield* restore(runTransition); } return yield* runTransition.pipe( Effect.catchCause((cause) => Effect.gen(function* () { const state = yield* Ref.get(runtimeTeardownState); if (state._tag === "InProgress") { yield* poisonTeardown( "ACP hard teardown failed unexpectedly; the session is poisoned", Cause.squash(cause), ); } return yield* Effect.failCause(cause); }), ), ); }).pipe(Effect.ensuring(runtimeTransitionPermit.release(1))); }), ); }, (effect, turnInput) => effect.pipe( Effect.mapError( (cause) => new ProviderAdapter.ProviderAdapterInterruptError({ driver, providerThreadId: turnInput.providerThread.id, providerTurnId: turnInput.providerTurnId, cause, }), ), ), ), respondToRuntimeRequest: (requestInput) => Effect.uninterruptibleMask((restore) => Effect.gen(function* () { yield* restore(runtimeTransitionPermit.take(1)); return yield* Effect.gen(function* () { yield* awaitRuntimeTeardown(); const generation = yield* Ref.get(runtimeCallbackGeneration); const pending = (yield* Ref.get(pendingRuntimeRequests)).get( String(requestInput.requestId), ); if (pending === undefined || pending.generation !== generation) { return yield* new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: `No pending ACP runtime request ${requestInput.requestId}`, }); } const settled = pending.type === "user_input" ? yield* Deferred.succeed(pending.answers, requestInput.answers ?? null) : requestInput.decision === undefined ? yield* new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: `ACP approval request ${requestInput.requestId} requires a decision`, }) : yield* Deferred.succeed(pending.decision, requestInput.decision); if (!settled) { return yield* new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: `ACP runtime request ${requestInput.requestId} was already resolved`, }); } yield* awaitNativeResponseAcknowledgements([ [pending.transportRequestId, pending.nativeResponseAcknowledgement], ]); yield* Deferred.await(pending.nativeResponseAcknowledgement); }).pipe(Effect.ensuring(runtimeTransitionPermit.release(1))); }), ).pipe( Effect.mapError( (cause) => new ProviderAdapter.ProviderAdapterRuntimeRequestResponseError({ driver, requestId: requestInput.requestId, cause, }), ), ), readThreadSnapshot: Effect.fn("AcpAdapterV2.readThreadSnapshot")( function* (snapshotInput) { return yield* runtimeTransitionPermit.withPermit( Effect.gen(function* () { yield* awaitRuntimeTeardown(); useProviderThreadIdentity(snapshotInput.providerThread); yield* restartRuntimeAfterTeardownIfRequired( snapshotInput.providerThread.appThreadId, ); const sessionId = yield* nativeThreadId(driver, snapshotInput.providerThread); if ((yield* Ref.get(activeSessionId)) !== sessionId) { if (!capabilities.threads.canReadThreadSnapshot) { return yield* new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: "ACP driver does not support session/load snapshots", }); } yield* Ref.set(snapshot, { order: [], messages: new Map(), loadingRole: null, loadingMessageId: null, loadingIndex: 0, }); const mcpContext = yield* readMcpContext( snapshotInput.providerThread.appThreadId, ); prepareTerminalEnvironment(mcpContext, sessionId); const activated = yield* runtime.loadSession( sessionId, acpMcpActivation(mcpContext), ); rememberTerminalEnvironment( activated.sessionId, yield* readMcpContext(snapshotInput.providerThread.appThreadId), ); yield* Ref.set(activeSessionId, activated.sessionId); yield* setActiveSessionSetup(activated); yield* Ref.set(activeSelection, null); yield* Ref.set(activeInteractionMode, null); } const state = yield* Ref.get(snapshot); const now = yield* DateTime.now; return { providerThread: { ...snapshotInput.providerThread, providerSessionId: input.providerSessionId, status: "idle" as const, updatedAt: now, }, providerTurns: [...(yield* Ref.get(providerTurns)).values()], messages: state.order.flatMap((key) => { const message = state.messages.get(key); return message === undefined ? [] : [message]; }), runtimeRequests: [], providerPayload: { protocol: ACP_PROTOCOL, sessionId }, }; }), ); }, (effect, snapshotInput) => effect.pipe( Effect.mapError( (cause) => new ProviderAdapter.ProviderAdapterReadThreadSnapshotError({ driver, providerThreadId: snapshotInput.providerThread.id, cause, }), ), ), ), rollbackThread: (rollbackInput) => runtimeTransitionPermit .withPermit( Effect.gen(function* () { const currentTurn = yield* Ref.get(activeTurn); if (currentTurn !== null) { return yield* new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: `Cannot roll back ACP provider thread ${rollbackInput.providerThread.id} while turn ${currentTurn.providerTurnId} is active`, }); } yield* awaitRuntimeTeardown(); itemIdentityVersion = 2; prepareTerminalEnvironment( yield* readMcpContext(rollbackInput.providerThread.appThreadId), ); const replacement = yield* startReplacementAcpRuntime( rollbackInput.providerThread.appThreadId, (candidate) => commitReplacementSession(rollbackInput.providerThread.appThreadId, candidate), ).pipe(Effect.retry({ times: 1 })); const now = yield* DateTime.now; return { providerThread: { ...rollbackInput.providerThread, nativeThreadRef: { driver, nativeId: replacement.sessionId, strength: "strong" as const, }, nativeConversationHeadRef: null, nativeMetadata: { ...rollbackInput.providerThread.nativeMetadata, itemIdentityVersion: 2 as const, }, status: "idle" as const, updatedAt: now, }, providerTurns: [], messages: [], runtimeRequests: [], }; }), ) .pipe( Effect.mapError( (cause) => new ProviderAdapter.ProviderAdapterRollbackThreadError({ driver, providerThreadId: rollbackInput.providerThread.id, checkpointId: rollbackInput.target.checkpointId, cause, }), ), ), forkThread: Effect.fn("AcpAdapterV2.forkThread")( function* (forkInput) { return yield* runtimeTransitionPermit.withPermit( Effect.gen(function* () { yield* awaitRuntimeTeardown(); useProviderThreadIdentity(forkInput.sourceProviderThread); yield* restartRuntimeAfterTeardownIfRequired( forkInput.sourceProviderThread.appThreadId, ); if (!capabilities.threads.canForkThread) { return yield* new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: "ACP driver did not negotiate session/fork", }); } if (forkInput.providerTurnId !== undefined) { return yield* new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail: "ACP session/fork can only fork the current session head", }); } const sourceSessionId = yield* nativeThreadId( driver, forkInput.sourceProviderThread, ); const mcpContext = yield* readMcpContext(forkInput.targetThreadId); prepareTerminalEnvironment(mcpContext); const forked = yield* runtime.forkSession( sourceSessionId, acpMcpActivation(mcpContext), ); rememberTerminalEnvironment( forked.sessionId, yield* readMcpContext(forkInput.targetThreadId), ); yield* Ref.set(activeSessionId, forked.sessionId); yield* setActiveSessionSetup(forked); yield* Ref.set(activeSelection, null); yield* Ref.set(activeInteractionMode, null); itemIdentityVersion = 2; const now = yield* DateTime.now; const providerThread = makeProviderThread({ driver, providerInstanceId: options.instanceId, idAllocator, appThreadId: forkInput.targetThreadId, providerSessionId: input.providerSessionId, nativeThreadId: forked.sessionId, itemIdentityVersion: 2, ...(forkInput.ownerNodeId === undefined ? {} : { ownerNodeId: forkInput.ownerNodeId }), forkedFrom: { providerThreadId: forkInput.sourceProviderThread.id, ...(forkInput.providerTurnId === undefined ? {} : { providerTurnId: forkInput.providerTurnId }), }, now, }); yield* Ref.update(providerThreadByNativeSessionId, (current) => new Map(current).set(forked.sessionId, providerThread), ); return providerThread; }), ); }, (effect, forkInput) => effect.pipe( Effect.mapError( (cause) => new ProviderAdapter.ProviderAdapterForkThreadError({ driver, providerThreadId: forkInput.sourceProviderThread.id, cause, }), ), ), ), }; return sessionRuntime; }, (effect, input) => effect.pipe( Effect.mapError( (cause) => new ProviderAdapter.ProviderAdapterOpenSessionError({ driver, providerSessionId: input.providerSessionId, cause, }), ), ), ), }); });