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,  ProviderSessionStartInput,  ProviderSessionId,  ProviderThreadId,  ProviderTurnId,  ThreadId,} 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,  ProviderSessionNotFoundError,  ProviderValidationError,  ProviderUnsupportedError,  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 { NodeServices } from "@effect/platform-node";import { ProviderSessionRuntimeRepositoryLive } from "../../persistence/Layers/ProviderSessionRuntime.ts";import { ProviderSessionRuntimeRepository } from "../../persistence/Services/ProviderSessionRuntime.ts";import {  makeSqlitePersistenceLive,  SqlitePersistenceMemory,} from "../../persistence/Layers/Sqlite.ts"; const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value);const asTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value);const asRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.makeUnsafe(value);const asEventId = (value: string): EventId => EventId.makeUnsafe(value);const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); function makeFakeCodexAdapter() {  const sessions = new Map<ProviderSessionId, ProviderSession>();  const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());  let nextSession = 1;   const startSession = vi.fn((input: ProviderSessionStartInput) =>    Effect.sync(() => {      const now = new Date().toISOString();      const next = nextSession;      const session: ProviderSession = {        sessionId: ProviderSessionId.makeUnsafe(`sess-${next}`),        provider: "codex",        status: "ready",        threadId: input.resumeThreadId ?? ProviderThreadId.makeUnsafe(`thread-${next}`),        cwd: input.cwd ?? process.cwd(),        createdAt: now,        updatedAt: now,      };      nextSession += 1;      sessions.set(session.sessionId, session);      return session;    }),  );   const sendTurn = vi.fn(    (      input: ProviderSendTurnInput,    ): Effect.Effect<ProviderTurnStartResult, ProviderAdapterError> => {      if (!sessions.has(input.sessionId)) {        return Effect.fail(          new ProviderAdapterSessionNotFoundError({            provider: "codex",            sessionId: input.sessionId,          }),        );      }       return Effect.succeed({        threadId: ProviderThreadId.makeUnsafe("thread-1"),        turnId: ProviderTurnId.makeUnsafe("turn-1"),      });    },  );   const interruptTurn = vi.fn(    (_sessionId: string, _turnId?: string): Effect.Effect<void, ProviderAdapterError> =>      Effect.void,  );   const respondToRequest = vi.fn(    (      _sessionId: string,      _requestId: string,      _decision: ProviderApprovalDecision,    ): Effect.Effect<void, ProviderAdapterError> => Effect.void,  );   const stopSession = vi.fn(    (sessionId: string): Effect.Effect<void, ProviderAdapterError> =>      Effect.sync(() => {        sessions.delete(ProviderSessionId.makeUnsafe(sessionId));      }),  );   const listSessions = vi.fn(    (): Effect.Effect<ReadonlyArray<ProviderSession>> =>      Effect.sync(() => Array.from(sessions.values())),  );   const hasSession = vi.fn(    (sessionId: string): Effect.Effect<boolean> =>      Effect.succeed(sessions.has(ProviderSessionId.makeUnsafe(sessionId))),  );   const readThread = vi.fn(    (      _sessionId: ProviderSessionId,    ): Effect.Effect<      {        threadId: ProviderThreadId;        turns: ReadonlyArray<{ id: ProviderTurnId; items: readonly [] }>;      },      ProviderAdapterError    > =>      Effect.succeed({        threadId: ProviderThreadId.makeUnsafe("thread-1"),        turns: [{ id: asTurnId("turn-1"), items: [] }],      }),  );   const rollbackThread = vi.fn(    (      _sessionId: ProviderSessionId,      _numTurns: number,    ): Effect.Effect<{ threadId: ProviderThreadId; turns: readonly [] }, ProviderAdapterError> =>      Effect.succeed({ threadId: ProviderThreadId.makeUnsafe("thread-1"), turns: [] }),  );   const stopAll = vi.fn(    (): Effect.Effect<void, ProviderAdapterError> =>      Effect.sync(() => {        sessions.clear();      }),  );   const adapter: ProviderAdapterShape<ProviderAdapterError> = {    provider: "codex",    startSession,    sendTurn,    interruptTurn,    respondToRequest,    stopSession,    listSessions,    hasSession,    readThread,    rollbackThread,    stopAll,    streamEvents: Stream.fromPubSub(runtimeEventPubSub),  };   const emit = (event: ProviderRuntimeEvent): void => {    Effect.runSync(PubSub.publish(runtimeEventPubSub, event));  };   return {    adapter,    emit,    startSession,    sendTurn,    interruptTurn,    respondToRequest,    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),      ),      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({        sessionId: asSessionId("sess-stale"),        provider: "codex",        threadId: ThreadId.makeUnsafe("thread-stale"),      });    }).pipe(Effect.provide(directoryLayer));     const providerLayer = makeProviderServiceLive().pipe(      Layer.provide(Layer.succeed(ProviderAdapterRegistry, registry)),      Layer.provide(directoryLayer),    );     yield* Effect.gen(function* () {      yield* ProviderService;    }).pipe(Effect.provide(providerLayer));     const persistedProvider = yield* Effect.gen(function* () {      const directory = yield* ProviderSessionDirectory;      return yield* directory.getProvider(asSessionId("sess-stale"));    }).pipe(Effect.provide(directoryLayer));    assert.equal(persistedProvider, "codex");     const runtime = yield* Effect.gen(function* () {      const repository = yield* ProviderSessionRuntimeRepository;      return yield* repository.getBySessionId({ providerSessionId: asSessionId("sess-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),      );       const startedSession = yield* Effect.gen(function* () {        const provider = yield* ProviderService;        return yield* provider.startSession(          asThreadId("thread-1"),          {            provider: "codex",            cwd: "/tmp/project",          },        );      }).pipe(Effect.provide(firstProviderLayer));       yield* Effect.gen(function* () {        const provider = yield* ProviderService;        yield* provider.stopAll();      }).pipe(Effect.provide(firstProviderLayer));       const persistedAfterStopAll = yield* Effect.gen(function* () {        const repository = yield* ProviderSessionRuntimeRepository;        return yield* repository.getBySessionId({          providerSessionId: startedSession.sessionId,        });      }).pipe(Effect.provide(runtimeRepositoryLayer));      assert.equal(Option.isSome(persistedAfterStopAll), true);      if (Option.isSome(persistedAfterStopAll)) {        assert.equal(persistedAfterStopAll.value.status, "stopped");        assert.equal(persistedAfterStopAll.value.resumeCursor, null);      }       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),      );       secondCodex.startSession.mockClear();      secondCodex.rollbackThread.mockClear();       yield* Effect.gen(function* () {        const provider = yield* ProviderService;        yield* provider.rollbackConversation({          sessionId: startedSession.sessionId,          numTurns: 1,        });      }).pipe(Effect.provide(secondProviderLayer));       assert.deepEqual(secondCodex.startSession.mock.calls, [        [          {            provider: "codex",            resumeThreadId: 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",          cwd: "/tmp/project",        },      );      assert.equal(session.provider, "codex");       const sessions = yield* provider.listSessions();      assert.equal(sessions.length, 1);       yield* provider.sendTurn({        sessionId: session.sessionId,        input: "hello",        attachments: [],      });      assert.equal(routing.codex.sendTurn.mock.calls.length, 1);       yield* provider.interruptTurn({ sessionId: session.sessionId });      assert.deepEqual(routing.codex.interruptTurn.mock.calls, [[session.sessionId, undefined]]);       yield* provider.respondToRequest({        sessionId: session.sessionId,        requestId: asRequestId("req-1"),        decision: "accept",      });      assert.deepEqual(routing.codex.respondToRequest.mock.calls, [        [session.sessionId, asRequestId("req-1"), "accept"],      ]);       yield* provider.rollbackConversation({        sessionId: session.sessionId,        numTurns: 0,      });       yield* provider.stopSession({ sessionId: session.sessionId });      const sendAfterStop = yield* Effect.result(        provider.sendTurn({          sessionId: session.sessionId,          input: "after-stop",          attachments: [],        }),      );      assertFailure(        sendAfterStop,        new ProviderSessionNotFoundError({ sessionId: session.sessionId }),      );    }),  );   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"),        {          provider: "codex",          cwd: "/tmp/project",        },      );      yield* routing.codex.stopSession(initial.sessionId);      routing.codex.startSession.mockClear();      routing.codex.rollbackThread.mockClear();       yield* provider.rollbackConversation({        sessionId: initial.sessionId,        numTurns: 1,      });       assert.deepEqual(routing.codex.startSession.mock.calls, [        [          {            provider: "codex",            resumeThreadId: initial.threadId,          },        ],      ]);      assert.equal(routing.codex.rollbackThread.mock.calls.length, 1);      const rollbackCall = routing.codex.rollbackThread.mock.calls[0];      assert.equal(rollbackCall?.[1], 1);    }),  );   it.effect("stops all sessions and clears adapter runtime", () =>    Effect.gen(function* () {      const provider = yield* ProviderService;       yield* provider.startSession(        asThreadId("thread-1"),        {          provider: "codex",        },      );      yield* provider.startSession(        asThreadId("thread-2"),        {          provider: "codex",        },

Lines 1–500 of 747. Load more