import * as NodeServices from "@effect/platform-node/NodeServices";import { assert, it } from "@effect/vitest";import {  EnvironmentId,  type ModelSelection,  type OrchestrationV2AppThread,  type OrchestrationV2DomainEvent,  type OrchestrationV2ProviderCapabilities,  type OrchestrationV2ProviderSession,  type OrchestrationV2ProviderThread,  ProviderDriverKind,  ProviderInstanceId,  type ProviderSessionId,  ThreadId,} from "@t3tools/contracts";import * as DateTime from "effect/DateTime";import * as Deferred from "effect/Deferred";import * as Effect from "effect/Effect";import * as Fiber from "effect/Fiber";import * as FileSystem from "effect/FileSystem";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as Queue from "effect/Queue";import * as Ref from "effect/Ref";import * as Stream from "effect/Stream";import { TestClock } from "effect/testing";import { HttpServer } from "effect/unstable/http"; import { ProviderWorkspaceMissingError } from "../provider/Errors.ts";import { ServerEnvironment } from "../environment/ServerEnvironment.ts";import * as McpProviderSession from "../mcp/McpProviderSession.ts";import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts";import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts";import * as ServerSettings from "../serverSettings.ts";import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";import { EventSinkV2, EventSinkWriteError, layer as eventSinkLayer } from "./EventSink.ts";import { layer as eventStoreLayer } from "./EventStore.ts";import {  IdAllocatorV2,  type IdAllocatorV2Shape,  layer as idAllocatorLayer,} from "./IdAllocator.ts";import { ProjectionStoreV2, layer as projectionStoreLayer } from "./ProjectionStore.ts";import {  ProviderAdapterEventStreamError,  type ProviderAdapterV2Event,  ProviderAdapterProtocolError,  type ProviderAdapterV2RuntimePolicy,  type ProviderAdapterV2SessionRuntime,  type ProviderAdapterV2Shape,} from "./ProviderAdapter.ts";import { makeSingleLayer as makeProviderAdapterRegistryLayer } from "./ProviderAdapterRegistry.ts";import {  ProviderSessionManagerV2,  layerWithOptions as providerSessionManagerLayerWithOptions,} from "./ProviderSessionManager.ts"; const TestDatabaseLayer = SqlitePersistenceMemory;const TestStoresLayer = Layer.merge(eventStoreLayer, projectionStoreLayer).pipe(  Layer.provide(TestDatabaseLayer),);const TestEventSinkLayer = eventSinkLayer.pipe(  Layer.provide(Layer.mergeAll(TestStoresLayer, TestDatabaseLayer)),);const FailingReleaseEventSinkLayer = Layer.effect(  EventSinkV2,  Effect.gen(function* () {    const delegate = yield* EventSinkV2;    return EventSinkV2.of({      ...delegate,      write: (input) =>        input.events.some(          (event) =>            event.type === "provider-session.updated" &&            (event.payload.status === "stopped" || event.payload.status === "error"),        )          ? Effect.fail(new EventSinkWriteError({ eventCount: input.events.length }))          : delegate.write(input),    });  }),).pipe(Layer.provide(TestEventSinkLayer)); const CodexCapabilities: OrchestrationV2ProviderCapabilities = CodexProviderCapabilitiesV2;const ExclusiveCapabilities: OrchestrationV2ProviderCapabilities = {  ...CodexCapabilities,  sessions: {    ...CodexCapabilities.sessions,    supportsMultipleProviderThreadsPerSession: false,  },}; interface TestProviderRuntimeState {  readonly openCount: number;  readonly closeCount: number;  readonly interruptCount: number;  readonly resumeCount: number;  readonly eventQueues: ReadonlyMap<string, Queue.Queue<ProviderAdapterV2Event>>;} const emptyState: TestProviderRuntimeState = {  openCount: 0,  closeCount: 0,  interruptCount: 0,  resumeCount: 0,  eventQueues: new Map(),}; const modelSelection = {  instanceId: ProviderInstanceId.make("codex"),  model: "gpt-5.4",} satisfies ModelSelection;const CODEX_DRIVER = ProviderDriverKind.make("codex"); const runtimePolicy = {  runtimeMode: "full-access",  interactionMode: "default",  cwd: process.cwd(),} satisfies ProviderAdapterV2RuntimePolicy; function makeProviderSession(input: {  readonly providerSessionId: ProviderSessionId;  readonly now: DateTime.Utc;  readonly capabilities?: OrchestrationV2ProviderCapabilities;}): OrchestrationV2ProviderSession {  return {    id: input.providerSessionId,    driver: CODEX_DRIVER,    providerInstanceId: modelSelection.instanceId,    status: "ready",    cwd: process.cwd(),    model: "gpt-5.4",    capabilities: input.capabilities ?? CodexCapabilities,    createdAt: input.now,    updatedAt: input.now,    lastError: null,  };} function makeThreadCreatedEvent(input: {  readonly idAllocator: IdAllocatorV2Shape;  readonly threadId: ThreadId;  readonly now: DateTime.Utc;}) {  return Effect.gen(function* () {    const projectId = yield* input.idAllocator.allocate.project({      fixtureName: "provider-session-manager",    });    const providerThreadId = input.idAllocator.derive.providerThread({      driver: CODEX_DRIVER,      nativeThreadId: "native-thread",    });    const thread: OrchestrationV2AppThread = {      createdBy: "user",      creationSource: "web",      id: input.threadId,      projectId,      title: "Provider session manager",      providerInstanceId: modelSelection.instanceId,      modelSelection,      runtimeMode: "full-access",      interactionMode: "default",      branch: null,      worktreePath: null,      activeProviderThreadId: providerThreadId,      lineage: {        parentThreadId: null,        relationshipToParent: null,        rootThreadId: input.threadId,      },      forkedFrom: null,      createdAt: input.now,      updatedAt: input.now,      archivedAt: null,      settledOverride: null,      settledAt: null,      lastVisitedAt: null,      deletedAt: null,    };    return {      id: yield* input.idAllocator.allocate.event({ threadId: input.threadId }),      type: "thread.created" as const,      threadId: input.threadId,      occurredAt: input.now,      payload: thread,    };  });} function makeProviderThread(input: {  readonly idAllocator: IdAllocatorV2Shape;  readonly threadId: ThreadId;  readonly providerSessionId: ProviderSessionId;  readonly now: DateTime.Utc;}): OrchestrationV2ProviderThread {  return {    id: input.idAllocator.derive.providerThread({      driver: CODEX_DRIVER,      nativeThreadId: "native-thread",    }),    driver: CODEX_DRIVER,    providerInstanceId: modelSelection.instanceId,    providerSessionId: input.providerSessionId,    appThreadId: input.threadId,    ownerNodeId: null,    nativeThreadRef: {      driver: CODEX_DRIVER,      nativeId: "native-thread",      strength: "strong",    },    nativeConversationHeadRef: null,    status: "idle",    firstRunOrdinal: null,    lastRunOrdinal: null,    handoffIds: [],    forkedFrom: null,    createdAt: input.now,    updatedAt: input.now,  };} function unimplemented(detail: string) {  return Effect.fail(    new ProviderAdapterProtocolError({      driver: CODEX_DRIVER,      detail,    }),  );} function makeProviderAdapter(  state: Ref.Ref<TestProviderRuntimeState>,  options: {    readonly failEventStream?: boolean;    readonly capabilities?: OrchestrationV2ProviderCapabilities;    readonly mcpConfigs?: Ref.Ref<      ReadonlyArray<McpProviderSession.McpProviderSessionConfig | undefined>    >;    readonly beforeOpen?: (input: {      readonly providerSessionId: ProviderSessionId;    }) => Effect.Effect<void>;    readonly hasPendingBackgroundWork?: Effect.Effect<boolean>;    readonly hangSessionScopeClose?: boolean;  } = {},): ProviderAdapterV2Shape {  return {    instanceId: ProviderInstanceId.make("codex"),    driver: CODEX_DRIVER,    getCapabilities: () => Effect.succeed(options.capabilities ?? CodexCapabilities),    planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),    openSession: (input) =>      Effect.gen(function* () {        if (options.beforeOpen !== undefined) {          yield* options.beforeOpen({ providerSessionId: input.providerSessionId });        }        if (options.mcpConfigs !== undefined) {          yield* Ref.update(options.mcpConfigs, (configs) => [            ...configs,            McpProviderSession.readMcpProviderSession(input.threadId),          ]);        }        const now = yield* DateTime.now;        const events = yield* Queue.unbounded<ProviderAdapterV2Event>();        const session = makeProviderSession({          providerSessionId: input.providerSessionId,          now,          ...(options.capabilities === undefined ? {} : { capabilities: options.capabilities }),        });        yield* Ref.update(state, (current) => {          const eventQueues = new Map(current.eventQueues);          eventQueues.set(String(input.providerSessionId), events);          return {            ...current,            openCount: current.openCount + 1,            eventQueues,          };        });        yield* Effect.addFinalizer(() =>          Ref.update(state, (current) => ({            ...current,            closeCount: current.closeCount + 1,          })),        );        if (options.hangSessionScopeClose === true) {          // Registered last so it runs first on scope close, wedging the          // close before the closeCount finalizer, like a provider process          // that never yields its message stream.          yield* Effect.addFinalizer(() => Effect.never);        }         return {          instanceId: ProviderInstanceId.make("codex"),          driver: CODEX_DRIVER,          providerSessionId: input.providerSessionId,          providerSession: session,          events: options.failEventStream            ? Stream.fail(                new ProviderAdapterEventStreamError({                  driver: CODEX_DRIVER,                  providerSessionId: input.providerSessionId,                  cause: "process exited",                }),              )            : Stream.fromQueue(events),          ...(options.hasPendingBackgroundWork === undefined            ? {}            : { hasPendingBackgroundWork: options.hasPendingBackgroundWork }),          ensureThread: () => unimplemented("ensureThread unused in test"),          resumeThread: (threadInput) =>            Ref.update(state, (current) => ({              ...current,              resumeCount: current.resumeCount + 1,            })).pipe(Effect.as(threadInput.providerThread)),          startTurn: () => Effect.void,          steerTurn: () => Effect.void,          interruptTurn: () =>            Ref.update(state, (current) => ({              ...current,              interruptCount: current.interruptCount + 1,            })),          respondToRuntimeRequest: () => Effect.void,          readThreadSnapshot: () => unimplemented("readThreadSnapshot unused in test"),          rollbackThread: () => unimplemented("rollbackThread unused in test"),          forkThread: () => unimplemented("forkThread unused in test"),        } satisfies ProviderAdapterV2SessionRuntime;      }),  };} function makeTestLayer(input: {  readonly state: Ref.Ref<TestProviderRuntimeState>;  readonly idleTimeoutMs: number;  readonly maxIdlePinMs?: number;  readonly failEventStream?: boolean;  readonly capabilities?: OrchestrationV2ProviderCapabilities;  readonly mcpConfigs?: Ref.Ref<    ReadonlyArray<McpProviderSession.McpProviderSessionConfig | undefined>  >;  readonly beforeOpen?: (input: {    readonly providerSessionId: ProviderSessionId;  }) => Effect.Effect<void>;  readonly failReleaseEventWrites?: boolean;  readonly hasPendingBackgroundWork?: Effect.Effect<boolean>;  readonly hangSessionScopeClose?: boolean;  readonly serverSettingsLayer?: ReturnType<typeof ServerSettings.layerTest>;}) {  const configuredEventSinkLayer = input.failReleaseEventWrites    ? FailingReleaseEventSinkLayer    : TestEventSinkLayer;  const registryLayer = makeProviderAdapterRegistryLayer(    makeProviderAdapter(input.state, {      failEventStream: input.failEventStream ?? false,      ...(input.capabilities === undefined ? {} : { capabilities: input.capabilities }),      ...(input.mcpConfigs === undefined ? {} : { mcpConfigs: input.mcpConfigs }),      ...(input.beforeOpen === undefined ? {} : { beforeOpen: input.beforeOpen }),      ...(input.hasPendingBackgroundWork === undefined        ? {}        : { hasPendingBackgroundWork: input.hasPendingBackgroundWork }),      ...(input.hangSessionScopeClose === undefined        ? {}        : { hangSessionScopeClose: input.hangSessionScopeClose }),    }),  );  return Layer.mergeAll(    TestStoresLayer,    configuredEventSinkLayer,    idAllocatorLayer,    TestMcpRegistryLayer,    providerSessionManagerLayerWithOptions({      idleTimeoutMs: input.idleTimeoutMs,      ...(input.maxIdlePinMs === undefined ? {} : { maxIdlePinMs: input.maxIdlePinMs }),    }).pipe(      Layer.provide(        Layer.mergeAll(          registryLayer,          configuredEventSinkLayer,          idAllocatorLayer,          TestMcpRegistryLayer,          TestStoresLayer,          ...(input.serverSettingsLayer === undefined ? [] : [input.serverSettingsLayer]),        ),      ),    ),  ).pipe(Layer.provide(NodeServices.layer));} const fakeHttpServer = HttpServer.HttpServer.of({  address: { _tag: "TcpAddress", hostname: "127.0.0.1", port: 43123 },  serve: (() => Effect.void) as HttpServer.HttpServer["Service"]["serve"],}); const fakeEnvironment = ServerEnvironment.of({  getEnvironmentId: Effect.succeed(EnvironmentId.make("environment-provider-session-manager")),  getDescriptor: Effect.die("unused"),}); const TestMcpRegistryLayer = Layer.effect(  McpSessionRegistry.McpSessionRegistry,  McpSessionRegistry.__testing.make(),).pipe(  Layer.provide(Layer.succeed(HttpServer.HttpServer, fakeHttpServer)),  Layer.provide(Layer.succeed(ServerEnvironment, fakeEnvironment)),  Layer.provide(NodeServices.layer),); function makePendingRuntimeRequestEvents(input: {  readonly idAllocator: IdAllocatorV2Shape;  readonly threadId: ThreadId;  readonly providerSessionId: ProviderSessionId;  readonly providerThread: OrchestrationV2ProviderThread;  readonly now: DateTime.Utc;}) {  return Effect.gen(function* () {    const requestId = yield* input.idAllocator.allocate.runtimeRequest({      driver: CODEX_DRIVER,      nativeRequestId: "pending-approval",    });    const nodeId = input.idAllocator.derive.approvalNode({ requestId });    const node = {      id: nodeId,      threadId: input.threadId,      runId: null,      parentNodeId: null,      rootNodeId: nodeId,      kind: "approval_request" as const,      status: "waiting" as const,      countsForRun: false,      providerThreadId: input.providerThread.id,      providerTurnId: null,      nativeItemRef: null,      runtimeRequestId: requestId,      checkpointScopeId: null,      startedAt: input.now,      completedAt: null,    };    const request = {      id: requestId,      nodeId,      providerTurnId: null,      nativeRequestRef: {        driver: CODEX_DRIVER,        nativeId: "pending-approval",        strength: "strong" as const,      },      kind: "command" as const,      status: "pending" as const,      responseCapability: {        type: "live" as const,        providerSessionId: input.providerSessionId,      },      createdAt: input.now,      resolvedAt: null,    };    const turnItem = {      id: input.idAllocator.derive.approvalTurnItem({ requestId }),      threadId: input.threadId,      runId: null,      nodeId,      providerThreadId: input.providerThread.id,      providerTurnId: null,      nativeItemRef: null,      parentItemId: null,      ordinal: 1,      status: "waiting" as const,      title: null,      startedAt: input.now,      completedAt: null,      updatedAt: input.now,      type: "approval_request" as const,      requestId,      requestKind: "command" as const,    };    return [      {        id: yield* input.idAllocator.allocate.event({          threadId: input.threadId,          providerSessionId: input.providerSessionId,        }),        type: "node.updated" as const,        threadId: input.threadId,        nodeId,        driver: CODEX_DRIVER,        occurredAt: input.now,        payload: node,      },      {        id: yield* input.idAllocator.allocate.event({          threadId: input.threadId,          providerSessionId: input.providerSessionId,        }),        type: "runtime-request.updated" as const,        threadId: input.threadId,        nodeId,        driver: CODEX_DRIVER,        occurredAt: input.now,        payload: request,      },      {        id: yield* input.idAllocator.allocate.event({          threadId: input.threadId,          providerSessionId: input.providerSessionId,

Lines 1–500 of 2448. Load more