import fs from "node:fs";import os from "node:os";import path from "node:path"; import type {  ProviderApprovalDecision,  ProviderRuntimeEvent,  ProviderSendTurnInput,  ProviderSession,  ProviderTurnStartResult,} from "@t3tools/contracts";import {  ApprovalRequestId,  EventId,  type ProviderKind,  ProviderSessionStartInput,  ThreadId,  TurnId,} from "@t3tools/contracts";import { it, assert, vi } from "@effect/vitest";import { assertFailure } from "@effect/vitest/utils"; import { Effect, Fiber, Layer, Option, PubSub, Ref, Stream } from "effect";import * as SqlClient from "effect/unstable/sql/SqlClient"; import {  ProviderAdapterSessionNotFoundError,  ProviderUnsupportedError,  ProviderValidationError,  type ProviderAdapterError,} from "../Errors.ts";import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts";import { ProviderAdapterRegistry } from "../Services/ProviderAdapterRegistry.ts";import { ProviderService } from "../Services/ProviderService.ts";import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts";import { makeProviderServiceLive } from "./ProviderService.ts";import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts";import * as NodeServices from "@effect/platform-node/NodeServices";import { ProviderSessionRuntimeRepositoryLive } from "../../persistence/Layers/ProviderSessionRuntime.ts";import { ProviderSessionRuntimeRepository } from "../../persistence/Services/ProviderSessionRuntime.ts";import {  makeSqlitePersistenceLive,  SqlitePersistenceMemory,} from "../../persistence/Layers/Sqlite.ts";import { AnalyticsService } from "../../telemetry/Services/AnalyticsService.ts"; const asRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.makeUnsafe(value);const asEventId = (value: string): EventId => EventId.makeUnsafe(value);const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value);const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); type LegacyProviderRuntimeEvent = {  readonly type: string;  readonly eventId: EventId;  readonly provider: "codex";  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: ProviderKind = "codex") {  const sessions = new Map<ThreadId, ProviderSession>();  const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());   const startSession = vi.fn((input: ProviderSessionStartInput) =>    Effect.sync(() => {      const now = new Date().toISOString();      const session: ProviderSession = {        provider,        status: "ready",        runtimeMode: input.runtimeMode,        threadId: input.threadId,        resumeCursor: input.resumeCursor ?? { opaque: `cursor-${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.makeUnsafe(`turn-${String(input.threadId)}`),      });    },  );   const interruptTurn = vi.fn(    (_threadId: ThreadId, _turnId?: TurnId): Effect.Effect<void, ProviderAdapterError> =>      Effect.void,  );   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 stopAll = vi.fn(    (): Effect.Effect<void, ProviderAdapterError> =>      Effect.sync(() => {        sessions.clear();      }),  );   const adapter: ProviderAdapterShape<ProviderAdapterError> = {    provider,    capabilities: {      sessionModelSwitch: "in-session",    },    startSession,    sendTurn,    interruptTurn,    respondToRequest,    respondToUserInput,    stopSession,    listSessions,    hasSession,    readThread,    rollbackThread,    stopAll,    streamEvents: Stream.fromPubSub(runtimeEventPubSub),  };   const emit = (event: LegacyProviderRuntimeEvent): void => {    Effect.runSync(PubSub.publish(runtimeEventPubSub, event as unknown as ProviderRuntimeEvent));  };   return {    adapter,    emit,    startSession,    sendTurn,    interruptTurn,    respondToRequest,    respondToUserInput,    stopSession,    listSessions,    hasSession,    readThread,    rollbackThread,    stopAll,  };} const sleep = (ms: number) =>  Effect.promise(() => new Promise<void>((resolve) => setTimeout(resolve, ms))); function makeProviderServiceLayer() {  const codex = makeFakeCodexAdapter();  const registry: typeof ProviderAdapterRegistry.Service = {    getByProvider: (provider) =>      provider === "codex"        ? Effect.succeed(codex.adapter)        : Effect.fail(new ProviderUnsupportedError({ provider })),    listProviders: () => Effect.succeed(["codex"]),  };   const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry);  const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe(    Layer.provide(SqlitePersistenceMemory),  );  const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer));   const layer = it.layer(    Layer.mergeAll(      makeProviderServiceLive().pipe(        Layer.provide(providerAdapterLayer),        Layer.provide(directoryLayer),        Layer.provideMerge(AnalyticsService.layerTest),      ),      directoryLayer,       runtimeRepositoryLayer,      NodeServices.layer,    ),  );   return {    codex,    layer,  };} const routing = makeProviderServiceLayer();it.effect("ProviderServiceLive keeps persisted resumable sessions on startup", () =>  Effect.gen(function* () {    const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-service-"));    const dbPath = path.join(tempDir, "orchestration.sqlite");     const codex = makeFakeCodexAdapter();    const registry: typeof ProviderAdapterRegistry.Service = {      getByProvider: (provider) =>        provider === "codex"          ? Effect.succeed(codex.adapter)          : Effect.fail(new ProviderUnsupportedError({ provider })),      listProviders: () => Effect.succeed(["codex"]),    };     const persistenceLayer = makeSqlitePersistenceLive(dbPath);    const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe(      Layer.provide(persistenceLayer),    );    const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer));     yield* Effect.gen(function* () {      const directory = yield* ProviderSessionDirectory;      yield* directory.upsert({        provider: "codex",        threadId: ThreadId.makeUnsafe("thread-stale"),      });    }).pipe(Effect.provide(directoryLayer));     const providerLayer = makeProviderServiceLive().pipe(      Layer.provide(Layer.succeed(ProviderAdapterRegistry, registry)),      Layer.provide(directoryLayer),      Layer.provide(AnalyticsService.layerTest),    );     yield* Effect.gen(function* () {      yield* ProviderService;    }).pipe(Effect.provide(providerLayer));     const persistedProvider = yield* Effect.gen(function* () {      const directory = yield* ProviderSessionDirectory;      return yield* directory.getProvider(asThreadId("thread-stale"));    }).pipe(Effect.provide(directoryLayer));    assert.equal(persistedProvider, "codex");     const runtime = yield* Effect.gen(function* () {      const repository = yield* ProviderSessionRuntimeRepository;      return yield* repository.getByThreadId({ threadId: asThreadId("thread-stale") });    }).pipe(Effect.provide(runtimeRepositoryLayer));    assert.equal(Option.isSome(runtime), true);     const legacyTableRows = yield* Effect.gen(function* () {      const sql = yield* SqlClient.SqlClient;      return yield* sql<{ readonly name: string }>`        SELECT name        FROM sqlite_master        WHERE type = 'table' AND name = 'provider_sessions'      `;    }).pipe(Effect.provide(persistenceLayer));    assert.equal(legacyTableRows.length, 0);     fs.rmSync(tempDir, { recursive: true, force: true });  }).pipe(Effect.provide(NodeServices.layer)),); it.effect(  "ProviderServiceLive restores rollback routing after restart using persisted thread mapping",  () =>    Effect.gen(function* () {      const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-service-restart-"));      const dbPath = path.join(tempDir, "orchestration.sqlite");      const persistenceLayer = makeSqlitePersistenceLive(dbPath);      const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe(        Layer.provide(persistenceLayer),      );       const firstCodex = makeFakeCodexAdapter();      const firstRegistry: typeof ProviderAdapterRegistry.Service = {        getByProvider: (provider) =>          provider === "codex"            ? Effect.succeed(firstCodex.adapter)            : Effect.fail(new ProviderUnsupportedError({ provider })),        listProviders: () => Effect.succeed(["codex"]),      };       const firstDirectoryLayer = ProviderSessionDirectoryLive.pipe(        Layer.provide(runtimeRepositoryLayer),      );      const firstProviderLayer = makeProviderServiceLive().pipe(        Layer.provide(Layer.succeed(ProviderAdapterRegistry, firstRegistry)),        Layer.provide(firstDirectoryLayer),        Layer.provide(AnalyticsService.layerTest),      );       const startedSession = yield* Effect.gen(function* () {        const provider = yield* ProviderService;        const threadId = asThreadId("thread-1");        return yield* provider.startSession(threadId, {          provider: "codex",          cwd: "/tmp/project",          runtimeMode: "full-access",          threadId,        });      }).pipe(Effect.provide(firstProviderLayer));       const persistedAfterStopAll = yield* Effect.gen(function* () {        const repository = yield* ProviderSessionRuntimeRepository;        return yield* repository.getByThreadId({ threadId: startedSession.threadId });      }).pipe(Effect.provide(runtimeRepositoryLayer));      assert.equal(Option.isSome(persistedAfterStopAll), true);      if (Option.isSome(persistedAfterStopAll)) {        assert.equal(persistedAfterStopAll.value.status, "stopped");        assert.deepEqual(persistedAfterStopAll.value.resumeCursor, startedSession.resumeCursor);      }       const secondCodex = makeFakeCodexAdapter();      const secondRegistry: typeof ProviderAdapterRegistry.Service = {        getByProvider: (provider) =>          provider === "codex"            ? Effect.succeed(secondCodex.adapter)            : Effect.fail(new ProviderUnsupportedError({ provider })),        listProviders: () => Effect.succeed(["codex"]),      };      const secondDirectoryLayer = ProviderSessionDirectoryLive.pipe(        Layer.provide(runtimeRepositoryLayer),      );      const secondProviderLayer = makeProviderServiceLive().pipe(        Layer.provide(Layer.succeed(ProviderAdapterRegistry, secondRegistry)),        Layer.provide(secondDirectoryLayer),        Layer.provide(AnalyticsService.layerTest),      );       secondCodex.startSession.mockClear();      secondCodex.rollbackThread.mockClear();       yield* Effect.gen(function* () {        const provider = yield* ProviderService;        yield* provider.rollbackConversation({          threadId: startedSession.threadId,          numTurns: 1,        });      }).pipe(Effect.provide(secondProviderLayer));       assert.equal(secondCodex.startSession.mock.calls.length, 1);      const resumedStartInput = secondCodex.startSession.mock.calls[0]?.[0];      assert.equal(typeof resumedStartInput === "object" && resumedStartInput !== null, true);      if (resumedStartInput && typeof resumedStartInput === "object") {        const startPayload = resumedStartInput as {          provider?: string;          cwd?: string;          resumeCursor?: unknown;          threadId?: string;        };        assert.equal(startPayload.provider, "codex");        assert.equal(startPayload.cwd, "/tmp/project");        assert.deepEqual(startPayload.resumeCursor, startedSession.resumeCursor);        assert.equal(startPayload.threadId, startedSession.threadId);      }      assert.equal(secondCodex.rollbackThread.mock.calls.length, 1);      const rollbackCall = secondCodex.rollbackThread.mock.calls[0];      assert.equal(typeof rollbackCall?.[0], "string");      assert.equal(rollbackCall?.[1], 1);       fs.rmSync(tempDir, { recursive: true, force: true });    }).pipe(Effect.provide(NodeServices.layer)),); routing.layer("ProviderServiceLive routing", (it) => {  it.effect("routes provider operations and rollback conversation", () =>    Effect.gen(function* () {      const provider = yield* ProviderService;       const session = yield* provider.startSession(asThreadId("thread-1"), {        provider: "codex",        threadId: asThreadId("thread-1"),        cwd: "/tmp/project",        runtimeMode: "full-access",      });      assert.equal(session.provider, "codex");       const sessions = yield* provider.listSessions();      assert.equal(sessions.length, 1);       yield* provider.sendTurn({        threadId: session.threadId,        input: "hello",        attachments: [],      });      assert.equal(routing.codex.sendTurn.mock.calls.length, 1);       yield* provider.interruptTurn({ threadId: session.threadId });      assert.deepEqual(routing.codex.interruptTurn.mock.calls, [[session.threadId, undefined]]);       yield* provider.respondToRequest({        threadId: session.threadId,        requestId: asRequestId("req-1"),        decision: "accept",      });      assert.deepEqual(routing.codex.respondToRequest.mock.calls, [        [session.threadId, asRequestId("req-1"), "accept"],      ]);       yield* provider.respondToUserInput({        threadId: session.threadId,        requestId: asRequestId("req-user-input-1"),        answers: {          sandbox_mode: "workspace-write",        },      });      assert.deepEqual(routing.codex.respondToUserInput.mock.calls, [        [          session.threadId,          asRequestId("req-user-input-1"),          {            sandbox_mode: "workspace-write",          },        ],      ]);       yield* provider.rollbackConversation({        threadId: session.threadId,        numTurns: 0,      });       yield* provider.stopSession({ threadId: session.threadId });      const sendAfterStop = yield* Effect.result(        provider.sendTurn({          threadId: session.threadId,          input: "after-stop",          attachments: [],        }),      );      assertFailure(        sendAfterStop,        new ProviderValidationError({          operation: "ProviderService.sendTurn",          issue: `Cannot route thread '${session.threadId}' because no persisted provider binding exists.`,        }),      );    }),  );   it.effect("recovers stale persisted sessions for rollback by resuming thread identity", () =>    Effect.gen(function* () {      const provider = yield* ProviderService;       const initial = yield* provider.startSession(asThreadId("thread-1"), {

Lines 1–500 of 874. Load more