// @effect-diagnostics nodeBuiltinImport:offimport * as NodeFS from "node:fs";import * as NodeOS from "node:os";import * as NodePath from "node:path";import * as NodeChildProcess from "node:child_process"; import {  OrchestrationReadModel,  ProviderDriverKind,  ProviderRuntimeEvent,  ProviderSession,  ProviderInstanceId,} from "@t3tools/contracts";import {  ApprovalRequestId,  CommandId,  DEFAULT_PROVIDER_INTERACTION_MODE,  EventId,  MessageId,  type OrchestrationCommand,  ProjectId,  ProviderItemId,  RuntimeRequestId,  type ServerSettings,  ThreadId,  TurnId,} from "@t3tools/contracts";import * as Clock from "effect/Clock";import * as Deferred from "effect/Deferred";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 Option from "effect/Option";import * as PubSub from "effect/PubSub";import * as Scope from "effect/Scope";import * as Stream from "effect/Stream";import * as Tracer from "effect/Tracer";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 * as CheckpointStore from "../../checkpointing/CheckpointStore.ts";import * as VcsDriverRegistry from "../../vcs/VcsDriverRegistry.ts";import * as VcsProcess from "../../vcs/VcsProcess.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";import { makeSqlStatementCounter } from "../../../integration/SqlStatementCounter.integration.ts"; 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<{      readonly events: ReadonlyArray<ProviderRuntimeEvent>;      readonly enqueued?: Deferred.Deferred<void>;    }>(),  );  const runtimeSessions: ProviderSession[] = [];   const unsupported = () => Effect.die(new Error("Unsupported provider call in test")) as never;  const service: ProviderServiceShape = {    startSession: () => unsupported(),    sendTurn: () => unsupported(),    compactThread: () => unsupported(),    interruptTurn: () => unsupported(),    respondToRequest: () => unsupported(),    respondToUserInput: () => unsupported(),    stopSession: () => unsupported(),    listSessions: () => Effect.succeed([...runtimeSessions]),    getCapabilities: () => Effect.succeed({ sessionModelSwitch: "in-session" }),    assertConversationRollbackSupported: () => unsupported(),    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).pipe(        Stream.flatMap(({ events, enqueued }) =>          Stream.concat(            Stream.fromIterable(events),            enqueued              ? Stream.fromEffect(Deferred.succeed(enqueued, undefined)).pipe(Stream.drain)              : Stream.empty,          ),        ),      );    },  };   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, { events: [normalizeLegacyEvent(event)] }));  };   const emitAndWaitForEnqueue = Effect.fnUntraced(function* (    events: ReadonlyArray<LegacyProviderRuntimeEvent>,  ) {    const enqueued = yield* Deferred.make<void>();    yield* PubSub.publish(runtimeEventPubSub, {      events: events.map(normalizeLegacyEvent),      enqueued,    });    yield* Deferred.await(enqueued);  });   return {    service,    emit,    emitAndWaitForEnqueue,    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;    workspaceSubdirectory?: string;  }) {    const repositoryRoot = makeTempDir("t3-provider-project-");    NodeChildProcess.execFileSync("git", ["init", "--initial-branch=main"], {      cwd: repositoryRoot,      stdio: "ignore",    });    const workspaceRoot = NodePath.join(repositoryRoot, options?.workspaceSubdirectory ?? "");    NodeFS.mkdirSync(workspaceRoot, { recursive: true });    const provider = createProviderServiceHarness();    const sqlCounter = makeSqlStatementCounter();    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 ingestionProjectionSnapshotLayer = Layer.effect(      ProjectionSnapshotQuery,      Effect.gen(function* () {        const query = yield* ProjectionSnapshotQuery;        return ProjectionSnapshotQuery.of({          ...query,          getThreadDetailById: () =>            Effect.die("provider runtime ingestion must not hydrate thread detail"),        });      }),    ).pipe(Layer.provide(projectionSnapshotLayer));    const layer = ProviderRuntimeIngestionLive.pipe(      Layer.provideMerge(orchestrationLayer),      Layer.provideMerge(ingestionProjectionSnapshotLayer),      // 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(CheckpointStore.layer.pipe(Layer.provide(VcsDriverRegistry.layer))),      Layer.provideMerge(VcsProcess.layer),      Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())),      Layer.provideMerge(NodeServices.layer),      Layer.provideMerge(Layer.succeed(Tracer.Tracer, sqlCounter.tracer)),    );    const testRuntime = ManagedRuntime.make(layer);    runtime = testRuntime;    const engine = await testRuntime.runPromise(Effect.service(OrchestrationEngineService));    const snapshotQuery = await testRuntime.runPromise(Effect.service(ProjectionSnapshotQuery));    const ingestion = await testRuntime.runPromise(Effect.service(ProviderRuntimeIngestionService));    scope = await Effect.runPromise(Scope.make("sequential"));    await testRuntime.runPromise(ingestion.start().pipe(Scope.provide(scope)));    const drain = () => testRuntime.runPromise(ingestion.drain);    const dispatch = (command: OrchestrationCommand) =>      testRuntime.runPromise(engine.dispatch(command));    const emitAndDrain = (events: ReadonlyArray<LegacyProviderRuntimeEvent>) =>      testRuntime.runPromise(        provider.emitAndWaitForEnqueue(events).pipe(Effect.andThen(ingestion.drain)),      );     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: () => testRuntime.runPromise(snapshotQuery.getSnapshot()),      readThreadShell: () =>        testRuntime.runPromise(          snapshotQuery            .getThreadShellById(asThreadId("thread-1"))            .pipe(Effect.map(Option.getOrThrow)),        ),      emit: provider.emit,      emitAndDrain,      sqlCount: sqlCounter.count,      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.each([    { delivery: "buffered", enableLegacyTokenStreaming: false },    { delivery: "streamed", enableLegacyTokenStreaming: true },  ])("settles OpenCode aborted turns and saves $delivery assistant text", async (settings) => {    const harness = await createHarness({      serverSettings: { enableLegacyTokenStreaming: settings.enableLegacyTokenStreaming },    });    const threadId = asThreadId("thread-1");    const turnId = asTurnId("opencode-aborted-turn");    const base = {      provider: ProviderDriverKind.make("opencode"),      threadId,      turnId,      createdAt: "2026-01-01T00:00:01.000Z",    };    harness.emit({ ...base, type: "turn.started", eventId: asEventId("opencode-started") });    harness.emit({      ...base,      type: "content.delta",      eventId: asEventId("opencode-partial-text"),      itemId: asItemId("opencode-text-part"),      payload: { streamKind: "assistant_text", delta: "Work before the stop." },    });    harness.emit({      ...base,      type: "turn.aborted",      eventId: asEventId("opencode-aborted"),      createdAt: "2026-01-01T00:00:02.000Z",      payload: { reason: "Interrupted by user." },    });     await harness.drain();    const thread = (await harness.readModel()).threads.find((entry) => entry.id === threadId);    expect(thread?.session).toMatchObject({      status: "interrupted",      activeTurnId: null,      lastError: null,    });    expect(thread?.latestTurn).toMatchObject({      turnId,      state: "interrupted",      completedAt: "2026-01-01T00:00:02.000Z",    });    expect(thread?.messages).toEqual([      expect.objectContaining({        role: "assistant",        turnId,        text: "Work before the stop.",        streaming: false,      }),    ]);  });   it.each(["turn.completed", "turn.aborted"] as const)(    "finalizes old buffered text on late %s without stopping the newer turn",    async (terminalType) => {      const harness = await createHarness({

Lines 1–500 of 4369. Load more