diff --git a/apps/mobile/src/components/ProviderIcon.tsx b/apps/mobile/src/components/ProviderIcon.tsx index 5eb69627f58d..3a5d4a8054f4 100644 --- a/apps/mobile/src/components/ProviderIcon.tsx +++ b/apps/mobile/src/components/ProviderIcon.tsx @@ -12,6 +12,32 @@ export function ProviderIcon(props: ProviderIconProps) { const size = props.size ?? 16; const mono = isDarkMode ? "#e5e5e5" : "#171717"; + if (props.provider === "pi") { + return ( + + + + + ); + } + + if (props.provider === "hermes") { + return ( + + + + ); + } + if (props.provider === "claudeAgent") { return ( diff --git a/apps/server/src/persistence/Migrations/040_ProjectionProjectFaviconPath.test.ts b/apps/server/src/persistence/Migrations/040_ProjectionProjectFaviconPath.test.ts index 7fd43d9b2ece..f161458b2a06 100644 --- a/apps/server/src/persistence/Migrations/040_ProjectionProjectFaviconPath.test.ts +++ b/apps/server/src/persistence/Migrations/040_ProjectionProjectFaviconPath.test.ts @@ -13,8 +13,8 @@ layer("040_ProjectionProjectFaviconPath", (it) => { Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; - yield* runMigrations({ toMigrationInclusive: 39 }); - yield* runMigrations({ toMigrationInclusive: 40 }); + yield* runMigrations({ toMigrationInclusive: 44 }); + yield* runMigrations({ toMigrationInclusive: 45 }); const columns = yield* sql<{ readonly name: string; readonly notnull: number }>` PRAGMA table_info(projection_projects) diff --git a/apps/server/src/provider/Drivers/HermesDriver.ts b/apps/server/src/provider/Drivers/HermesDriver.ts new file mode 100644 index 000000000000..03a79c11907b --- /dev/null +++ b/apps/server/src/provider/Drivers/HermesDriver.ts @@ -0,0 +1,22 @@ +import { HermesSettings, ProviderDriverKind } from "@t3tools/contracts"; +import * as Schema from "effect/Schema"; + +import { makeHermesAdapter } from "../Layers/HermesAdapter.ts"; +import { makeStandardAcpCliDriver, type StandardAcpCliDriverEnv } from "./StandardAcpCliDriver.ts"; + +const DRIVER_KIND = ProviderDriverKind.make("hermes"); +const decodeSettings = Schema.decodeSync(HermesSettings); + +export type HermesDriverEnv = StandardAcpCliDriverEnv; + +export const HermesDriver = makeStandardAcpCliDriver({ + driverKind: DRIVER_KIND, + displayName: "Hermes Agent", + defaultBinary: "hermes-acp", + settingsSchema: HermesSettings, + defaultSettings: () => decodeSettings({}), + makeAdapter: makeHermesAdapter, + excludedAuthMethodIds: new Set(["hermes-setup"]), + setupHint: "Run `hermes-acp --setup` to configure Hermes credentials.", + missingCommandMessage: "Hermes Agent (`hermes-acp`) is not installed or not on PATH.", +}); diff --git a/apps/server/src/provider/Drivers/PiDriver.ts b/apps/server/src/provider/Drivers/PiDriver.ts new file mode 100644 index 000000000000..7d0ba9f92c58 --- /dev/null +++ b/apps/server/src/provider/Drivers/PiDriver.ts @@ -0,0 +1,22 @@ +import { PiSettings, ProviderDriverKind } from "@t3tools/contracts"; +import * as Schema from "effect/Schema"; + +import { makePiAdapter } from "../Layers/PiAdapter.ts"; +import { makeStandardAcpCliDriver, type StandardAcpCliDriverEnv } from "./StandardAcpCliDriver.ts"; + +const DRIVER_KIND = ProviderDriverKind.make("pi"); +const decodeSettings = Schema.decodeSync(PiSettings); + +export type PiDriverEnv = StandardAcpCliDriverEnv; + +export const PiDriver = makeStandardAcpCliDriver({ + driverKind: DRIVER_KIND, + displayName: "Pi", + defaultBinary: "pi-acp", + settingsSchema: PiSettings, + defaultSettings: () => decodeSettings({}), + makeAdapter: makePiAdapter, + setupHint: + "Install both `pi-acp` and `@earendil-works/pi-coding-agent`, then authenticate with `pi`.", + missingCommandMessage: "Pi ACP (`pi-acp`) is not installed or not on PATH.", +}); diff --git a/apps/server/src/provider/Drivers/StandardAcpCliDriver.ts b/apps/server/src/provider/Drivers/StandardAcpCliDriver.ts new file mode 100644 index 000000000000..aedc415f806f --- /dev/null +++ b/apps/server/src/provider/Drivers/StandardAcpCliDriver.ts @@ -0,0 +1,191 @@ +import { + type ProviderDriverKind, + type ServerProvider, + TextGenerationError, +} from "@t3tools/contracts"; +import * as Crypto from "effect/Crypto"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; +import * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; +import { ChildProcessSpawner } from "effect/unstable/process"; + +import * as BackgroundPolicy from "../../background/BackgroundPolicy.ts"; +import { ServerConfig } from "../../config.ts"; +import { ServerSettingsService } from "../../serverSettings.ts"; +import type { TextGenerationShape } from "../../textGeneration/TextGeneration.ts"; +import { ProviderDriverError } from "../Errors.ts"; +import type { StandardAcpAdapterLiveOptions } from "../Layers/StandardAcpAdapter.ts"; +import { + buildInitialStandardAcpCliProviderSnapshot, + checkStandardAcpCliProviderStatus, + type StandardAcpCliProviderConfig, +} from "../Layers/StandardAcpCliProvider.ts"; +import { ProviderEventLoggers } from "../Layers/ProviderEventLoggers.ts"; +import { makeManagedServerProvider } from "../makeManagedServerProvider.ts"; +import { makeManualOnlyProviderMaintenanceCapabilities } from "../providerMaintenance.ts"; +import { + defaultProviderContinuationIdentity, + type ProviderDriver, + type ProviderInstance, +} from "../ProviderDriver.ts"; +import { mergeProviderInstanceEnvironment } from "../ProviderInstanceEnvironment.ts"; +import type { ServerProviderDraft } from "../providerSnapshot.ts"; + +const SNAPSHOT_REFRESH_INTERVAL = Duration.minutes(5); + +export interface StandardAcpCliSettings { + readonly enabled: boolean; + readonly binaryPath: string; + readonly customModels: ReadonlyArray; +} + +export type StandardAcpCliDriverEnv = + | BackgroundPolicy.BackgroundPolicy + | ChildProcessSpawner.ChildProcessSpawner + | Crypto.Crypto + | FileSystem.FileSystem + | Path.Path + | ProviderEventLoggers + | ServerConfig + | ServerSettingsService; + +export interface StandardAcpCliDriverConfig { + readonly driverKind: ProviderDriverKind; + readonly displayName: string; + readonly defaultBinary: string; + readonly settingsSchema: Schema.Codec; + readonly defaultSettings: () => Settings; + readonly makeAdapter: ( + settings: Settings, + options: StandardAcpAdapterLiveOptions, + ) => Effect.Effect; + readonly setupHint: string; + readonly missingCommandMessage: string; + readonly excludedAuthMethodIds?: ReadonlySet; +} + +function makeUnsupportedTextGeneration(displayName: string): TextGenerationShape { + const fail = (operation: TextGenerationError["operation"]) => + Effect.fail( + new TextGenerationError({ + operation, + detail: `${displayName} ACP is not supported for headless text generation. Select a different provider instance for generated metadata.`, + }), + ); + return { + generateCommitMessage: () => fail("generateCommitMessage"), + generatePrContent: () => fail("generatePrContent"), + generateBranchName: () => fail("generateBranchName"), + generateThreadTitle: () => fail("generateThreadTitle"), + }; +} + +export function makeStandardAcpCliDriver( + driverConfig: StandardAcpCliDriverConfig, +): ProviderDriver { + const maintenanceCapabilities = makeManualOnlyProviderMaintenanceCapabilities({ + provider: driverConfig.driverKind, + packageName: null, + }); + + const withInstanceIdentity = + (input: { + readonly instanceId: ProviderInstance["instanceId"]; + readonly displayName: string | undefined; + readonly accentColor: string | undefined; + readonly continuationGroupKey: string; + }) => + (snapshot: ServerProviderDraft): ServerProvider => ({ + ...snapshot, + instanceId: input.instanceId, + driver: driverConfig.driverKind, + ...(input.displayName ? { displayName: input.displayName } : {}), + ...(input.accentColor ? { accentColor: input.accentColor } : {}), + continuation: { groupKey: input.continuationGroupKey }, + }); + + return { + driverKind: driverConfig.driverKind, + metadata: { displayName: driverConfig.displayName, supportsMultipleInstances: true }, + configSchema: driverConfig.settingsSchema, + defaultConfig: driverConfig.defaultSettings, + create: ({ instanceId, displayName, accentColor, environment, enabled, config }) => + Effect.gen(function* () { + const crypto = yield* Crypto.Crypto; + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const processEnv = mergeProviderInstanceEnvironment(environment); + const effectiveConfig = { ...config, enabled } satisfies Settings; + const continuationIdentity = defaultProviderContinuationIdentity({ + driverKind: driverConfig.driverKind, + instanceId, + }); + const stampIdentity = withInstanceIdentity({ + instanceId, + displayName, + accentColor, + continuationGroupKey: continuationIdentity.continuationKey, + }); + const providerConfig: StandardAcpCliProviderConfig = { + provider: driverConfig.driverKind, + displayName: driverConfig.displayName, + command: effectiveConfig.binaryPath || driverConfig.defaultBinary, + enabled: effectiveConfig.enabled, + customModels: effectiveConfig.customModels, + environment: processEnv, + setupHint: driverConfig.setupHint, + missingCommandMessage: driverConfig.missingCommandMessage, + ...(driverConfig.excludedAuthMethodIds + ? { excludedAuthMethodIds: driverConfig.excludedAuthMethodIds } + : {}), + }; + + const eventLoggers = yield* ProviderEventLoggers; + const adapter = yield* driverConfig.makeAdapter(effectiveConfig, { + instanceId, + environment: processEnv, + ...(eventLoggers.native ? { nativeEventLogger: eventLoggers.native } : {}), + }); + const checkProvider = checkStandardAcpCliProviderStatus(providerConfig).pipe( + Effect.map(stampIdentity), + Effect.provideService(Crypto.Crypto, crypto), + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + ); + const snapshot = yield* makeManagedServerProvider({ + maintenanceCapabilities, + getSettings: Effect.succeed(providerConfig), + streamSettings: Stream.never, + haveSettingsChanged: () => false, + initialSnapshot: (settings) => + buildInitialStandardAcpCliProviderSnapshot(settings).pipe(Effect.map(stampIdentity)), + checkProvider, + refreshInterval: SNAPSHOT_REFRESH_INTERVAL, + }).pipe( + Effect.mapError( + (cause) => + new ProviderDriverError({ + driver: driverConfig.driverKind, + instanceId, + detail: `Failed to build ${driverConfig.displayName} snapshot: ${cause.message ?? String(cause)}`, + cause, + }), + ), + ); + + return { + instanceId, + driverKind: driverConfig.driverKind, + continuationIdentity, + displayName, + accentColor, + enabled, + snapshot, + adapter, + textGeneration: makeUnsupportedTextGeneration(driverConfig.displayName), + } satisfies ProviderInstance; + }), + }; +} diff --git a/apps/server/src/provider/Layers/GrokAdapter.ts b/apps/server/src/provider/Layers/GrokAdapter.ts index 858d862e6d5f..206ea605bad1 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.ts @@ -1,58 +1,5 @@ -import { - ApprovalRequestId, - type GrokSettings, - EventId, - type ProviderApprovalDecision, - type ProviderRuntimeEvent, - type ProviderSession, - type ProviderUserInputAnswers, - ProviderDriverKind, - ProviderInstanceId, - RuntimeRequestId, - type ThreadId, - TurnId, -} from "@t3tools/contracts"; -import * as Crypto from "effect/Crypto"; -import * as DateTime from "effect/DateTime"; -import * as Deferred from "effect/Deferred"; -import * as Effect from "effect/Effect"; -import * as Exit from "effect/Exit"; -import * as Fiber from "effect/Fiber"; -import * as FileSystem from "effect/FileSystem"; -import * as Option from "effect/Option"; -import * as Path from "effect/Path"; -import * as PubSub from "effect/PubSub"; -import * as Ref from "effect/Ref"; -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 SynchronizedRef from "effect/SynchronizedRef"; -import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner"; -import * as EffectAcpErrors from "effect-acp/errors"; -import type * as EffectAcpSchema from "effect-acp/schema"; +import { type GrokSettings, ProviderDriverKind, ProviderInstanceId } from "@t3tools/contracts"; -import { resolveAttachmentPath } from "../../attachmentStore.ts"; -import { ServerConfig } from "../../config.ts"; -import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; -import { - ProviderAdapterProcessError, - ProviderAdapterRequestError, - ProviderAdapterSessionNotFoundError, - ProviderAdapterValidationError, -} from "../Errors.ts"; -import { mapAcpToAdapterError } from "../acp/AcpAdapterSupport.ts"; -import type * as AcpSessionRuntime from "../acp/AcpSessionRuntime.ts"; -import { - makeAcpAssistantItemEvent, - makeAcpContentDeltaEvent, - makeAcpPlanUpdatedEvent, - makeAcpRequestOpenedEvent, - makeAcpRequestResolvedEvent, - makeAcpToolCallEvent, -} from "../acp/AcpCoreRuntimeEvents.ts"; -import { parsePermissionRequest } from "../acp/AcpRuntimeModel.ts"; -import { makeAcpNativeLoggerFactory } from "../acp/AcpNativeLogging.ts"; import { applyGrokAcpModelSelection, currentGrokModelIdFromSessionSetup, @@ -66,1405 +13,42 @@ import { promptResponseHasMissingXAiStopReason, XAiAskUserQuestionRequest, } from "../acp/XAiAcpExtension.ts"; -import { type GrokAdapterShape } from "../Services/GrokAdapter.ts"; -import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; - -const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.fromJsonString(Schema.Unknown)); - -const PROVIDER = ProviderDriverKind.make("grok"); -const GROK_RESUME_VERSION = 1 as const; - -function encodeJsonStringForDiagnostics(input: unknown): string | undefined { - const result = encodeUnknownJsonStringExit(input); - return Exit.isSuccess(result) ? result.value : undefined; -} - -export interface GrokAdapterLiveOptions { - readonly environment?: NodeJS.ProcessEnv; - readonly nativeEventLogPath?: string; - readonly nativeEventLogger?: EventNdjsonLogger; - readonly instanceId?: ProviderInstanceId; -} - -interface PendingApproval { - readonly decision: Deferred.Deferred; -} - -type PendingUserInputResolution = - | { readonly _tag: "answered"; readonly answers: ProviderUserInputAnswers } - | { readonly _tag: "cancelled" }; - -interface PendingUserInput { - readonly resolution: Deferred.Deferred; -} - -interface GrokSessionContext { - readonly threadId: ThreadId; - readonly acpSessionId: string; - session: ProviderSession; - readonly scope: Scope.Closeable; - readonly acp: AcpSessionRuntime.AcpSessionRuntime["Service"]; - notificationFiber: Fiber.Fiber | undefined; - readonly pendingApprovals: Map; - readonly pendingUserInputs: Map; - turns: Array<{ id: TurnId; items: Array }>; - lastPlanFingerprint: string | undefined; - activeTurnId: TurnId | undefined; - /** Turns already interrupted; late prompt RPCs must not resurrect them. */ - interruptedTurnIds: Set; - /** Number of sendTurn prompts currently in flight or being prepared. - * >0 means a turn is actively running, so a new sendTurn is a steer that - * continues it, and only the last remaining prompt settles the turn. */ - promptsInFlight: number; - currentModelId: string | undefined; - stopped: boolean; -} - -function settlePendingApprovalsAsCancelled( - pendingApprovals: ReadonlyMap, -): Effect.Effect { - return Effect.forEach( - Array.from(pendingApprovals.values()), - (pending) => Deferred.succeed(pending.decision, "cancel").pipe(Effect.ignore), - { discard: true }, - ); -} - -function settlePendingUserInputsAsCancelled( - pendingUserInputs: ReadonlyMap, -): Effect.Effect { - return Effect.forEach( - Array.from(pendingUserInputs.values()), - (pending) => Deferred.succeed(pending.resolution, { _tag: "cancelled" }).pipe(Effect.ignore), - { discard: true }, - ); -} - -function appendPromptResultToTurn( - ctx: GrokSessionContext, - turnId: TurnId, - promptParts: ReadonlyArray, - result: EffectAcpSchema.PromptResponse, -): void { - const existingTurnRecord = ctx.turns.find((turn) => turn.id === turnId); - ctx.turns = existingTurnRecord - ? ctx.turns.map((turn) => - turn.id === turnId - ? { ...turn, items: [...turn.items, { prompt: promptParts, result }] } - : turn, - ) - : [...ctx.turns, { id: turnId, items: [{ prompt: promptParts, result }] }]; -} - -function isRecord(value: unknown): value is Record { - return typeof value === "object" && value !== null && !Array.isArray(value); -} - -const resolveNotificationTurnId = (ctx: GrokSessionContext): TurnId | undefined => ctx.activeTurnId; - -const resolveCallbackTurnId = (ctx: GrokSessionContext): TurnId | undefined => ctx.activeTurnId; - -const resolveSessionCallbackTurnId = ( - sessions: ReadonlyMap, - threadId: ThreadId, -): TurnId | undefined => { - const ctx = sessions.get(threadId); - return ctx ? resolveCallbackTurnId(ctx) : undefined; -}; - -function parseGrokResume(raw: unknown): { sessionId: string } | undefined { - if (!isRecord(raw)) return undefined; - if (raw.schemaVersion !== GROK_RESUME_VERSION) return undefined; - if (typeof raw.sessionId !== "string" || !raw.sessionId.trim()) return undefined; - return { sessionId: raw.sessionId.trim() }; -} - -function selectPermissionOptionId( - request: EffectAcpSchema.RequestPermissionRequest, - decision: Exclude, -): string | undefined { - const kind = - decision === "acceptForSession" - ? "allow_always" - : decision === "accept" - ? "allow_once" - : "reject_once"; - const option = request.options.find((entry) => entry.kind === kind); - return option?.optionId.trim() || undefined; -} - -function selectAutoApprovedPermissionOption( - request: EffectAcpSchema.RequestPermissionRequest, -): string | undefined { - return ( - selectPermissionOptionId(request, "acceptForSession") ?? - selectPermissionOptionId(request, "accept") - ); -} +import { + makeStandardAcpAdapter, + standardAcpPromptSettlementBelongsToContext, + type StandardAcpAdapterLiveOptions, +} from "./StandardAcpAdapter.ts"; -function completedStopReasonFromPromptResponse( - response: EffectAcpSchema.PromptResponse | undefined, -): EffectAcpSchema.StopReason | null { - if (response === undefined || promptResponseHasMissingXAiStopReason(response)) { - return null; - } - return response.stopReason; -} +export interface GrokAdapterLiveOptions extends StandardAcpAdapterLiveOptions {} -export function grokPromptSettlementBelongsToContext(input: { - readonly liveAcpSessionId: string; - readonly expectedAcpSessionId: string; - readonly liveActiveTurnId: TurnId | undefined; - readonly liveSessionActiveTurnId: TurnId | undefined; - readonly turnId: TurnId; -}): boolean { - return ( - input.liveAcpSessionId === input.expectedAcpSessionId && - (input.liveActiveTurnId === input.turnId || input.liveSessionActiveTurnId === input.turnId) - ); -} +export const grokPromptSettlementBelongsToContext = standardAcpPromptSettlementBelongsToContext; export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapterLiveOptions) { - return Effect.gen(function* () { - const boundInstanceId = options?.instanceId ?? ProviderInstanceId.make("grok"); - const fileSystem = yield* FileSystem.FileSystem; - const path = yield* Path.Path; - const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; - const serverConfig = yield* Effect.service(ServerConfig); - const crypto = yield* Crypto.Crypto; - const nativeEventLogger = - options?.nativeEventLogger ?? - (options?.nativeEventLogPath !== undefined - ? yield* makeEventNdjsonLogger(options.nativeEventLogPath, { stream: "native" }) - : undefined); - const managedNativeEventLogger = - options?.nativeEventLogger === undefined ? nativeEventLogger : undefined; - const makeAcpNativeLoggers = yield* makeAcpNativeLoggerFactory(); - - const sessions = new Map(); - const threadLocksRef = yield* SynchronizedRef.make(new Map()); - const runtimeEventPubSub = yield* PubSub.unbounded(); - - const nowIso = Effect.map(DateTime.now, DateTime.formatIso); - const randomUUIDv4 = crypto.randomUUIDv4.pipe( - Effect.mapError( - (cause) => - new ProviderAdapterRequestError({ - provider: PROVIDER, - method: "crypto/randomUUIDv4", - detail: "Failed to generate Grok runtime identifier.", - cause, - }), - ), - ); - const nextEventId = Effect.map(randomUUIDv4, (id) => EventId.make(id)); - const makeEventStamp = () => Effect.all({ eventId: nextEventId, createdAt: nowIso }); - const mapAcpCallbackFailure = (effect: Effect.Effect) => - effect.pipe( - Effect.mapError( - (cause) => - new EffectAcpErrors.AcpTransportError({ - detail: "Failed to process Grok ACP callback.", - cause, - }), - ), - ); - - const offerRuntimeEvent = (event: ProviderRuntimeEvent) => - PubSub.publish(runtimeEventPubSub, event).pipe(Effect.asVoid); - - const getThreadSemaphore = (threadId: string) => - SynchronizedRef.modifyEffect(threadLocksRef, (current) => { - const existing: Option.Option = Option.fromNullishOr( - current.get(threadId), - ); - return Option.match(existing, { - onNone: () => - Semaphore.make(1).pipe( - Effect.map((semaphore) => { - const next = new Map(current); - next.set(threadId, semaphore); - return [semaphore, next] as const; - }), - ), - onSome: (semaphore) => Effect.succeed([semaphore, current] as const), - }); - }); - - const withThreadLock = (threadId: string, effect: Effect.Effect) => - Effect.flatMap(getThreadSemaphore(threadId), (semaphore) => semaphore.withPermit(effect)); - - const settlePromptInFlight = ( - threadId: ThreadId, - turnId: TurnId, - expectedAcpSessionId: string, - options?: { - readonly errorMessage?: string; - readonly completedStopReason?: EffectAcpSchema.StopReason | null; - readonly emitTurnCompletion?: boolean; - /** Interrupt/cancel: drop every outstanding prompt slot and settle once. */ - readonly settleAllPrompts?: boolean; - }, - ) => - Effect.gen(function* () { - const liveCtx = sessions.get(threadId); - if (!liveCtx) { - return; - } - const settlementBelongsToLiveContext = grokPromptSettlementBelongsToContext({ - liveAcpSessionId: liveCtx.acpSessionId, - expectedAcpSessionId, - liveActiveTurnId: liveCtx.activeTurnId, - liveSessionActiveTurnId: liveCtx.session.activeTurnId, - turnId, - }); - if (!settlementBelongsToLiveContext) { - // interruptTurn already consumed every prompt slot for this turn. A - // late prompt result must neither emit a second terminal event nor - // consume a slot belonging to a newer turn on the same ACP session. - if ( - liveCtx.acpSessionId !== expectedAcpSessionId || - liveCtx.interruptedTurnIds.has(turnId) - ) { - return; - } - if (options?.emitTurnCompletion !== false) { - if (options?.errorMessage !== undefined) { - yield* offerRuntimeEvent({ - type: "turn.completed", - ...(yield* makeEventStamp()), - provider: PROVIDER, - threadId, - turnId, - payload: { - state: "failed", - errorMessage: options.errorMessage, - }, - }); - } else if (options?.completedStopReason !== undefined) { - yield* offerRuntimeEvent({ - type: "turn.completed", - ...(yield* makeEventStamp()), - provider: PROVIDER, - threadId, - turnId, - payload: { - state: options.completedStopReason === "cancelled" ? "cancelled" : "completed", - stopReason: options.completedStopReason ?? null, - }, - }); - } - } - return; - } - let settleTurnId = turnId; - if (options?.settleAllPrompts) { - liveCtx.promptsInFlight = 0; - if (liveCtx.activeTurnId !== turnId && liveCtx.session.activeTurnId !== turnId) { - const fallbackTurnId = liveCtx.activeTurnId ?? liveCtx.session.activeTurnId; - if (!fallbackTurnId) { - if (liveCtx.session.status === "running" || liveCtx.session.status === "connecting") { - const updatedAt = yield* nowIso; - const { activeTurnId: _activeTurnId, ...readySession } = liveCtx.session; - liveCtx.activeTurnId = undefined; - liveCtx.session = { - ...readySession, - status: "ready", - updatedAt, - }; - } - return; - } - settleTurnId = fallbackTurnId; - } - } else { - const remainingPrompts = Math.max(0, liveCtx.promptsInFlight - 1); - if ( - remainingPrompts > 0 || - liveCtx.activeTurnId !== settleTurnId || - liveCtx.session.activeTurnId !== settleTurnId - ) { - liveCtx.promptsInFlight = remainingPrompts; - return; - } - liveCtx.promptsInFlight = remainingPrompts; - } - const updatedAt = yield* nowIso; - const canEmitTurnCompletion = - liveCtx.session.status === "running" || liveCtx.session.status === "connecting"; - const shouldEmitFailedTurn = options?.errorMessage !== undefined && canEmitTurnCompletion; - const shouldEmitCompletedTurn = - options?.completedStopReason !== undefined && canEmitTurnCompletion; - const { activeTurnId: _activeTurnId, ...readySession } = liveCtx.session; - liveCtx.activeTurnId = undefined; - liveCtx.session = { - ...readySession, - status: "ready", - updatedAt, - }; - if (options?.emitTurnCompletion === false) { - return; - } - if (shouldEmitFailedTurn) { - yield* offerRuntimeEvent({ - type: "turn.completed", - ...(yield* makeEventStamp()), - provider: PROVIDER, - threadId, - turnId: settleTurnId, - payload: { - state: "failed", - errorMessage: options.errorMessage, - }, - }); - } else if (shouldEmitCompletedTurn) { - yield* offerRuntimeEvent({ - type: "turn.completed", - ...(yield* makeEventStamp()), - provider: PROVIDER, - threadId, - turnId: settleTurnId, - payload: { - state: options.completedStopReason === "cancelled" ? "cancelled" : "completed", - stopReason: options.completedStopReason ?? null, - }, - }); - } - }); - - const logNative = (threadId: ThreadId, method: string, payload: unknown) => - Effect.gen(function* () { - if (!nativeEventLogger) return; - const observedAt = yield* nowIso; - yield* nativeEventLogger.write( - { - observedAt, - event: { - id: yield* randomUUIDv4, - kind: "notification", - provider: PROVIDER, - createdAt: observedAt, - method, - threadId, - payload, - }, - }, - threadId, - ); - }).pipe( - Effect.catchCause((cause) => - Effect.logWarning("Failed to write native Grok notification log.", { - cause, - threadId, - method, - }), - ), - ); - - const emitPlanUpdate = ( - ctx: GrokSessionContext, - turnId: TurnId | undefined, - stamp: { readonly eventId: EventId; readonly createdAt: string }, - payload: { - readonly explanation?: string | null; - readonly plan: ReadonlyArray<{ - readonly step: string; - readonly status: "pending" | "inProgress" | "completed"; - }>; - }, - rawPayload: unknown, - method: string, - ) => - Effect.gen(function* () { - const fingerprint = `${turnId ?? "no-turn"}:${encodeJsonStringForDiagnostics(payload) ?? "[unserializable payload]"}`; - if (ctx.lastPlanFingerprint === fingerprint) { - return; - } - ctx.lastPlanFingerprint = fingerprint; - yield* offerRuntimeEvent( - makeAcpPlanUpdatedEvent({ - stamp, - provider: PROVIDER, - threadId: ctx.threadId, - turnId, - payload, - source: "acp.jsonrpc", - method, - rawPayload, - }), - ); - }); - - const requireSession = ( - threadId: ThreadId, - ): Effect.Effect => { - const ctx = sessions.get(threadId); - if (!ctx || ctx.stopped) { - return Effect.fail( - new ProviderAdapterSessionNotFoundError({ provider: PROVIDER, threadId }), - ); - } - return Effect.succeed(ctx); - }; - - const stopSessionInternal = (ctx: GrokSessionContext) => - Effect.gen(function* () { - if (ctx.stopped) return; - ctx.stopped = true; - yield* settlePendingApprovalsAsCancelled(ctx.pendingApprovals); - yield* settlePendingUserInputsAsCancelled(ctx.pendingUserInputs); - if (ctx.notificationFiber) { - yield* Fiber.interrupt(ctx.notificationFiber); - } - yield* Effect.ignore(Scope.close(ctx.scope, Exit.void)); - sessions.delete(ctx.threadId); - yield* offerRuntimeEvent({ - type: "session.exited", - ...(yield* makeEventStamp()), - provider: PROVIDER, - threadId: ctx.threadId, - payload: { exitKind: "graceful" }, - }); - }); - - const startSession: GrokAdapterShape["startSession"] = (input) => - withThreadLock( - input.threadId, - Effect.gen(function* () { - if (input.provider !== undefined && input.provider !== PROVIDER) { - return yield* new ProviderAdapterValidationError({ - provider: PROVIDER, - operation: "startSession", - issue: `Expected provider '${PROVIDER}' but received '${input.provider}'.`, - }); - } - if (!input.cwd?.trim()) { - return yield* new ProviderAdapterValidationError({ - provider: PROVIDER, - operation: "startSession", - issue: "cwd is required and must be non-empty.", - }); - } - - const cwd = path.resolve(input.cwd.trim()); - const grokModelSelection = - input.modelSelection?.instanceId === boundInstanceId ? input.modelSelection : undefined; - const existing = sessions.get(input.threadId); - if (existing && !existing.stopped) { - yield* stopSessionInternal(existing); - } - - const pendingApprovals = new Map(); - const pendingUserInputs = new Map(); - const sessionScope = yield* Scope.make("sequential"); - let sessionScopeTransferred = false; - yield* Effect.addFinalizer(() => - sessionScopeTransferred ? Effect.void : Scope.close(sessionScope, Exit.void), - ); - - const resumeSessionId = parseGrokResume(input.resumeCursor)?.sessionId; - const acpNativeLoggers = makeAcpNativeLoggers({ - nativeEventLogger, - provider: PROVIDER, - threadId: input.threadId, - }); - - const mcpSession = McpProviderSession.readMcpProviderSession(input.threadId); - const acp = yield* makeGrokAcpRuntime({ - grokSettings, - ...(options?.environment ? { environment: options.environment } : {}), - childProcessSpawner, - cwd, - ...(resumeSessionId ? { resumeSessionId } : {}), - clientInfo: { name: "t3-code", version: "0.0.0" }, - ...(mcpSession - ? { - mcpServers: [ - { - type: "http" as const, - name: "t3-code", - url: mcpSession.endpoint, - headers: [ - { - name: "Authorization", - value: mcpSession.authorizationHeader, - }, - ], - }, - ], - } - : {}), - ...acpNativeLoggers, - }).pipe( - Effect.provideService(Crypto.Crypto, crypto), - Effect.provideService(Scope.Scope, sessionScope), - Effect.mapError( - (cause) => - new ProviderAdapterProcessError({ - provider: PROVIDER, - threadId: input.threadId, - detail: cause.message, - cause, - }), - ), - ); - const started = yield* Effect.gen(function* () { - yield* Effect.forEach( - ["x.ai/ask_user_question", "_x.ai/ask_user_question"] as const, - (method) => - acp.handleExtRequest(method, XAiAskUserQuestionRequest, (params) => - mapAcpCallbackFailure( - Effect.gen(function* () { - yield* logNative(input.threadId, method, params); - const requestId = ApprovalRequestId.make(yield* randomUUIDv4); - const runtimeRequestId = RuntimeRequestId.make(requestId); - const resolution = yield* Deferred.make(); - const turnId = resolveSessionCallbackTurnId(sessions, input.threadId); - pendingUserInputs.set(requestId, { resolution }); - yield* offerRuntimeEvent({ - type: "user-input.requested", - ...(yield* makeEventStamp()), - provider: PROVIDER, - threadId: input.threadId, - turnId, - requestId: runtimeRequestId, - payload: { questions: extractXAiAskUserQuestions(params) }, - raw: { - source: "acp.grok.extension", - method, - payload: params, - }, - }); - const resolved = yield* Deferred.await(resolution); - pendingUserInputs.delete(requestId); - const resolvedAnswers = resolved._tag === "answered" ? resolved.answers : {}; - yield* offerRuntimeEvent({ - type: "user-input.resolved", - ...(yield* makeEventStamp()), - provider: PROVIDER, - threadId: input.threadId, - turnId, - requestId: runtimeRequestId, - payload: { answers: resolvedAnswers }, - raw: { - source: "acp.grok.extension", - method, - payload: params, - }, - }); - switch (resolved._tag) { - case "answered": - return makeXAiAskUserQuestionResponse(params, resolved.answers); - case "cancelled": - return makeXAiAskUserQuestionCancelledResponse(); - } - }), - ), - ), - { discard: true }, - ); - yield* acp.handleRequestPermission((params) => - mapAcpCallbackFailure( - Effect.gen(function* () { - yield* logNative(input.threadId, "session/request_permission", params); - if (input.runtimeMode === "full-access") { - const autoApprovedOptionId = selectAutoApprovedPermissionOption(params); - if (autoApprovedOptionId !== undefined) { - return { - outcome: { - outcome: "selected" as const, - optionId: autoApprovedOptionId, - }, - }; - } - } - const permissionRequest = parsePermissionRequest(params); - const requestId = ApprovalRequestId.make(yield* randomUUIDv4); - const runtimeRequestId = RuntimeRequestId.make(requestId); - const decision = yield* Deferred.make(); - const turnId = resolveSessionCallbackTurnId(sessions, input.threadId); - pendingApprovals.set(requestId, { decision }); - yield* offerRuntimeEvent( - makeAcpRequestOpenedEvent({ - stamp: yield* makeEventStamp(), - provider: PROVIDER, - threadId: input.threadId, - turnId, - requestId: runtimeRequestId, - permissionRequest, - detail: - permissionRequest.detail ?? - encodeJsonStringForDiagnostics(params)?.slice(0, 2000) ?? - "[unserializable params]", - args: params, - source: "acp.jsonrpc", - method: "session/request_permission", - rawPayload: params, - }), - ); - const resolved = yield* Deferred.await(decision); - pendingApprovals.delete(requestId); - yield* offerRuntimeEvent( - makeAcpRequestResolvedEvent({ - stamp: yield* makeEventStamp(), - provider: PROVIDER, - threadId: input.threadId, - turnId, - requestId: runtimeRequestId, - permissionRequest, - decision: resolved, - }), - ); - const selectedOptionId = - resolved === "cancel" ? undefined : selectPermissionOptionId(params, resolved); - return { - outcome: selectedOptionId - ? { - outcome: "selected" as const, - optionId: selectedOptionId, - } - : ({ outcome: "cancelled" } as const), - }; - }), - ), - ); - return yield* acp.start(); - }).pipe( - Effect.mapError((error) => - mapAcpToAdapterError(PROVIDER, input.threadId, "session/start", error), - ), - ); - - const requestedStartModelId = grokModelSelection?.model - ? resolveGrokAcpBaseModelId(grokModelSelection.model) - : undefined; - const boundModelId = yield* applyGrokAcpModelSelection({ - runtime: acp, - currentModelId: currentGrokModelIdFromSessionSetup(started.sessionSetupResult), - requestedModelId: requestedStartModelId, - mapError: (cause) => - mapAcpToAdapterError(PROVIDER, input.threadId, "session/set_model", cause), - }); - - const now = yield* nowIso; - const session: ProviderSession = { - provider: PROVIDER, - providerInstanceId: boundInstanceId, - status: "ready", - runtimeMode: input.runtimeMode, - cwd, - ...(boundModelId ? { model: resolveGrokAcpBaseModelId(boundModelId) } : {}), - threadId: input.threadId, - resumeCursor: { - schemaVersion: GROK_RESUME_VERSION, - sessionId: started.sessionId, - }, - createdAt: now, - updatedAt: now, - }; - - const ctx: GrokSessionContext = { - threadId: input.threadId, - acpSessionId: started.sessionId, - session, - scope: sessionScope, - acp, - notificationFiber: undefined, - pendingApprovals, - pendingUserInputs, - turns: [], - lastPlanFingerprint: undefined, - activeTurnId: undefined, - interruptedTurnIds: new Set(), - promptsInFlight: 0, - currentModelId: boundModelId, - stopped: false, - }; - - const nf = yield* Stream.runDrain( - Stream.mapEffect(acp.getEvents(), (event) => - Effect.gen(function* () { - if (event._tag === "EventStreamBarrier") { - yield* Deferred.succeed(event.acknowledge, undefined); - return; - } - if ( - event._tag === "PlanUpdated" || - event._tag === "ToolCallUpdated" || - event._tag === "ContentDelta" - ) { - yield* logNative(ctx.threadId, "session/update", event.rawPayload); - } - - if (event._tag === "ModeChanged") { - return; - } - - const notificationTurnId = resolveNotificationTurnId(ctx); - if ( - notificationTurnId === undefined || - ctx.interruptedTurnIds.has(notificationTurnId) - ) { - return; - } - const stamp = yield* makeEventStamp(); - - switch (event._tag) { - case "AssistantItemStarted": - yield* offerRuntimeEvent( - makeAcpAssistantItemEvent({ - stamp, - provider: PROVIDER, - threadId: ctx.threadId, - turnId: notificationTurnId, - itemId: event.itemId, - lifecycle: "item.started", - }), - ); - return; - case "AssistantItemCompleted": - yield* offerRuntimeEvent( - makeAcpAssistantItemEvent({ - stamp, - provider: PROVIDER, - threadId: ctx.threadId, - turnId: notificationTurnId, - itemId: event.itemId, - lifecycle: "item.completed", - }), - ); - return; - case "PlanUpdated": - yield* emitPlanUpdate( - ctx, - notificationTurnId, - stamp, - event.payload, - event.rawPayload, - "session/update", - ); - return; - case "ToolCallUpdated": - yield* offerRuntimeEvent( - makeAcpToolCallEvent({ - stamp, - provider: PROVIDER, - threadId: ctx.threadId, - turnId: notificationTurnId, - toolCall: event.toolCall, - rawPayload: event.rawPayload, - }), - ); - return; - case "ContentDelta": - yield* offerRuntimeEvent( - makeAcpContentDeltaEvent({ - stamp, - provider: PROVIDER, - threadId: ctx.threadId, - turnId: notificationTurnId, - ...(event.itemId ? { itemId: event.itemId } : {}), - text: event.text, - rawPayload: event.rawPayload, - }), - ); - return; - } - }), - ), - ).pipe( - Effect.catch((cause) => - Effect.logError("Failed to process Grok runtime notification.", { cause }), - ), - // Fork into the session scope, not the calling fiber. `forkChild` - // makes this a child of `startSession`, and Effect interrupts a - // fiber's children when it completes, so the consumer died as soon - // as `startSession` returned and every later notification was - // dropped. The scope is created, stored on the context and closed - // on teardown already; only the fork target was wrong. - Effect.forkIn(ctx.scope), - ); - - ctx.notificationFiber = nf; - sessions.set(input.threadId, ctx); - sessionScopeTransferred = true; - - yield* offerRuntimeEvent({ - type: "session.started", - ...(yield* makeEventStamp()), - provider: PROVIDER, - threadId: input.threadId, - payload: { resume: started.initializeResult }, - }); - yield* offerRuntimeEvent({ - type: "session.state.changed", - ...(yield* makeEventStamp()), - provider: PROVIDER, - threadId: input.threadId, - payload: { state: "ready", reason: "Grok ACP session ready" }, - }); - yield* offerRuntimeEvent({ - type: "thread.started", - ...(yield* makeEventStamp()), - provider: PROVIDER, - threadId: input.threadId, - payload: { providerThreadId: started.sessionId }, - }); - - return session; - }).pipe(Effect.scoped), - ); - - const sendTurn: GrokAdapterShape["sendTurn"] = (input) => - Effect.gen(function* () { - const prepared = yield* withThreadLock( - input.threadId, - Effect.gen(function* () { - const ctx = yield* requireSession(input.threadId); - // A sendTurn while a prompt is in flight is a steer: the agent - // folds the new prompt into the ongoing work, so the active turn - // id is reused instead of opening a new turn. - const steeringTurnId = ctx.promptsInFlight > 0 ? ctx.activeTurnId : undefined; - const turnId = steeringTurnId ?? TurnId.make(yield* randomUUIDv4); - // Count this prompt immediately so a superseded in-flight prompt - // resolving from here on does not settle the turn; decremented on - // preparation failure here, and after the prompt below otherwise. - ctx.promptsInFlight += 1; - // Bind the turn id before cooperative yields so interruptTurn can - // settle this prompt even if stop arrives during preparation. - ctx.activeTurnId = turnId; - ctx.session = { - ...ctx.session, - status: steeringTurnId === undefined ? "connecting" : "running", - activeTurnId: turnId, - updatedAt: yield* nowIso, - }; - - return yield* Effect.gen(function* () { - const turnModelSelection = - input.modelSelection?.instanceId === boundInstanceId - ? input.modelSelection - : undefined; - const requestedTurnModelId = turnModelSelection?.model - ? resolveGrokAcpBaseModelId(turnModelSelection.model) - : undefined; - const currentModelId = yield* applyGrokAcpModelSelection({ - runtime: ctx.acp, - currentModelId: ctx.currentModelId, - requestedModelId: requestedTurnModelId, - mapError: (cause) => - mapAcpToAdapterError(PROVIDER, input.threadId, "session/set_model", cause), - }); - - const text = input.input?.trim(); - const imagePromptParts = yield* Effect.forEach( - input.attachments ?? [], - (attachment) => - Effect.gen(function* () { - const attachmentPath = resolveAttachmentPath({ - attachmentsDir: serverConfig.attachmentsDir, - attachment, - }); - if (!attachmentPath) { - return yield* new ProviderAdapterRequestError({ - provider: PROVIDER, - method: "session/prompt", - detail: `Invalid attachment id '${attachment.id}'.`, - }); - } - const bytes = yield* fileSystem.readFile(attachmentPath).pipe( - Effect.mapError( - (cause) => - new ProviderAdapterRequestError({ - provider: PROVIDER, - method: "session/prompt", - detail: cause.message, - cause, - }), - ), - ); - return { - type: "image", - data: Buffer.from(bytes).toString("base64"), - mimeType: attachment.mimeType, - } satisfies EffectAcpSchema.ContentBlock; - }), - ); - const promptParts: Array = [ - ...(text ? [{ type: "text" as const, text }] : []), - ...imagePromptParts, - ]; - - if (promptParts.length === 0) { - return yield* new ProviderAdapterValidationError({ - provider: PROVIDER, - operation: "sendTurn", - issue: "Turn requires non-empty text or attachments.", - }); - } - - ctx.currentModelId = currentModelId; - const displayModel = currentModelId - ? resolveGrokAcpBaseModelId(currentModelId) - : undefined; - for (let yieldAttempt = 0; yieldAttempt < 8; yieldAttempt += 1) { - yield* Effect.yieldNow; - } - if (ctx.interruptedTurnIds.has(turnId)) { - yield* settlePromptInFlight(input.threadId, turnId, ctx.acpSessionId, { - completedStopReason: "cancelled", - emitTurnCompletion: false, - settleAllPrompts: true, - }); - return yield* new ProviderAdapterRequestError({ - provider: PROVIDER, - method: "session/prompt", - detail: "Grok prompt was interrupted during preparation.", - }); - } - if (steeringTurnId === undefined) { - ctx.lastPlanFingerprint = undefined; - } - ctx.session = { - ...ctx.session, - status: "running", - activeTurnId: turnId, - updatedAt: yield* nowIso, - ...(displayModel ? { model: displayModel } : {}), - }; - - if (steeringTurnId === undefined) { - yield* offerRuntimeEvent({ - type: "turn.started", - ...(yield* makeEventStamp()), - provider: PROVIDER, - threadId: input.threadId, - turnId, - payload: displayModel ? { model: displayModel } : {}, - }); - } - - return { - acp: ctx.acp, - acpSessionId: ctx.acpSessionId, - displayModel, - promptParts, - turnId, - }; - }).pipe( - Effect.tapCause(() => - Effect.gen(function* () { - const liveCtx = sessions.get(input.threadId); - if (!liveCtx) { - return; - } - yield* settlePromptInFlight(input.threadId, turnId, liveCtx.acpSessionId, { - errorMessage: "Grok prompt preparation failed.", - emitTurnCompletion: false, - }); - }), - ), - ); - }), - ); - const promptSettled = yield* Ref.make(false); - const promptRpcSucceeded = yield* Ref.make(false); - const promptResultRef = yield* Ref.make( - undefined, - ); - - const promptFailureMessageRef = yield* Ref.make(undefined); - - return yield* Effect.gen(function* () { - const result = yield* prepared.acp - .prompt({ - prompt: prepared.promptParts, - }) - .pipe( - Effect.tap((promptResult) => - Effect.all([ - Ref.set(promptRpcSucceeded, true), - Ref.set(promptResultRef, promptResult), - ]), - ), - Effect.tapError((error) => - Ref.set( - promptFailureMessageRef, - mapAcpToAdapterError(PROVIDER, input.threadId, "session/prompt", error).message, - ).pipe(Effect.andThen(prepared.acp.drainEvents)), - ), - Effect.mapError((error) => - mapAcpToAdapterError(PROVIDER, input.threadId, "session/prompt", error), - ), - ); - - return yield* withThreadLock( - input.threadId, - Effect.gen(function* () { - const ctx = yield* requireSession(input.threadId); - if (ctx.acpSessionId !== prepared.acpSessionId) { - yield* settlePromptInFlight( - input.threadId, - prepared.turnId, - prepared.acpSessionId, - { - errorMessage: "Grok session changed before the turn completed.", - settleAllPrompts: true, - }, - ); - yield* Ref.set(promptSettled, true); - return yield* new ProviderAdapterRequestError({ - provider: PROVIDER, - method: "session/prompt", - detail: "Grok session changed before the turn completed.", - }); - } - // Keep prompt settlement atomic with respect to Stop and steering. - // interruptTurn marks its target before waiting for this lock, so - // cancellation can still win while queued ACP events are drained. - for (let yieldAttempt = 0; yieldAttempt < 8; yieldAttempt += 1) { - yield* Effect.yieldNow; - } - yield* prepared.acp.drainEvents; - if (ctx.interruptedTurnIds.has(prepared.turnId)) { - yield* Ref.set(promptSettled, true); - return { - threadId: input.threadId, - turnId: prepared.turnId, - resumeCursor: ctx.session.resumeCursor, - }; - } - - if ( - ctx.promptsInFlight <= 0 || - ctx.activeTurnId !== prepared.turnId || - ctx.session.activeTurnId !== prepared.turnId - ) { - yield* Ref.set(promptSettled, true); - return { - threadId: input.threadId, - turnId: prepared.turnId, - resumeCursor: ctx.session.resumeCursor, - }; - } - - appendPromptResultToTurn(ctx, prepared.turnId, prepared.promptParts, result); - ctx.session = { - ...ctx.session, - status: "running", - activeTurnId: prepared.turnId, - updatedAt: yield* nowIso, - ...(prepared.displayModel ? { model: prepared.displayModel } : {}), - }; - const remainingPrompts = Math.max(0, ctx.promptsInFlight - 1); - ctx.promptsInFlight = remainingPrompts; - - // Only the last remaining prompt settles the turn. A steer- - // superseded prompt resolving while another is in flight or - // pending must leave the merged turn running. - if ( - remainingPrompts === 0 && - ctx.activeTurnId === prepared.turnId && - ctx.session.activeTurnId === prepared.turnId - ) { - if (ctx.interruptedTurnIds.has(prepared.turnId)) { - yield* Ref.set(promptSettled, true); - return { - threadId: input.threadId, - turnId: prepared.turnId, - resumeCursor: ctx.session.resumeCursor, - }; - } - const completedAt = yield* nowIso; - const { activeTurnId: _completedTurnId, ...readySession } = ctx.session; - ctx.activeTurnId = undefined; - ctx.session = { - ...readySession, - status: "ready", - updatedAt: completedAt, - ...(prepared.displayModel ? { model: prepared.displayModel } : {}), - }; - const completedStopReason = completedStopReasonFromPromptResponse(result); - yield* offerRuntimeEvent({ - type: "turn.completed", - ...(yield* makeEventStamp()), - provider: PROVIDER, - threadId: input.threadId, - turnId: prepared.turnId, - payload: { - state: result.stopReason === "cancelled" ? "cancelled" : "completed", - stopReason: completedStopReason, - }, - }); - ctx.interruptedTurnIds.delete(prepared.turnId); - yield* Ref.set(promptSettled, true); - } else if (remainingPrompts > 0) { - yield* Ref.set(promptSettled, true); - } - - return { - threadId: input.threadId, - turnId: prepared.turnId, - resumeCursor: ctx.session.resumeCursor, - }; - }), - ); - }).pipe( - Effect.ensuring( - Effect.gen(function* () { - if (yield* Ref.get(promptSettled)) { - return; - } - - if (yield* Ref.get(promptRpcSucceeded)) { - const promptResult = yield* Ref.get(promptResultRef); - if (promptResult === undefined) { - return; - } - yield* withThreadLock( - input.threadId, - Effect.gen(function* () { - const ctx = yield* requireSession(input.threadId); - if (ctx.acpSessionId !== prepared.acpSessionId) { - yield* settlePromptInFlight( - input.threadId, - prepared.turnId, - prepared.acpSessionId, - { - errorMessage: "Grok session changed before the turn completed.", - settleAllPrompts: true, - }, - ); - return; - } - if (ctx.interruptedTurnIds.has(prepared.turnId)) { - return; - } - if ( - ctx.promptsInFlight <= 0 || - ctx.activeTurnId !== prepared.turnId || - ctx.session.activeTurnId !== prepared.turnId - ) { - return; - } - appendPromptResultToTurn( - ctx, - prepared.turnId, - prepared.promptParts, - promptResult, - ); - yield* settlePromptInFlight( - input.threadId, - prepared.turnId, - prepared.acpSessionId, - { - completedStopReason: completedStopReasonFromPromptResponse(promptResult), - }, - ); - }), - ); - return; - } - - const errorMessage = yield* Ref.get(promptFailureMessageRef); - yield* withThreadLock( - input.threadId, - settlePromptInFlight(input.threadId, prepared.turnId, prepared.acpSessionId, { - errorMessage: errorMessage ?? "Grok prompt request failed.", - }), - ); - }).pipe(Effect.catch(() => Effect.void)), - ), - ); - }); - - const interruptTurn: GrokAdapterShape["interruptTurn"] = (threadId, turnId) => - Effect.gen(function* () { - const observed = yield* Effect.sync(() => { - const ctx = sessions.get(threadId); - if (!ctx || ctx.stopped) { - return { - _tag: "Proceed" as const, - acpSessionId: undefined, - interruptedTurnId: turnId, - }; - } - const activeTurnId = ctx.activeTurnId ?? ctx.session.activeTurnId; - if (turnId !== undefined && activeTurnId !== undefined && activeTurnId !== turnId) { - return { _tag: "Ignore" as const }; - } - const interruptedTurnId = turnId ?? activeTurnId; - if (interruptedTurnId !== undefined) { - ctx.interruptedTurnIds.add(interruptedTurnId); - } - return { - _tag: "Proceed" as const, - acpSessionId: ctx.acpSessionId, - interruptedTurnId, - }; - }); - if (observed._tag === "Ignore") { - return; - } - - yield* withThreadLock( - threadId, - Effect.gen(function* () { - const ctx = yield* requireSession(threadId); - if (observed.acpSessionId !== undefined && ctx.acpSessionId !== observed.acpSessionId) { - return; - } - const activeTurnId = ctx.activeTurnId ?? ctx.session.activeTurnId; - if (turnId !== undefined && activeTurnId !== undefined && activeTurnId !== turnId) { - return; - } - if ( - observed.interruptedTurnId !== undefined && - activeTurnId !== undefined && - activeTurnId !== observed.interruptedTurnId - ) { - return; - } - const interruptedTurnId = - observed.interruptedTurnId ?? turnId ?? activeTurnId ?? ctx.session.activeTurnId; - yield* settlePendingApprovalsAsCancelled(ctx.pendingApprovals); - yield* settlePendingUserInputsAsCancelled(ctx.pendingUserInputs); - yield* Effect.ignore( - ctx.acp.cancel.pipe( - Effect.mapError((error) => - mapAcpToAdapterError(PROVIDER, threadId, "session/cancel", error), - ), - ), - ); - if (interruptedTurnId) { - ctx.interruptedTurnIds.add(interruptedTurnId); - yield* settlePromptInFlight(threadId, interruptedTurnId, ctx.acpSessionId, { - completedStopReason: "cancelled", - settleAllPrompts: true, - }); - } else if ( - ctx.promptsInFlight > 0 || - ctx.session.status === "running" || - ctx.session.status === "connecting" - ) { - const updatedAt = yield* nowIso; - ctx.promptsInFlight = 0; - ctx.activeTurnId = undefined; - const { activeTurnId: _activeTurnId, ...readySession } = ctx.session; - ctx.session = { - ...readySession, - status: "ready", - updatedAt, - }; - } - }), - ); - }); - - const respondToRequest: GrokAdapterShape["respondToRequest"] = ( - threadId, - requestId, - decision, - ) => - Effect.gen(function* () { - const ctx = yield* requireSession(threadId); - const pending = ctx.pendingApprovals.get(requestId); - if (!pending) { - return yield* new ProviderAdapterRequestError({ - provider: PROVIDER, - method: "session/request_permission", - detail: `Unknown pending approval request: ${requestId}`, - }); - } - yield* Deferred.succeed(pending.decision, decision); - }); - - const respondToUserInput: GrokAdapterShape["respondToUserInput"] = ( - threadId, - requestId, - answers, - ) => - Effect.gen(function* () { - const ctx = yield* requireSession(threadId); - const pending = ctx.pendingUserInputs.get(requestId); - if (!pending) { - return yield* new ProviderAdapterRequestError({ - provider: PROVIDER, - method: "_x.ai/ask_user_question", - detail: `Unknown pending user-input request: ${requestId}`, - }); - } - yield* Deferred.succeed(pending.resolution, { _tag: "answered", answers }); - }); - - const readThread: GrokAdapterShape["readThread"] = (threadId) => - Effect.gen(function* () { - const ctx = yield* requireSession(threadId); - return { threadId, turns: ctx.turns }; - }); - - const rollbackThread: GrokAdapterShape["rollbackThread"] = (threadId, numTurns) => - Effect.gen(function* () { - yield* requireSession(threadId); - if (!Number.isInteger(numTurns) || numTurns < 1) { - return yield* new ProviderAdapterValidationError({ - provider: PROVIDER, - operation: "rollbackThread", - issue: "numTurns must be an integer >= 1.", - }); - } - return yield* new ProviderAdapterRequestError({ - provider: PROVIDER, - method: "thread/rollback", - detail: "Grok ACP sessions do not support provider-side rollback yet.", - }); - }); - - const stopSession: GrokAdapterShape["stopSession"] = (threadId) => - withThreadLock( - threadId, - Effect.gen(function* () { - const ctx = yield* requireSession(threadId); - yield* stopSessionInternal(ctx); + return makeStandardAcpAdapter( + { + provider: ProviderDriverKind.make("grok"), + defaultInstanceId: ProviderInstanceId.make("grok"), + label: "Grok", + makeRuntime: (input) => + makeGrokAcpRuntime({ + ...input, + grokSettings, + ...(options?.environment ? { environment: options.environment } : {}), }), - ); - - const listSessions: GrokAdapterShape["listSessions"] = () => - Effect.sync(() => Array.from(sessions.values(), (c) => ({ ...c.session }))); - - const hasSession: GrokAdapterShape["hasSession"] = (threadId) => - Effect.sync(() => { - const c = sessions.get(threadId); - return c !== undefined && !c.stopped; - }); - - const stopAll: GrokAdapterShape["stopAll"] = () => - Effect.forEach(Array.from(sessions.values()), stopSessionInternal, { discard: true }); - - yield* Effect.addFinalizer(() => - Effect.ignore(stopAll()).pipe( - Effect.tap(() => PubSub.shutdown(runtimeEventPubSub)), - Effect.tap(() => managedNativeEventLogger?.close() ?? Effect.void), - ), - ); - - const streamEvents = Stream.fromPubSub(runtimeEventPubSub); - - return { - provider: PROVIDER, - capabilities: { sessionModelSwitch: "in-session" }, - startSession, - sendTurn, - interruptTurn, - readThread, - rollbackThread, - respondToRequest, - respondToUserInput, - stopSession, - listSessions, - hasSession, - stopAll, - streamEvents, - } satisfies GrokAdapterShape; - }); + normalizeModel: resolveGrokAcpBaseModelId, + currentModelFromSetup: currentGrokModelIdFromSessionSetup, + applyModelSelection: applyGrokAcpModelSelection, + promptStopReason: (response) => + promptResponseHasMissingXAiStopReason(response) ? null : response.stopReason, + userInput: { + methods: ["x.ai/ask_user_question", "_x.ai/ask_user_question"], + schema: XAiAskUserQuestionRequest, + extractQuestions: extractXAiAskUserQuestions, + makeAnsweredResponse: makeXAiAskUserQuestionResponse, + makeCancelledResponse: makeXAiAskUserQuestionCancelledResponse, + source: "acp.grok.extension", + }, + }, + options, + ); } diff --git a/apps/server/src/provider/Layers/HermesAdapter.ts b/apps/server/src/provider/Layers/HermesAdapter.ts new file mode 100644 index 000000000000..3c77a83f7cd9 --- /dev/null +++ b/apps/server/src/provider/Layers/HermesAdapter.ts @@ -0,0 +1,41 @@ +import { ProviderDriverKind, ProviderInstanceId, type HermesSettings } from "@t3tools/contracts"; + +import { + applyStandardAcpModelSelection, + currentStandardAcpModelFromSetup, + firstAdvertisedAuthMethod, + makeStandardAcpCliRuntime, + normalizeStandardAcpModel, +} from "../acp/StandardAcpCliSupport.ts"; +import { + makeStandardAcpAdapter, + type StandardAcpAdapterLiveOptions, +} from "./StandardAcpAdapter.ts"; + +const PROVIDER = ProviderDriverKind.make("hermes"); +const SETUP_AUTH_METHODS = new Set(["hermes-setup"]); + +export function makeHermesAdapter( + settings: HermesSettings, + options?: StandardAcpAdapterLiveOptions, +) { + return makeStandardAcpAdapter( + { + provider: PROVIDER, + defaultInstanceId: ProviderInstanceId.make("hermes"), + label: "Hermes Agent", + makeRuntime: (input) => + makeStandardAcpCliRuntime({ + ...input, + command: settings.binaryPath || "hermes-acp", + ...(options?.environment ? { environment: options.environment } : {}), + resolveAuthMethodId: (initializeResult) => + firstAdvertisedAuthMethod(initializeResult, SETUP_AUTH_METHODS), + }), + normalizeModel: (model) => normalizeStandardAcpModel(model, PROVIDER), + currentModelFromSetup: currentStandardAcpModelFromSetup, + applyModelSelection: applyStandardAcpModelSelection, + }, + options, + ); +} diff --git a/apps/server/src/provider/Layers/PiAdapter.ts b/apps/server/src/provider/Layers/PiAdapter.ts new file mode 100644 index 000000000000..53043977642e --- /dev/null +++ b/apps/server/src/provider/Layers/PiAdapter.ts @@ -0,0 +1,34 @@ +import { ProviderDriverKind, ProviderInstanceId, type PiSettings } from "@t3tools/contracts"; + +import { + applyStandardAcpModelSelection, + currentStandardAcpModelFromSetup, + makeStandardAcpCliRuntime, + normalizeStandardAcpModel, +} from "../acp/StandardAcpCliSupport.ts"; +import { + makeStandardAcpAdapter, + type StandardAcpAdapterLiveOptions, +} from "./StandardAcpAdapter.ts"; + +const PROVIDER = ProviderDriverKind.make("pi"); + +export function makePiAdapter(settings: PiSettings, options?: StandardAcpAdapterLiveOptions) { + return makeStandardAcpAdapter( + { + provider: PROVIDER, + defaultInstanceId: ProviderInstanceId.make("pi"), + label: "Pi", + makeRuntime: (input) => + makeStandardAcpCliRuntime({ + ...input, + command: settings.binaryPath || "pi-acp", + ...(options?.environment ? { environment: options.environment } : {}), + }), + normalizeModel: (model) => normalizeStandardAcpModel(model, PROVIDER), + currentModelFromSetup: currentStandardAcpModelFromSetup, + applyModelSelection: applyStandardAcpModelSelection, + }, + options, + ); +} diff --git a/apps/server/src/provider/Layers/ProviderRegistry.test.ts b/apps/server/src/provider/Layers/ProviderRegistry.test.ts index 48226e7bcb27..0b88191b41d6 100644 --- a/apps/server/src/provider/Layers/ProviderRegistry.test.ts +++ b/apps/server/src/provider/Layers/ProviderRegistry.test.ts @@ -1381,7 +1381,9 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te claudeAgent: { enabled: false }, cursor: { enabled: false }, grok: { enabled: false }, + hermes: { enabled: false }, opencode: { enabled: false }, + pi: { enabled: false }, }, // `providerInstances` keys are branded `ProviderInstanceId`; // the branded index signature rejects plain string literals @@ -1760,8 +1762,10 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te "droid", "geminiCli", "grok", + "hermes", "kilo", "opencode", + "pi", ]); assert.strictEqual(cursorProvider?.enabled, false); assert.strictEqual(cursorProvider?.status, "disabled"); diff --git a/apps/server/src/provider/Layers/StandardAcpAdapter.ts b/apps/server/src/provider/Layers/StandardAcpAdapter.ts new file mode 100644 index 000000000000..767c08e8972a --- /dev/null +++ b/apps/server/src/provider/Layers/StandardAcpAdapter.ts @@ -0,0 +1,1552 @@ +import { + ApprovalRequestId, + EventId, + type ProviderApprovalDecision, + type ProviderRuntimeEvent, + type ProviderSession, + type ProviderUserInputAnswers, + ProviderDriverKind, + ProviderInstanceId, + RuntimeRequestId, + type ThreadId, + TurnId, + type UserInputQuestion, +} from "@t3tools/contracts"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as PubSub from "effect/PubSub"; +import * as Ref from "effect/Ref"; +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 SynchronizedRef from "effect/SynchronizedRef"; +import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner"; +import * as EffectAcpErrors from "effect-acp/errors"; +import type * as EffectAcpSchema from "effect-acp/schema"; + +import { resolveAttachmentPath } from "../../attachmentStore.ts"; +import { ServerConfig } from "../../config.ts"; +import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; +import { + type ProviderAdapterError, + ProviderAdapterProcessError, + ProviderAdapterRequestError, + ProviderAdapterSessionNotFoundError, + ProviderAdapterValidationError, +} from "../Errors.ts"; +import { mapAcpToAdapterError } from "../acp/AcpAdapterSupport.ts"; +import type * as AcpSessionRuntime from "../acp/AcpSessionRuntime.ts"; +import { + makeAcpAssistantItemEvent, + makeAcpContentDeltaEvent, + makeAcpPlanUpdatedEvent, + makeAcpRequestOpenedEvent, + makeAcpRequestResolvedEvent, + makeAcpToolCallEvent, +} from "../acp/AcpCoreRuntimeEvents.ts"; +import { parsePermissionRequest } from "../acp/AcpRuntimeModel.ts"; +import { makeAcpNativeLoggerFactory } from "../acp/AcpNativeLogging.ts"; +import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts"; +import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; + +const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.fromJsonString(Schema.Unknown)); + +const STANDARD_ACP_RESUME_VERSION = 1 as const; +/** Lets queued ACP events and a pending Stop win prompt-settlement races. */ +const SETTLEMENT_YIELD_ATTEMPTS = 8; +type StandardAcpAdapterShape = ProviderAdapterShape; + +interface ThreadLockEntry { + readonly semaphore: Semaphore.Semaphore; + readonly leases: number; +} + +function encodeJsonStringForDiagnostics(input: unknown): string | undefined { + const result = encodeUnknownJsonStringExit(input); + return Exit.isSuccess(result) ? result.value : undefined; +} + +export interface StandardAcpAdapterLiveOptions { + readonly environment?: NodeJS.ProcessEnv; + readonly nativeEventLogPath?: string; + readonly nativeEventLogger?: EventNdjsonLogger; + readonly instanceId?: ProviderInstanceId; +} + +export interface StandardAcpUserInputRegistration { + readonly methods: ReadonlyArray; + readonly schema: Schema.Codec; + readonly extractQuestions: (params: Params) => ReadonlyArray; + readonly makeAnsweredResponse: (params: Params, answers: ProviderUserInputAnswers) => unknown; + readonly makeCancelledResponse: () => unknown; + readonly source: `acp.${string}.extension`; +} + +export interface StandardAcpAdapterConfig { + readonly provider: ProviderDriverKind; + readonly defaultInstanceId: ProviderInstanceId; + readonly label: string; + readonly makeRuntime: (input: { + readonly childProcessSpawner: ChildProcessSpawner.ChildProcessSpawner["Service"]; + readonly cwd: string; + readonly resumeSessionId?: string; + readonly clientInfo: { readonly name: string; readonly version: string }; + readonly mcpServers?: ReadonlyArray; + readonly requestLogger?: NonNullable< + AcpSessionRuntime.AcpSessionRuntimeOptions["requestLogger"] + >; + readonly protocolLogging?: NonNullable< + AcpSessionRuntime.AcpSessionRuntimeOptions["protocolLogging"] + >; + }) => Effect.Effect< + AcpSessionRuntime.AcpSessionRuntime["Service"], + EffectAcpErrors.AcpError, + Crypto.Crypto | Scope.Scope + >; + readonly normalizeModel: (model: string | null | undefined) => string; + readonly currentModelFromSetup: ( + setup: + | EffectAcpSchema.LoadSessionResponse + | EffectAcpSchema.NewSessionResponse + | EffectAcpSchema.ResumeSessionResponse, + ) => string | undefined; + readonly applyModelSelection: (input: { + readonly runtime: AcpSessionRuntime.AcpSessionRuntime["Service"]; + readonly currentModelId: string | undefined; + readonly requestedModelId: string | undefined; + readonly mapError: (cause: EffectAcpErrors.AcpError) => E; + }) => Effect.Effect; + readonly promptStopReason?: ( + response: EffectAcpSchema.PromptResponse, + ) => EffectAcpSchema.StopReason | null; + readonly userInput?: StandardAcpUserInputRegistration; +} + +interface PendingApproval { + readonly decision: Deferred.Deferred; +} + +type PendingUserInputResolution = + | { readonly _tag: "answered"; readonly answers: ProviderUserInputAnswers } + | { readonly _tag: "cancelled" }; + +interface PendingUserInput { + readonly resolution: Deferred.Deferred; +} + +interface StandardAcpSessionContext { + readonly threadId: ThreadId; + readonly acpSessionId: string; + session: ProviderSession; + readonly scope: Scope.Closeable; + readonly acp: AcpSessionRuntime.AcpSessionRuntime["Service"]; + notificationFiber: Fiber.Fiber | undefined; + readonly pendingApprovals: Map; + readonly pendingUserInputs: Map; + turns: Array<{ id: TurnId; items: Array }>; + lastPlanFingerprint: string | undefined; + activeTurnId: TurnId | undefined; + /** Turns already interrupted; late prompt RPCs must not resurrect them. */ + interruptedTurnIds: Set; + /** Number of sendTurn prompts currently in flight or being prepared. + * >0 means a turn is actively running, so a new sendTurn is a steer that + * continues it, and only the last remaining prompt settles the turn. */ + promptsInFlight: number; + currentModelId: string | undefined; + stopped: boolean; +} + +function settlePendingApprovalsAsCancelled( + pendingApprovals: ReadonlyMap, +): Effect.Effect { + return Effect.forEach( + Array.from(pendingApprovals.values()), + (pending) => Deferred.succeed(pending.decision, "cancel").pipe(Effect.ignore), + { discard: true }, + ); +} + +function settlePendingUserInputsAsCancelled( + pendingUserInputs: ReadonlyMap, +): Effect.Effect { + return Effect.forEach( + Array.from(pendingUserInputs.values()), + (pending) => Deferred.succeed(pending.resolution, { _tag: "cancelled" }).pipe(Effect.ignore), + { discard: true }, + ); +} + +function appendPromptResultToTurn( + ctx: StandardAcpSessionContext, + turnId: TurnId, + promptParts: ReadonlyArray, + result: EffectAcpSchema.PromptResponse, +): void { + const existingTurnRecord = ctx.turns.find((turn) => turn.id === turnId); + ctx.turns = existingTurnRecord + ? ctx.turns.map((turn) => + turn.id === turnId + ? { ...turn, items: [...turn.items, { prompt: promptParts, result }] } + : turn, + ) + : [...ctx.turns, { id: turnId, items: [{ prompt: promptParts, result }] }]; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +const resolveNotificationTurnId = (ctx: StandardAcpSessionContext): TurnId | undefined => + ctx.activeTurnId; + +const resolveCallbackTurnId = (ctx: StandardAcpSessionContext): TurnId | undefined => + ctx.activeTurnId; + +const resolveSessionCallbackTurnId = ( + sessions: ReadonlyMap, + threadId: ThreadId, +): TurnId | undefined => { + const ctx = sessions.get(threadId); + return ctx ? resolveCallbackTurnId(ctx) : undefined; +}; + +function parseStandardAcpResume(raw: unknown): { sessionId: string } | undefined { + if (!isRecord(raw)) return undefined; + if (raw.schemaVersion !== STANDARD_ACP_RESUME_VERSION) return undefined; + if (typeof raw.sessionId !== "string" || !raw.sessionId.trim()) return undefined; + return { sessionId: raw.sessionId.trim() }; +} + +function selectPermissionOptionId( + request: EffectAcpSchema.RequestPermissionRequest, + decision: Exclude, +): string | undefined { + const kind = + decision === "acceptForSession" + ? "allow_always" + : decision === "accept" + ? "allow_once" + : "reject_once"; + const option = request.options.find((entry) => entry.kind === kind); + return option?.optionId.trim() || undefined; +} + +function selectAutoApprovedPermissionOption( + request: EffectAcpSchema.RequestPermissionRequest, +): string | undefined { + return ( + selectPermissionOptionId(request, "acceptForSession") ?? + selectPermissionOptionId(request, "accept") + ); +} + +function completedStopReasonFromPromptResponse( + response: EffectAcpSchema.PromptResponse | undefined, + resolveStopReason?: StandardAcpAdapterConfig["promptStopReason"], +): EffectAcpSchema.StopReason | null { + if (response === undefined) { + return null; + } + return resolveStopReason ? resolveStopReason(response) : response.stopReason; +} + +export function standardAcpPromptSettlementBelongsToContext(input: { + readonly liveAcpSessionId: string; + readonly expectedAcpSessionId: string; + readonly liveActiveTurnId: TurnId | undefined; + readonly liveSessionActiveTurnId: TurnId | undefined; + readonly turnId: TurnId; +}): boolean { + return ( + input.liveAcpSessionId === input.expectedAcpSessionId && + (input.liveActiveTurnId === input.turnId || input.liveSessionActiveTurnId === input.turnId) + ); +} + +export function makeStandardAcpAdapter( + config: StandardAcpAdapterConfig, + options?: StandardAcpAdapterLiveOptions, +) { + return Effect.gen(function* () { + const PROVIDER = config.provider; + const boundInstanceId = options?.instanceId ?? config.defaultInstanceId; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const serverConfig = yield* Effect.service(ServerConfig); + const crypto = yield* Crypto.Crypto; + const nativeEventLogger = + options?.nativeEventLogger ?? + (options?.nativeEventLogPath !== undefined + ? yield* makeEventNdjsonLogger(options.nativeEventLogPath, { stream: "native" }) + : undefined); + const managedNativeEventLogger = + options?.nativeEventLogger === undefined ? nativeEventLogger : undefined; + const makeAcpNativeLoggers = yield* makeAcpNativeLoggerFactory(); + + const sessions = new Map(); + const threadLocksRef = yield* SynchronizedRef.make(new Map()); + const runtimeEventPubSub = yield* PubSub.unbounded(); + + const nowIso = Effect.map(DateTime.now, DateTime.formatIso); + const randomUUIDv4 = crypto.randomUUIDv4.pipe( + Effect.mapError( + (cause) => + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "crypto/randomUUIDv4", + detail: `Failed to generate ${config.label} runtime identifier.`, + cause, + }), + ), + ); + const nextEventId = Effect.map(randomUUIDv4, (id) => EventId.make(id)); + const makeEventStamp = () => Effect.all({ eventId: nextEventId, createdAt: nowIso }); + const mapAcpCallbackFailure = (effect: Effect.Effect) => + effect.pipe( + Effect.mapError( + (cause) => + new EffectAcpErrors.AcpTransportError({ + detail: `Failed to process ${config.label} ACP callback.`, + cause, + }), + ), + ); + + const offerRuntimeEvent = (event: ProviderRuntimeEvent) => + PubSub.publish(runtimeEventPubSub, event).pipe(Effect.asVoid); + + const acquireThreadLock = (threadId: string) => + SynchronizedRef.modifyEffect(threadLocksRef, (current) => { + const existing = current.get(threadId); + if (existing) { + const leased = { ...existing, leases: existing.leases + 1 }; + const next = new Map(current); + next.set(threadId, leased); + return Effect.succeed([leased, next] as const); + } + return Semaphore.make(1).pipe( + Effect.map((semaphore) => { + const entry = { semaphore, leases: 1 } satisfies ThreadLockEntry; + const next = new Map(current); + next.set(threadId, entry); + return [entry, next] as const; + }), + ); + }); + + const releaseThreadLock = (threadId: string, entry: ThreadLockEntry) => + SynchronizedRef.update(threadLocksRef, (current) => { + const live = current.get(threadId); + if (!live || live.semaphore !== entry.semaphore) return current; + const next = new Map(current); + if (live.leases === 1) { + next.delete(threadId); + } else { + next.set(threadId, { ...live, leases: live.leases - 1 }); + } + return next; + }); + + const withThreadLock = (threadId: string, effect: Effect.Effect) => + Effect.acquireUseRelease( + acquireThreadLock(threadId), + (entry) => entry.semaphore.withPermit(effect), + (entry) => releaseThreadLock(threadId, entry), + ); + + const settlePromptInFlight = ( + threadId: ThreadId, + turnId: TurnId, + expectedAcpSessionId: string, + options?: { + readonly errorMessage?: string; + readonly completedStopReason?: EffectAcpSchema.StopReason | null; + readonly emitTurnCompletion?: boolean; + /** Interrupt/cancel: drop every outstanding prompt slot and settle once. */ + readonly settleAllPrompts?: boolean; + }, + ) => + Effect.gen(function* () { + const liveCtx = sessions.get(threadId); + if (!liveCtx) { + return; + } + const settlementBelongsToLiveContext = standardAcpPromptSettlementBelongsToContext({ + liveAcpSessionId: liveCtx.acpSessionId, + expectedAcpSessionId, + liveActiveTurnId: liveCtx.activeTurnId, + liveSessionActiveTurnId: liveCtx.session.activeTurnId, + turnId, + }); + if (!settlementBelongsToLiveContext) { + // interruptTurn already consumed every prompt slot for this turn. A + // late prompt result must neither emit a second terminal event nor + // consume a slot belonging to a newer turn on the same ACP session. + if ( + liveCtx.acpSessionId !== expectedAcpSessionId || + liveCtx.interruptedTurnIds.has(turnId) + ) { + return; + } + if (options?.emitTurnCompletion !== false) { + if (options?.errorMessage !== undefined) { + yield* offerRuntimeEvent({ + type: "turn.completed", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId, + turnId, + payload: { + state: "failed", + errorMessage: options.errorMessage, + }, + }); + } else if (options?.completedStopReason !== undefined) { + yield* offerRuntimeEvent({ + type: "turn.completed", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId, + turnId, + payload: { + state: options.completedStopReason === "cancelled" ? "cancelled" : "completed", + stopReason: options.completedStopReason ?? null, + }, + }); + } + } + return; + } + let settleTurnId = turnId; + if (options?.settleAllPrompts) { + liveCtx.promptsInFlight = 0; + if (liveCtx.activeTurnId !== turnId && liveCtx.session.activeTurnId !== turnId) { + const fallbackTurnId = liveCtx.activeTurnId ?? liveCtx.session.activeTurnId; + if (!fallbackTurnId) { + if (liveCtx.session.status === "running" || liveCtx.session.status === "connecting") { + const updatedAt = yield* nowIso; + const { activeTurnId: _activeTurnId, ...readySession } = liveCtx.session; + liveCtx.activeTurnId = undefined; + liveCtx.session = { + ...readySession, + status: "ready", + updatedAt, + }; + } + return; + } + settleTurnId = fallbackTurnId; + } + } else { + const remainingPrompts = Math.max(0, liveCtx.promptsInFlight - 1); + if ( + remainingPrompts > 0 || + liveCtx.activeTurnId !== settleTurnId || + liveCtx.session.activeTurnId !== settleTurnId + ) { + liveCtx.promptsInFlight = remainingPrompts; + return; + } + liveCtx.promptsInFlight = remainingPrompts; + } + const updatedAt = yield* nowIso; + const canEmitTurnCompletion = + liveCtx.session.status === "running" || liveCtx.session.status === "connecting"; + const shouldEmitFailedTurn = options?.errorMessage !== undefined && canEmitTurnCompletion; + const shouldEmitCompletedTurn = + options?.completedStopReason !== undefined && canEmitTurnCompletion; + const { activeTurnId: _activeTurnId, ...readySession } = liveCtx.session; + liveCtx.activeTurnId = undefined; + liveCtx.session = { + ...readySession, + status: "ready", + updatedAt, + }; + if (options?.emitTurnCompletion === false) { + return; + } + if (shouldEmitFailedTurn) { + yield* offerRuntimeEvent({ + type: "turn.completed", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId, + turnId: settleTurnId, + payload: { + state: "failed", + errorMessage: options.errorMessage, + }, + }); + } else if (shouldEmitCompletedTurn) { + yield* offerRuntimeEvent({ + type: "turn.completed", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId, + turnId: settleTurnId, + payload: { + state: options.completedStopReason === "cancelled" ? "cancelled" : "completed", + stopReason: options.completedStopReason ?? null, + }, + }); + } + }); + + const logNative = (threadId: ThreadId, method: string, payload: unknown) => + Effect.gen(function* () { + if (!nativeEventLogger) return; + const observedAt = yield* nowIso; + yield* nativeEventLogger.write( + { + observedAt, + event: { + id: yield* randomUUIDv4, + kind: "notification", + provider: PROVIDER, + createdAt: observedAt, + method, + threadId, + payload, + }, + }, + threadId, + ); + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning(`Failed to write native ${config.label} notification log.`, { + cause, + threadId, + method, + }), + ), + ); + + const emitPlanUpdate = ( + ctx: StandardAcpSessionContext, + turnId: TurnId | undefined, + stamp: { readonly eventId: EventId; readonly createdAt: string }, + payload: { + readonly explanation?: string | null; + readonly plan: ReadonlyArray<{ + readonly step: string; + readonly status: "pending" | "inProgress" | "completed"; + }>; + }, + rawPayload: unknown, + method: string, + ) => + Effect.gen(function* () { + const fingerprint = `${turnId ?? "no-turn"}:${encodeJsonStringForDiagnostics(payload) ?? "[unserializable payload]"}`; + if (ctx.lastPlanFingerprint === fingerprint) { + return; + } + ctx.lastPlanFingerprint = fingerprint; + yield* offerRuntimeEvent( + makeAcpPlanUpdatedEvent({ + stamp, + provider: PROVIDER, + threadId: ctx.threadId, + turnId, + payload, + source: "acp.jsonrpc", + method, + rawPayload, + }), + ); + }); + + const requireSession = ( + threadId: ThreadId, + ): Effect.Effect => { + const ctx = sessions.get(threadId); + if (!ctx || ctx.stopped) { + return Effect.fail( + new ProviderAdapterSessionNotFoundError({ provider: PROVIDER, threadId }), + ); + } + return Effect.succeed(ctx); + }; + + const stopSessionInternal = (ctx: StandardAcpSessionContext) => + Effect.gen(function* () { + if (ctx.stopped) return; + ctx.stopped = true; + yield* settlePendingApprovalsAsCancelled(ctx.pendingApprovals); + yield* settlePendingUserInputsAsCancelled(ctx.pendingUserInputs); + if (ctx.notificationFiber) { + yield* Fiber.interrupt(ctx.notificationFiber); + } + yield* Effect.ignore(Scope.close(ctx.scope, Exit.void)); + sessions.delete(ctx.threadId); + yield* offerRuntimeEvent({ + type: "session.exited", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: ctx.threadId, + payload: { exitKind: "graceful" }, + }); + }); + + const startSession: StandardAcpAdapterShape["startSession"] = (input) => + withThreadLock( + input.threadId, + Effect.gen(function* () { + if (input.provider !== undefined && input.provider !== PROVIDER) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "startSession", + issue: `Expected provider '${PROVIDER}' but received '${input.provider}'.`, + }); + } + if (!input.cwd?.trim()) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "startSession", + issue: "cwd is required and must be non-empty.", + }); + } + + const cwd = path.resolve(input.cwd.trim()); + const modelSelection = + input.modelSelection?.instanceId === boundInstanceId ? input.modelSelection : undefined; + const existing = sessions.get(input.threadId); + if (existing && !existing.stopped) { + yield* stopSessionInternal(existing); + } + + const pendingApprovals = new Map(); + const pendingUserInputs = new Map(); + const sessionScope = yield* Scope.make("sequential"); + let sessionScopeTransferred = false; + yield* Effect.addFinalizer(() => + sessionScopeTransferred ? Effect.void : Scope.close(sessionScope, Exit.void), + ); + + const resumeSessionId = parseStandardAcpResume(input.resumeCursor)?.sessionId; + const acpNativeLoggers = makeAcpNativeLoggers({ + nativeEventLogger, + provider: PROVIDER, + threadId: input.threadId, + }); + + const mcpSession = McpProviderSession.readMcpProviderSession(input.threadId); + const acp = yield* config + .makeRuntime({ + childProcessSpawner, + cwd, + ...(resumeSessionId ? { resumeSessionId } : {}), + clientInfo: { name: "t3-code", version: "0.0.0" }, + ...(mcpSession + ? { + mcpServers: [ + { + type: "http" as const, + name: "t3-code", + url: mcpSession.endpoint, + headers: [ + { + name: "Authorization", + value: mcpSession.authorizationHeader, + }, + ], + }, + ], + } + : {}), + ...acpNativeLoggers, + }) + .pipe( + Effect.provideService(Crypto.Crypto, crypto), + Effect.provideService(Scope.Scope, sessionScope), + Effect.mapError( + (cause) => + new ProviderAdapterProcessError({ + provider: PROVIDER, + threadId: input.threadId, + detail: cause.message, + cause, + }), + ), + ); + const started = yield* Effect.gen(function* () { + const userInput = config.userInput; + if (userInput) { + yield* Effect.forEach( + userInput.methods, + (method) => + acp.handleExtRequest(method, userInput.schema, (params) => + mapAcpCallbackFailure( + Effect.gen(function* () { + yield* logNative(input.threadId, method, params); + const requestId = ApprovalRequestId.make(yield* randomUUIDv4); + const runtimeRequestId = RuntimeRequestId.make(requestId); + const resolution = yield* Deferred.make(); + const turnId = resolveSessionCallbackTurnId(sessions, input.threadId); + pendingUserInputs.set(requestId, { resolution }); + yield* offerRuntimeEvent({ + type: "user-input.requested", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + turnId, + requestId: runtimeRequestId, + payload: { questions: userInput.extractQuestions(params) }, + raw: { + source: userInput.source, + method, + payload: params, + }, + }); + const resolved = yield* Deferred.await(resolution); + pendingUserInputs.delete(requestId); + const resolvedAnswers = + resolved._tag === "answered" ? resolved.answers : {}; + yield* offerRuntimeEvent({ + type: "user-input.resolved", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + turnId, + requestId: runtimeRequestId, + payload: { answers: resolvedAnswers }, + raw: { + source: userInput.source, + method, + payload: params, + }, + }); + switch (resolved._tag) { + case "answered": + return userInput.makeAnsweredResponse(params, resolved.answers); + case "cancelled": + return userInput.makeCancelledResponse(); + } + }), + ), + ), + { discard: true }, + ); + } + yield* acp.handleRequestPermission((params) => + mapAcpCallbackFailure( + Effect.gen(function* () { + yield* logNative(input.threadId, "session/request_permission", params); + if (input.runtimeMode === "full-access") { + const autoApprovedOptionId = selectAutoApprovedPermissionOption(params); + if (autoApprovedOptionId !== undefined) { + return { + outcome: { + outcome: "selected" as const, + optionId: autoApprovedOptionId, + }, + }; + } + } + const permissionRequest = parsePermissionRequest(params); + const requestId = ApprovalRequestId.make(yield* randomUUIDv4); + const runtimeRequestId = RuntimeRequestId.make(requestId); + const decision = yield* Deferred.make(); + const turnId = resolveSessionCallbackTurnId(sessions, input.threadId); + pendingApprovals.set(requestId, { decision }); + yield* offerRuntimeEvent( + makeAcpRequestOpenedEvent({ + stamp: yield* makeEventStamp(), + provider: PROVIDER, + threadId: input.threadId, + turnId, + requestId: runtimeRequestId, + permissionRequest, + detail: + permissionRequest.detail ?? + encodeJsonStringForDiagnostics(params)?.slice(0, 2000) ?? + "[unserializable params]", + args: params, + source: "acp.jsonrpc", + method: "session/request_permission", + rawPayload: params, + }), + ); + const resolved = yield* Deferred.await(decision); + pendingApprovals.delete(requestId); + yield* offerRuntimeEvent( + makeAcpRequestResolvedEvent({ + stamp: yield* makeEventStamp(), + provider: PROVIDER, + threadId: input.threadId, + turnId, + requestId: runtimeRequestId, + permissionRequest, + decision: resolved, + }), + ); + const selectedOptionId = + resolved === "cancel" ? undefined : selectPermissionOptionId(params, resolved); + return { + outcome: selectedOptionId + ? { + outcome: "selected" as const, + optionId: selectedOptionId, + } + : ({ outcome: "cancelled" } as const), + }; + }), + ), + ); + return yield* acp.start(); + }).pipe( + Effect.mapError((error) => + mapAcpToAdapterError(PROVIDER, input.threadId, "session/start", error), + ), + ); + + const requestedStartModelId = modelSelection?.model + ? config.normalizeModel(modelSelection.model) + : undefined; + const boundModelId = yield* config.applyModelSelection({ + runtime: acp, + currentModelId: config.currentModelFromSetup(started.sessionSetupResult), + requestedModelId: requestedStartModelId, + mapError: (cause) => + mapAcpToAdapterError(PROVIDER, input.threadId, "session/set_model", cause), + }); + + const now = yield* nowIso; + const session: ProviderSession = { + provider: PROVIDER, + providerInstanceId: boundInstanceId, + status: "ready", + runtimeMode: input.runtimeMode, + cwd, + ...(boundModelId ? { model: config.normalizeModel(boundModelId) } : {}), + threadId: input.threadId, + resumeCursor: { + schemaVersion: STANDARD_ACP_RESUME_VERSION, + sessionId: started.sessionId, + }, + createdAt: now, + updatedAt: now, + }; + + const ctx: StandardAcpSessionContext = { + threadId: input.threadId, + acpSessionId: started.sessionId, + session, + scope: sessionScope, + acp, + notificationFiber: undefined, + pendingApprovals, + pendingUserInputs, + turns: [], + lastPlanFingerprint: undefined, + activeTurnId: undefined, + interruptedTurnIds: new Set(), + promptsInFlight: 0, + currentModelId: boundModelId, + stopped: false, + }; + + const nf = yield* Stream.runDrain( + Stream.mapEffect(acp.getEvents(), (event) => + Effect.gen(function* () { + if (event._tag === "EventStreamBarrier") { + yield* Deferred.succeed(event.acknowledge, undefined); + return; + } + if ( + event._tag === "PlanUpdated" || + event._tag === "ToolCallUpdated" || + event._tag === "ContentDelta" + ) { + yield* logNative(ctx.threadId, "session/update", event.rawPayload); + } + + if (event._tag === "ModeChanged") { + return; + } + + const notificationTurnId = resolveNotificationTurnId(ctx); + if ( + notificationTurnId === undefined || + ctx.interruptedTurnIds.has(notificationTurnId) + ) { + return; + } + const stamp = yield* makeEventStamp(); + + switch (event._tag) { + case "AssistantItemStarted": + yield* offerRuntimeEvent( + makeAcpAssistantItemEvent({ + stamp, + provider: PROVIDER, + threadId: ctx.threadId, + turnId: notificationTurnId, + itemId: event.itemId, + lifecycle: "item.started", + }), + ); + return; + case "AssistantItemCompleted": + yield* offerRuntimeEvent( + makeAcpAssistantItemEvent({ + stamp, + provider: PROVIDER, + threadId: ctx.threadId, + turnId: notificationTurnId, + itemId: event.itemId, + lifecycle: "item.completed", + }), + ); + return; + case "PlanUpdated": + yield* emitPlanUpdate( + ctx, + notificationTurnId, + stamp, + event.payload, + event.rawPayload, + "session/update", + ); + return; + case "ToolCallUpdated": + yield* offerRuntimeEvent( + makeAcpToolCallEvent({ + stamp, + provider: PROVIDER, + threadId: ctx.threadId, + turnId: notificationTurnId, + toolCall: event.toolCall, + rawPayload: event.rawPayload, + }), + ); + return; + case "ContentDelta": + yield* offerRuntimeEvent( + makeAcpContentDeltaEvent({ + stamp, + provider: PROVIDER, + threadId: ctx.threadId, + turnId: notificationTurnId, + ...(event.itemId ? { itemId: event.itemId } : {}), + text: event.text, + rawPayload: event.rawPayload, + }), + ); + return; + } + }), + ), + ).pipe( + Effect.catch((cause) => + Effect.logError(`Failed to process ${config.label} runtime notification.`, { cause }), + ), + // The consumer must outlive startSession's calling fiber. Its + // session scope is retained on the context and closed on teardown. + Effect.forkIn(ctx.scope), + ); + + ctx.notificationFiber = nf; + sessions.set(input.threadId, ctx); + sessionScopeTransferred = true; + + yield* offerRuntimeEvent({ + type: "session.started", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + payload: { resume: started.initializeResult }, + }); + yield* offerRuntimeEvent({ + type: "session.state.changed", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + payload: { state: "ready", reason: `${config.label} ACP session ready` }, + }); + yield* offerRuntimeEvent({ + type: "thread.started", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + payload: { providerThreadId: started.sessionId }, + }); + + return session; + }).pipe(Effect.scoped), + ); + + const sendTurn: StandardAcpAdapterShape["sendTurn"] = (input) => + Effect.gen(function* () { + const prepared = yield* withThreadLock( + input.threadId, + Effect.gen(function* () { + const ctx = yield* requireSession(input.threadId); + // A sendTurn while a prompt is in flight is a steer: the agent + // folds the new prompt into the ongoing work, so the active turn + // id is reused instead of opening a new turn. + const steeringTurnId = ctx.promptsInFlight > 0 ? ctx.activeTurnId : undefined; + const turnId = steeringTurnId ?? TurnId.make(yield* randomUUIDv4); + // Count this prompt immediately so a superseded in-flight prompt + // resolving from here on does not settle the turn; decremented on + // preparation failure here, and after the prompt below otherwise. + ctx.promptsInFlight += 1; + // Bind the turn id before cooperative yields so interruptTurn can + // settle this prompt even if stop arrives during preparation. + ctx.activeTurnId = turnId; + ctx.session = { + ...ctx.session, + status: steeringTurnId === undefined ? "connecting" : "running", + activeTurnId: turnId, + updatedAt: yield* nowIso, + }; + + return yield* Effect.gen(function* () { + const turnModelSelection = + input.modelSelection?.instanceId === boundInstanceId + ? input.modelSelection + : undefined; + const requestedTurnModelId = turnModelSelection?.model + ? config.normalizeModel(turnModelSelection.model) + : undefined; + const currentModelId = yield* config.applyModelSelection({ + runtime: ctx.acp, + currentModelId: ctx.currentModelId, + requestedModelId: requestedTurnModelId, + mapError: (cause) => + mapAcpToAdapterError(PROVIDER, input.threadId, "session/set_model", cause), + }); + + const text = input.input?.trim(); + const imagePromptParts = yield* Effect.forEach( + input.attachments ?? [], + (attachment) => + Effect.gen(function* () { + const attachmentPath = resolveAttachmentPath({ + attachmentsDir: serverConfig.attachmentsDir, + attachment, + }); + if (!attachmentPath) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session/prompt", + detail: `Invalid attachment id '${attachment.id}'.`, + }); + } + const bytes = yield* fileSystem.readFile(attachmentPath).pipe( + Effect.mapError( + (cause) => + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session/prompt", + detail: cause.message, + cause, + }), + ), + ); + return { + type: "image", + data: Buffer.from(bytes).toString("base64"), + mimeType: attachment.mimeType, + } satisfies EffectAcpSchema.ContentBlock; + }), + ); + const promptParts: Array = [ + ...(text ? [{ type: "text" as const, text }] : []), + ...imagePromptParts, + ]; + + if (promptParts.length === 0) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "sendTurn", + issue: "Turn requires non-empty text or attachments.", + }); + } + + ctx.currentModelId = currentModelId; + const displayModel = currentModelId + ? config.normalizeModel(currentModelId) + : undefined; + for ( + let yieldAttempt = 0; + yieldAttempt < SETTLEMENT_YIELD_ATTEMPTS; + yieldAttempt += 1 + ) { + yield* Effect.yieldNow; + } + if (ctx.interruptedTurnIds.has(turnId)) { + yield* settlePromptInFlight(input.threadId, turnId, ctx.acpSessionId, { + completedStopReason: "cancelled", + emitTurnCompletion: false, + settleAllPrompts: true, + }); + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session/prompt", + detail: `${config.label} prompt was interrupted during preparation.`, + }); + } + if (steeringTurnId === undefined) { + ctx.lastPlanFingerprint = undefined; + } + ctx.session = { + ...ctx.session, + status: "running", + activeTurnId: turnId, + updatedAt: yield* nowIso, + ...(displayModel ? { model: displayModel } : {}), + }; + + if (steeringTurnId === undefined) { + yield* offerRuntimeEvent({ + type: "turn.started", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + turnId, + payload: displayModel ? { model: displayModel } : {}, + }); + } + + return { + acp: ctx.acp, + acpSessionId: ctx.acpSessionId, + displayModel, + promptParts, + turnId, + }; + }).pipe( + Effect.tapCause(() => + Effect.gen(function* () { + const liveCtx = sessions.get(input.threadId); + if (!liveCtx) { + return; + } + yield* settlePromptInFlight(input.threadId, turnId, liveCtx.acpSessionId, { + errorMessage: `${config.label} prompt preparation failed.`, + emitTurnCompletion: false, + }); + }), + ), + ); + }), + ); + const promptSettled = yield* Ref.make(false); + const promptRpcSucceeded = yield* Ref.make(false); + const promptResultRef = yield* Ref.make( + undefined, + ); + + const promptFailureMessageRef = yield* Ref.make(undefined); + + return yield* Effect.gen(function* () { + const result = yield* prepared.acp + .prompt({ + prompt: prepared.promptParts, + }) + .pipe( + Effect.tap((promptResult) => + Effect.all([ + Ref.set(promptRpcSucceeded, true), + Ref.set(promptResultRef, promptResult), + ]), + ), + Effect.tapError((error) => + Ref.set( + promptFailureMessageRef, + mapAcpToAdapterError(PROVIDER, input.threadId, "session/prompt", error).message, + ).pipe(Effect.andThen(prepared.acp.drainEvents)), + ), + Effect.mapError((error) => + mapAcpToAdapterError(PROVIDER, input.threadId, "session/prompt", error), + ), + ); + + return yield* withThreadLock( + input.threadId, + Effect.gen(function* () { + const ctx = yield* requireSession(input.threadId); + if (ctx.acpSessionId !== prepared.acpSessionId) { + yield* settlePromptInFlight( + input.threadId, + prepared.turnId, + prepared.acpSessionId, + { + errorMessage: `${config.label} session changed before the turn completed.`, + settleAllPrompts: true, + }, + ); + yield* Ref.set(promptSettled, true); + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session/prompt", + detail: `${config.label} session changed before the turn completed.`, + }); + } + // Keep prompt settlement atomic with respect to Stop and steering. + // interruptTurn marks its target before waiting for this lock, so + // cancellation can still win while queued ACP events are drained. + for ( + let yieldAttempt = 0; + yieldAttempt < SETTLEMENT_YIELD_ATTEMPTS; + yieldAttempt += 1 + ) { + yield* Effect.yieldNow; + } + yield* prepared.acp.drainEvents; + if (ctx.interruptedTurnIds.has(prepared.turnId)) { + yield* Ref.set(promptSettled, true); + return { + threadId: input.threadId, + turnId: prepared.turnId, + resumeCursor: ctx.session.resumeCursor, + }; + } + + if ( + ctx.promptsInFlight <= 0 || + ctx.activeTurnId !== prepared.turnId || + ctx.session.activeTurnId !== prepared.turnId + ) { + yield* Ref.set(promptSettled, true); + return { + threadId: input.threadId, + turnId: prepared.turnId, + resumeCursor: ctx.session.resumeCursor, + }; + } + + appendPromptResultToTurn(ctx, prepared.turnId, prepared.promptParts, result); + ctx.session = { + ...ctx.session, + status: "running", + activeTurnId: prepared.turnId, + updatedAt: yield* nowIso, + ...(prepared.displayModel ? { model: prepared.displayModel } : {}), + }; + const remainingPrompts = Math.max(0, ctx.promptsInFlight - 1); + ctx.promptsInFlight = remainingPrompts; + + // Only the last remaining prompt settles the turn. A steer- + // superseded prompt resolving while another is in flight or + // pending must leave the merged turn running. + if ( + remainingPrompts === 0 && + ctx.activeTurnId === prepared.turnId && + ctx.session.activeTurnId === prepared.turnId + ) { + if (ctx.interruptedTurnIds.has(prepared.turnId)) { + yield* Ref.set(promptSettled, true); + return { + threadId: input.threadId, + turnId: prepared.turnId, + resumeCursor: ctx.session.resumeCursor, + }; + } + const completedAt = yield* nowIso; + const { activeTurnId: _completedTurnId, ...readySession } = ctx.session; + ctx.activeTurnId = undefined; + ctx.session = { + ...readySession, + status: "ready", + updatedAt: completedAt, + ...(prepared.displayModel ? { model: prepared.displayModel } : {}), + }; + const completedStopReason = completedStopReasonFromPromptResponse( + result, + config.promptStopReason, + ); + yield* offerRuntimeEvent({ + type: "turn.completed", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + turnId: prepared.turnId, + payload: { + state: result.stopReason === "cancelled" ? "cancelled" : "completed", + stopReason: completedStopReason, + }, + }); + ctx.interruptedTurnIds.delete(prepared.turnId); + yield* Ref.set(promptSettled, true); + } else if (remainingPrompts > 0) { + yield* Ref.set(promptSettled, true); + } + + return { + threadId: input.threadId, + turnId: prepared.turnId, + resumeCursor: ctx.session.resumeCursor, + }; + }), + ); + }).pipe( + Effect.ensuring( + Effect.gen(function* () { + if (yield* Ref.get(promptSettled)) { + return; + } + + if (yield* Ref.get(promptRpcSucceeded)) { + const promptResult = yield* Ref.get(promptResultRef); + if (promptResult === undefined) { + return; + } + yield* withThreadLock( + input.threadId, + Effect.gen(function* () { + const ctx = yield* requireSession(input.threadId); + if (ctx.acpSessionId !== prepared.acpSessionId) { + yield* settlePromptInFlight( + input.threadId, + prepared.turnId, + prepared.acpSessionId, + { + errorMessage: `${config.label} session changed before the turn completed.`, + settleAllPrompts: true, + }, + ); + return; + } + if (ctx.interruptedTurnIds.has(prepared.turnId)) { + return; + } + if ( + ctx.promptsInFlight <= 0 || + ctx.activeTurnId !== prepared.turnId || + ctx.session.activeTurnId !== prepared.turnId + ) { + return; + } + appendPromptResultToTurn( + ctx, + prepared.turnId, + prepared.promptParts, + promptResult, + ); + yield* settlePromptInFlight( + input.threadId, + prepared.turnId, + prepared.acpSessionId, + { + completedStopReason: completedStopReasonFromPromptResponse( + promptResult, + config.promptStopReason, + ), + }, + ); + }), + ); + return; + } + + const errorMessage = yield* Ref.get(promptFailureMessageRef); + yield* withThreadLock( + input.threadId, + settlePromptInFlight(input.threadId, prepared.turnId, prepared.acpSessionId, { + errorMessage: errorMessage ?? `${config.label} prompt request failed.`, + }), + ); + }).pipe(Effect.catch(() => Effect.void)), + ), + ); + }); + + const interruptTurn: StandardAcpAdapterShape["interruptTurn"] = (threadId, turnId) => + Effect.gen(function* () { + const observed = yield* Effect.sync(() => { + const ctx = sessions.get(threadId); + if (!ctx || ctx.stopped) { + return { + _tag: "Proceed" as const, + acpSessionId: undefined, + interruptedTurnId: turnId, + }; + } + const activeTurnId = ctx.activeTurnId ?? ctx.session.activeTurnId; + if (turnId !== undefined && activeTurnId !== undefined && activeTurnId !== turnId) { + return { _tag: "Ignore" as const }; + } + const interruptedTurnId = turnId ?? activeTurnId; + if (interruptedTurnId !== undefined) { + ctx.interruptedTurnIds.add(interruptedTurnId); + } + return { + _tag: "Proceed" as const, + acpSessionId: ctx.acpSessionId, + interruptedTurnId, + }; + }); + if (observed._tag === "Ignore") { + return; + } + + yield* withThreadLock( + threadId, + Effect.gen(function* () { + const ctx = yield* requireSession(threadId); + if (observed.acpSessionId !== undefined && ctx.acpSessionId !== observed.acpSessionId) { + return; + } + const activeTurnId = ctx.activeTurnId ?? ctx.session.activeTurnId; + if (turnId !== undefined && activeTurnId !== undefined && activeTurnId !== turnId) { + return; + } + if ( + observed.interruptedTurnId !== undefined && + activeTurnId !== undefined && + activeTurnId !== observed.interruptedTurnId + ) { + return; + } + const interruptedTurnId = + observed.interruptedTurnId ?? turnId ?? activeTurnId ?? ctx.session.activeTurnId; + yield* settlePendingApprovalsAsCancelled(ctx.pendingApprovals); + yield* settlePendingUserInputsAsCancelled(ctx.pendingUserInputs); + yield* Effect.ignore( + ctx.acp.cancel.pipe( + Effect.mapError((error) => + mapAcpToAdapterError(PROVIDER, threadId, "session/cancel", error), + ), + ), + ); + if (interruptedTurnId) { + ctx.interruptedTurnIds.add(interruptedTurnId); + yield* settlePromptInFlight(threadId, interruptedTurnId, ctx.acpSessionId, { + completedStopReason: "cancelled", + settleAllPrompts: true, + }); + } else if ( + ctx.promptsInFlight > 0 || + ctx.session.status === "running" || + ctx.session.status === "connecting" + ) { + const updatedAt = yield* nowIso; + ctx.promptsInFlight = 0; + ctx.activeTurnId = undefined; + const { activeTurnId: _activeTurnId, ...readySession } = ctx.session; + ctx.session = { + ...readySession, + status: "ready", + updatedAt, + }; + } + }), + ); + }); + + const respondToRequest: StandardAcpAdapterShape["respondToRequest"] = ( + threadId, + requestId, + decision, + ) => + Effect.gen(function* () { + const ctx = yield* requireSession(threadId); + const pending = ctx.pendingApprovals.get(requestId); + if (!pending) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session/request_permission", + detail: `Unknown pending approval request: ${requestId}`, + }); + } + yield* Deferred.succeed(pending.decision, decision); + }); + + const respondToUserInput: StandardAcpAdapterShape["respondToUserInput"] = ( + threadId, + requestId, + answers, + ) => + Effect.gen(function* () { + const ctx = yield* requireSession(threadId); + const pending = ctx.pendingUserInputs.get(requestId); + if (!pending) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: config.userInput?.methods[0] ?? "session/user_input", + detail: `Unknown pending user-input request: ${requestId}`, + }); + } + yield* Deferred.succeed(pending.resolution, { _tag: "answered", answers }); + }); + + const readThread: StandardAcpAdapterShape["readThread"] = (threadId) => + Effect.gen(function* () { + const ctx = yield* requireSession(threadId); + return { threadId, turns: ctx.turns }; + }); + + const rollbackThread: StandardAcpAdapterShape["rollbackThread"] = (threadId, numTurns) => + Effect.gen(function* () { + yield* requireSession(threadId); + if (!Number.isInteger(numTurns) || numTurns < 1) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "rollbackThread", + issue: "numTurns must be an integer >= 1.", + }); + } + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "thread/rollback", + detail: `${config.label} ACP sessions do not support provider-side rollback yet.`, + }); + }); + + const stopSession: StandardAcpAdapterShape["stopSession"] = (threadId) => + withThreadLock( + threadId, + Effect.gen(function* () { + const ctx = yield* requireSession(threadId); + yield* stopSessionInternal(ctx); + }), + ); + + const listSessions: StandardAcpAdapterShape["listSessions"] = () => + Effect.sync(() => Array.from(sessions.values(), (c) => ({ ...c.session }))); + + const hasSession: StandardAcpAdapterShape["hasSession"] = (threadId) => + Effect.sync(() => { + const c = sessions.get(threadId); + return c !== undefined && !c.stopped; + }); + + const stopAll: StandardAcpAdapterShape["stopAll"] = () => + Effect.forEach(Array.from(sessions.values()), stopSessionInternal, { discard: true }); + + yield* Effect.addFinalizer(() => + Effect.ignore(stopAll()).pipe( + Effect.tap(() => PubSub.shutdown(runtimeEventPubSub)), + Effect.tap(() => managedNativeEventLogger?.close() ?? Effect.void), + ), + ); + + const streamEvents = Stream.fromPubSub(runtimeEventPubSub); + + return { + provider: PROVIDER, + capabilities: { sessionModelSwitch: "in-session" }, + startSession, + sendTurn, + interruptTurn, + readThread, + rollbackThread, + respondToRequest, + respondToUserInput, + stopSession, + listSessions, + hasSession, + stopAll, + streamEvents, + } satisfies StandardAcpAdapterShape; + }); +} diff --git a/apps/server/src/provider/Layers/StandardAcpCliProvider.test.ts b/apps/server/src/provider/Layers/StandardAcpCliProvider.test.ts new file mode 100644 index 000000000000..3077f8e8242c --- /dev/null +++ b/apps/server/src/provider/Layers/StandardAcpCliProvider.test.ts @@ -0,0 +1,41 @@ +import { describe, expect, it } from "@effect/vitest"; + +import { __testing } from "./StandardAcpCliProvider.ts"; + +describe("standard ACP CLI provider errors", () => { + it("classifies platform not-found spawn errors as missing commands", () => { + expect( + __testing.hasMissingCommandCause({ + _tag: "AcpSpawnError", + cause: { _tag: "PlatformError", reason: { _tag: "NotFound" } }, + }), + ).toBe(true); + }); + + it("classifies raw ENOENT spawn errors as missing commands", () => { + expect( + __testing.hasMissingCommandCause({ + _tag: "AcpSpawnError", + cause: { code: "ENOENT" }, + }), + ).toBe(true); + }); + + it("keeps permission-denied spawn errors on the startup-failure path", () => { + expect( + __testing.hasMissingCommandCause({ + _tag: "AcpSpawnError", + cause: { _tag: "PlatformError", reason: { _tag: "PermissionDenied" } }, + }), + ).toBe(false); + }); + + it("does not treat unrelated not-found errors as missing commands", () => { + expect( + __testing.hasMissingCommandCause({ + _tag: "AcpRequestError", + cause: { _tag: "PlatformError", reason: { _tag: "NotFound" } }, + }), + ).toBe(false); + }); +}); diff --git a/apps/server/src/provider/Layers/StandardAcpCliProvider.ts b/apps/server/src/provider/Layers/StandardAcpCliProvider.ts new file mode 100644 index 000000000000..57fa9eb3cd03 --- /dev/null +++ b/apps/server/src/provider/Layers/StandardAcpCliProvider.ts @@ -0,0 +1,224 @@ +import type { + ModelCapabilities, + ProviderDriverKind, + ServerProviderModel, +} from "@t3tools/contracts"; +import { createModelCapabilities } from "@t3tools/shared/model"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Option from "effect/Option"; +import { ChildProcessSpawner } from "effect/unstable/process"; +import type * as EffectAcpSchema from "effect-acp/schema"; + +import { + firstAdvertisedAuthMethod, + makeStandardAcpCliRuntime, + normalizeStandardAcpModel, +} from "../acp/StandardAcpCliSupport.ts"; +import { + buildServerProvider, + providerModelsFromSettings, + type ServerProviderDraft, +} from "../providerSnapshot.ts"; + +const MODEL_DISCOVERY_TIMEOUT_MS = 15_000; +const EMPTY_CAPABILITIES: ModelCapabilities = createModelCapabilities({ optionDescriptors: [] }); + +export interface StandardAcpCliProviderConfig { + readonly provider: ProviderDriverKind; + readonly displayName: string; + readonly command: string; + readonly enabled: boolean; + readonly customModels: ReadonlyArray; + readonly environment: NodeJS.ProcessEnv; + readonly setupHint: string; + readonly missingCommandMessage: string; + readonly excludedAuthMethodIds?: ReadonlySet; +} + +function modelsFromSessionState( + state: EffectAcpSchema.SessionModelState | null | undefined, + provider: ProviderDriverKind, +): ReadonlyArray { + if (!state) return []; + const seen = new Set(); + return state.availableModels.flatMap((model) => { + const slug = normalizeStandardAcpModel(model.modelId, provider); + if (!slug || seen.has(slug)) return []; + seen.add(slug); + return [ + { + slug, + name: model.name.trim() || slug, + isCustom: false, + capabilities: EMPTY_CAPABILITIES, + } satisfies ServerProviderModel, + ]; + }); +} + +function hasMissingCommandCause(cause: unknown, seen = new WeakSet()): boolean { + if (!cause || typeof cause !== "object" || seen.has(cause)) return false; + seen.add(cause); + if (Array.isArray(cause)) { + return cause.some((entry) => hasMissingCommandCause(entry, seen)); + } + const record = cause as Record; + if (record._tag === "AcpSpawnError") { + return hasNotFoundCause(record.cause); + } + return Object.values(record).some((entry) => hasMissingCommandCause(entry, seen)); +} + +function hasNotFoundCause(cause: unknown, seen = new WeakSet()): boolean { + if (!cause || typeof cause !== "object" || seen.has(cause)) return false; + seen.add(cause); + if (Array.isArray(cause)) { + return cause.some((entry) => hasNotFoundCause(entry, seen)); + } + const record = cause as Record; + if (record.code === "ENOENT") return true; + if (record._tag === "PlatformError") { + const reason = record.reason; + if ( + reason && + typeof reason === "object" && + (reason as Record)._tag === "NotFound" + ) { + return true; + } + } + return Object.values(record).some((entry) => hasNotFoundCause(entry, seen)); +} + +export function buildInitialStandardAcpCliProviderSnapshot( + config: StandardAcpCliProviderConfig, +): Effect.Effect { + return Effect.map(DateTime.now, (now) => + buildServerProvider({ + presentation: { + displayName: config.displayName, + badgeLabel: "Early Access", + showInteractionModeToggle: true, + requiresNewThreadForModelChange: false, + }, + enabled: config.enabled, + checkedAt: DateTime.formatIso(now), + models: providerModelsFromSettings([], config.customModels, EMPTY_CAPABILITIES), + probe: config.enabled + ? { + installed: true, + version: null, + status: "warning", + auth: { status: "unknown" }, + message: `Checking ${config.displayName} ACP availability...`, + } + : { + installed: false, + version: null, + status: "warning", + auth: { status: "unknown" }, + message: `${config.displayName} is disabled in T3 Code settings.`, + }, + }), + ); +} + +export const checkStandardAcpCliProviderStatus = Effect.fn("checkStandardAcpCliProviderStatus")( + function* ( + config: StandardAcpCliProviderConfig, + ): Effect.fn.Return< + ServerProviderDraft, + never, + ChildProcessSpawner.ChildProcessSpawner | Crypto.Crypto + > { + const checkedAt = DateTime.formatIso(yield* DateTime.now); + const fallbackModels = providerModelsFromSettings([], config.customModels, EMPTY_CAPABILITIES); + const presentation = { + displayName: config.displayName, + badgeLabel: "Early Access", + showInteractionModeToggle: true, + requiresNewThreadForModelChange: false, + } as const; + + if (!config.enabled) { + return buildServerProvider({ + presentation, + enabled: false, + checkedAt, + models: fallbackModels, + probe: { + installed: false, + version: null, + status: "warning", + auth: { status: "unknown" }, + message: `${config.displayName} is disabled in T3 Code settings.`, + }, + }); + } + + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const discovery = yield* makeStandardAcpCliRuntime({ + childProcessSpawner, + command: config.command, + cwd: process.cwd(), + environment: config.environment, + clientInfo: { name: "t3-code-provider-probe", version: "0.0.0" }, + ...(config.excludedAuthMethodIds + ? { + resolveAuthMethodId: (initializeResult) => + firstAdvertisedAuthMethod(initializeResult, config.excludedAuthMethodIds), + } + : {}), + }).pipe( + Effect.flatMap((runtime) => runtime.start()), + Effect.map((started) => + modelsFromSessionState(started.sessionSetupResult.models, config.provider), + ), + Effect.scoped, + Effect.timeoutOption(MODEL_DISCOVERY_TIMEOUT_MS), + Effect.exit, + ); + + if (Exit.isSuccess(discovery) && Option.isSome(discovery.value)) { + const discoveredModels = discovery.value.value; + return buildServerProvider({ + presentation, + enabled: true, + checkedAt, + models: + discoveredModels.length > 0 + ? providerModelsFromSettings(discoveredModels, config.customModels, EMPTY_CAPABILITIES) + : fallbackModels, + probe: { + installed: true, + version: null, + status: "ready", + auth: { status: "unknown" }, + }, + }); + } + + const missing = Exit.isFailure(discovery) && hasMissingCommandCause(discovery.cause); + return buildServerProvider({ + presentation, + enabled: true, + checkedAt, + models: fallbackModels, + probe: { + installed: !missing, + version: null, + status: "error", + auth: { status: "unknown" }, + message: missing + ? config.missingCommandMessage + : `${config.displayName} ACP startup failed or timed out. ${config.setupHint}`, + }, + }); + }, +); + +/** Exposed for focused error-classification tests. */ +export const __testing = { hasMissingCommandCause }; diff --git a/apps/server/src/provider/Layers/StandardAcpProviders.test.ts b/apps/server/src/provider/Layers/StandardAcpProviders.test.ts new file mode 100644 index 000000000000..f2c38b94e48d --- /dev/null +++ b/apps/server/src/provider/Layers/StandardAcpProviders.test.ts @@ -0,0 +1,107 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeFSP from "node:fs/promises"; +import * as NodeOS from "node:os"; +import * as NodePath from "node:path"; +import * as NodeURL from "node:url"; + +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; + +import { + HermesSettings, + PiSettings, + ProviderDriverKind, + ThreadId, + type ProviderRuntimeEvent, +} from "@t3tools/contracts"; + +import { ServerConfig } from "../../config.ts"; +import { makeHermesAdapter } from "./HermesAdapter.ts"; +import { makePiAdapter } from "./PiAdapter.ts"; + +const decodeHermesSettings = Schema.decodeUnknownEffect(HermesSettings); +const decodePiSettings = Schema.decodeUnknownEffect(PiSettings); + +const __dirname = NodePath.dirname(NodeURL.fileURLToPath(import.meta.url)); +const mockAgentPath = NodePath.join(__dirname, "../../../scripts/acp-mock-agent.ts"); + +async function makeMockAcpWrapper() { + const dir = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "standard-acp-mock-")); + const wrapperPath = NodePath.join(dir, "fake-acp.sh"); + await NodeFSP.writeFile( + wrapperPath, + `#!/bin/sh\nexec ${JSON.stringify(process.execPath)} ${JSON.stringify(mockAgentPath)} "$@"\n`, + "utf8", + ); + await NodeFSP.chmod(wrapperPath, 0o755); + return wrapperPath; +} + +const testLayer = ServerConfig.layerTest(process.cwd(), { + prefix: "t3code-standard-acp-adapter-test-", +}).pipe(Layer.provideMerge(NodeServices.layer)); + +it.layer(testLayer)("standard ACP provider adapters", (it) => { + for (const provider of ["hermes", "pi"] as const) { + it.effect(`${provider} starts, streams a prompt, and stops`, () => + Effect.gen(function* () { + const binaryPath = yield* Effect.promise(makeMockAcpWrapper); + const adapter = yield* provider === "hermes" + ? decodeHermesSettings({ binaryPath }).pipe(Effect.flatMap(makeHermesAdapter)) + : decodePiSettings({ binaryPath }).pipe(Effect.flatMap(makePiAdapter)); + const threadId = ThreadId.make(`${provider}-mock-thread`); + const providerKind = + provider === "hermes" ? ProviderDriverKind.make("hermes") : ProviderDriverKind.make("pi"); + const events: ProviderRuntimeEvent[] = []; + const turnCompleted = yield* Deferred.make(); + const eventFiber = yield* adapter.streamEvents.pipe( + Stream.runForEach((event) => + Effect.sync(() => events.push(event)).pipe( + Effect.andThen( + event.type === "turn.completed" + ? Deferred.succeed(turnCompleted, undefined).pipe(Effect.asVoid) + : Effect.void, + ), + ), + ), + Effect.forkChild, + ); + + yield* Effect.gen(function* () { + const session = yield* adapter.startSession({ + threadId, + provider: providerKind, + cwd: process.cwd(), + runtimeMode: "full-access", + }); + assert.equal(String(session.provider), provider); + assert.deepStrictEqual(session.resumeCursor, { + schemaVersion: 1, + sessionId: "mock-session-1", + }); + + yield* adapter.sendTurn({ threadId, input: `hello ${provider}`, attachments: [] }); + yield* Deferred.await(turnCompleted); + + assert.includeMembers( + events.map((event) => event.type), + [ + "session.started", + "thread.started", + "turn.started", + "content.delta", + "turn.completed", + ], + ); + yield* adapter.stopSession(threadId); + }).pipe(Effect.ensuring(Fiber.interrupt(eventFiber))); + }), + ); + } +}); diff --git a/apps/server/src/provider/acp/AcpSessionRuntime.ts b/apps/server/src/provider/acp/AcpSessionRuntime.ts index 82cad3921a49..ff0f1c16bab1 100644 --- a/apps/server/src/provider/acp/AcpSessionRuntime.ts +++ b/apps/server/src/provider/acp/AcpSessionRuntime.ts @@ -68,7 +68,16 @@ export interface AcpSessionRuntimeOptions { readonly name: string; readonly version: string; }; - readonly authMethodId: string; + /** + * Authentication method to select after initialization. A resolver is + * useful for agents whose configured provider determines the advertised + * method at runtime. Returning `undefined` skips the optional ACP + * authenticate request. + */ + readonly authMethodId: + | string + | undefined + | ((initializeResult: EffectAcpSchema.InitializeResponse) => string | undefined); readonly mcpServers?: ReadonlyArray; readonly requestLogger?: (event: AcpSessionRequestLogEvent) => Effect.Effect; readonly protocolLogging?: { @@ -547,15 +556,21 @@ export const make = ( acp.agent.initialize(initializePayload), ); - const authenticatePayload = { - methodId: options.authMethodId, - } satisfies EffectAcpSchema.AuthenticateRequest; + const authMethodId = + typeof options.authMethodId === "function" + ? options.authMethodId(initializeResult) + : options.authMethodId; + if (authMethodId !== undefined) { + const authenticatePayload = { + methodId: authMethodId, + } satisfies EffectAcpSchema.AuthenticateRequest; - yield* runLoggedRequest( - "authenticate", - authenticatePayload, - acp.agent.authenticate(authenticatePayload), - ); + yield* runLoggedRequest( + "authenticate", + authenticatePayload, + acp.agent.authenticate(authenticatePayload), + ); + } let sessionId: string; let sessionSetupResult: diff --git a/apps/server/src/provider/acp/StandardAcpCliSupport.test.ts b/apps/server/src/provider/acp/StandardAcpCliSupport.test.ts new file mode 100644 index 000000000000..a10125009a37 --- /dev/null +++ b/apps/server/src/provider/acp/StandardAcpCliSupport.test.ts @@ -0,0 +1,61 @@ +import { describe, expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; + +import { ProviderDriverKind } from "@t3tools/contracts"; + +import { + applyStandardAcpModelSelection, + firstAdvertisedAuthMethod, + normalizeStandardAcpModel, +} from "./StandardAcpCliSupport.ts"; + +describe("standard ACP CLI support", () => { + it("selects advertised auth while excluding interactive setup methods", () => { + const selected = firstAdvertisedAuthMethod( + { + protocolVersion: 1, + agentCapabilities: {}, + authMethods: [ + { id: "hermes-setup", name: "Setup" }, + { id: "openrouter", name: "OpenRouter" }, + ], + }, + new Set(["hermes-setup"]), + ); + expect(selected).toBe("openrouter"); + }); + + it("preserves provider-qualified model ids owned by the ACP agent", () => { + expect( + normalizeStandardAcpModel(" anthropic:claude-sonnet ", ProviderDriverKind.make("hermes")), + ).toBe("anthropic:claude-sonnet"); + }); + + it.effect("switches models only when the requested model changes", () => + Effect.gen(function* () { + const calls: string[] = []; + const runtime = { + setSessionModel: (modelId: string) => + Effect.sync(() => { + calls.push(modelId); + return {}; + }), + }; + const selected = yield* applyStandardAcpModelSelection({ + runtime, + currentModelId: "old", + requestedModelId: "new", + mapError: String, + }); + const unchanged = yield* applyStandardAcpModelSelection({ + runtime, + currentModelId: "new", + requestedModelId: "new", + mapError: String, + }); + expect(selected).toBe("new"); + expect(unchanged).toBe("new"); + expect(calls).toEqual(["new"]); + }), + ); +}); diff --git a/apps/server/src/provider/acp/StandardAcpCliSupport.ts b/apps/server/src/provider/acp/StandardAcpCliSupport.ts new file mode 100644 index 000000000000..995f436e56d0 --- /dev/null +++ b/apps/server/src/provider/acp/StandardAcpCliSupport.ts @@ -0,0 +1,92 @@ +import type { ProviderDriverKind } from "@t3tools/contracts"; +import { normalizeModelSlug } from "@t3tools/shared/model"; +import * as Crypto from "effect/Crypto"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Scope from "effect/Scope"; +import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner"; +import * as EffectAcpErrors from "effect-acp/errors"; +import type * as EffectAcpSchema from "effect-acp/schema"; + +import * as AcpSessionRuntime from "./AcpSessionRuntime.ts"; + +export interface StandardAcpCliRuntimeInput extends Omit< + AcpSessionRuntime.AcpSessionRuntimeOptions, + "authMethodId" | "clientCapabilities" | "spawn" +> { + readonly childProcessSpawner: ChildProcessSpawner.ChildProcessSpawner["Service"]; + readonly command: string; + readonly args?: ReadonlyArray; + readonly environment?: NodeJS.ProcessEnv; + readonly resolveAuthMethodId?: ( + initializeResult: EffectAcpSchema.InitializeResponse, + ) => string | undefined; +} + +export const makeStandardAcpCliRuntime = ( + input: StandardAcpCliRuntimeInput, +): Effect.Effect< + AcpSessionRuntime.AcpSessionRuntime["Service"], + EffectAcpErrors.AcpError, + Crypto.Crypto | Scope.Scope +> => + Effect.gen(function* () { + const acpContext = yield* Layer.build( + AcpSessionRuntime.layer({ + ...input, + spawn: { + command: input.command, + args: input.args ?? [], + cwd: input.cwd, + ...(input.environment ? { env: input.environment } : {}), + }, + authMethodId: input.resolveAuthMethodId, + }).pipe( + Layer.provide( + Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, input.childProcessSpawner), + ), + ), + ); + return yield* Effect.service(AcpSessionRuntime.AcpSessionRuntime).pipe( + Effect.provide(acpContext), + ); + }); + +export function firstAdvertisedAuthMethod( + initializeResult: EffectAcpSchema.InitializeResponse, + excludedIds: ReadonlySet = new Set(), +): string | undefined { + return initializeResult.authMethods?.find((method) => !excludedIds.has(method.id))?.id; +} + +export function normalizeStandardAcpModel( + model: string | null | undefined, + provider: ProviderDriverKind, +): string { + const trimmed = model?.trim(); + if (!trimmed) return ""; + return normalizeModelSlug(trimmed, provider) ?? trimmed; +} + +export function currentStandardAcpModelFromSetup( + setup: + | EffectAcpSchema.LoadSessionResponse + | EffectAcpSchema.NewSessionResponse + | EffectAcpSchema.ResumeSessionResponse, +): string | undefined { + return setup.models?.currentModelId?.trim() || undefined; +} + +export function applyStandardAcpModelSelection(input: { + readonly runtime: Pick; + readonly currentModelId: string | undefined; + readonly requestedModelId: string | undefined; + readonly mapError: (cause: EffectAcpErrors.AcpError) => E; +}): Effect.Effect { + if (input.requestedModelId === undefined || input.requestedModelId === input.currentModelId) { + return Effect.succeed(input.currentModelId); + } + return input.runtime + .setSessionModel(input.requestedModelId) + .pipe(Effect.mapError(input.mapError), Effect.as(input.requestedModelId)); +} diff --git a/apps/server/src/provider/builtInDrivers.ts b/apps/server/src/provider/builtInDrivers.ts index 2aa18f4f1148..03eba9ca67a6 100644 --- a/apps/server/src/provider/builtInDrivers.ts +++ b/apps/server/src/provider/builtInDrivers.ts @@ -28,8 +28,10 @@ import { CursorDriver, type CursorDriverEnv } from "./Drivers/CursorDriver.ts"; import { DroidDriver, type DroidDriverEnv } from "./Drivers/DroidDriver.ts"; import { GeminiCliDriver, type GeminiCliDriverEnv } from "./Drivers/GeminiCliDriver.ts"; import { GrokDriver, type GrokDriverEnv } from "./Drivers/GrokDriver.ts"; +import { HermesDriver, type HermesDriverEnv } from "./Drivers/HermesDriver.ts"; import { KiloDriver, type KiloDriverEnv } from "./Drivers/KiloDriver.ts"; import { OpenCodeDriver, type OpenCodeDriverEnv } from "./Drivers/OpenCodeDriver.ts"; +import { PiDriver, type PiDriverEnv } from "./Drivers/PiDriver.ts"; import type { AnyProviderDriver } from "./ProviderDriver.ts"; /** @@ -43,11 +45,13 @@ export type BuiltInDriversEnv = | CursorDriverEnv | DroidDriverEnv | GrokDriverEnv + | HermesDriverEnv | OpenCodeDriverEnv | AmpDriverEnv | CopilotDriverEnv | GeminiCliDriverEnv - | KiloDriverEnv; + | KiloDriverEnv + | PiDriverEnv; /** * Ordered list of built-in drivers. Order matters only for tie-breaking in @@ -67,5 +71,7 @@ export const BUILT_IN_DRIVERS: ReadonlyArray; +} as const satisfies Partial>; const MIRRORED_CLIENT_KEYS = new Set([ "confirmThreadDelete", "diffWordWrap", @@ -325,7 +325,10 @@ function toUnifiedPatch(patch: Partial): Partial { }; } const providerModelEntries = Object.entries(APP_SETTINGS_PROVIDER_CUSTOM_MODEL_KEYS) as Array< - [ProviderKind, (typeof APP_SETTINGS_PROVIDER_CUSTOM_MODEL_KEYS)[ProviderKind]] + [ + keyof typeof APP_SETTINGS_PROVIDER_CUSTOM_MODEL_KEYS, + (typeof APP_SETTINGS_PROVIDER_CUSTOM_MODEL_KEYS)[keyof typeof APP_SETTINGS_PROVIDER_CUSTOM_MODEL_KEYS], + ] >; for (const [provider, settingsKey] of providerModelEntries) { const models = patch[settingsKey]; diff --git a/apps/web/src/components/GhosttyTerminalSplitView.tsx b/apps/web/src/components/GhosttyTerminalSplitView.tsx index 3c21ed30e16b..e5839ed6c7b9 100644 --- a/apps/web/src/components/GhosttyTerminalSplitView.tsx +++ b/apps/web/src/components/GhosttyTerminalSplitView.tsx @@ -43,6 +43,7 @@ import { resolveAccentColorRgba, } from "../accentColor"; import { resolveTerminalFontFamily } from "../lib/terminalFont"; +import { Tooltip, TooltipPopup, TooltipTrigger } from "./ui/tooltip"; // ─── Ghostty WASM Initialization ──────────────────────────────────────────── // ghostty-web requires a one-time async init to load the WASM module. @@ -630,44 +631,79 @@ export default function GhosttyTerminalSplitView({
{/* Split button */} - + + + } + > + + + + {canSplit ? `Split (max ${MAX_PANES})` : `Max ${MAX_PANES} panes`} + + {/* Add new pane */} - + + + } + > + + + {canSplit ? "New pane" : `Max ${MAX_PANES} panes`} + {/* Collapse */} - + + setIsCollapsed(true)} + aria-label="Collapse terminal split view" + /> + } + > + + + Collapse +
diff --git a/apps/web/src/components/Icons.tsx b/apps/web/src/components/Icons.tsx index 2857ee13f59f..09f2737ac134 100644 --- a/apps/web/src/components/Icons.tsx +++ b/apps/web/src/components/Icons.tsx @@ -904,6 +904,16 @@ export const PiAgentIcon: Icon = ({ className, ...props }) => ( ); +// Hermes — original winged-H glyph, kept monochrome to match provider chrome. +export const HermesIcon: Icon = ({ className, ...props }) => ( + + + +); + // Amp / Sourcegraph — orange ampersand-inspired wedge mark. export const AmpIcon: Icon = ({ className, monochrome, ...props }) => ( = { @@ -20,10 +24,14 @@ const PROVIDER_ICON_BY_PROVIDER: Record = { copilot: GitHubIcon, claudeAgent: ClaudeAI, cursor: CursorIcon, + droid: DroidIcon, + grok: GrokIcon, opencode: OpenCodeIcon, geminiCli: Gemini, amp: AmpIcon, kilo: KiloIcon, + hermes: HermesIcon, + pi: PiAgentIcon, }; export type ProviderLogoProps = ComponentProps & { diff --git a/apps/web/src/components/chat/providerIconUtils.ts b/apps/web/src/components/chat/providerIconUtils.ts index c524a564eefc..0b628929370a 100644 --- a/apps/web/src/components/chat/providerIconUtils.ts +++ b/apps/web/src/components/chat/providerIconUtils.ts @@ -1,5 +1,15 @@ import { ProviderDriverKind } from "@t3tools/contracts"; -import { ClaudeAI, CursorIcon, DroidIcon, GrokIcon, Icon, OpenAI, OpenCodeIcon } from "../Icons"; +import { + ClaudeAI, + CursorIcon, + DroidIcon, + GrokIcon, + HermesIcon, + Icon, + OpenAI, + OpenCodeIcon, + PiAgentIcon, +} from "../Icons"; import { PROVIDER_OPTIONS } from "../../session-logic"; export const PROVIDER_ICON_BY_PROVIDER: Partial> = { @@ -9,6 +19,8 @@ export const PROVIDER_ICON_BY_PROVIDER: Partial [ProviderDriverKind.make("cursor")]: CursorIcon, [ProviderDriverKind.make("droid")]: DroidIcon, [ProviderDriverKind.make("grok")]: GrokIcon, + [ProviderDriverKind.make("hermes")]: HermesIcon, + [ProviderDriverKind.make("pi")]: PiAgentIcon, }; function isAvailableProviderOption(option: (typeof PROVIDER_OPTIONS)[number]): option is { diff --git a/apps/web/src/components/settings/AddProviderInstanceDialog.tsx b/apps/web/src/components/settings/AddProviderInstanceDialog.tsx index 158908b5e942..9428a01301de 100644 --- a/apps/web/src/components/settings/AddProviderInstanceDialog.tsx +++ b/apps/web/src/components/settings/AddProviderInstanceDialog.tsx @@ -14,7 +14,7 @@ import { useEnvironmentSettings, useUpdateEnvironmentSettings } from "../../hook import { cn } from "../../lib/utils"; import { normalizeProviderAccentColor } from "../../providerInstances"; import { Button } from "../ui/button"; -import { ACPRegistryIcon, Gemini, GithubCopilotIcon, PiAgentIcon, type Icon } from "../Icons"; +import { ACPRegistryIcon, Gemini, GithubCopilotIcon, type Icon } from "../Icons"; import { Dialog, DialogDescription, @@ -93,11 +93,6 @@ const COMING_SOON_DRIVER_OPTIONS: readonly ComingSoonDriverOption[] = [ label: "ACP Registry", icon: ACPRegistryIcon, }, - { - value: ProviderDriverKind.make("piAgent"), - label: "Pi Agent", - icon: PiAgentIcon, - }, ]; /** diff --git a/apps/web/src/components/settings/providerDriverMeta.ts b/apps/web/src/components/settings/providerDriverMeta.ts index 80076e2c37ee..f809a1b13ebe 100644 --- a/apps/web/src/components/settings/providerDriverMeta.ts +++ b/apps/web/src/components/settings/providerDriverMeta.ts @@ -7,8 +7,10 @@ import { DroidSettings, GeminiCliSettings, GrokSettings, + HermesSettings, KiloSettings, OpenCodeSettings, + PiSettings, ProviderDriverKind, } from "@t3tools/contracts"; import type * as Schema from "effect/Schema"; @@ -20,10 +22,12 @@ import { DroidIcon, GeminiCliIcon, GrokIcon, + HermesIcon, type Icon, KiloIcon, OpenAI, OpenCodeIcon, + PiAgentIcon, } from "../Icons"; type ProviderSettingsSchema = { @@ -119,6 +123,20 @@ export const PROVIDER_CLIENT_DEFINITIONS: readonly ProviderClientDefinition[] = settingsSchema: KiloSettings, badgeLabel: "Fork Extension", }, + { + value: ProviderDriverKind.make("hermes"), + label: "Hermes", + icon: HermesIcon, + settingsSchema: HermesSettings, + badgeLabel: "Fork Extension", + }, + { + value: ProviderDriverKind.make("pi"), + label: "Pi", + icon: PiAgentIcon, + settingsSchema: PiSettings, + badgeLabel: "Fork Extension", + }, ]; export const PROVIDER_CLIENT_DEFINITION_BY_VALUE: Partial< diff --git a/apps/web/src/providerKind.ts b/apps/web/src/providerKind.ts index fbee7ea7042f..9fcb80772541 100644 --- a/apps/web/src/providerKind.ts +++ b/apps/web/src/providerKind.ts @@ -3,7 +3,7 @@ * * Upstream's #2277 refactor removed the `ProviderKind` union from * `@t3tools/contracts` in favor of `ProviderDriverKind` (an open branded - * slug). The fork still relies on a closed union of the eight provider names + * slug). The fork still relies on a closed union of built-in provider names * it ships adapters for — used as keys in `Record` shapes * and as the discriminator in exhaustive `switch` statements throughout the * web app. @@ -18,10 +18,14 @@ export const PROVIDER_KINDS = [ "copilot", "claudeAgent", "cursor", + "droid", + "grok", "opencode", "geminiCli", "amp", "kilo", + "hermes", + "pi", ] as const; export type ProviderKind = (typeof PROVIDER_KINDS)[number]; diff --git a/apps/web/src/session-logic.test.ts b/apps/web/src/session-logic.test.ts index f1b6ecd5c957..fdae915626e1 100644 --- a/apps/web/src/session-logic.test.ts +++ b/apps/web/src/session-logic.test.ts @@ -1773,6 +1773,8 @@ describe("PROVIDER_OPTIONS", () => { { value: "geminiCli", label: "Gemini CLI", available: true }, { value: "amp", label: "AMPcode", available: true }, { value: "kilo", label: "Kilo", available: true }, + { value: "hermes", label: "Hermes", available: true }, + { value: "pi", label: "Pi", available: true }, ]); expect(copilot).toEqual({ value: "copilot", diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index f58d08c6548b..1d8519b27f3f 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -62,6 +62,8 @@ export const PROVIDER_OPTIONS: Array<{ { value: ProviderDriverKind.make("geminiCli"), label: "Gemini CLI", available: true }, { value: ProviderDriverKind.make("amp"), label: "AMPcode", available: true }, { value: ProviderDriverKind.make("kilo"), label: "Kilo", available: true }, + { value: ProviderDriverKind.make("hermes"), label: "Hermes", available: true }, + { value: ProviderDriverKind.make("pi"), label: "Pi", available: true }, ]; export type WorkLogToolLifecycleStatus = diff --git a/packages/contracts/src/model.ts b/packages/contracts/src/model.ts index ad336dd25b64..0e3f93108b58 100644 --- a/packages/contracts/src/model.ts +++ b/packages/contracts/src/model.ts @@ -132,7 +132,9 @@ const CLAUDE_DRIVER_KIND = ProviderDriverKind.make("claudeAgent"); const CURSOR_DRIVER_KIND = ProviderDriverKind.make("cursor"); const DROID_DRIVER_KIND = ProviderDriverKind.make("droid"); const GROK_DRIVER_KIND = ProviderDriverKind.make("grok"); +const HERMES_DRIVER_KIND = ProviderDriverKind.make("hermes"); const OPENCODE_DRIVER_KIND = ProviderDriverKind.make("opencode"); +const PI_DRIVER_KIND = ProviderDriverKind.make("pi"); export const DEFAULT_MODEL = "gpt-5.6-sol"; @@ -229,5 +231,7 @@ export const PROVIDER_DISPLAY_NAMES: Partial> [CURSOR_DRIVER_KIND]: "Cursor", [DROID_DRIVER_KIND]: "Droid", [GROK_DRIVER_KIND]: "Grok", + [HERMES_DRIVER_KIND]: "Hermes Agent", [OPENCODE_DRIVER_KIND]: "OpenCode", + [PI_DRIVER_KIND]: "Pi", }; diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index ce2aa7da687f..4e2347323aae 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -641,6 +641,49 @@ export const KiloSettings = makeProviderSettingsSchema( ); export type KiloSettings = typeof KiloSettings.Type; +function makeAcpCliProviderSettingsSchema(input: { + readonly defaultBinary: string; + readonly description: string; +}) { + return makeProviderSettingsSchema( + { + enabled: Schema.Boolean.pipe( + Schema.withDecodingDefault(Effect.succeed(true)), + Schema.annotateKey({ providerSettingsForm: { hidden: true } }), + ), + binaryPath: makeBinaryPathSetting(input.defaultBinary).pipe( + Schema.annotateKey({ + title: "ACP binary path", + description: input.description, + providerSettingsForm: { + placeholder: input.defaultBinary, + clearWhenEmpty: "omit", + }, + }), + ), + customModels: Schema.Array(Schema.String).pipe( + Schema.withDecodingDefault(Effect.succeed([])), + Schema.annotateKey({ providerSettingsForm: { hidden: true } }), + ), + }, + { + order: ["binaryPath"], + }, + ); +} + +export const HermesSettings = makeAcpCliProviderSettingsSchema({ + defaultBinary: "hermes-acp", + description: "Path to the Hermes Agent ACP executable.", +}); +export type HermesSettings = typeof HermesSettings.Type; + +export const PiSettings = makeAcpCliProviderSettingsSchema({ + defaultBinary: "pi-acp", + description: "Path to the pi-acp executable. The Pi coding agent must also be installed.", +}); +export type PiSettings = typeof PiSettings.Type; + export const GenericProviderSettings = Schema.Struct({ enabled: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))), customModels: Schema.Array(Schema.String).pipe(Schema.withDecodingDefault(Effect.succeed([]))), @@ -783,7 +826,9 @@ export const ServerSettings = Schema.Struct({ opencode: OpenCodeSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), geminiCli: GeminiCliSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), amp: AmpSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), + hermes: HermesSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), kilo: KiloSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), + pi: PiSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), }).pipe(Schema.withDecodingDefault(Effect.succeed({}))), // New driver-agnostic instance map. Keyed by `ProviderInstanceId`; values // are `ProviderInstanceConfig` envelopes. The driver-specific config blob @@ -944,7 +989,9 @@ export const ServerSettingsPatch = Schema.Struct({ opencode: Schema.optionalKey(OpenCodeSettingsPatch), geminiCli: Schema.optionalKey(GenericProviderSettingsPatch), amp: Schema.optionalKey(GenericProviderSettingsPatch), + hermes: Schema.optionalKey(GenericProviderSettingsPatch), kilo: Schema.optionalKey(GenericProviderSettingsPatch), + pi: Schema.optionalKey(GenericProviderSettingsPatch), }), ), // Whole-map replacement for the new instance config. Patching individual