import { Cause, Deferred, Effect, Exit, Layer, Queue, Ref, Scope, Context, Stream } from "effect";import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";import * as EffectAcpClient from "effect-acp/client";import * as EffectAcpErrors from "effect-acp/errors";import type * as EffectAcpSchema from "effect-acp/schema";import type * as EffectAcpProtocol from "effect-acp/protocol"; import { collectSessionConfigOptionValues, extractModelConfigId, findSessionConfigOption, mergeToolCallState, parseSessionModeState, parseSessionUpdateEvent, type AcpParsedSessionEvent, type AcpSessionModeState, type AcpToolCallState,} from "./AcpRuntimeModel.ts"; export interface AcpSpawnInput { readonly command: string; readonly args: ReadonlyArray<string>; readonly cwd?: string; readonly env?: NodeJS.ProcessEnv;} export interface AcpSessionRuntimeOptions { readonly spawn: AcpSpawnInput; readonly cwd: string; readonly resumeSessionId?: string; readonly clientCapabilities?: EffectAcpSchema.InitializeRequest["clientCapabilities"]; readonly clientInfo: { readonly name: string; readonly version: string; }; readonly authMethodId: string; readonly requestLogger?: (event: AcpSessionRequestLogEvent) => Effect.Effect<void, never>; readonly protocolLogging?: { readonly logIncoming?: boolean; readonly logOutgoing?: boolean; readonly logger?: (event: EffectAcpProtocol.AcpProtocolLogEvent) => Effect.Effect<void, never>; };} export interface AcpSessionRequestLogEvent { readonly method: string; readonly payload: unknown; readonly status: "started" | "succeeded" | "failed"; readonly result?: unknown; readonly cause?: Cause.Cause<EffectAcpErrors.AcpError>;} export interface AcpSessionRuntimeStartResult { readonly sessionId: string; readonly initializeResult: EffectAcpSchema.InitializeResponse; readonly sessionSetupResult: | EffectAcpSchema.LoadSessionResponse | EffectAcpSchema.NewSessionResponse | EffectAcpSchema.ResumeSessionResponse; readonly modelConfigId: string | undefined;} export interface AcpSessionRuntimeShape { readonly handleRequestPermission: EffectAcpClient.AcpClientShape["handleRequestPermission"]; readonly handleElicitation: EffectAcpClient.AcpClientShape["handleElicitation"]; readonly handleReadTextFile: EffectAcpClient.AcpClientShape["handleReadTextFile"]; readonly handleWriteTextFile: EffectAcpClient.AcpClientShape["handleWriteTextFile"]; readonly handleCreateTerminal: EffectAcpClient.AcpClientShape["handleCreateTerminal"]; readonly handleTerminalOutput: EffectAcpClient.AcpClientShape["handleTerminalOutput"]; readonly handleTerminalWaitForExit: EffectAcpClient.AcpClientShape["handleTerminalWaitForExit"]; readonly handleTerminalKill: EffectAcpClient.AcpClientShape["handleTerminalKill"]; readonly handleTerminalRelease: EffectAcpClient.AcpClientShape["handleTerminalRelease"]; readonly handleSessionUpdate: EffectAcpClient.AcpClientShape["handleSessionUpdate"]; readonly handleElicitationComplete: EffectAcpClient.AcpClientShape["handleElicitationComplete"]; readonly handleUnknownExtRequest: EffectAcpClient.AcpClientShape["handleUnknownExtRequest"]; readonly handleUnknownExtNotification: EffectAcpClient.AcpClientShape["handleUnknownExtNotification"]; readonly handleExtRequest: EffectAcpClient.AcpClientShape["handleExtRequest"]; readonly handleExtNotification: EffectAcpClient.AcpClientShape["handleExtNotification"]; readonly start: () => Effect.Effect<AcpSessionRuntimeStartResult, EffectAcpErrors.AcpError>; readonly getEvents: () => Stream.Stream<AcpParsedSessionEvent, never>; readonly getModeState: Effect.Effect<AcpSessionModeState | undefined>; readonly getConfigOptions: Effect.Effect<ReadonlyArray<EffectAcpSchema.SessionConfigOption>>; readonly prompt: ( payload: Omit<EffectAcpSchema.PromptRequest, "sessionId">, ) => Effect.Effect<EffectAcpSchema.PromptResponse, EffectAcpErrors.AcpError>; readonly cancel: Effect.Effect<void, EffectAcpErrors.AcpError>; readonly setMode: ( modeId: string, ) => Effect.Effect<EffectAcpSchema.SetSessionModeResponse, EffectAcpErrors.AcpError>; readonly setConfigOption: ( configId: string, value: string | boolean, ) => Effect.Effect<EffectAcpSchema.SetSessionConfigOptionResponse, EffectAcpErrors.AcpError>; readonly setModel: (model: string) => Effect.Effect<void, EffectAcpErrors.AcpError>; readonly request: ( method: string, payload: unknown, ) => Effect.Effect<unknown, EffectAcpErrors.AcpError>; readonly notify: ( method: string, payload: unknown, ) => Effect.Effect<void, EffectAcpErrors.AcpError>;} interface AcpStartedState extends AcpSessionRuntimeStartResult {} type AcpStartState = | { readonly _tag: "NotStarted" } | { readonly _tag: "Starting"; readonly deferred: Deferred.Deferred<AcpSessionRuntimeStartResult, EffectAcpErrors.AcpError>; } | { readonly _tag: "Started"; readonly result: AcpStartedState }; interface AcpAssistantSegmentState { readonly nextSegmentIndex: number; readonly activeItemId?: string;} interface EnsureActiveAssistantSegmentResult { readonly itemId: string; readonly startedEvent?: Extract<AcpParsedSessionEvent, { readonly _tag: "AssistantItemStarted" }>;} export class AcpSessionRuntime extends Context.Service<AcpSessionRuntime, AcpSessionRuntimeShape>()( "t3/provider/acp/AcpSessionRuntime",) { static layer( options: AcpSessionRuntimeOptions, ): Layer.Layer< AcpSessionRuntime, EffectAcpErrors.AcpError, ChildProcessSpawner.ChildProcessSpawner > { return Layer.effect(AcpSessionRuntime, makeAcpSessionRuntime(options)); }} const makeAcpSessionRuntime = ( options: AcpSessionRuntimeOptions,): Effect.Effect< AcpSessionRuntimeShape, EffectAcpErrors.AcpError, ChildProcessSpawner.ChildProcessSpawner | Scope.Scope> => Effect.gen(function* () { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; const runtimeScope = yield* Scope.Scope; const eventQueue = yield* Queue.unbounded<AcpParsedSessionEvent>(); const modeStateRef = yield* Ref.make<AcpSessionModeState | undefined>(undefined); const toolCallsRef = yield* Ref.make(new Map<string, AcpToolCallState>()); const assistantSegmentRef = yield* Ref.make<AcpAssistantSegmentState>({ nextSegmentIndex: 0 }); const configOptionsRef = yield* Ref.make(sessionConfigOptionsFromSetup(undefined)); const startStateRef = yield* Ref.make<AcpStartState>({ _tag: "NotStarted" }); const logRequest = (event: AcpSessionRequestLogEvent) => options.requestLogger ? options.requestLogger(event) : Effect.void; const runLoggedRequest = <A>( method: string, payload: unknown, effect: Effect.Effect<A, EffectAcpErrors.AcpError>, ): Effect.Effect<A, EffectAcpErrors.AcpError> => logRequest({ method, payload, status: "started" }).pipe( Effect.flatMap(() => effect.pipe( Effect.tap((result) => logRequest({ method, payload, status: "succeeded", result, }), ), Effect.onError((cause) => logRequest({ method, payload, status: "failed", cause, }), ), ), ), ); const child = yield* spawner .spawn( ChildProcess.make(options.spawn.command, [...options.spawn.args], { ...(options.spawn.cwd ? { cwd: options.spawn.cwd } : {}), ...(options.spawn.env ? { env: { ...process.env, ...options.spawn.env } } : {}), shell: process.platform === "win32", }), ) .pipe( Effect.provideService(Scope.Scope, runtimeScope), Effect.mapError( (cause) => new EffectAcpErrors.AcpSpawnError({ command: options.spawn.command, cause, }), ), ); const acpContext = yield* Layer.build( EffectAcpClient.layerChildProcess(child, { ...(options.protocolLogging?.logIncoming !== undefined ? { logIncoming: options.protocolLogging.logIncoming } : {}), ...(options.protocolLogging?.logOutgoing !== undefined ? { logOutgoing: options.protocolLogging.logOutgoing } : {}), ...(options.protocolLogging?.logger ? { logger: options.protocolLogging.logger } : {}), }), ).pipe(Effect.provideService(Scope.Scope, runtimeScope)); const acp = yield* Effect.service(EffectAcpClient.AcpClient).pipe(Effect.provide(acpContext)); yield* acp.handleSessionUpdate((notification) => handleSessionUpdate({ queue: eventQueue, modeStateRef, toolCallsRef, assistantSegmentRef, params: notification, }), ); const initializeClientCapabilities = { fs: { readTextFile: false, writeTextFile: false, ...options.clientCapabilities?.fs, }, terminal: options.clientCapabilities?.terminal ?? false, ...(options.clientCapabilities?.auth ? { auth: options.clientCapabilities.auth } : {}), ...(options.clientCapabilities?.elicitation ? { elicitation: options.clientCapabilities.elicitation } : {}), ...(options.clientCapabilities?._meta ? { _meta: options.clientCapabilities._meta } : {}), } satisfies NonNullable<EffectAcpSchema.InitializeRequest["clientCapabilities"]>; const getStartedState = Effect.gen(function* () { const state = yield* Ref.get(startStateRef); if (state._tag === "Started") { return state.result; } return yield* new EffectAcpErrors.AcpTransportError({ detail: "ACP session runtime has not been started", cause: new Error("ACP session runtime has not been started"), }); }); const validateConfigOptionValue = ( configId: string, value: string | boolean, ): Effect.Effect<void, EffectAcpErrors.AcpError> => Effect.gen(function* () { const configOption = findSessionConfigOption(yield* Ref.get(configOptionsRef), configId); if (!configOption) { return; } if (configOption.type === "boolean") { if (typeof value === "boolean") { return; } return yield* new EffectAcpErrors.AcpRequestError({ code: -32602, errorMessage: `Invalid value ${JSON.stringify(value)} for session config option "${configOption.id}": expected boolean`, data: { configId: configOption.id, expectedType: "boolean", receivedValue: value, }, }); } if (typeof value !== "string") { return yield* new EffectAcpErrors.AcpRequestError({ code: -32602, errorMessage: `Invalid value ${JSON.stringify(value)} for session config option "${configOption.id}": expected string`, data: { configId: configOption.id, expectedType: "string", receivedValue: value, }, }); } const allowedValues = collectSessionConfigOptionValues(configOption); if (allowedValues.includes(value)) { return; } return yield* new EffectAcpErrors.AcpRequestError({ code: -32602, errorMessage: `Invalid value ${JSON.stringify(value)} for session config option "${configOption.id}": expected one of ${allowedValues.join(", ")}`, data: { configId: configOption.id, allowedValues, receivedValue: value, }, }); }); const updateConfigOptions = ( response: | EffectAcpSchema.SetSessionConfigOptionResponse | EffectAcpSchema.LoadSessionResponse | EffectAcpSchema.NewSessionResponse | EffectAcpSchema.ResumeSessionResponse, ): Effect.Effect<void> => Ref.set(configOptionsRef, sessionConfigOptionsFromSetup(response)); const updateCurrentModeId = (modeId: string): Effect.Effect<void> => Ref.update(modeStateRef, (current) => current ? { ...current, currentModeId: modeId } : current, ); const setConfigOption = ( configId: string, value: string | boolean, ): Effect.Effect<EffectAcpSchema.SetSessionConfigOptionResponse, EffectAcpErrors.AcpError> => validateConfigOptionValue(configId, value).pipe( Effect.flatMap(() => getStartedState), Effect.flatMap((started) => Ref.get(configOptionsRef).pipe( Effect.flatMap((configOptions) => { const existing = findSessionConfigOption(configOptions, configId); if (existing && configOptionCurrentValueMatches(existing, value)) { return Effect.succeed({ configOptions, } satisfies EffectAcpSchema.SetSessionConfigOptionResponse); } const requestPayload = typeof value === "boolean" ? ({ sessionId: started.sessionId, configId, type: "boolean", value, } satisfies EffectAcpSchema.SetSessionConfigOptionRequest) : ({ sessionId: started.sessionId, configId, value: String(value), } satisfies EffectAcpSchema.SetSessionConfigOptionRequest); return runLoggedRequest( "session/set_config_option", requestPayload, acp.agent.setSessionConfigOption(requestPayload), ).pipe(Effect.tap((response) => updateConfigOptions(response))); }), ), ), ); const startOnce = Effect.gen(function* () { const initializePayload = { protocolVersion: 1, clientCapabilities: initializeClientCapabilities, clientInfo: options.clientInfo, } satisfies EffectAcpSchema.InitializeRequest; const initializeResult = yield* runLoggedRequest( "initialize", initializePayload, acp.agent.initialize(initializePayload), ); const authenticatePayload = { methodId: options.authMethodId, } satisfies EffectAcpSchema.AuthenticateRequest; yield* runLoggedRequest( "authenticate", authenticatePayload, acp.agent.authenticate(authenticatePayload), ); let sessionId: string; let sessionSetupResult: | EffectAcpSchema.LoadSessionResponse | EffectAcpSchema.NewSessionResponse | EffectAcpSchema.ResumeSessionResponse; if (options.resumeSessionId) { const loadPayload = { sessionId: options.resumeSessionId, cwd: options.cwd, mcpServers: [], } satisfies EffectAcpSchema.LoadSessionRequest; const resumed = yield* runLoggedRequest( "session/load", loadPayload, acp.agent.loadSession(loadPayload), ).pipe(Effect.exit); if (Exit.isSuccess(resumed)) { sessionId = options.resumeSessionId; sessionSetupResult = resumed.value; } else { const createPayload = { cwd: options.cwd, mcpServers: [], } satisfies EffectAcpSchema.NewSessionRequest; const created = yield* runLoggedRequest( "session/new", createPayload, acp.agent.createSession(createPayload), ); sessionId = created.sessionId; sessionSetupResult = created; } } else { const createPayload = { cwd: options.cwd, mcpServers: [], } satisfies EffectAcpSchema.NewSessionRequest; const created = yield* runLoggedRequest( "session/new", createPayload, acp.agent.createSession(createPayload), ); sessionId = created.sessionId; sessionSetupResult = created; } yield* Ref.set(modeStateRef, parseSessionModeState(sessionSetupResult)); yield* Ref.set(configOptionsRef, sessionConfigOptionsFromSetup(sessionSetupResult)); const nextState = { sessionId, initializeResult, sessionSetupResult, modelConfigId: extractModelConfigId(sessionSetupResult), } satisfies AcpStartedState; return nextState; }); const start = Effect.gen(function* () { const deferred = yield* Deferred.make< AcpSessionRuntimeStartResult, EffectAcpErrors.AcpError >(); const effect = yield* Ref.modify(startStateRef, (state) => { switch (state._tag) { case "Started": return [Effect.succeed(state.result), state] as const; case "Starting": return [Deferred.await(state.deferred), state] as const; case "NotStarted": return [ startOnce.pipe( Effect.tap((result) => Ref.set(startStateRef, { _tag: "Started", result }).pipe( Effect.andThen(Deferred.succeed(deferred, result)), ), ), Effect.onError((cause) => Deferred.failCause(deferred, cause).pipe( Effect.andThen(Ref.set(startStateRef, { _tag: "NotStarted" })), ), ), ), { _tag: "Starting", deferred } satisfies AcpStartState, ] as const; } }); return yield* effect; }); return { handleRequestPermission: acp.handleRequestPermission, handleElicitation: acp.handleElicitation, handleReadTextFile: acp.handleReadTextFile, handleWriteTextFile: acp.handleWriteTextFile, handleCreateTerminal: acp.handleCreateTerminal, handleTerminalOutput: acp.handleTerminalOutput, handleTerminalWaitForExit: acp.handleTerminalWaitForExit, handleTerminalKill: acp.handleTerminalKill, handleTerminalRelease: acp.handleTerminalRelease, handleSessionUpdate: acp.handleSessionUpdate, handleElicitationComplete: acp.handleElicitationComplete, handleUnknownExtRequest: acp.handleUnknownExtRequest, handleUnknownExtNotification: acp.handleUnknownExtNotification, handleExtRequest: acp.handleExtRequest, handleExtNotification: acp.handleExtNotification, start: () => start, getEvents: () => Stream.fromQueue(eventQueue), getModeState: Ref.get(modeStateRef), getConfigOptions: Ref.get(configOptionsRef), prompt: (payload) => getStartedState.pipe( Effect.flatMap((started) => { const requestPayload = { sessionId: started.sessionId, ...payload, } satisfies EffectAcpSchema.PromptRequest; return closeActiveAssistantSegment({ queue: eventQueue, assistantSegmentRef, }).pipe( Effect.andThen( runLoggedRequest(Lines 1–500 of 722. Load more
import { Cause, Deferred, Effect, Exit, Layer, Queue, Ref, Scope, Context, Stream } from "effect"; import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; import * as EffectAcpClient from "effect-acp/client"; import * as EffectAcpErrors from "effect-acp/errors"; import type * as EffectAcpSchema from "effect-acp/schema"; import type * as EffectAcpProtocol from "effect-acp/protocol"; import { collectSessionConfigOptionValues, extractModelConfigId, findSessionConfigOption, mergeToolCallState, parseSessionModeState, parseSessionUpdateEvent, type AcpParsedSessionEvent, type AcpSessionModeState, type AcpToolCallState, } from "./AcpRuntimeModel.ts"; export interface AcpSpawnInput { readonly command: string; readonly args: ReadonlyArray<string>; readonly cwd?: string; readonly env?: NodeJS.ProcessEnv; } export interface AcpSessionRuntimeOptions { readonly spawn: AcpSpawnInput; readonly cwd: string; readonly resumeSessionId?: string; readonly clientCapabilities?: EffectAcpSchema.InitializeRequest["clientCapabilities"]; readonly clientInfo: { readonly name: string; readonly version: string; }; readonly authMethodId: string; readonly requestLogger?: (event: AcpSessionRequestLogEvent) => Effect.Effect<void, never>; readonly protocolLogging?: { readonly logIncoming?: boolean; readonly logOutgoing?: boolean; readonly logger?: (event: EffectAcpProtocol.AcpProtocolLogEvent) => Effect.Effect<void, never>; }; } export interface AcpSessionRequestLogEvent { readonly method: string; readonly payload: unknown; readonly status: "started" | "succeeded" | "failed"; readonly result?: unknown; readonly cause?: Cause.Cause<EffectAcpErrors.AcpError>; } export interface AcpSessionRuntimeStartResult { readonly sessionId: string; readonly initializeResult: EffectAcpSchema.InitializeResponse; readonly sessionSetupResult: | EffectAcpSchema.LoadSessionResponse | EffectAcpSchema.NewSessionResponse | EffectAcpSchema.ResumeSessionResponse; readonly modelConfigId: string | undefined; } export interface AcpSessionRuntimeShape { readonly handleRequestPermission: EffectAcpClient.AcpClientShape["handleRequestPermission"]; readonly handleElicitation: EffectAcpClient.AcpClientShape["handleElicitation"]; readonly handleReadTextFile: EffectAcpClient.AcpClientShape["handleReadTextFile"]; readonly handleWriteTextFile: EffectAcpClient.AcpClientShape["handleWriteTextFile"]; readonly handleCreateTerminal: EffectAcpClient.AcpClientShape["handleCreateTerminal"]; readonly handleTerminalOutput: EffectAcpClient.AcpClientShape["handleTerminalOutput"]; readonly handleTerminalWaitForExit: EffectAcpClient.AcpClientShape["handleTerminalWaitForExit"]; readonly handleTerminalKill: EffectAcpClient.AcpClientShape["handleTerminalKill"]; readonly handleTerminalRelease: EffectAcpClient.AcpClientShape["handleTerminalRelease"]; readonly handleSessionUpdate: EffectAcpClient.AcpClientShape["handleSessionUpdate"]; readonly handleElicitationComplete: EffectAcpClient.AcpClientShape["handleElicitationComplete"]; readonly handleUnknownExtRequest: EffectAcpClient.AcpClientShape["handleUnknownExtRequest"]; readonly handleUnknownExtNotification: EffectAcpClient.AcpClientShape["handleUnknownExtNotification"]; readonly handleExtRequest: EffectAcpClient.AcpClientShape["handleExtRequest"]; readonly handleExtNotification: EffectAcpClient.AcpClientShape["handleExtNotification"]; readonly start: () => Effect.Effect<AcpSessionRuntimeStartResult, EffectAcpErrors.AcpError>; readonly getEvents: () => Stream.Stream<AcpParsedSessionEvent, never>; readonly getModeState: Effect.Effect<AcpSessionModeState | undefined>; readonly getConfigOptions: Effect.Effect<ReadonlyArray<EffectAcpSchema.SessionConfigOption>>; readonly prompt: ( payload: Omit<EffectAcpSchema.PromptRequest, "sessionId">, ) => Effect.Effect<EffectAcpSchema.PromptResponse, EffectAcpErrors.AcpError>; readonly cancel: Effect.Effect<void, EffectAcpErrors.AcpError>; readonly setMode: ( modeId: string, ) => Effect.Effect<EffectAcpSchema.SetSessionModeResponse, EffectAcpErrors.AcpError>; readonly setConfigOption: ( configId: string, value: string | boolean, ) => Effect.Effect<EffectAcpSchema.SetSessionConfigOptionResponse, EffectAcpErrors.AcpError>; readonly setModel: (model: string) => Effect.Effect<void, EffectAcpErrors.AcpError>; readonly request: ( method: string, payload: unknown, ) => Effect.Effect<unknown, EffectAcpErrors.AcpError>; readonly notify: ( method: string, payload: unknown, ) => Effect.Effect<void, EffectAcpErrors.AcpError>; } interface AcpStartedState extends AcpSessionRuntimeStartResult {} type AcpStartState = | { readonly _tag: "NotStarted" } | { readonly _tag: "Starting"; readonly deferred: Deferred.Deferred<AcpSessionRuntimeStartResult, EffectAcpErrors.AcpError>; } | { readonly _tag: "Started"; readonly result: AcpStartedState }; interface AcpAssistantSegmentState { readonly nextSegmentIndex: number; readonly activeItemId?: string; } interface EnsureActiveAssistantSegmentResult { readonly itemId: string; readonly startedEvent?: Extract<AcpParsedSessionEvent, { readonly _tag: "AssistantItemStarted" }>; } export class AcpSessionRuntime extends Context.Service<AcpSessionRuntime, AcpSessionRuntimeShape>()( "t3/provider/acp/AcpSessionRuntime", ) { static layer( options: AcpSessionRuntimeOptions, ): Layer.Layer< AcpSessionRuntime, EffectAcpErrors.AcpError, ChildProcessSpawner.ChildProcessSpawner > { return Layer.effect(AcpSessionRuntime, makeAcpSessionRuntime(options)); } } const makeAcpSessionRuntime = ( options: AcpSessionRuntimeOptions, ): Effect.Effect< AcpSessionRuntimeShape, EffectAcpErrors.AcpError, ChildProcessSpawner.ChildProcessSpawner | Scope.Scope > => Effect.gen(function* () { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; const runtimeScope = yield* Scope.Scope; const eventQueue = yield* Queue.unbounded<AcpParsedSessionEvent>(); const modeStateRef = yield* Ref.make<AcpSessionModeState | undefined>(undefined); const toolCallsRef = yield* Ref.make(new Map<string, AcpToolCallState>()); const assistantSegmentRef = yield* Ref.make<AcpAssistantSegmentState>({ nextSegmentIndex: 0 }); const configOptionsRef = yield* Ref.make(sessionConfigOptionsFromSetup(undefined)); const startStateRef = yield* Ref.make<AcpStartState>({ _tag: "NotStarted" }); const logRequest = (event: AcpSessionRequestLogEvent) => options.requestLogger ? options.requestLogger(event) : Effect.void; const runLoggedRequest = <A>( method: string, payload: unknown, effect: Effect.Effect<A, EffectAcpErrors.AcpError>, ): Effect.Effect<A, EffectAcpErrors.AcpError> => logRequest({ method, payload, status: "started" }).pipe( Effect.flatMap(() => effect.pipe( Effect.tap((result) => logRequest({ method, payload, status: "succeeded", result, }), ), Effect.onError((cause) => logRequest({ method, payload, status: "failed", cause, }), ), ), ), ); const child = yield* spawner .spawn( ChildProcess.make(options.spawn.command, [...options.spawn.args], { ...(options.spawn.cwd ? { cwd: options.spawn.cwd } : {}), ...(options.spawn.env ? { env: { ...process.env, ...options.spawn.env } } : {}), shell: process.platform === "win32", }), ) .pipe( Effect.provideService(Scope.Scope, runtimeScope), Effect.mapError( (cause) => new EffectAcpErrors.AcpSpawnError({ command: options.spawn.command, cause, }), ), ); const acpContext = yield* Layer.build( EffectAcpClient.layerChildProcess(child, { ...(options.protocolLogging?.logIncoming !== undefined ? { logIncoming: options.protocolLogging.logIncoming } : {}), ...(options.protocolLogging?.logOutgoing !== undefined ? { logOutgoing: options.protocolLogging.logOutgoing } : {}), ...(options.protocolLogging?.logger ? { logger: options.protocolLogging.logger } : {}), }), ).pipe(Effect.provideService(Scope.Scope, runtimeScope)); const acp = yield* Effect.service(EffectAcpClient.AcpClient).pipe(Effect.provide(acpContext)); yield* acp.handleSessionUpdate((notification) => handleSessionUpdate({ queue: eventQueue, modeStateRef, toolCallsRef, assistantSegmentRef, params: notification, }), ); const initializeClientCapabilities = { fs: { readTextFile: false, writeTextFile: false, ...options.clientCapabilities?.fs, }, terminal: options.clientCapabilities?.terminal ?? false, ...(options.clientCapabilities?.auth ? { auth: options.clientCapabilities.auth } : {}), ...(options.clientCapabilities?.elicitation ? { elicitation: options.clientCapabilities.elicitation } : {}), ...(options.clientCapabilities?._meta ? { _meta: options.clientCapabilities._meta } : {}), } satisfies NonNullable<EffectAcpSchema.InitializeRequest["clientCapabilities"]>; const getStartedState = Effect.gen(function* () { const state = yield* Ref.get(startStateRef); if (state._tag === "Started") { return state.result; } return yield* new EffectAcpErrors.AcpTransportError({ detail: "ACP session runtime has not been started", cause: new Error("ACP session runtime has not been started"), }); }); const validateConfigOptionValue = ( configId: string, value: string | boolean, ): Effect.Effect<void, EffectAcpErrors.AcpError> => Effect.gen(function* () { const configOption = findSessionConfigOption(yield* Ref.get(configOptionsRef), configId); if (!configOption) { return; } if (configOption.type === "boolean") { if (typeof value === "boolean") { return; } return yield* new EffectAcpErrors.AcpRequestError({ code: -32602, errorMessage: `Invalid value ${JSON.stringify(value)} for session config option "${configOption.id}": expected boolean`, data: { configId: configOption.id, expectedType: "boolean", receivedValue: value, }, }); } if (typeof value !== "string") { return yield* new EffectAcpErrors.AcpRequestError({ code: -32602, errorMessage: `Invalid value ${JSON.stringify(value)} for session config option "${configOption.id}": expected string`, data: { configId: configOption.id, expectedType: "string", receivedValue: value, }, }); } const allowedValues = collectSessionConfigOptionValues(configOption); if (allowedValues.includes(value)) { return; } return yield* new EffectAcpErrors.AcpRequestError({ code: -32602, errorMessage: `Invalid value ${JSON.stringify(value)} for session config option "${configOption.id}": expected one of ${allowedValues.join(", ")}`, data: { configId: configOption.id, allowedValues, receivedValue: value, }, }); }); const updateConfigOptions = ( response: | EffectAcpSchema.SetSessionConfigOptionResponse | EffectAcpSchema.LoadSessionResponse | EffectAcpSchema.NewSessionResponse | EffectAcpSchema.ResumeSessionResponse, ): Effect.Effect<void> => Ref.set(configOptionsRef, sessionConfigOptionsFromSetup(response)); const updateCurrentModeId = (modeId: string): Effect.Effect<void> => Ref.update(modeStateRef, (current) => current ? { ...current, currentModeId: modeId } : current, ); const setConfigOption = ( configId: string, value: string | boolean, ): Effect.Effect<EffectAcpSchema.SetSessionConfigOptionResponse, EffectAcpErrors.AcpError> => validateConfigOptionValue(configId, value).pipe( Effect.flatMap(() => getStartedState), Effect.flatMap((started) => Ref.get(configOptionsRef).pipe( Effect.flatMap((configOptions) => { const existing = findSessionConfigOption(configOptions, configId); if (existing && configOptionCurrentValueMatches(existing, value)) { return Effect.succeed({ configOptions, } satisfies EffectAcpSchema.SetSessionConfigOptionResponse); } const requestPayload = typeof value === "boolean" ? ({ sessionId: started.sessionId, configId, type: "boolean", value, } satisfies EffectAcpSchema.SetSessionConfigOptionRequest) : ({ sessionId: started.sessionId, configId, value: String(value), } satisfies EffectAcpSchema.SetSessionConfigOptionRequest); return runLoggedRequest( "session/set_config_option", requestPayload, acp.agent.setSessionConfigOption(requestPayload), ).pipe(Effect.tap((response) => updateConfigOptions(response))); }), ), ), ); const startOnce = Effect.gen(function* () { const initializePayload = { protocolVersion: 1, clientCapabilities: initializeClientCapabilities, clientInfo: options.clientInfo, } satisfies EffectAcpSchema.InitializeRequest; const initializeResult = yield* runLoggedRequest( "initialize", initializePayload, acp.agent.initialize(initializePayload), ); const authenticatePayload = { methodId: options.authMethodId, } satisfies EffectAcpSchema.AuthenticateRequest; yield* runLoggedRequest( "authenticate", authenticatePayload, acp.agent.authenticate(authenticatePayload), ); let sessionId: string; let sessionSetupResult: | EffectAcpSchema.LoadSessionResponse | EffectAcpSchema.NewSessionResponse | EffectAcpSchema.ResumeSessionResponse; if (options.resumeSessionId) { const loadPayload = { sessionId: options.resumeSessionId, cwd: options.cwd, mcpServers: [], } satisfies EffectAcpSchema.LoadSessionRequest; const resumed = yield* runLoggedRequest( "session/load", loadPayload, acp.agent.loadSession(loadPayload), ).pipe(Effect.exit); if (Exit.isSuccess(resumed)) { sessionId = options.resumeSessionId; sessionSetupResult = resumed.value; } else { const createPayload = { cwd: options.cwd, mcpServers: [], } satisfies EffectAcpSchema.NewSessionRequest; const created = yield* runLoggedRequest( "session/new", createPayload, acp.agent.createSession(createPayload), ); sessionId = created.sessionId; sessionSetupResult = created; } } else { const createPayload = { cwd: options.cwd, mcpServers: [], } satisfies EffectAcpSchema.NewSessionRequest; const created = yield* runLoggedRequest( "session/new", createPayload, acp.agent.createSession(createPayload), ); sessionId = created.sessionId; sessionSetupResult = created; } yield* Ref.set(modeStateRef, parseSessionModeState(sessionSetupResult)); yield* Ref.set(configOptionsRef, sessionConfigOptionsFromSetup(sessionSetupResult)); const nextState = { sessionId, initializeResult, sessionSetupResult, modelConfigId: extractModelConfigId(sessionSetupResult), } satisfies AcpStartedState; return nextState; }); const start = Effect.gen(function* () { const deferred = yield* Deferred.make< AcpSessionRuntimeStartResult, EffectAcpErrors.AcpError >(); const effect = yield* Ref.modify(startStateRef, (state) => { switch (state._tag) { case "Started": return [Effect.succeed(state.result), state] as const; case "Starting": return [Deferred.await(state.deferred), state] as const; case "NotStarted": return [ startOnce.pipe( Effect.tap((result) => Ref.set(startStateRef, { _tag: "Started", result }).pipe( Effect.andThen(Deferred.succeed(deferred, result)), ), ), Effect.onError((cause) => Deferred.failCause(deferred, cause).pipe( Effect.andThen(Ref.set(startStateRef, { _tag: "NotStarted" })), ), ), ), { _tag: "Starting", deferred } satisfies AcpStartState, ] as const; } }); return yield* effect; }); return { handleRequestPermission: acp.handleRequestPermission, handleElicitation: acp.handleElicitation, handleReadTextFile: acp.handleReadTextFile, handleWriteTextFile: acp.handleWriteTextFile, handleCreateTerminal: acp.handleCreateTerminal, handleTerminalOutput: acp.handleTerminalOutput, handleTerminalWaitForExit: acp.handleTerminalWaitForExit, handleTerminalKill: acp.handleTerminalKill, handleTerminalRelease: acp.handleTerminalRelease, handleSessionUpdate: acp.handleSessionUpdate, handleElicitationComplete: acp.handleElicitationComplete, handleUnknownExtRequest: acp.handleUnknownExtRequest, handleUnknownExtNotification: acp.handleUnknownExtNotification, handleExtRequest: acp.handleExtRequest, handleExtNotification: acp.handleExtNotification, start: () => start, getEvents: () => Stream.fromQueue(eventQueue), getModeState: Ref.get(modeStateRef), getConfigOptions: Ref.get(configOptionsRef), prompt: (payload) => getStartedState.pipe( Effect.flatMap((started) => { const requestPayload = { sessionId: started.sessionId, ...payload, } satisfies EffectAcpSchema.PromptRequest; return closeActiveAssistantSegment({ queue: eventQueue, assistantSegmentRef, }).pipe( Effect.andThen( runLoggedRequest(