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 * 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"; const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value);const asTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value);const asProviderThreadId = (value: string): ProviderThreadId => ProviderThreadId.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", approvalPolicy: "never", sandboxMode: "danger-full-access", }); }).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", cwd: "/tmp/project", resumeThreadId: startedSession.threadId, approvalPolicy: "never", sandboxMode: "danger-full-access", }, ], ]); 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", approvalPolicy: "never", sandboxMode: "danger-full-access", }); 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", cwd: "/tmp/project", resumeThreadId: initial.threadId, approvalPolicy: "never", sandboxMode: "danger-full-access", }, ], ]); assert.equal(routing.codex.rollbackThread.mock.calls.length, 1); const rollbackCall = routing.codex.rollbackThread.mock.calls[0]; assert.equal(rollbackCall?.[1], 1); }), ); it.effect("recovers stale sessions for sendTurn using persisted cwd", () => Effect.gen(function* () { const provider = yield* ProviderService; const initial = yield* provider.startSession(asThreadId("thread-1"), { provider: "codex", cwd: "/tmp/project-send-turn", approvalPolicy: "never", sandboxMode: "danger-full-access", }); yield* provider.stopAll(); routing.codex.startSession.mockClear();Lines 1–500 of 764. Load more
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 * 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"; const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); const asTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value); const asProviderThreadId = (value: string): ProviderThreadId => ProviderThreadId.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", approvalPolicy: "never", sandboxMode: "danger-full-access", }); }).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", cwd: "/tmp/project", resumeThreadId: startedSession.threadId, approvalPolicy: "never", sandboxMode: "danger-full-access", }, ], ]); 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", approvalPolicy: "never", sandboxMode: "danger-full-access", }); 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", cwd: "/tmp/project", resumeThreadId: initial.threadId, approvalPolicy: "never", sandboxMode: "danger-full-access", }, ], ]); assert.equal(routing.codex.rollbackThread.mock.calls.length, 1); const rollbackCall = routing.codex.rollbackThread.mock.calls[0]; assert.equal(rollbackCall?.[1], 1); }), ); it.effect("recovers stale sessions for sendTurn using persisted cwd", () => Effect.gen(function* () { const provider = yield* ProviderService; const initial = yield* provider.startSession(asThreadId("thread-1"), { provider: "codex", cwd: "/tmp/project-send-turn", approvalPolicy: "never", sandboxMode: "danger-full-access", }); yield* provider.stopAll(); routing.codex.startSession.mockClear();