import {  ApprovalRequestId,  type GrokSettings,  EventId,  type ProviderApprovalDecision,  type ProviderRuntimeEvent,  type ProviderSession,  type ProviderUserInputAnswers,  ProviderDriverKind,  ProviderInstanceId,  RuntimeRequestId,  type ThreadId,  TurnId,} from "@t3tools/contracts";import { HostProcessEnvironment, HostProcessPlatform } from "@t3tools/shared/hostProcess";import { getModelSelectionStringOptionValue } from "@t3tools/shared/model";import { stableStringify } from "@t3tools/shared/relaySigning";import * as Clock from "effect/Clock";import * as Crypto from "effect/Crypto";import * as DateTime from "effect/DateTime";import * as Deferred from "effect/Deferred";import * as Duration from "effect/Duration";import * as Effect from "effect/Effect";import * as Exit from "effect/Exit";import * as Fiber from "effect/Fiber";import * as FileSystem from "effect/FileSystem";import * as Option from "effect/Option";import * as Path from "effect/Path";import * as PubSub from "effect/PubSub";import * as Queue from "effect/Queue";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 { buildRuntimeInstructions } from "../RuntimeInstructions.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,  currentGrokReasoningEffortFromSessionSetup,  makeGrokAcpRuntime,  normalizeGrokReasoningEffort,  resolveGrokAcpBaseModelId,} from "../acp/GrokAcpSupport.ts";import {  extractGrokPlanMarkdownFromToolCallData,  extractXAiAskUserQuestions,  extractXAiExitPlanMarkdown,  makeXAiAskUserQuestionCancelledResponse,  makeXAiAskUserQuestionResponse,  makeXAiExitPlanModeCapturedResponse,  promptResponseHasMissingXAiStopReason,  XAiAskUserQuestionRequest,  XAiExitPlanModeRequest,} 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;const NANOS_PER_MILLI = 1_000_000n;// ACP does not expose Grok's private `streaming_reasoning` phase. Once it has// emitted standard ACP progress, ten silent minutes is long enough to avoid// treating legitimate reasoning as a stalled stream.const DEFAULT_GROK_TURN_INACTIVITY_TIMEOUT_MS = 10 * 60 * 1_000;// A tool can legitimately run without emitting text for much longer than// reasoning. It still needs a deadline so a lost tool update cannot leave the// turn working forever.const DEFAULT_GROK_ACTIVE_TOOL_INACTIVITY_TIMEOUT_MS = 30 * 60 * 1_000; 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;  /** Override the conservative ACP turn liveness timeout in focused tests. */  readonly turnInactivityTimeoutMs?: number;  /** Override the longer active-tool liveness timeout in focused tests. */  readonly activeToolInactivityTimeoutMs?: number;} 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 GrokTurnLivenessSignal {  readonly turnId: TurnId;} 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;  /**   * Latest plan.md body + turn it was emitted for. Dedupe is turn-scoped so a   * later turn re-proposing the same text still gets a new proposed-plan card.   */  lastKnownProposedPlanMarkdown: string | undefined;  lastKnownProposedPlanTurnId: TurnId | undefined;  /** True after enter_plan_mode until the turn ends or exit_plan_mode resolves. */  planModeActive: boolean;  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   * cancels the in-flight prompt and continues the same turn. Only the last   * remaining prompt settles the turn. */  promptsInFlight: number;  /** Monotonic id assigned to each sendTurn. Steers discard older epochs. */  promptEpoch: number;  /** Prompt epochs below this value must not start an ACP session/prompt. */  discardBeforeEpoch: number;  /** Serializes cancel-then-prompt so a steer cannot miss or hit the wrong RPC. */  readonly promptLifecycle: Semaphore.Semaphore;  readonly livenessSignals: Queue.Queue<GrokTurnLivenessSignal>;  livenessTurnId: TurnId | undefined;  lastTurnActivityAtNanos: bigint | undefined;  readonly activeToolCallIds: Set<string>;  livenessUpdatesInFlight: number;  /** Prompt RPCs that returned before their turn settlement acquired the lock. */  promptResponsesReady: number;  currentModelId: string | undefined;  currentReasoningEffort: 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; function clearProposedPlanFallback(ctx: GrokSessionContext): void {  ctx.lastKnownProposedPlanMarkdown = undefined;  ctx.lastKnownProposedPlanTurnId = undefined;  ctx.planModeActive = false;} /** Detect Grok's enter_plan_mode tool call from ACP tool state. */export function isGrokEnterPlanModeToolCall(toolCall: {  readonly title?: string;  readonly data: Record<string, unknown>;}): boolean {  const title = toolCall.title?.trim().toLowerCase() ?? "";  if (    title === "enter_plan_mode" ||    title === "plan: enter" ||    title === "plan mode entered" ||    title.includes("enter_plan_mode")  ) {    return true;  }  const rawInput = toolCall.data.rawInput;  if (isRecord(rawInput) && rawInput.variant === "EnterPlanMode") {    return true;  }  return false;} /** Failed enter_plan_mode must not leave planModeActive stuck on. */export function nextGrokPlanModeActive(  currentlyActive: boolean,  toolCall: {    readonly title?: string;    readonly status?: "pending" | "inProgress" | "completed" | "failed";    readonly data: Record<string, unknown>;  },): boolean {  if (!isGrokEnterPlanModeToolCall(toolCall)) {    return currentlyActive;  }  if (toolCall.status === "failed") {    return false;  }  if (toolCall.status === "completed" || toolCall.status === "inProgress") {    return true;  }  return currentlyActive;} 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() };} export function selectGrokPermissionOptionId(  request: EffectAcpSchema.RequestPermissionRequest,  decision: Exclude<ProviderApprovalDecision, "cancel">,): string | undefined {  const preferredKind =    decision === "acceptForSession"      ? "allow_always"      : decision === "accept"        ? "allow_once"        : "reject_once";  const preferred = request.options.find((entry) => entry.kind === preferredKind);  const preferredId = preferred?.optionId.trim();  if (preferredId) {    return preferredId;  }  // Grok 4.6 often omits allow_always. T3 still offers "Always allow this session".  if (decision === "acceptForSession") {    const once = request.options.find((entry) => entry.kind === "allow_once");    const onceId = once?.optionId.trim();    if (onceId) {      return onceId;    }  }  return undefined;} function selectAutoApprovedPermissionOption(  request: EffectAcpSchema.RequestPermissionRequest,): string | undefined {  return (    selectGrokPermissionOptionId(request, "acceptForSession") ??    selectGrokPermissionOptionId(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 hostPlatform = yield* HostProcessPlatform;    const hostEnvironment = yield* HostProcessEnvironment;    const grokPlanPathHost = {      platform: hostPlatform,      environment: options?.environment ?? hostEnvironment,    };     const sessions = new Map<ThreadId, GrokSessionContext>();    const threadLocksRef = yield* SynchronizedRef.make(new Map<string, Semaphore.Semaphore>());    const runtimeEventPubSub = yield* PubSub.unbounded<ProviderRuntimeEvent>();    const requestedTurnInactivityTimeoutMs = options?.turnInactivityTimeoutMs;    const turnInactivityTimeoutMs =      typeof requestedTurnInactivityTimeoutMs === "number" &&      Number.isFinite(requestedTurnInactivityTimeoutMs)        ? Math.max(1, Math.floor(requestedTurnInactivityTimeoutMs))        : DEFAULT_GROK_TURN_INACTIVITY_TIMEOUT_MS;    const turnInactivityTimeoutNanos = BigInt(turnInactivityTimeoutMs) * NANOS_PER_MILLI;    const requestedActiveToolInactivityTimeoutMs = options?.activeToolInactivityTimeoutMs;    const activeToolInactivityTimeoutMs =      typeof requestedActiveToolInactivityTimeoutMs === "number" &&      Number.isFinite(requestedActiveToolInactivityTimeoutMs)        ? Math.max(1, Math.floor(requestedActiveToolInactivityTimeoutMs))        : DEFAULT_GROK_ACTIVE_TOOL_INACTIVITY_TIMEOUT_MS;    const activeToolInactivityTimeoutNanos =      BigInt(activeToolInactivityTimeoutMs) * NANOS_PER_MILLI;     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 signalTurnLiveness = (ctx: GrokSessionContext, turnId: TurnId) =>      Queue.offer(ctx.livenessSignals, { turnId }).pipe(Effect.asVoid);     const beginTurnLiveness = (ctx: GrokSessionContext, turnId: TurnId) =>      Effect.sync(() => {        ctx.livenessTurnId = turnId;        // Do not start a deadline until ACP has made observable progress.        // Grok's private reasoning phase is not present in the ACP stream.        ctx.lastTurnActivityAtNanos = undefined;        ctx.activeToolCallIds.clear();      });     const clearTurnLiveness = (ctx: GrokSessionContext) => {      const turnId = ctx.livenessTurnId;      ctx.livenessTurnId = undefined;      ctx.lastTurnActivityAtNanos = undefined;      ctx.activeToolCallIds.clear();      ctx.livenessUpdatesInFlight = 0;      ctx.promptResponsesReady = 0;      return turnId === undefined ? Effect.void : signalTurnLiveness(ctx, turnId);    };     const recordTurnActivity = Effect.fn("GrokAdapter.recordTurnActivity")(function* (      ctx: GrokSessionContext,      turnId: TurnId,      event: Extract<        AcpSessionRuntime.AcpSessionRuntimeEvent,        {          _tag:            | "AssistantItemStarted"            | "AssistantItemCompleted"            | "PlanUpdated"            | "ToolCallUpdated"            | "ContentDelta";        }      >,    ) {      if (        ctx.livenessTurnId !== turnId ||        (event._tag === "ContentDelta" && event.text.length === 0)      ) {        return;      }      ctx.livenessUpdatesInFlight += 1;      try {        const activityAtNanos = yield* Clock.monotonicTimeNanos;        if (ctx.livenessTurnId !== turnId || ctx.interruptedTurnIds.has(turnId)) {          return;        }        if (event._tag === "ToolCallUpdated") {          if (event.toolCall.status === "completed" || event.toolCall.status === "failed") {            ctx.activeToolCallIds.delete(event.toolCall.toolCallId);          } else {            // A tool update without a terminal status receives a longer            // deadline so a long-running tool is not mistaken for a stall.            ctx.activeToolCallIds.add(event.toolCall.toolCallId);          }        }        ctx.lastTurnActivityAtNanos = activityAtNanos;      } finally {        // Decrement before signaling. The watchdog treats in-flight updates as a        // pause; if it consumed a signal while the counter was still > 0 it would        // wait on the next take with no follow-up wake after this decrement.        ctx.livenessUpdatesInFlight = Math.max(0, ctx.livenessUpdatesInFlight - 1);        yield* signalTurnLiveness(ctx, turnId);      }    });     const hasLivenessPause = (ctx: GrokSessionContext) =>      ctx.pendingApprovals.size > 0 ||

Lines 1–500 of 2153. Load more