// @effect-diagnostics nodeBuiltinImport:offimport * as NodeFS from "node:fs";import * as NodeOS from "node:os";import * as NodePath from "node:path"; import {  OrchestrationReadModel,  ProviderDriverKind,  ProviderRuntimeEvent,  ProviderSession,  ProviderInstanceId,} from "@t3tools/contracts";import {  ApprovalRequestId,  CommandId,  DEFAULT_PROVIDER_INTERACTION_MODE,  EventId,  MessageId,  type OrchestrationCommand,  ProjectId,  ProviderItemId,  type ServerSettings,  ThreadId,  TurnId,} from "@t3tools/contracts";import * as Clock from "effect/Clock";import * as Effect from "effect/Effect";import * as Exit from "effect/Exit";import * as Layer from "effect/Layer";import * as ManagedRuntime from "effect/ManagedRuntime";import * as PubSub from "effect/PubSub";import * as Scope from "effect/Scope";import * as Stream from "effect/Stream";import { it as effectIt } from "@effect/vitest";import { afterEach, describe, expect, it } from "vite-plus/test"; import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts";import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";import {  ProviderService,  type ProviderServiceShape,} from "../../provider/Services/ProviderService.ts";import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityResolver.ts";import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts";import * as ThreadPlanProgress from "../ThreadPlanProgress.ts";import { ProviderRuntimeIngestionLive } from "./ProviderRuntimeIngestion.ts";import { DEFAULT_THREAD_TITLE } from "../threadTitles.ts";import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts";import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";import { ServerConfig } from "../../config.ts";import { ServerSettingsService } from "../../serverSettings.ts";import * as NodeServices from "@effect/platform-node/NodeServices"; function makeTestServerSettingsLayer(overrides: Partial<ServerSettings> = {}) {  return ServerSettingsService.layerTest(overrides);} const asProjectId = (value: string): ProjectId => ProjectId.make(value);const asItemId = (value: string): ProviderItemId => ProviderItemId.make(value);const asEventId = (value: string): EventId => EventId.make(value);const asMessageId = (value: string): MessageId => MessageId.make(value);const asThreadId = (value: string): ThreadId => ThreadId.make(value);const asTurnId = (value: string): TurnId => TurnId.make(value); type LegacyProviderRuntimeEvent = {  readonly type: string;  readonly eventId: EventId;  readonly provider: ProviderRuntimeEvent["provider"];  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;}; type LegacyTurnCompletedEvent = LegacyProviderRuntimeEvent & {  readonly type: "turn.completed";  readonly payload?: undefined;  readonly status: "completed" | "failed" | "interrupted" | "cancelled";  readonly errorMessage?: string | undefined;}; function isLegacyTurnCompletedEvent(  event: LegacyProviderRuntimeEvent,): event is LegacyTurnCompletedEvent {  return (    event.type === "turn.completed" &&    event.payload === undefined &&    typeof event.status === "string"  );} function createProviderServiceHarness() {  const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());  const runtimeSessions: ProviderSession[] = [];   const unsupported = () => Effect.die(new Error("Unsupported provider call in test")) as never;  const service: ProviderServiceShape = {    startSession: () => unsupported(),    sendTurn: () => unsupported(),    interruptTurn: () => unsupported(),    respondToRequest: () => unsupported(),    respondToUserInput: () => unsupported(),    stopSession: () => unsupported(),    listSessions: () => Effect.succeed([...runtimeSessions]),    getCapabilities: () => Effect.succeed({ sessionModelSwitch: "in-session" }),    getInstanceInfo: (instanceId) => {      const driverKind = ProviderDriverKind.make(String(instanceId));      return Effect.succeed({        instanceId,        driverKind,        displayName: undefined,        enabled: true,        continuationIdentity: {          driverKind,          continuationKey: `${driverKind}:instance:${instanceId}`,        },      });    },    rollbackConversation: () => unsupported(),    uploadFeedback: () => unsupported(),    get streamEvents() {      return Stream.fromPubSub(runtimeEventPubSub);    },  };   const setSession = (session: ProviderSession): void => {    const existingIndex = runtimeSessions.findIndex((entry) => entry.threadId === session.threadId);    if (existingIndex >= 0) {      runtimeSessions[existingIndex] = session;      return;    }    runtimeSessions.push(session);  };   const normalizeLegacyEvent = (event: LegacyProviderRuntimeEvent): ProviderRuntimeEvent => {    if (isLegacyTurnCompletedEvent(event)) {      const normalized: Extract<ProviderRuntimeEvent, { type: "turn.completed" }> = {        ...(event as Omit<Extract<ProviderRuntimeEvent, { type: "turn.completed" }>, "payload">),        payload: {          state: event.status,          ...(typeof event.errorMessage === "string" ? { errorMessage: event.errorMessage } : {}),        },      };      return normalized;    }     return event as ProviderRuntimeEvent;  };   const emit = (event: LegacyProviderRuntimeEvent): void => {    Effect.runSync(PubSub.publish(runtimeEventPubSub, normalizeLegacyEvent(event)));  };   return {    service,    emit,    setSession,  };} type ProviderRuntimeTestReadModel = OrchestrationReadModel;type ProviderRuntimeTestThread = ProviderRuntimeTestReadModel["threads"][number];type ProviderRuntimeTestMessage = ProviderRuntimeTestThread["messages"][number];type ProviderRuntimeTestProposedPlan = ProviderRuntimeTestThread["proposedPlans"][number];type ProviderRuntimeTestActivity = ProviderRuntimeTestThread["activities"][number];type ProviderRuntimeTestCheckpoint = ProviderRuntimeTestThread["checkpoints"][number]; async function waitForThread(  readModel: () => Promise<ProviderRuntimeTestReadModel>,  predicate: (thread: ProviderRuntimeTestThread) => boolean,  timeoutMs = 2000,  threadId: ThreadId = asThreadId("thread-1"),) {  const deadline = (await Effect.runPromise(Clock.currentTimeMillis)) + timeoutMs;  const poll = async (): Promise<ProviderRuntimeTestThread> => {    const snapshot = await readModel();    const thread = snapshot.threads.find((entry) => entry.id === threadId);    if (thread && predicate(thread)) {      return thread;    }    if ((await Effect.runPromise(Clock.currentTimeMillis)) >= deadline) {      throw new Error("Timed out waiting for thread state");    }    await Effect.runPromise(Effect.yieldNow);    return poll();  };  return poll();} describe("ProviderRuntimeIngestion", () => {  let runtime: ManagedRuntime.ManagedRuntime<    OrchestrationEngineService | ProviderRuntimeIngestionService | ProjectionSnapshotQuery,    unknown  > | null = null;  let scope: Scope.Closeable | null = null;  const tempDirs: string[] = [];   function makeTempDir(prefix: string): string {    const dir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), prefix));    tempDirs.push(dir);    return dir;  }   afterEach(async () => {    if (scope) {      await Effect.runPromise(Scope.close(scope, Exit.void));    }    scope = null;    if (runtime) {      await runtime.dispose();    }    runtime = null;    for (const dir of tempDirs.splice(0)) {      NodeFS.rmSync(dir, { recursive: true, force: true });    }  });   async function createHarness(options?: {    serverSettings?: Partial<ServerSettings>;    threadTitle?: string;  }) {    const workspaceRoot = makeTempDir("t3-provider-project-");    NodeFS.mkdirSync(NodePath.join(workspaceRoot, ".git"));    const provider = createProviderServiceHarness();    const orchestrationLayer = OrchestrationEngineLive.pipe(      Layer.provide(OrchestrationProjectionSnapshotQueryLive),      Layer.provide(OrchestrationProjectionPipelineLive),      Layer.provide(OrchestrationEventStoreLive),      Layer.provide(OrchestrationCommandReceiptRepositoryLive),      Layer.provide(RepositoryIdentityResolver.layer),      Layer.provide(SqlitePersistenceMemory),    );    const projectionSnapshotLayer = OrchestrationProjectionSnapshotQueryLive.pipe(      Layer.provide(RepositoryIdentityResolver.layer),      Layer.provide(SqlitePersistenceMemory),    );    const layer = ProviderRuntimeIngestionLive.pipe(      Layer.provideMerge(orchestrationLayer),      Layer.provideMerge(projectionSnapshotLayer),      // Single shared liveness instance across ingestion (writer), the      // engine, and the snapshot query (reader).      Layer.provideMerge(ThreadBackgroundLiveness.layer),      Layer.provideMerge(ThreadPlanProgress.layer),      Layer.provideMerge(SqlitePersistenceMemory),      Layer.provideMerge(Layer.succeed(ProviderService, provider.service)),      Layer.provideMerge(makeTestServerSettingsLayer(options?.serverSettings)),      Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())),      Layer.provideMerge(NodeServices.layer),    );    runtime = ManagedRuntime.make(layer);    const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));    const snapshotQuery = await runtime.runPromise(Effect.service(ProjectionSnapshotQuery));    const ingestion = await runtime.runPromise(Effect.service(ProviderRuntimeIngestionService));    scope = await Effect.runPromise(Scope.make("sequential"));    await Effect.runPromise(ingestion.start().pipe(Scope.provide(scope)));    const drain = () => Effect.runPromise(ingestion.drain);    const dispatch = (command: OrchestrationCommand) => Effect.runPromise(engine.dispatch(command));     const createdAt = "2026-01-01T00:00:00.000Z";    await dispatch({      type: "project.create",      commandId: CommandId.make("cmd-provider-project-create"),      projectId: asProjectId("project-1"),      title: "Provider Project",      workspaceRoot,      defaultModelSelection: {        instanceId: ProviderInstanceId.make("codex"),        model: "gpt-5-codex",      },      createdAt,    });    await dispatch({      type: "thread.create",      commandId: CommandId.make("cmd-thread-create"),      threadId: ThreadId.make("thread-1"),      projectId: asProjectId("project-1"),      title: options?.threadTitle ?? "Thread",      modelSelection: {        instanceId: ProviderInstanceId.make("codex"),        model: "gpt-5-codex",      },      interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,      runtimeMode: "approval-required",      branch: null,      worktreePath: null,      createdAt,    });    await dispatch({      type: "thread.session.set",      commandId: CommandId.make("cmd-session-seed"),      threadId: ThreadId.make("thread-1"),      session: {        threadId: ThreadId.make("thread-1"),        status: "ready",        providerName: "codex",        runtimeMode: "approval-required",        activeTurnId: null,        updatedAt: createdAt,        lastError: null,      },      createdAt,    });    provider.setSession({      provider: ProviderDriverKind.make("codex"),      status: "ready",      runtimeMode: "approval-required",      threadId: ThreadId.make("thread-1"),      createdAt,      updatedAt: createdAt,    });     return {      engine,      dispatch,      readModel: () => Effect.runPromise(snapshotQuery.getSnapshot()),      emit: provider.emit,      setProviderSession: provider.setSession,      drain,    };  }   it("maps turn started/completed events into thread session updates", async () => {    const harness = await createHarness();    const now = "2026-01-01T00:00:00.000Z";     harness.emit({      type: "turn.started",      eventId: asEventId("evt-turn-started"),      provider: ProviderDriverKind.make("codex"),      threadId: asThreadId("thread-1"),      createdAt: now,      turnId: asTurnId("turn-1"),    });     await waitForThread(      harness.readModel,      (thread) => thread.session?.status === "running" && thread.session?.activeTurnId === "turn-1",    );     harness.emit({      type: "turn.completed",      eventId: asEventId("evt-turn-completed"),      provider: ProviderDriverKind.make("codex"),      threadId: asThreadId("thread-1"),      createdAt: "2026-01-01T00:00:00.000Z",      turnId: asTurnId("turn-1"),      payload: {        state: "failed",        errorMessage: "turn failed",      },    });     const thread = await waitForThread(      harness.readModel,      (entry) =>        entry.session?.status === "error" &&        entry.session?.activeTurnId === null &&        entry.session?.lastError === "turn failed",    );    expect(thread.session?.status).toBe("error");    expect(thread.session?.lastError).toBe("turn failed");  });   it("applies provider session.state.changed transitions directly", async () => {    const harness = await createHarness();    const waitingAt = "2026-01-01T00:00:00.000Z";     harness.emit({      type: "session.state.changed",      eventId: asEventId("evt-session-state-waiting"),      provider: ProviderDriverKind.make("codex"),      threadId: asThreadId("thread-1"),      createdAt: waitingAt,      payload: {        state: "waiting",        reason: "awaiting approval",      },    });     let thread = await waitForThread(      harness.readModel,      (entry) => entry.session?.status === "running" && entry.session?.activeTurnId === null,    );    expect(thread.session?.status).toBe("running");    expect(thread.session?.lastError).toBeNull();     harness.emit({      type: "session.state.changed",      eventId: asEventId("evt-session-state-error"),      provider: ProviderDriverKind.make("codex"),      threadId: asThreadId("thread-1"),      createdAt: "2026-01-01T00:00:00.000Z",      payload: {        state: "error",        reason: "provider crashed",      },    });     thread = await waitForThread(      harness.readModel,      (entry) =>        entry.session?.status === "error" &&        entry.session?.activeTurnId === null &&        entry.session?.lastError === "provider crashed",    );    expect(thread.session?.status).toBe("error");    expect(thread.session?.lastError).toBe("provider crashed");     harness.emit({      type: "session.state.changed",      eventId: asEventId("evt-session-state-stopped"),      provider: ProviderDriverKind.make("codex"),      threadId: asThreadId("thread-1"),      createdAt: "2026-01-01T00:00:00.000Z",      payload: {        state: "stopped",      },    });     thread = await waitForThread(      harness.readModel,      (entry) =>        entry.session?.status === "stopped" &&        entry.session?.activeTurnId === null &&        entry.session?.lastError === "provider crashed",    );    expect(thread.session?.status).toBe("stopped");    expect(thread.session?.lastError).toBe("provider crashed");     harness.emit({      type: "session.state.changed",      eventId: asEventId("evt-session-state-ready"),      provider: ProviderDriverKind.make("codex"),      threadId: asThreadId("thread-1"),      createdAt: "2026-01-01T00:00:00.000Z",      payload: {        state: "ready",      },    });     thread = await waitForThread(      harness.readModel,      (entry) =>        entry.session?.status === "ready" &&        entry.session?.activeTurnId === null &&        entry.session?.lastError === null,    );    expect(thread.session?.status).toBe("ready");    expect(thread.session?.lastError).toBeNull();  });   it("clears active turn when provider session becomes ready", async () => {    const harness = await createHarness();    const now = "2026-01-01T00:00:00.000Z";     harness.emit({      type: "turn.started",      eventId: asEventId("evt-turn-started-session-ready"),      provider: ProviderDriverKind.make("codex"),      createdAt: now,      threadId: asThreadId("thread-1"),      turnId: asTurnId("turn-session-ready"),    });     await waitForThread(      harness.readModel,      (thread) =>        thread.session?.status === "running" &&        thread.session?.activeTurnId === "turn-session-ready",      10_000,    );     harness.emit({      type: "session.state.changed",      eventId: asEventId("evt-session-state-ready-with-active-turn"),      provider: ProviderDriverKind.make("codex"),      threadId: asThreadId("thread-1"),      createdAt: "2026-01-01T00:00:01.000Z",      payload: {        state: "ready",      },    });     const thread = await waitForThread(      harness.readModel,      (entry) =>        entry.session?.status === "ready" &&        entry.session?.activeTurnId === null &&        entry.session?.lastError === null,      10_000,    );    expect(thread.session?.status).toBe("ready");

Lines 1–500 of 3627. Load more