import { randomUUID } from "node:crypto"; import {  EventId,  type ProviderRuntimeEvent,  type ProviderSession,  RuntimeItemId,  RuntimeRequestId,  ThreadId,  type ToolLifecycleItemType,  TurnId,  type UserInputQuestion,} from "@t3tools/contracts";import { Cause, Effect, Exit, Layer, Queue, Ref, Scope, Stream } from "effect";import type { OpencodeClient, Part, PermissionRequest, QuestionRequest } from "@opencode-ai/sdk/v2";import { getModelSelectionStringOptionValue } from "@t3tools/shared/model"; import { resolveAttachmentPath } from "../../attachmentStore.ts";import { ServerConfig } from "../../config.ts";import { ServerSettingsService } from "../../serverSettings.ts";import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";import {  ProviderAdapterProcessError,  ProviderAdapterRequestError,  ProviderAdapterSessionClosedError,  ProviderAdapterSessionNotFoundError,  ProviderAdapterValidationError,} from "../Errors.ts";import { OpenCodeAdapter, type OpenCodeAdapterShape } from "../Services/OpenCodeAdapter.ts";import {  buildOpenCodePermissionRules,  OpenCodeRuntime,  OpenCodeRuntimeError,  openCodeQuestionId,  openCodeRuntimeErrorDetail,  parseOpenCodeModelSlug,  runOpenCodeSdk,  toOpenCodeFileParts,  toOpenCodePermissionReply,  toOpenCodeQuestionAnswers,  type OpenCodeServerConnection,} from "../opencodeRuntime.ts"; const PROVIDER = "opencode" as const; interface OpenCodeTurnSnapshot {  readonly id: TurnId;  readonly items: Array<unknown>;} type OpenCodeSubscribedEvent =  Awaited<ReturnType<OpencodeClient["event"]["subscribe"]>> extends {    readonly stream: AsyncIterable<infer TEvent>;  }    ? TEvent    : never; interface OpenCodeSessionContext {  session: ProviderSession;  readonly client: OpencodeClient;  readonly server: OpenCodeServerConnection;  readonly directory: string;  readonly openCodeSessionId: string;  readonly pendingPermissions: Map<string, PermissionRequest>;  readonly pendingQuestions: Map<string, QuestionRequest>;  readonly messageRoleById: Map<string, "user" | "assistant">;  readonly partById: Map<string, Part>;  readonly emittedTextByPartId: Map<string, string>;  readonly completedAssistantPartIds: Set<string>;  readonly turns: Array<OpenCodeTurnSnapshot>;  activeTurnId: TurnId | undefined;  activeAgent: string | undefined;  activeVariant: string | undefined;  /**   * One-shot guard flipped by `stopOpenCodeContext` / `emitUnexpectedExit`.   * The session lifecycle is owned by `sessionScope`; this Ref exists only   * so concurrent callers can race the transition safely via `getAndSet`.   */  readonly stopped: Ref.Ref<boolean>;  /**   * Sole lifecycle handle for the session. Closing this scope:   *   - aborts the `AbortController` registered as a finalizer   *     (cancels the in-flight `event.subscribe` fetch),   *   - interrupts the event-pump and server-exit fibers forked   *     via `Effect.forkIn(sessionScope)`,   *   - tears down the OpenCode server process for scope-owned servers.   */  readonly sessionScope: Scope.Closeable;} export interface OpenCodeAdapterLiveOptions {  readonly nativeEventLogPath?: string;  readonly nativeEventLogger?: EventNdjsonLogger;} function nowIso(): string {  return new Date().toISOString();} /** * Map a tagged OpenCodeRuntimeError produced by {@link runOpenCodeSdk} into * the adapter-boundary `ProviderAdapterRequestError`. SDK-method-level call * sites pipe through this in `Effect.mapError` so they never build the error * shape by hand. */const toRequestError = (cause: OpenCodeRuntimeError): ProviderAdapterRequestError =>  new ProviderAdapterRequestError({    provider: PROVIDER,    method: cause.operation,    detail: cause.detail,    cause: cause.cause,  }); /** * Map a `Cause.squash`-ed failure into a `ProviderAdapterProcessError`. The * typed cause is usually an `OpenCodeRuntimeError` (from {@link runOpenCodeSdk}), * in which case we preserve its `detail`; otherwise we fall back to * {@link openCodeRuntimeErrorDetail} for unknown causes (defects, etc.). */const toProcessError = (threadId: ThreadId, cause: unknown): ProviderAdapterProcessError =>  new ProviderAdapterProcessError({    provider: PROVIDER,    threadId,    detail: OpenCodeRuntimeError.is(cause) ? cause.detail : openCodeRuntimeErrorDetail(cause),    cause,  }); function buildEventBase(input: {  readonly threadId: ThreadId;  readonly turnId?: TurnId | undefined;  readonly itemId?: string | undefined;  readonly requestId?: string | undefined;  readonly createdAt?: string | undefined;  readonly raw?: unknown;}): Pick<  ProviderRuntimeEvent,  "eventId" | "provider" | "threadId" | "createdAt" | "turnId" | "itemId" | "requestId" | "raw"> {  return {    eventId: EventId.make(randomUUID()),    provider: PROVIDER,    threadId: input.threadId,    createdAt: input.createdAt ?? nowIso(),    ...(input.turnId ? { turnId: input.turnId } : {}),    ...(input.itemId ? { itemId: RuntimeItemId.make(input.itemId) } : {}),    ...(input.requestId ? { requestId: RuntimeRequestId.make(input.requestId) } : {}),    ...(input.raw !== undefined      ? {          raw: {            source: "opencode.sdk.event",            payload: input.raw,          },        }      : {}),  };} function toToolLifecycleItemType(toolName: string): ToolLifecycleItemType {  const normalized = toolName.toLowerCase();  if (normalized.includes("bash") || normalized.includes("command")) {    return "command_execution";  }  if (    normalized.includes("edit") ||    normalized.includes("write") ||    normalized.includes("patch") ||    normalized.includes("multiedit")  ) {    return "file_change";  }  if (normalized.includes("web")) {    return "web_search";  }  if (normalized.includes("mcp")) {    return "mcp_tool_call";  }  if (normalized.includes("image")) {    return "image_view";  }  if (    normalized.includes("task") ||    normalized.includes("agent") ||    normalized.includes("subtask")  ) {    return "collab_agent_tool_call";  }  return "dynamic_tool_call";} function mapPermissionToRequestType(  permission: string,): "command_execution_approval" | "file_read_approval" | "file_change_approval" | "unknown" {  switch (permission) {    case "bash":      return "command_execution_approval";    case "read":      return "file_read_approval";    case "edit":      return "file_change_approval";    default:      return "unknown";  }} function mapPermissionDecision(reply: "once" | "always" | "reject"): string {  switch (reply) {    case "once":      return "accept";    case "always":      return "acceptForSession";    case "reject":    default:      return "decline";  }} function resolveTurnSnapshot(  context: OpenCodeSessionContext,  turnId: TurnId,): OpenCodeTurnSnapshot {  const existing = context.turns.find((turn) => turn.id === turnId);  if (existing) {    return existing;  }   const created: OpenCodeTurnSnapshot = { id: turnId, items: [] };  context.turns.push(created);  return created;} function appendTurnItem(  context: OpenCodeSessionContext,  turnId: TurnId | undefined,  item: unknown,): void {  if (!turnId) {    return;  }  resolveTurnSnapshot(context, turnId).items.push(item);} function ensureSessionContext(  sessions: ReadonlyMap<ThreadId, OpenCodeSessionContext>,  threadId: ThreadId,): OpenCodeSessionContext {  const session = sessions.get(threadId);  if (!session) {    throw new ProviderAdapterSessionNotFoundError({ provider: PROVIDER, threadId });  }  // `ensureSessionContext` is a sync gate used from both sync helpers and  // Effect bodies. `Ref.getUnsafe` is an atomic read of the backing cell —  // no fiber suspension required, which keeps this callable everywhere.  if (Ref.getUnsafe(session.stopped)) {    throw new ProviderAdapterSessionClosedError({ provider: PROVIDER, threadId });  }  return session;} function normalizeQuestionRequest(request: QuestionRequest): ReadonlyArray<UserInputQuestion> {  return request.questions.map((question, index) => ({    id: openCodeQuestionId(index, question),    header: question.header,    question: question.question,    options: question.options.map((option) => ({      label: option.label,      description: option.description,    })),    ...(question.multiple ? { multiSelect: true } : {}),  }));} function resolveTextStreamKind(part: Part | undefined): "assistant_text" | "reasoning_text" {  return part?.type === "reasoning" ? "reasoning_text" : "assistant_text";} function textFromPart(part: Part): string | undefined {  switch (part.type) {    case "text":    case "reasoning":      return part.text;    default:      return undefined;  }} function commonPrefixLength(left: string, right: string): number {  let index = 0;  while (index < left.length && index < right.length && left[index] === right[index]) {    index += 1;  }  return index;} function suffixPrefixOverlap(text: string, delta: string): number {  const maxLength = Math.min(text.length, delta.length);  for (let length = maxLength; length > 0; length -= 1) {    if (text.endsWith(delta.slice(0, length))) {      return length;    }  }  return 0;} function resolveLatestAssistantText(previousText: string | undefined, nextText: string): string {  if (previousText && previousText.length > nextText.length && previousText.startsWith(nextText)) {    return previousText;  }  return nextText;} export function mergeOpenCodeAssistantText(  previousText: string | undefined,  nextText: string,): {  readonly latestText: string;  readonly deltaToEmit: string;} {  const latestText = resolveLatestAssistantText(previousText, nextText);  return {    latestText,    deltaToEmit: latestText.slice(commonPrefixLength(previousText ?? "", latestText)),  };} export function appendOpenCodeAssistantTextDelta(  previousText: string,  delta: string,): {  readonly nextText: string;  readonly deltaToEmit: string;} {  const deltaToEmit = delta.slice(suffixPrefixOverlap(previousText, delta));  return {    nextText: previousText + deltaToEmit,    deltaToEmit,  };} function isoFromEpochMs(value: number | undefined): string | undefined {  if (typeof value !== "number" || !Number.isFinite(value) || value <= 0) {    return undefined;  }  return new Date(value).toISOString();} function messageRoleForPart(  context: OpenCodeSessionContext,  part: Pick<Part, "messageID" | "type">,): "assistant" | "user" | undefined {  const known = context.messageRoleById.get(part.messageID);  if (known) {    return known;  }  return part.type === "tool" ? "assistant" : undefined;} function detailFromToolPart(part: Extract<Part, { type: "tool" }>): string | undefined {  switch (part.state.status) {    case "completed":      return part.state.output;    case "error":      return part.state.error;    case "running":      return part.state.title;    default:      return undefined;  }} function toolStateCreatedAt(part: Extract<Part, { type: "tool" }>): string | undefined {  switch (part.state.status) {    case "running":      return isoFromEpochMs(part.state.time.start);    case "completed":    case "error":      return isoFromEpochMs(part.state.time.end);    default:      return undefined;  }} function sessionErrorMessage(error: unknown): string {  if (!error || typeof error !== "object") {    return "OpenCode session failed.";  }  const data = "data" in error && error.data && typeof error.data === "object" ? error.data : null;  const message = data && "message" in data ? data.message : null;  return typeof message === "string" && message.trim().length > 0    ? message    : "OpenCode session failed.";} function updateProviderSession(  context: OpenCodeSessionContext,  patch: Partial<ProviderSession>,  options?: {    readonly clearActiveTurnId?: boolean;    readonly clearLastError?: boolean;  },): ProviderSession {  const nextSession = {    ...context.session,    ...patch,    updatedAt: nowIso(),  } as ProviderSession & Record<string, unknown>;  const mutableSession = nextSession as Record<string, unknown>;  if (options?.clearActiveTurnId) {    delete mutableSession.activeTurnId;  }  if (options?.clearLastError) {    delete mutableSession.lastError;  }  context.session = nextSession;  return nextSession;} const stopOpenCodeContext = Effect.fn("stopOpenCodeContext")(function* (  context: OpenCodeSessionContext,) {  // Race-safe one-shot: first caller flips the flag, everyone else no-ops.  if (yield* Ref.getAndSet(context.stopped, true)) {    return;  }   // Best-effort remote abort. The scope close below tears down the local  // handles (event-pump fiber, server-exit fiber, event-subscribe fetch),  // but we still want to tell OpenCode that this session is done.  yield* runOpenCodeSdk("session.abort", () =>    context.client.session.abort({ sessionID: context.openCodeSessionId }),  ).pipe(Effect.ignore({ log: true }));   // Closing the session scope interrupts every fiber forked into it and  // runs each finalizer we registered — the `AbortController.abort()` call,  // the child-process termination, etc.  yield* Scope.close(context.sessionScope, Exit.void);}); export function makeOpenCodeAdapterLive(options?: OpenCodeAdapterLiveOptions) {  return Layer.effect(    OpenCodeAdapter,    Effect.gen(function* () {      const serverConfig = yield* ServerConfig;      const serverSettings = yield* ServerSettingsService;      const openCodeRuntime = yield* OpenCodeRuntime;      const nativeEventLogger =        options?.nativeEventLogger ??        (options?.nativeEventLogPath !== undefined          ? yield* makeEventNdjsonLogger(options.nativeEventLogPath, {              stream: "native",            })          : undefined);      // Only close loggers we created. If the caller passed one in via      // `options.nativeEventLogger`, they own its lifecycle.      const managedNativeEventLogger =        options?.nativeEventLogger === undefined ? nativeEventLogger : undefined;      const runtimeEvents = yield* Queue.unbounded<ProviderRuntimeEvent>();      const sessions = new Map<ThreadId, OpenCodeSessionContext>();       // Layer-level finalizer: when the adapter layer shuts down, stop every      // session. Each session's `Scope.close` tears down its spawned OpenCode      // server (via the `ChildProcessSpawner` finalizer installed in      // `startOpenCodeServerProcess`) and interrupts the forked event/exit      // fibers. Consumers that can't reason about Effect scopes therefore      // cannot leak OpenCode child processes by forgetting to call `stopAll`.      yield* Effect.addFinalizer(() =>        Effect.gen(function* () {          const contexts = [...sessions.values()];          sessions.clear();          // `ignoreCause` swallows both typed failures (none here) and defects          // from throwing scope finalizers so a sibling's death can't interrupt          // the remaining cleanups.          yield* Effect.forEach(            contexts,            (context) => Effect.ignoreCause(stopOpenCodeContext(context)),            { concurrency: "unbounded", discard: true },          );          // Close the logger AFTER session teardown so any final lifecycle          // events emitted during shutdown still get written. `close` flushes          // the `Logger.batched` window and closes each per-thread          // `RotatingFileSink` handle owned by the logger's internal scope.          if (managedNativeEventLogger !== undefined) {            yield* managedNativeEventLogger.close();          }        }),      );       const emit = (event: ProviderRuntimeEvent) =>        Queue.offer(runtimeEvents, event).pipe(Effect.asVoid);      const writeNativeEvent = (        threadId: ThreadId,        event: {          readonly observedAt: string;          readonly event: Record<string, unknown>;        },      ) => (nativeEventLogger ? nativeEventLogger.write(event, threadId) : Effect.void);      const writeNativeEventBestEffort = (        threadId: ThreadId,        event: {          readonly observedAt: string;          readonly event: Record<string, unknown>;

Lines 1–500 of 1389. Load more