import * as Cause from "effect/Cause";import * as Clock from "effect/Clock";import * as Context from "effect/Context";import * as Crypto from "effect/Crypto";import * as Deferred from "effect/Deferred";import * as Duration from "effect/Duration";import * as Effect from "effect/Effect";import * as Exit from "effect/Exit";import * as Fiber from "effect/Fiber";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as Queue from "effect/Queue";import * as Ref from "effect/Ref";import * as Scope from "effect/Scope";import * as Semaphore from "effect/Semaphore";import * as Stream from "effect/Stream";import * as ChildProcess from "effect/unstable/process/ChildProcess";import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner";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 { resolveSpawnCommand } from "@t3tools/shared/shell"; import {  collectSessionConfigOptionValues,  decideToolCallUpdateEmission,  extractModelConfigId,  findSessionConfigOption,  mergeToolCallState,  toolCallProgressLength,  parseSessionModeState,  parseSessionUpdateEvent,  sessionUpdateIsReplay,  waitForSessionLoadReplayIdle,  type SessionLoadGate,  type AcpParsedSessionEvent,  type AcpSessionModeState,  type AcpToolCallState,} from "./AcpRuntimeModel.ts"; interface AcpToolCallTrackedState {  readonly state: AcpToolCallState;  readonly lastEmittedDetailLength: number | undefined;  readonly skippedSinceEmit: number;} function formatConfigOptionValue(value: string | boolean): string {  return JSON.stringify(value);} export interface AcpSessionEventStreamBarrier {  readonly _tag: "EventStreamBarrier";  readonly acknowledge: Deferred.Deferred<void>;} export type AcpSessionRuntimeEvent =  | AcpParsedSessionEvent  | AcpSessionEventStreamBarrier  | {      readonly _tag: "ConnectionTerminated";      readonly error: EffectAcpErrors.AcpError;    }; const defaultSessionLoadTimeout = Duration.seconds(90);const defaultSessionLoadReplayIdleGap = Duration.seconds(2);const defaultCancelTimeout = Duration.seconds(15);const maxStartupMetadataUpdates = 32;// Antigravity can emit an accepted 16 KiB Google authorization URL on stderr.const maxStderrChunkLength = 32_768; export interface AcpSpawnInput {  readonly command: string;  readonly args: ReadonlyArray<string>;  readonly cwd?: string;  readonly env?: NodeJS.ProcessEnv;  readonly extendEnv?: boolean;} export interface AcpSessionRuntimeOptions {  readonly spawn: AcpSpawnInput;  readonly cwd: string;  readonly resumeSessionId?: string;  readonly resumeMethod?: "load" | "resume";  readonly sessionLoadTimeout?: Duration.Input;  readonly sessionLoadReplayIdleGap?: Duration.Input;  /** Native cancellation waits for the prompt response and the getEvents consumer to drain. */  readonly cancelBehavior?: "interrupt" | "wait-for-prompt";  readonly cancelTimeout?: Duration.Input;  readonly clientCapabilities?: EffectAcpSchema.InitializeRequest["clientCapabilities"];  readonly clientInfo: {    readonly name: string;    readonly version: string;  };  readonly authMethodId: string;  readonly mcpServers?: ReadonlyArray<EffectAcpSchema.McpServer>;  /** Extra workspace roots the agent may read and write besides `cwd`. */  readonly additionalDirectories?: ReadonlyArray<string>;  /** Transforms provider stdout before protocol parsing and protocol logging. */  readonly transformStdout?: EffectAcpClient.AcpClientOptions["transformStdout"];  /** Normalizes provider-specific fields before notification queues or runtime state retain them. */  readonly transformSessionUpdate?: (    notification: EffectAcpSchema.SessionNotification,  ) => EffectAcpSchema.SessionNotification;  /** Receives bounded stderr chunks. Redact secrets before logging. A failure closes the runtime. */  readonly onStderr?: (text: string) => Effect.Effect<void, EffectAcpErrors.AcpError>;  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 class AcpSessionRuntime extends Context.Service<  AcpSessionRuntime,  {    /**     * Registers a handler for `session/request_permission`.     * @see https://agentclientprotocol.com/protocol/schema#session/request_permission     */    readonly handleRequestPermission: EffectAcpClient.AcpClient["Service"]["handleRequestPermission"];    /**     * Registers a handler for `session/elicitation`.     * @see https://agentclientprotocol.com/protocol/schema#session/elicitation     */    readonly handleElicitation: EffectAcpClient.AcpClient["Service"]["handleElicitation"];    /**     * Registers a handler for `fs/read_text_file`.     * @see https://agentclientprotocol.com/protocol/schema#fs/read_text_file     */    readonly handleReadTextFile: EffectAcpClient.AcpClient["Service"]["handleReadTextFile"];    /**     * Registers a handler for `fs/write_text_file`.     * @see https://agentclientprotocol.com/protocol/schema#fs/write_text_file     */    readonly handleWriteTextFile: EffectAcpClient.AcpClient["Service"]["handleWriteTextFile"];    /**     * Registers a handler for `terminal/create`.     * @see https://agentclientprotocol.com/protocol/schema#terminal/create     */    readonly handleCreateTerminal: EffectAcpClient.AcpClient["Service"]["handleCreateTerminal"];    /**     * Registers a handler for `terminal/output`.     * @see https://agentclientprotocol.com/protocol/schema#terminal/output     */    readonly handleTerminalOutput: EffectAcpClient.AcpClient["Service"]["handleTerminalOutput"];    /**     * Registers a handler for `terminal/wait_for_exit`.     * @see https://agentclientprotocol.com/protocol/schema#terminal/wait_for_exit     */    readonly handleTerminalWaitForExit: EffectAcpClient.AcpClient["Service"]["handleTerminalWaitForExit"];    /**     * Registers a handler for `terminal/kill`.     * @see https://agentclientprotocol.com/protocol/schema#terminal/kill     */    readonly handleTerminalKill: EffectAcpClient.AcpClient["Service"]["handleTerminalKill"];    /**     * Registers a handler for `terminal/release`.     * @see https://agentclientprotocol.com/protocol/schema#terminal/release     */    readonly handleTerminalRelease: EffectAcpClient.AcpClient["Service"]["handleTerminalRelease"];    /**     * Registers a handler for `session/update`.     * @see https://agentclientprotocol.com/protocol/schema#session/update     */    readonly handleSessionUpdate: EffectAcpClient.AcpClient["Service"]["handleSessionUpdate"];    /**     * Registers a handler for `session/elicitation/complete`.     * @see https://agentclientprotocol.com/protocol/schema#session/elicitation/complete     */    readonly handleElicitationComplete: EffectAcpClient.AcpClient["Service"]["handleElicitationComplete"];    /**     * Registers a fallback extension request handler.     * @see https://agentclientprotocol.com/protocol/extensibility     */    readonly handleUnknownExtRequest: EffectAcpClient.AcpClient["Service"]["handleUnknownExtRequest"];    /**     * Registers a fallback extension notification handler.     * @see https://agentclientprotocol.com/protocol/extensibility     */    readonly handleUnknownExtNotification: EffectAcpClient.AcpClient["Service"]["handleUnknownExtNotification"];    /**     * Registers a typed extension request handler.     * @see https://agentclientprotocol.com/protocol/extensibility     */    readonly handleExtRequest: EffectAcpClient.AcpClient["Service"]["handleExtRequest"];    /**     * Registers a typed extension notification handler.     * @see https://agentclientprotocol.com/protocol/extensibility     */    readonly handleExtNotification: EffectAcpClient.AcpClient["Service"]["handleExtNotification"];    /**     * Sends only `initialize` and returns the agent's response. Health probes use this to read     * advertised capabilities without authenticating or opening a session, so a probe can never     * start an interactive login or boot MCP servers.     * @see https://agentclientprotocol.com/protocol/schema#initialize     */    readonly initialize: () => Effect.Effect<      EffectAcpSchema.InitializeResponse,      EffectAcpErrors.AcpError    >;    /**     * Initializes the ACP connection, authenticates, and loads, resumes, or creates the session.     * Concurrent calls share the same in-flight startup and a failed startup may be retried.     */    readonly start: () => Effect.Effect<AcpSessionRuntimeStartResult, EffectAcpErrors.AcpError>;    /** Stream of parsed root-session events and connection failures. */    readonly getEvents: () => Stream.Stream<AcpSessionRuntimeEvent, never>;    /** Waits for queued events to be processed, or for the runtime scope to close. */    readonly drainEvents: Effect.Effect<void>;    /** Latest mode state observed from session setup and `session/update` notifications. */    readonly getModeState: Effect.Effect<AcpSessionModeState | undefined>;    /** Latest configuration options observed from session setup and configuration writes. */    readonly getConfigOptions: Effect.Effect<ReadonlyArray<EffectAcpSchema.SessionConfigOption>>;    /**     * Sends a prompt turn to the active session. `options.dispatched` settles once the     * `session/prompt` RPC is registered as the active prompt, so a caller that forks this     * effect knows when a later `cancel` will target this prompt.     * @see https://agentclientprotocol.com/protocol/schema#session/prompt     */    readonly prompt: (      payload: Omit<EffectAcpSchema.PromptRequest, "sessionId">,      options?: { readonly dispatched?: Deferred.Deferred<void> },    ) => Effect.Effect<EffectAcpSchema.PromptResponse, EffectAcpErrors.AcpError>;    /**     * Sends a real ACP `session/cancel` notification for the active session.     * @see https://agentclientprotocol.com/protocol/schema#session/cancel     */    readonly cancel: Effect.Effect<void, EffectAcpErrors.AcpError>;    /**     * Selects the active mode through the negotiated `mode` configuration option.     * This is a no-op when the requested mode is already active.     * @see https://agentclientprotocol.com/protocol/schema#session/set_config_option     */    readonly setMode: (      modeId: string,    ) => Effect.Effect<EffectAcpSchema.SetSessionModeResponse, EffectAcpErrors.AcpError>;    /**     * Updates a session configuration option and the runtime configuration snapshot.     * @see https://agentclientprotocol.com/protocol/schema#session/set_config_option     */    readonly setConfigOption: (      configId: string,      value: string | boolean,    ) => Effect.Effect<EffectAcpSchema.SetSessionConfigOptionResponse, EffectAcpErrors.AcpError>;    /**     * Selects the base model through the negotiated model configuration option.     * @see https://agentclientprotocol.com/protocol/schema#session/set_config_option     */    readonly setModel: (model: string) => Effect.Effect<void, EffectAcpErrors.AcpError>;    /**     * Selects the active model through the unstable ACP `session/set_model` capability.     * @see https://agentclientprotocol.com/protocol/schema#session/set_model     */    readonly setSessionModel: (      modelId: string,      meta?: EffectAcpSchema.SetSessionModelRequest["_meta"],    ) => Effect.Effect<EffectAcpSchema.SetSessionModelResponse, EffectAcpErrors.AcpError>;    /**     * Sends a generic ACP extension request and records it through the request logger.     * @see https://agentclientprotocol.com/protocol/extensibility     */    readonly request: (      method: string,      payload: unknown,    ) => Effect.Effect<unknown, EffectAcpErrors.AcpError>;    /**     * Sends a generic ACP extension notification.     * @see https://agentclientprotocol.com/protocol/extensibility     */    readonly notify: (      method: string,      payload: unknown,    ) => Effect.Effect<void, EffectAcpErrors.AcpError>;  }>()("t3/provider/acp/AcpSessionRuntime") {} 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" }>;} interface AcpActivePrompt {  readonly fiber: Fiber.Fiber<EffectAcpSchema.PromptResponse, EffectAcpErrors.AcpError>;  readonly completed: Deferred.Deferred<void>;} export const make = (  options: AcpSessionRuntimeOptions,): Effect.Effect<  AcpSessionRuntime["Service"],  EffectAcpErrors.AcpError,  ChildProcessSpawner.ChildProcessSpawner | Crypto.Crypto | Scope.Scope> =>  Effect.gen(function* () {    const crypto = yield* Crypto.Crypto;    const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;    const runtimeScope = yield* Scope.Scope;    const eventQueue = yield* Queue.unbounded<AcpSessionRuntimeEvent>();    const modeStateRef = yield* Ref.make<AcpSessionModeState | undefined>(undefined);    const toolCallsRef = yield* Ref.make(new Map<string, AcpToolCallTrackedState>());    const assistantItemRuntimeId = yield* crypto.randomUUIDv4.pipe(      Effect.mapError(        (cause) =>          new EffectAcpErrors.AcpTransportError({            detail: "Failed to generate an ACP assistant item runtime identifier.",            cause,          }),      ),    );    const assistantSegmentRef = yield* Ref.make<AcpAssistantSegmentState>({ nextSegmentIndex: 0 });    const configOptionsRef = yield* Ref.make(sessionConfigOptionsFromSetup(undefined));    const startStateRef = yield* Ref.make<AcpStartState>({ _tag: "NotStarted" });    const startupMetadataRef = yield* Ref.make<ReadonlyArray<EffectAcpSchema.SessionNotification>>(      [],    );    const notificationSemaphore = yield* Semaphore.make(1);    const terminationErrorRef = yield* Ref.make<Option.Option<EffectAcpErrors.AcpError>>(      Option.none(),    );    const stoppingRef = yield* Ref.make(false);    const stderrFailure = yield* Deferred.make<never, EffectAcpErrors.AcpError>();    const runtimeClosed = yield* Deferred.make<void>();    const promptSerializationSemaphore = yield* Semaphore.make(1);    const promptDispatchSemaphore = yield* Semaphore.make(1);    const activePromptRef = yield* Ref.make<Option.Option<AcpActivePrompt>>(Option.none());    const sessionLoadGateRef = yield* Ref.make<Option.Option<SessionLoadGate>>(Option.none());     const ensureConnected = Effect.gen(function* () {      const error = yield* Ref.get(terminationErrorRef);      if (Option.isSome(error)) {        return yield* error.value;      }      if (yield* Ref.get(stoppingRef)) {        return yield* new EffectAcpErrors.AcpTransportError({          detail: "The ACP session runtime is closed.",          cause: undefined,        });      }    });     const recordTermination = Effect.fn("AcpSessionRuntime.recordTermination")(function* (      error: EffectAcpErrors.AcpError,    ) {      if (yield* Ref.get(stoppingRef)) {        return;      }      const firstTermination = yield* Ref.modify(terminationErrorRef, (current) =>        Option.isSome(current)          ? ([false, current] as const)          : ([true, Option.some(error)] as const),      );      if (!firstTermination) {        return;      }      yield* closeActiveAssistantSegment({ queue: eventQueue, assistantSegmentRef });      yield* Queue.offer(eventQueue, { _tag: "ConnectionTerminated", error });    });     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(() =>          (options.onStderr            ? Effect.raceFirst(effect, Deferred.await(stderrFailure))            : effect          ).pipe(            Effect.tap((result) =>              logRequest({                method,                payload,                status: "succeeded",                result,              }),            ),            Effect.onError((cause) =>              logRequest({                method,                payload,                status: "failed",                cause,              }),            ),          ),        ),      );     const spawnCommand = yield* resolveSpawnCommand(options.spawn.command, options.spawn.args, {      ...(options.spawn.env ? { env: options.spawn.env } : {}),      extendEnv: options.spawn.extendEnv ?? true,    });    const child = yield* spawner      .spawn(        ChildProcess.make(spawnCommand.command, spawnCommand.args, {          ...(options.spawn.cwd ? { cwd: options.spawn.cwd } : {}),          ...(options.spawn.env ? { env: options.spawn.env } : {}),          extendEnv: options.spawn.extendEnv ?? true,          shell: spawnCommand.shell,        }),      )      .pipe(        Effect.provideService(Scope.Scope, runtimeScope),        Effect.mapError(          (cause) =>            new EffectAcpErrors.AcpSpawnError({              command: options.spawn.command,              cause,            }),        ),      );     yield* child.stderr.pipe(      Stream.decodeText(),      Stream.runForEach((chunk) =>        (options.onStderr          ? options.onStderr(chunk.slice(-maxStderrChunkLength))          : Effect.void        ).pipe(          Effect.catch((error) =>            Effect.gen(function* () {              yield* Deferred.fail(stderrFailure, error);              yield* recordTermination(error);              yield* child.kill({ forceKillAfter: "1 second" }).pipe(Effect.ignore);            }),          ),        ),      ),      Effect.ignore,      Effect.forkIn(runtimeScope),    );     const acpContext = yield* Layer.build(      EffectAcpClient.layerChildProcess(child, {        ...(options.transformStdout ? { transformStdout: options.transformStdout } : {}),        ...(options.transformSessionUpdate          ? { transformSessionUpdate: options.transformSessionUpdate }          : {}),        onTermination: recordTermination,        ...(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));     const processSessionUpdate = (notification: EffectAcpSchema.SessionNotification) =>      handleSessionUpdate({        queue: eventQueue,        modeStateRef,        configOptionsRef,        toolCallsRef,        assistantSegmentRef,        assistantItemRuntimeId,        params: notification,      });

Lines 1–500 of 1294. Load more