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 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 { 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, makeGrokAcpRuntime, resolveGrokAcpBaseModelId,} from "../acp/GrokAcpSupport.ts";import { extractXAiAskUserQuestions, makeXAiAskUserQuestionCancelledResponse, makeXAiAskUserQuestionResponse, XAiAskUserQuestionRequest,} from "../acp/XAiAcpExtension.ts";import { type GrokAdapterShape } from "../Services/GrokAdapter.ts";import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.UnknownFromJsonString); 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<ProviderApprovalDecision>;} type PendingUserInputResolution = | { readonly _tag: "answered"; readonly answers: ProviderUserInputAnswers } | { readonly _tag: "cancelled" }; interface PendingUserInput { readonly resolution: Deferred.Deferred<PendingUserInputResolution>;} interface GrokSessionContext { readonly threadId: ThreadId; readonly acpSessionId: string; session: ProviderSession; readonly scope: Scope.Closeable; readonly acp: AcpSessionRuntime.AcpSessionRuntime["Service"]; notificationFiber: Fiber.Fiber<void, never> | undefined; readonly pendingApprovals: Map<ApprovalRequestId, PendingApproval>; readonly pendingUserInputs: Map<ApprovalRequestId, PendingUserInput>; turns: Array<{ id: TurnId; items: Array<unknown> }>; lastPlanFingerprint: string | undefined; activeTurnId: TurnId | undefined; /** 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<ApprovalRequestId, PendingApproval>,): Effect.Effect<void> { return Effect.forEach( Array.from(pendingApprovals.values()), (pending) => Deferred.succeed(pending.decision, "cancel").pipe(Effect.ignore), { discard: true }, );} function settlePendingUserInputsAsCancelled( pendingUserInputs: ReadonlyMap<ApprovalRequestId, PendingUserInput>,): Effect.Effect<void> { return Effect.forEach( Array.from(pendingUserInputs.values()), (pending) => Deferred.succeed(pending.resolution, { _tag: "cancelled" }).pipe(Effect.ignore), { discard: true }, );} function isRecord(value: unknown): value is Record<string, unknown> { return typeof value === "object" && value !== null && !Array.isArray(value);} 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<ProviderApprovalDecision, "cancel">,): 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") );} 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<ThreadId, GrokSessionContext>(); const threadLocksRef = yield* SynchronizedRef.make(new Map<string, Semaphore.Semaphore>()); const runtimeEventPubSub = yield* PubSub.unbounded<ProviderRuntimeEvent>(); 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 = <A, E, R>(effect: Effect.Effect<A, E, R>) => 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<Semaphore.Semaphore> = 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 = <A, E, R>(threadId: string, effect: Effect.Effect<A, E, R>) => Effect.flatMap(getThreadSemaphore(threadId), (semaphore) => semaphore.withPermit(effect)); 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, 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 = `${ctx.activeTurnId ?? "no-turn"}:${encodeJsonStringForDiagnostics(payload) ?? "[unserializable payload]"}`; if (ctx.lastPlanFingerprint === fingerprint) { return; } ctx.lastPlanFingerprint = fingerprint; yield* offerRuntimeEvent( makeAcpPlanUpdatedEvent({ stamp: yield* makeEventStamp(), provider: PROVIDER, threadId: ctx.threadId, turnId: ctx.activeTurnId, payload, source: "acp.jsonrpc", method, rawPayload, }), ); }); const requireSession = ( threadId: ThreadId, ): Effect.Effect<GrokSessionContext, ProviderAdapterSessionNotFoundError> => { 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<ApprovalRequestId, PendingApproval>(); const pendingUserInputs = new Map<ApprovalRequestId, PendingUserInput>(); 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(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<PendingUserInputResolution>(); pendingUserInputs.set(requestId, { resolution }); yield* offerRuntimeEvent({ type: "user-input.requested", ...(yield* makeEventStamp()), provider: PROVIDER, threadId: input.threadId, turnId: sessions.get(input.threadId)?.activeTurnId, 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: sessions.get(input.threadId)?.activeTurnId, 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<ProviderApprovalDecision>(); pendingApprovals.set(requestId, { decision }); yield* offerRuntimeEvent( makeAcpRequestOpenedEvent({ stamp: yield* makeEventStamp(), provider: PROVIDER, threadId: input.threadId, turnId: sessions.get(input.threadId)?.activeTurnId, requestId: runtimeRequestId, permissionRequest, detail: permissionRequest.detail ?? encodeJsonStringForDiagnostics(params)?.slice(0, 2000) ??Lines 1–500 of 1007. Load more
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 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 { 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, makeGrokAcpRuntime, resolveGrokAcpBaseModelId, } from "../acp/GrokAcpSupport.ts"; import { extractXAiAskUserQuestions, makeXAiAskUserQuestionCancelledResponse, makeXAiAskUserQuestionResponse, XAiAskUserQuestionRequest, } from "../acp/XAiAcpExtension.ts"; import { type GrokAdapterShape } from "../Services/GrokAdapter.ts"; import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.UnknownFromJsonString); 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<ProviderApprovalDecision>; } type PendingUserInputResolution = | { readonly _tag: "answered"; readonly answers: ProviderUserInputAnswers } | { readonly _tag: "cancelled" }; interface PendingUserInput { readonly resolution: Deferred.Deferred<PendingUserInputResolution>; } interface GrokSessionContext { readonly threadId: ThreadId; readonly acpSessionId: string; session: ProviderSession; readonly scope: Scope.Closeable; readonly acp: AcpSessionRuntime.AcpSessionRuntime["Service"]; notificationFiber: Fiber.Fiber<void, never> | undefined; readonly pendingApprovals: Map<ApprovalRequestId, PendingApproval>; readonly pendingUserInputs: Map<ApprovalRequestId, PendingUserInput>; turns: Array<{ id: TurnId; items: Array<unknown> }>; lastPlanFingerprint: string | undefined; activeTurnId: TurnId | undefined; /** 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<ApprovalRequestId, PendingApproval>, ): Effect.Effect<void> { return Effect.forEach( Array.from(pendingApprovals.values()), (pending) => Deferred.succeed(pending.decision, "cancel").pipe(Effect.ignore), { discard: true }, ); } function settlePendingUserInputsAsCancelled( pendingUserInputs: ReadonlyMap<ApprovalRequestId, PendingUserInput>, ): Effect.Effect<void> { return Effect.forEach( Array.from(pendingUserInputs.values()), (pending) => Deferred.succeed(pending.resolution, { _tag: "cancelled" }).pipe(Effect.ignore), { discard: true }, ); } function isRecord(value: unknown): value is Record<string, unknown> { return typeof value === "object" && value !== null && !Array.isArray(value); } 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<ProviderApprovalDecision, "cancel">, ): 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") ); } 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<ThreadId, GrokSessionContext>(); const threadLocksRef = yield* SynchronizedRef.make(new Map<string, Semaphore.Semaphore>()); const runtimeEventPubSub = yield* PubSub.unbounded<ProviderRuntimeEvent>(); 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 = <A, E, R>(effect: Effect.Effect<A, E, R>) => 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<Semaphore.Semaphore> = 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 = <A, E, R>(threadId: string, effect: Effect.Effect<A, E, R>) => Effect.flatMap(getThreadSemaphore(threadId), (semaphore) => semaphore.withPermit(effect)); 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, 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 = `${ctx.activeTurnId ?? "no-turn"}:${encodeJsonStringForDiagnostics(payload) ?? "[unserializable payload]"}`; if (ctx.lastPlanFingerprint === fingerprint) { return; } ctx.lastPlanFingerprint = fingerprint; yield* offerRuntimeEvent( makeAcpPlanUpdatedEvent({ stamp: yield* makeEventStamp(), provider: PROVIDER, threadId: ctx.threadId, turnId: ctx.activeTurnId, payload, source: "acp.jsonrpc", method, rawPayload, }), ); }); const requireSession = ( threadId: ThreadId, ): Effect.Effect<GrokSessionContext, ProviderAdapterSessionNotFoundError> => { 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<ApprovalRequestId, PendingApproval>(); const pendingUserInputs = new Map<ApprovalRequestId, PendingUserInput>(); 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(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<PendingUserInputResolution>(); pendingUserInputs.set(requestId, { resolution }); yield* offerRuntimeEvent({ type: "user-input.requested", ...(yield* makeEventStamp()), provider: PROVIDER, threadId: input.threadId, turnId: sessions.get(input.threadId)?.activeTurnId, 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: sessions.get(input.threadId)?.activeTurnId, 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<ProviderApprovalDecision>(); pendingApprovals.set(requestId, { decision }); yield* offerRuntimeEvent( makeAcpRequestOpenedEvent({ stamp: yield* makeEventStamp(), provider: PROVIDER, threadId: input.threadId, turnId: sessions.get(input.threadId)?.activeTurnId, requestId: runtimeRequestId, permissionRequest, detail: permissionRequest.detail ?? encodeJsonStringForDiagnostics(params)?.slice(0, 2000) ??