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 { createModelSelection } from "@t3tools/shared/model";import { it, assert, vi } from "@effect/vitest"; import { Effect, Fiber, Layer, Metric, 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 { ServerSettingsService } from "../../serverSettings.ts";import { AnalyticsService } from "../../telemetry/Services/AnalyticsService.ts"; const defaultServerSettingsLayer = ServerSettingsService.layerTest(); const asRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.make(value);const asEventId = (value: string): EventId => EventId.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: ProviderKind; 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: `resume-${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.make(`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, get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); }, }; const emit = (event: LegacyProviderRuntimeEvent): void => { Effect.runSync(PubSub.publish(runtimeEventPubSub, event as unknown as ProviderRuntimeEvent)); }; const updateSession = ( threadId: ThreadId, update: (session: ProviderSession) => ProviderSession, ): void => { const existing = sessions.get(threadId); if (!existing) { return; } sessions.set(threadId, update(existing)); }; return { adapter, emit, updateSession, startSession, sendTurn, interruptTurn, respondToRequest, respondToUserInput, stopSession, listSessions, hasSession, readThread, rollbackThread, stopAll, };} const sleep = (ms: number) => Effect.promise(() => new Promise<void>((resolve) => setTimeout(resolve, ms))); const hasMetricSnapshot = ( snapshots: ReadonlyArray<Metric.Metric.Snapshot>, id: string, attributes: Readonly<Record<string, string>>,) => snapshots.some( (snapshot) => snapshot.id === id && Object.entries(attributes).every(([key, value]) => snapshot.attributes?.[key] === value), ); function makeProviderServiceLayer() { const codex = makeFakeCodexAdapter(); const claude = makeFakeCodexAdapter("claudeAgent"); const cursor = makeFakeCodexAdapter("cursor"); const registry: typeof ProviderAdapterRegistry.Service = { getByProvider: (provider) => provider === "codex" ? Effect.succeed(codex.adapter) : provider === "claudeAgent" ? Effect.succeed(claude.adapter) : provider === "cursor" ? Effect.succeed(cursor.adapter) : Effect.fail(new ProviderUnsupportedError({ provider })), listProviders: () => Effect.succeed(["codex", "claudeAgent", "cursor"]), }; 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.provide(defaultServerSettingsLayer), Layer.provideMerge(AnalyticsService.layerTest), ), directoryLayer, runtimeRepositoryLayer, NodeServices.layer, ), ); return { codex, claude, cursor, layer, };} it.effect("ProviderServiceLive rejects new sessions for disabled providers", () => Effect.gen(function* () { const codex = makeFakeCodexAdapter(); const claude = makeFakeCodexAdapter("claudeAgent"); const registry: typeof ProviderAdapterRegistry.Service = { getByProvider: (provider) => provider === "codex" ? Effect.succeed(codex.adapter) : provider === "claudeAgent" ? Effect.succeed(claude.adapter) : Effect.fail(new ProviderUnsupportedError({ provider })), listProviders: () => Effect.succeed(["codex", "claudeAgent"]), }; const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry); const serverSettingsLayer = ServerSettingsService.layerTest({ providers: { claudeAgent: { enabled: false, }, }, }); const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe( Layer.provide(SqlitePersistenceMemory), ); const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer)); const providerLayer = makeProviderServiceLive().pipe( Layer.provide(providerAdapterLayer), Layer.provide(directoryLayer), Layer.provide(serverSettingsLayer), Layer.provide(AnalyticsService.layerTest), ); const failure = yield* Effect.flip( Effect.gen(function* () { const provider = yield* ProviderService; return yield* provider.startSession(asThreadId("thread-disabled"), { provider: "claudeAgent", threadId: asThreadId("thread-disabled"), runtimeMode: "full-access", }); }).pipe(Effect.provide(providerLayer)), ); assert.instanceOf(failure, ProviderValidationError); assert.include(failure.issue, "Provider 'claudeAgent' is disabled in T3 Code settings."); assert.equal(claude.startSession.mock.calls.length, 0); }).pipe(Effect.provide(NodeServices.layer)),); const routing = makeProviderServiceLayer(); it.effect("ProviderServiceLive writes canonical events to the emitting thread segment", () => Effect.gen(function* () { const codex = makeFakeCodexAdapter(); const canonicalEvents: ProviderRuntimeEvent[] = []; const canonicalThreadIds: Array<string | null> = []; const registry: typeof ProviderAdapterRegistry.Service = { getByProvider: (provider) => provider === "codex" ? Effect.succeed(codex.adapter) : Effect.fail(new ProviderUnsupportedError({ provider })), listProviders: () => Effect.succeed(["codex"]), }; const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe( Layer.provide(SqlitePersistenceMemory), ); const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer)); const providerLayer = makeProviderServiceLive({ canonicalEventLogger: { filePath: "memory://provider-canonical-events", write: (event, threadId) => { canonicalEvents.push(event as ProviderRuntimeEvent); canonicalThreadIds.push(threadId ?? null); return Effect.void; }, close: () => Effect.void, }, }).pipe( Layer.provide(Layer.succeed(ProviderAdapterRegistry, registry)), Layer.provide(directoryLayer), Layer.provide(defaultServerSettingsLayer), Layer.provide(AnalyticsService.layerTest), ); yield* Effect.gen(function* () { yield* ProviderService; yield* sleep(10); codex.emit({ eventId: asEventId("evt-canonical-thread-segment"), provider: "codex", threadId: asThreadId("thread-canonical-thread-segment"), createdAt: new Date().toISOString(), type: "turn.completed", payload: { state: "completed", }, }); yield* sleep(20); }).pipe(Effect.provide(providerLayer)); assert.equal(canonicalEvents.length, 1); assert.equal(canonicalEvents[0]?.threadId, "thread-canonical-thread-segment"); assert.deepEqual(canonicalThreadIds, ["thread-canonical-thread-segment"]); }).pipe(Effect.provide(NodeServices.layer)),); 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.make("thread-stale"), }); }).pipe(Effect.provide(directoryLayer)); const providerLayer = makeProviderServiceLive().pipe( Layer.provide(Layer.succeed(ProviderAdapterRegistry, registry)), Layer.provide(directoryLayer), Layer.provide(defaultServerSettingsLayer), 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(defaultServerSettingsLayer), Layer.provide(AnalyticsService.layerTest), ); const updatedResumeCursor = { threadId: asThreadId("thread-1"), resume: "resume-session-1", resumeSessionAt: "assistant-message-1", turnCount: 1, }; const startedSession = yield* Effect.gen(function* () { const provider = yield* ProviderService; const threadId = asThreadId("thread-1");Lines 1–500 of 1498. 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, type ProviderKind, ProviderSessionStartInput, ThreadId, TurnId, } from "@t3tools/contracts"; import { createModelSelection } from "@t3tools/shared/model"; import { it, assert, vi } from "@effect/vitest"; import { Effect, Fiber, Layer, Metric, 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 { ServerSettingsService } from "../../serverSettings.ts"; import { AnalyticsService } from "../../telemetry/Services/AnalyticsService.ts"; const defaultServerSettingsLayer = ServerSettingsService.layerTest(); const asRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.make(value); const asEventId = (value: string): EventId => EventId.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: ProviderKind; 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: `resume-${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.make(`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, get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); }, }; const emit = (event: LegacyProviderRuntimeEvent): void => { Effect.runSync(PubSub.publish(runtimeEventPubSub, event as unknown as ProviderRuntimeEvent)); }; const updateSession = ( threadId: ThreadId, update: (session: ProviderSession) => ProviderSession, ): void => { const existing = sessions.get(threadId); if (!existing) { return; } sessions.set(threadId, update(existing)); }; return { adapter, emit, updateSession, startSession, sendTurn, interruptTurn, respondToRequest, respondToUserInput, stopSession, listSessions, hasSession, readThread, rollbackThread, stopAll, }; } const sleep = (ms: number) => Effect.promise(() => new Promise<void>((resolve) => setTimeout(resolve, ms))); const hasMetricSnapshot = ( snapshots: ReadonlyArray<Metric.Metric.Snapshot>, id: string, attributes: Readonly<Record<string, string>>, ) => snapshots.some( (snapshot) => snapshot.id === id && Object.entries(attributes).every(([key, value]) => snapshot.attributes?.[key] === value), ); function makeProviderServiceLayer() { const codex = makeFakeCodexAdapter(); const claude = makeFakeCodexAdapter("claudeAgent"); const cursor = makeFakeCodexAdapter("cursor"); const registry: typeof ProviderAdapterRegistry.Service = { getByProvider: (provider) => provider === "codex" ? Effect.succeed(codex.adapter) : provider === "claudeAgent" ? Effect.succeed(claude.adapter) : provider === "cursor" ? Effect.succeed(cursor.adapter) : Effect.fail(new ProviderUnsupportedError({ provider })), listProviders: () => Effect.succeed(["codex", "claudeAgent", "cursor"]), }; 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.provide(defaultServerSettingsLayer), Layer.provideMerge(AnalyticsService.layerTest), ), directoryLayer, runtimeRepositoryLayer, NodeServices.layer, ), ); return { codex, claude, cursor, layer, }; } it.effect("ProviderServiceLive rejects new sessions for disabled providers", () => Effect.gen(function* () { const codex = makeFakeCodexAdapter(); const claude = makeFakeCodexAdapter("claudeAgent"); const registry: typeof ProviderAdapterRegistry.Service = { getByProvider: (provider) => provider === "codex" ? Effect.succeed(codex.adapter) : provider === "claudeAgent" ? Effect.succeed(claude.adapter) : Effect.fail(new ProviderUnsupportedError({ provider })), listProviders: () => Effect.succeed(["codex", "claudeAgent"]), }; const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry); const serverSettingsLayer = ServerSettingsService.layerTest({ providers: { claudeAgent: { enabled: false, }, }, }); const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe( Layer.provide(SqlitePersistenceMemory), ); const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer)); const providerLayer = makeProviderServiceLive().pipe( Layer.provide(providerAdapterLayer), Layer.provide(directoryLayer), Layer.provide(serverSettingsLayer), Layer.provide(AnalyticsService.layerTest), ); const failure = yield* Effect.flip( Effect.gen(function* () { const provider = yield* ProviderService; return yield* provider.startSession(asThreadId("thread-disabled"), { provider: "claudeAgent", threadId: asThreadId("thread-disabled"), runtimeMode: "full-access", }); }).pipe(Effect.provide(providerLayer)), ); assert.instanceOf(failure, ProviderValidationError); assert.include(failure.issue, "Provider 'claudeAgent' is disabled in T3 Code settings."); assert.equal(claude.startSession.mock.calls.length, 0); }).pipe(Effect.provide(NodeServices.layer)), ); const routing = makeProviderServiceLayer(); it.effect("ProviderServiceLive writes canonical events to the emitting thread segment", () => Effect.gen(function* () { const codex = makeFakeCodexAdapter(); const canonicalEvents: ProviderRuntimeEvent[] = []; const canonicalThreadIds: Array<string | null> = []; const registry: typeof ProviderAdapterRegistry.Service = { getByProvider: (provider) => provider === "codex" ? Effect.succeed(codex.adapter) : Effect.fail(new ProviderUnsupportedError({ provider })), listProviders: () => Effect.succeed(["codex"]), }; const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe( Layer.provide(SqlitePersistenceMemory), ); const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer)); const providerLayer = makeProviderServiceLive({ canonicalEventLogger: { filePath: "memory://provider-canonical-events", write: (event, threadId) => { canonicalEvents.push(event as ProviderRuntimeEvent); canonicalThreadIds.push(threadId ?? null); return Effect.void; }, close: () => Effect.void, }, }).pipe( Layer.provide(Layer.succeed(ProviderAdapterRegistry, registry)), Layer.provide(directoryLayer), Layer.provide(defaultServerSettingsLayer), Layer.provide(AnalyticsService.layerTest), ); yield* Effect.gen(function* () { yield* ProviderService; yield* sleep(10); codex.emit({ eventId: asEventId("evt-canonical-thread-segment"), provider: "codex", threadId: asThreadId("thread-canonical-thread-segment"), createdAt: new Date().toISOString(), type: "turn.completed", payload: { state: "completed", }, }); yield* sleep(20); }).pipe(Effect.provide(providerLayer)); assert.equal(canonicalEvents.length, 1); assert.equal(canonicalEvents[0]?.threadId, "thread-canonical-thread-segment"); assert.deepEqual(canonicalThreadIds, ["thread-canonical-thread-segment"]); }).pipe(Effect.provide(NodeServices.layer)), ); 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.make("thread-stale"), }); }).pipe(Effect.provide(directoryLayer)); const providerLayer = makeProviderServiceLive().pipe( Layer.provide(Layer.succeed(ProviderAdapterRegistry, registry)), Layer.provide(directoryLayer), Layer.provide(defaultServerSettingsLayer), 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(defaultServerSettingsLayer), Layer.provide(AnalyticsService.layerTest), ); const updatedResumeCursor = { threadId: asThreadId("thread-1"), resume: "resume-session-1", resumeSessionAt: "assistant-message-1", turnCount: 1, }; const startedSession = yield* Effect.gen(function* () { const provider = yield* ProviderService; const threadId = asThreadId("thread-1");