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 { 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,  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.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;  /** Turns already interrupted; late prompt RPCs must not resurrect them. */  interruptedTurnIds: Set<TurnId>;  /** 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 appendPromptResultToTurn(  ctx: GrokSessionContext,  turnId: TurnId,  promptParts: ReadonlyArray<EffectAcpSchema.ContentBlock>,  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<string, unknown> {  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, GrokSessionContext>,  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<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")  );} function completedStopReasonFromPromptResponse(  response: EffectAcpSchema.PromptResponse | undefined,): EffectAcpSchema.StopReason | null {  if (response === undefined || promptResponseHasMissingXAiStopReason(response)) {    return null;  }  return response.stopReason;} 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 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 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<GrokSessionContext, ProviderAdapterSessionNotFoundError> => {

Lines 1–500 of 1464. Load more