// @effect-diagnostics nodeBuiltinImport:offimport * as NodeFS from "node:fs";import * as NodeOS from "node:os";import * as NodePath from "node:path"; import type {  ProviderApprovalDecision,  ProviderRuntimeEvent,  ProviderSendTurnInput,  ProviderSession,  ProviderTurnStartResult,  ProviderUploadFeedbackInput,  ProviderUploadFeedbackResult,} from "@t3tools/contracts";import {  ASSISTANT_CITATION_MAX_TEXT_LENGTH,  AssistantCitation,  ApprovalRequestId,  EnvironmentId,  EventId,  MessageId,  OrchestrationThreadShell,  ProjectId,  PROVIDER_SEND_TURN_MAX_INPUT_CHARS,  ProviderDriverKind,  ProviderInstanceId,  ProviderSessionStartInput,  ThreadId,  TurnId,} from "@t3tools/contracts";import {  expandAssistantCitationsForProvider,  serializeAssistantCitation,} from "@t3tools/shared/assistantCitations";import { createModelSelection } from "@t3tools/shared/model";import { it, assert, describe, vi } from "@effect/vitest";import { afterAll } from "vite-plus/test"; import * as Cause from "effect/Cause";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 Layer from "effect/Layer";import * as Metric from "effect/Metric";import * as Option from "effect/Option";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 Stream from "effect/Stream";import * as TestClock from "effect/testing/TestClock";import * as SqlClient from "effect/unstable/sql/SqlClient"; import {  ProviderAdapterRequestError,  ProviderAdapterSessionNotFoundError,  ProviderUnsupportedError,  ProviderValidationError,  ProviderWorkspaceMissingError,  type ProviderAdapterError,} from "../Errors.ts";import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts";import * as ProviderAdapterRegistry from "../Services/ProviderAdapterRegistry.ts";import * as ProviderService from "../Services/ProviderService.ts";import * as ProviderSessionDirectory from "../Services/ProviderSessionDirectory.ts";import { makeProviderServiceLive } from "./ProviderService.ts";import * as ProviderEventLoggers from "./ProviderEventLoggers.ts";import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts";import * as NodeServices from "@effect/platform-node/NodeServices";import * as ProviderSessionRuntime from "../../persistence/ProviderSessionRuntime.ts";import {  makeSqlitePersistenceLive,  SqlitePersistenceMemory,} from "../../persistence/Layers/Sqlite.ts";import * as ServerConfig from "../../config.ts";import * as ServerSettings from "../../serverSettings.ts";import * as AnalyticsService from "../../telemetry/AnalyticsService.ts";import { makeAdapterRegistryMock } from "../testUtils/providerAdapterRegistryMock.ts";import * as ProjectionSnapshotQuery from "../../orchestration/Services/ProjectionSnapshotQuery.ts"; const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));const defaultServerSettingsLayer = ServerSettings.ServerSettingsService.layerTest();const serverConfigTestLayer = ServerConfig.layerTest(process.cwd(), process.cwd()).pipe(  Layer.provide(NodeServices.layer),); // startSession verifies the workspace folder exists before dispatching to an// adapter, so session cwd fixtures must be real directories.const fixtureCwdRoot = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "provider-service-test-"));afterAll(() => NodeFS.rmSync(fixtureCwdRoot, { recursive: true, force: true }));function fixtureCwd(name: string): string {  const dir = NodePath.join(fixtureCwdRoot, name);  NodeFS.mkdirSync(dir, { recursive: true });  return dir;} const asRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.make(value);const asEventId = (value: string): EventId => EventId.make(value);const asThreadId = (value: string): ThreadId => ThreadId.make(value);const asTurnId = (value: string): TurnId => TurnId.make(value);const codexInstanceId = ProviderInstanceId.make("codex");const claudeAgentInstanceId = ProviderInstanceId.make("claudeAgent");const CODEX_DRIVER = ProviderDriverKind.make("codex");const CLAUDE_AGENT_DRIVER = ProviderDriverKind.make("claudeAgent");const CURSOR_DRIVER = ProviderDriverKind.make("cursor"); const assistantQuoteText = 'Keep the shared parser for "résumé".\nPreserve line breaks.';const assistantCitation = {  version: 1,  environmentId: EnvironmentId.make("source-environment/remote"),  threadId: asThreadId("source-thread/earlier"),  messageId: MessageId.make("source-message/first"),  text: assistantQuoteText,  start: 17,  end: 17 + assistantQuoteText.length,  prefix: "Previous advice. ",  suffix: " Next steps.",} satisfies AssistantCitation;const decodeAssistantQuoteContext = Schema.decodeUnknownEffect(  Schema.fromJsonString(    Schema.Array(Schema.Struct({ id: Schema.String, citation: AssistantCitation })),  ),); type LegacyProviderRuntimeEvent = {  readonly type: string;  readonly eventId: EventId;  readonly provider: ProviderDriverKind;  readonly createdAt: string;  readonly threadId: ThreadId;  readonly turnId?: string | undefined;  readonly itemId?: string | undefined;  readonly requestId?: string | undefined;  readonly payload?: unknown | undefined;  readonly [key: string]: unknown;}; function makeFakeCodexAdapter(  provider: ProviderDriverKind = CODEX_DRIVER,  supportsConversationRollback?: boolean,) {  const sessions = new Map<ThreadId, ProviderSession>();  const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());   const startSession = vi.fn((input: ProviderSessionStartInput) =>    Effect.sync(() => {      const now = "2026-01-01T00:00:00.000Z";      const session: ProviderSession = {        provider,        ...(input.providerInstanceId !== undefined          ? { providerInstanceId: input.providerInstanceId }          : {}),        status: "ready",        runtimeMode: input.runtimeMode,        threadId: input.threadId,        resumeCursor: input.resumeCursor ?? {          opaque: `resume-${String(input.threadId)}`,        },        cwd: input.cwd ?? process.cwd(),        createdAt: now,        updatedAt: now,      };      sessions.set(session.threadId, session);      return session;    }),  );   const sendTurn = vi.fn(    (      input: ProviderSendTurnInput,    ): Effect.Effect<ProviderTurnStartResult, ProviderAdapterError> => {      if (!sessions.has(input.threadId)) {        return Effect.fail(          new ProviderAdapterSessionNotFoundError({            provider,            threadId: input.threadId,          }),        );      }       return Effect.succeed({        threadId: input.threadId,        turnId: TurnId.make(`turn-${String(input.threadId)}`),      });    },  );   const interruptTurn = vi.fn(    (_threadId: ThreadId, _turnId?: TurnId): Effect.Effect<void, ProviderAdapterError> =>      Effect.void,  );   const compactThread = vi.fn((threadId: ThreadId): Effect.Effect<void, ProviderAdapterError> =>    Effect.sync(() =>      emit({        type: "thread.state.changed",        eventId: asEventId("evt-native-compact"),        provider,        createdAt: "2026-01-01T00:00:00.000Z",        threadId,        payload: { state: "compacted" },      }),    ),  );  const respondToRequest = vi.fn(    (      _threadId: ThreadId,      _requestId: string,      _decision: ProviderApprovalDecision,    ): Effect.Effect<void, ProviderAdapterError> => Effect.void,  );   const respondToUserInput = vi.fn(    (      _threadId: ThreadId,      _requestId: string,      _answers: Record<string, unknown>,    ): Effect.Effect<void, ProviderAdapterError> => Effect.void,  );   const stopSession = vi.fn((threadId: ThreadId): Effect.Effect<void, ProviderAdapterError> =>    Effect.sync(() => {      sessions.delete(threadId);    }),  );   const listSessions = vi.fn((): Effect.Effect<ReadonlyArray<ProviderSession>> =>    Effect.sync(() => Array.from(sessions.values())),  );   const hasSession = vi.fn((threadId: ThreadId): Effect.Effect<boolean> =>    Effect.succeed(sessions.has(threadId)),  );   const readThread = vi.fn(    (      threadId: ThreadId,    ): Effect.Effect<      {        threadId: ThreadId;        turns: ReadonlyArray<{ id: TurnId; items: readonly [] }>;      },      ProviderAdapterError    > =>      Effect.succeed({        threadId,        turns: [{ id: asTurnId("turn-1"), items: [] }],      }),  );   const rollbackThread = vi.fn(    (      threadId: ThreadId,      _numTurns: number,    ): Effect.Effect<{ threadId: ThreadId; turns: readonly [] }, ProviderAdapterError> =>      Effect.succeed({ threadId, turns: [] }),  );   const uploadFeedback = vi.fn(    (      input: ProviderUploadFeedbackInput,    ): Effect.Effect<ProviderUploadFeedbackResult, ProviderAdapterError> =>      Effect.succeed({ feedbackId: `feedback-${input.threadId}` }),  );   const stopAll = vi.fn((): Effect.Effect<void, ProviderAdapterError> =>    Effect.sync(() => {      sessions.clear();    }),  );   const adapter: ProviderAdapterShape<ProviderAdapterError> = {    provider,    capabilities: {      sessionModelSwitch: "in-session",      ...(supportsConversationRollback !== undefined ? { supportsConversationRollback } : {}),      ...(provider === CODEX_DRIVER ? { promptlessTurnContinuation: true } : {}),    },    startSession,    sendTurn,    ...(provider === CODEX_DRIVER      ? { compaction: { type: "native", start: compactThread } }      : provider === CURSOR_DRIVER        ? { compaction: { type: "slash-command", command: "/compress" } }        : provider === CLAUDE_AGENT_DRIVER          ? { compaction: { type: "slash-command", command: "/compact" } }          : {}),    interruptTurn,    respondToRequest,    respondToUserInput,    stopSession,    listSessions,    hasSession,    readThread,    rollbackThread,    ...(provider === CODEX_DRIVER ? { uploadFeedback } : {}),    stopAll,    get streamEvents() {      return Stream.fromPubSub(runtimeEventPubSub);    },  };   const emit = (event: LegacyProviderRuntimeEvent): void => {    Effect.runSync(PubSub.publish(runtimeEventPubSub, event as unknown as ProviderRuntimeEvent));  };   const updateSession = (    threadId: ThreadId,    update: (session: ProviderSession) => ProviderSession,  ): void => {    const existing = sessions.get(threadId);    if (!existing) {      return;    }    sessions.set(threadId, update(existing));  };   return {    adapter,    emit,    updateSession,    startSession,    sendTurn,    compactThread,    interruptTurn,    respondToRequest,    respondToUserInput,    stopSession,    listSessions,    hasSession,    readThread,    rollbackThread,    uploadFeedback,    stopAll,  };} interface RecordedAnalyticsEvent {  readonly event: string;  readonly properties?: Readonly<Record<string, unknown>>;} function makeRecordingAnalytics() {  const events: Array<RecordedAnalyticsEvent> = [];  const layer = Layer.succeed(    AnalyticsService.AnalyticsService,    AnalyticsService.AnalyticsService.of({      record: (event, properties) =>        Effect.sync(() => {          events.push({ event, ...(properties ? { properties } : {}) });        }),      flush: Effect.void,    }),  );   return {    layer,    reset: () => {      events.length = 0;    },    eventsByName: (event: string) => events.filter((entry) => entry.event === event),  };} function makeStaticInstanceRegistry(  entries: ReadonlyArray<readonly [ProviderInstanceId, ProviderAdapterShape<ProviderAdapterError>]>,): ProviderAdapterRegistry.ProviderAdapterRegistry["Service"] {  const adapters = new Map(entries);  const unsupported = (instanceId: ProviderInstanceId) =>    new ProviderUnsupportedError({      provider: ProviderDriverKind.make(instanceId),    });   return {    getByInstance: (instanceId) => {      const adapter = adapters.get(instanceId);      return adapter ? Effect.succeed(adapter) : Effect.fail(unsupported(instanceId));    },    getInstanceInfo: (instanceId) => {      const adapter = adapters.get(instanceId);      return adapter        ? Effect.succeed({            instanceId,            driverKind: adapter.provider,            displayName: undefined,            enabled: true,            continuationIdentity: {              driverKind: adapter.provider,              continuationKey: `${adapter.provider}:instance:${instanceId}`,            },          })        : Effect.fail(unsupported(instanceId));    },    listInstances: () => Effect.succeed(Array.from(adapters.keys())),    subscribeChanges: Effect.flatMap(PubSub.unbounded<void>(), (pubsub) =>      PubSub.subscribe(pubsub),    ),  };} const advanceTestClock = (ms: number) =>  TestClock.adjust(`${ms} millis`).pipe(Effect.andThen(Effect.yieldNow)); const hasMetricSnapshot = (  snapshots: ReadonlyArray<Metric.Metric.Snapshot>,  id: string,  attributes: Readonly<Record<string, string>>,) =>  snapshots.some(    (snapshot) =>      snapshot.id === id &&      Object.entries(attributes).every(([key, value]) => snapshot.attributes?.[key] === value),  ); function makeProviderServiceLayer(  input: {    readonly directory?: ProviderSessionDirectory.ProviderSessionDirectory["Service"];    readonly supportsConversationRollback?: boolean;    readonly analyticsLayer?: Layer.Layer<AnalyticsService.AnalyticsService>;    readonly registry?: ProviderAdapterRegistry.ProviderAdapterRegistry["Service"];  } = {},) {  const codex = makeFakeCodexAdapter(CODEX_DRIVER, input.supportsConversationRollback);  const claude = makeFakeCodexAdapter(CLAUDE_AGENT_DRIVER);  const cursor = makeFakeCodexAdapter(CURSOR_DRIVER);  const registry =    input.registry ??    makeAdapterRegistryMock({      [ProviderDriverKind.make("codex")]: codex.adapter,      [ProviderDriverKind.make("claudeAgent")]: claude.adapter,      [ProviderDriverKind.make("cursor")]: cursor.adapter,    });   const providerAdapterLayer = Layer.succeed(    ProviderAdapterRegistry.ProviderAdapterRegistry,    registry,  );  const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe(    Layer.provide(SqlitePersistenceMemory),  );  const directoryLayer =    input.directory === undefined      ? ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer))      : Layer.succeed(ProviderSessionDirectory.ProviderSessionDirectory, input.directory);   const layer = it.layer(    Layer.mergeAll(      makeProviderServiceLive().pipe(        Layer.provide(NodeServices.layer),        Layer.provide(providerAdapterLayer),        Layer.provide(directoryLayer),        Layer.provide(defaultServerSettingsLayer),        Layer.provide(serverConfigTestLayer),        Layer.provideMerge(input.analyticsLayer ?? AnalyticsService.layerTest),        Layer.provide(          Layer.succeed(            ProviderEventLoggers.ProviderEventLoggers,            ProviderEventLoggers.NoOpProviderEventLoggers,          ),        ),      ),      directoryLayer,       runtimeRepositoryLayer,      NodeServices.layer,    ),  );   return {    codex,    claude,    cursor,    layer,  };} for (const [enabled, completed] of [  [false, false],  [true, false],  [true, true],] as const) {  it.effect(    `persists shutdown recovery before stopping providers when enabled=${enabled}, completed=${completed}`,    () =>      Effect.gen(function* () {        const codex = makeFakeCodexAdapter();        const persistence = yield* Layer.build(          ProviderSessionDirectoryLive.pipe(            Layer.provide(              ProviderSessionRuntime.layer.pipe(Layer.provide(SqlitePersistenceMemory)),            ),          ),        );        const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory.pipe(          Effect.provide(persistence),        );        const threadId = asThreadId("shutdown-recovery");        const turnId = asTurnId("shutdown-recovery-turn");        const scope = yield* Scope.make();

Lines 1–500 of 4968. Load more