diff --git a/apps/server/integration/providerService.integration.test.ts b/apps/server/integration/providerService.integration.test.ts index af4326bc6acd901f839668946cd6208799571d83..b32853277ae72eed9320c4ca55abca03ba5ded11 100644 --- a/apps/server/integration/providerService.integration.test.ts +++ b/apps/server/integration/providerService.integration.test.ts @@ -16,6 +16,7 @@ import { } from "../src/provider/Services/ProviderService.ts"; import { SqlitePersistenceMemory } from "../src/persistence/Layers/Sqlite.ts"; import { CheckpointRepositoryLive } from "../src/persistence/Layers/Checkpoints.ts"; +import { ProviderSessionRepositoryLive } from "../src/persistence/Layers/ProviderSessions.ts"; import { makeTestProviderAdapterHarness, @@ -77,12 +78,16 @@ const makeIntegrationFixture = Effect.gen(function* () { listProviders: () => Effect.succeed(["codex"]), }; + const directoryLayer = ProviderSessionDirectoryLive.pipe( + Layer.provide(ProviderSessionRepositoryLive), + ); + const shared = Layer.mergeAll( - ProviderSessionDirectoryLive, + directoryLayer, Layer.succeed(ProviderAdapterRegistry, registry), Layer.provide(CheckpointStoreLive, NodeServices.layer), - Layer.provide(CheckpointRepositoryLive, SqlitePersistenceMemory), - ); + CheckpointRepositoryLive, + ).pipe(Layer.provide(SqlitePersistenceMemory)); const checkpointLayer = CheckpointServiceLive.pipe(Layer.provide(shared)); diff --git a/apps/server/src/checkpointing/Layers/CheckpointService.test.ts b/apps/server/src/checkpointing/Layers/CheckpointService.test.ts index 7aa4f4ec0999302861787fab9dfcbca48f1d315b..983ecdd33b91e1c7f9a67683b073994cc98c9933 100644 --- a/apps/server/src/checkpointing/Layers/CheckpointService.test.ts +++ b/apps/server/src/checkpointing/Layers/CheckpointService.test.ts @@ -32,6 +32,7 @@ import { ProviderSessionDirectoryLive } from "../../provider/Layers/ProviderSess import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; import { CheckpointRepository } from "../../persistence/Services/Checkpoints.ts"; import { CheckpointRepositoryLive } from "../../persistence/Layers/Checkpoints.ts"; +import { ProviderSessionRepositoryLive } from "../../persistence/Layers/ProviderSessions.ts"; const runGit = (cwd: string, args: readonly string[]) => runGitProcess({ @@ -211,13 +212,17 @@ function makeFixture(turns: ReadonlyArray) { listProviders: () => Effect.succeed(["codex"]), }; + const directoryLayer = ProviderSessionDirectoryLive.pipe( + Layer.provide(ProviderSessionRepositoryLive), + ); + const dependencies = Layer.mergeAll( Layer.provide(CheckpointStoreLive, NodeServices.layer), - Layer.provide(CheckpointRepositoryLive, SqlitePersistenceMemory), + CheckpointRepositoryLive, Layer.succeed(ProviderAdapterRegistry, registry), - ProviderSessionDirectoryLive, + directoryLayer, NodeServices.layer, - ); + ).pipe(Layer.provide(SqlitePersistenceMemory)); const layer = CheckpointServiceLive.pipe(Layer.provideMerge(dependencies)); diff --git a/apps/server/src/checkpointing/Layers/CheckpointService.ts b/apps/server/src/checkpointing/Layers/CheckpointService.ts index c0841ddc2fd4f93754b753fd9ace00caa11cb55b..907e7df71034efa1648c7f035b6eedddb6e056e3 100644 --- a/apps/server/src/checkpointing/Layers/CheckpointService.ts +++ b/apps/server/src/checkpointing/Layers/CheckpointService.ts @@ -392,16 +392,40 @@ const makeCheckpointService = Effect.gen(function* () { Effect.map((current) => Option.fromNullishOr(current.get(sessionId))), ); + const mapDirectoryLookupError = ( + sessionId: string, + operation: string, + cause: unknown, + ): CheckpointSessionNotFoundError | CheckpointInvariantError => { + if ( + typeof cause === "object" && + cause !== null && + "_tag" in cause && + (cause as { _tag?: string })._tag === "ProviderSessionNotFoundError" + ) { + return new CheckpointSessionNotFoundError({ + sessionId, + cause, + }); + } + + return invariantError( + operation, + `Failed to resolve provider session metadata for session ${sessionId}.`, + cause, + ); + }; + const resolveAdapterForSession = ( sessionId: string, ): Effect.Effect, CheckpointServiceError> => directory.getProvider(sessionId).pipe( - Effect.mapError( - (cause) => - new CheckpointSessionNotFoundError({ - sessionId, - cause, - }), + Effect.mapError((cause) => + mapDirectoryLookupError( + sessionId, + "CheckpointService.resolveAdapterForSession:getProvider", + cause, + ), ), Effect.flatMap((provider) => registry @@ -456,12 +480,12 @@ const makeCheckpointService = Effect.gen(function* () { operation: string, ): Effect.Effect => directory.getThreadId(sessionId).pipe( - Effect.mapError( - (cause) => - new CheckpointSessionNotFoundError({ - sessionId, - cause, - }), + Effect.mapError((cause) => + mapDirectoryLookupError( + sessionId, + `${operation}:getThreadId`, + cause, + ), ), Effect.flatMap((threadId) => Option.match(threadId, { diff --git a/apps/server/src/persistence/Errors.ts b/apps/server/src/persistence/Errors.ts index a9073951453aaf70f639ada2bb5f2f431aaba8a9..a1c38e74c12a4567a4f89e2f4d059c02f5424873 100644 --- a/apps/server/src/persistence/Errors.ts +++ b/apps/server/src/persistence/Errors.ts @@ -150,6 +150,36 @@ export class CheckpointRepositoryPersistenceError extends Schema.TaggedErrorClas } } +// =============================== +// Provider Session Repository Errors +// =============================== + +export class ProviderSessionRepositoryValidationError extends Schema.TaggedErrorClass()( + "ProviderSessionRepositoryValidationError", + { + operation: Schema.String, + issue: Schema.String, + cause: Schema.optional(Schema.Defect), + }, +) { + override get message(): string { + return `Provider session repository validation failed in ${this.operation}: ${this.issue}`; + } +} + +export class ProviderSessionRepositoryPersistenceError extends Schema.TaggedErrorClass()( + "ProviderSessionRepositoryPersistenceError", + { + operation: Schema.String, + detail: Schema.String, + cause: Schema.optional(Schema.Defect), + }, +) { + override get message(): string { + return `Provider session repository persistence error in ${this.operation}: ${this.detail}`; + } +} + export type ProjectRepositoryError = | PersistenceSqlError | PersistenceDecodeError @@ -166,3 +196,7 @@ export type OrchestrationEventStoreError = export type CheckpointRepositoryError = | CheckpointRepositoryValidationError | CheckpointRepositoryPersistenceError; + +export type ProviderSessionRepositoryError = + | ProviderSessionRepositoryValidationError + | ProviderSessionRepositoryPersistenceError; diff --git a/apps/server/src/persistence/Layers/ProviderSessions.test.ts b/apps/server/src/persistence/Layers/ProviderSessions.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..60dbe0974c55bc01be2b63d37eb3fab3a3d36657 --- /dev/null +++ b/apps/server/src/persistence/Layers/ProviderSessions.test.ts @@ -0,0 +1,81 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import { NodeServices } from "@effect/platform-node"; +import { it, assert } from "@effect/vitest"; +import { Effect, Layer, Option } from "effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { makeSqlitePersistenceLive, SqlitePersistenceMemory } from "./Sqlite.ts"; +import { ProviderSessionRepository } from "../Services/ProviderSessions.ts"; +import { ProviderSessionRepositoryLive } from "./ProviderSessions.ts"; + +function makeRepositoryLayer( + persistenceLayer: Layer.Layer, +) { + return ProviderSessionRepositoryLive.pipe(Layer.provide(persistenceLayer)); +} + +it.effect("ProviderSessionRepositoryLive upserts, lists, gets, and deletes sessions", () => + Effect.gen(function* () { + const repository = yield* ProviderSessionRepository; + + yield* repository.upsertSession({ + sessionId: "sess-1", + provider: "codex", + threadId: "thread-1", + }); + yield* repository.upsertSession({ + sessionId: "sess-2", + provider: "claudeCode", + }); + + const all = yield* repository.listSessions(); + assert.equal(all.length, 2); + assert.equal(all[0]?.sessionId, "sess-1"); + assert.equal(all[1]?.sessionId, "sess-2"); + + const existing = yield* repository.getSession({ sessionId: "sess-1" }); + assert.equal(Option.isSome(existing), true); + if (Option.isSome(existing)) { + assert.equal(existing.value.provider, "codex"); + assert.equal(existing.value.threadId, "thread-1"); + } + + yield* repository.deleteSession({ sessionId: "sess-2" }); + const afterDelete = yield* repository.listSessions(); + assert.deepEqual(afterDelete.map((entry) => entry.sessionId), ["sess-1"]); + }).pipe(Effect.provide(makeRepositoryLayer(SqlitePersistenceMemory))), +); + +it.effect("ProviderSessionRepositoryLive persists sessions across layer restarts", () => + Effect.gen(function* () { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-sessions-")); + const dbPath = path.join(tempDir, "orchestration.sqlite"); + + const repositoryLayer = makeRepositoryLayer(makeSqlitePersistenceLive(dbPath)); + + yield* Effect.gen(function* () { + const repository = yield* ProviderSessionRepository; + yield* repository.upsertSession({ + sessionId: "sess-persisted", + provider: "codex", + threadId: "thread-persisted", + }); + }).pipe(Effect.provide(repositoryLayer)); + + const persisted = yield* Effect.gen(function* () { + const repository = yield* ProviderSessionRepository; + return yield* repository.getSession({ sessionId: "sess-persisted" }); + }).pipe(Effect.provide(repositoryLayer)); + + assert.equal(Option.isSome(persisted), true); + if (Option.isSome(persisted)) { + assert.equal(persisted.value.provider, "codex"); + assert.equal(persisted.value.threadId, "thread-persisted"); + } + + fs.rmSync(tempDir, { recursive: true, force: true }); + }).pipe(Effect.provide(NodeServices.layer)), +); diff --git a/apps/server/src/persistence/Layers/ProviderSessions.ts b/apps/server/src/persistence/Layers/ProviderSessions.ts new file mode 100644 index 0000000000000000000000000000000000000000..e1a1510a4d50d9105362e4f41eba1ea532b0d9b8 --- /dev/null +++ b/apps/server/src/persistence/Layers/ProviderSessions.ts @@ -0,0 +1,232 @@ +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as SqlSchema from "effect/unstable/sql/SqlSchema"; +import { Effect, Layer, Option, Schema } from "effect"; + +import { + ProviderSessionRepositoryPersistenceError, + ProviderSessionRepositoryValidationError, +} from "../Errors.ts"; +import { + ProviderSessionRepository, + type ProviderSessionEntry, + type ProviderSessionRepositoryShape, +} from "../Services/ProviderSessions.ts"; + +const NonEmptyString = Schema.NonEmptyString; + +const ProviderKindSchema = Schema.Literals(["codex", "claudeCode"]); + +const ProviderSessionRowSchema = Schema.Struct({ + sessionId: Schema.String, + provider: ProviderKindSchema, + threadId: Schema.NullOr(Schema.String), + createdAt: Schema.String, + updatedAt: Schema.String, +}); + +const SessionIdRequestSchema = Schema.Struct({ + sessionId: NonEmptyString, +}); + +const UpsertSessionRequestSchema = Schema.Struct({ + sessionId: NonEmptyString, + provider: ProviderKindSchema, + threadId: Schema.NullOr(NonEmptyString), +}); + +function errorMessage(cause: unknown, fallback: string): string { + if (cause instanceof Error && cause.message.length > 0) { + return cause.message; + } + return fallback; +} + +function toValidationError( + operation: string, + cause: unknown, +): ProviderSessionRepositoryValidationError { + return new ProviderSessionRepositoryValidationError({ + operation, + issue: errorMessage(cause, "Invalid provider session repository input."), + cause, + }); +} + +function decodeInput( + schema: Schema.Schema, + input: unknown, + operation: string, +): Effect.Effect { + return Schema.decodeUnknownEffect(schema)(input).pipe( + Effect.mapError((cause) => toValidationError(operation, cause)), + ) as Effect.Effect; +} + +function toPersistenceError( + operation: string, + cause: unknown, +): ProviderSessionRepositoryPersistenceError { + return new ProviderSessionRepositoryPersistenceError({ + operation, + detail: `Failed to execute ${operation}.`, + cause, + }); +} + +function toEntry( + row: Schema.Schema.Type, +): ProviderSessionEntry { + return { + sessionId: row.sessionId, + provider: row.provider, + ...(row.threadId !== null ? { threadId: row.threadId } : {}), + createdAt: row.createdAt, + updatedAt: row.updatedAt, + }; +} + +const makeProviderSessionRepository = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + const upsertSessionRow = SqlSchema.findOne({ + Request: UpsertSessionRequestSchema, + Result: ProviderSessionRowSchema, + execute: (request) => + sql` + INSERT INTO provider_sessions ( + session_id, + provider, + thread_id, + created_at, + updated_at + ) + VALUES ( + ${request.sessionId}, + ${request.provider}, + ${request.threadId}, + CURRENT_TIMESTAMP, + CURRENT_TIMESTAMP + ) + ON CONFLICT (session_id) + DO UPDATE SET + provider = excluded.provider, + thread_id = excluded.thread_id, + updated_at = CURRENT_TIMESTAMP + RETURNING + session_id AS "sessionId", + provider, + thread_id AS "threadId", + created_at AS "createdAt", + updated_at AS "updatedAt" + `, + }); + + const findSessionRow = SqlSchema.findOneOption({ + Request: SessionIdRequestSchema, + Result: ProviderSessionRowSchema, + execute: ({ sessionId }) => + sql` + SELECT + session_id AS "sessionId", + provider, + thread_id AS "threadId", + created_at AS "createdAt", + updated_at AS "updatedAt" + FROM provider_sessions + WHERE session_id = ${sessionId} + `, + }); + + const listSessionRows = SqlSchema.findAll({ + Request: Schema.Void, + Result: ProviderSessionRowSchema, + execute: () => + sql` + SELECT + session_id AS "sessionId", + provider, + thread_id AS "threadId", + created_at AS "createdAt", + updated_at AS "updatedAt" + FROM provider_sessions + ORDER BY created_at ASC, session_id ASC + `, + }); + + const deleteSessionRow = SqlSchema.void({ + Request: SessionIdRequestSchema, + execute: ({ sessionId }) => sql`DELETE FROM provider_sessions WHERE session_id = ${sessionId}`, + }); + + const upsertSession: ProviderSessionRepositoryShape["upsertSession"] = (input) => + Effect.gen(function* () { + const parsed = yield* decodeInput( + UpsertSessionRequestSchema, + { + sessionId: input.sessionId.trim(), + provider: input.provider, + threadId: input.threadId?.trim() || null, + }, + "ProviderSessionRepository.upsertSession", + ); + + yield* upsertSessionRow(parsed).pipe( + Effect.mapError((cause) => + toPersistenceError("ProviderSessionRepository.upsertSession:query", cause), + ), + Effect.asVoid, + ); + }); + + const getSession: ProviderSessionRepositoryShape["getSession"] = (input) => + Effect.gen(function* () { + const parsed = yield* decodeInput( + SessionIdRequestSchema, + { sessionId: input.sessionId.trim() }, + "ProviderSessionRepository.getSession", + ); + + const row = yield* findSessionRow(parsed).pipe( + Effect.mapError((cause) => + toPersistenceError("ProviderSessionRepository.getSession:query", cause), + ), + ); + + return Option.map(row, toEntry); + }); + + const listSessions: ProviderSessionRepositoryShape["listSessions"] = () => + listSessionRows(undefined).pipe( + Effect.mapError((cause) => + toPersistenceError("ProviderSessionRepository.listSessions:query", cause), + ), + Effect.map((rows) => rows.map(toEntry)), + ); + + const deleteSession: ProviderSessionRepositoryShape["deleteSession"] = (input) => + Effect.gen(function* () { + const parsed = yield* decodeInput( + SessionIdRequestSchema, + { sessionId: input.sessionId.trim() }, + "ProviderSessionRepository.deleteSession", + ); + + yield* deleteSessionRow(parsed).pipe( + Effect.mapError((cause) => + toPersistenceError("ProviderSessionRepository.deleteSession:query", cause), + ), + ); + }); + + return { + upsertSession, + getSession, + listSessions, + deleteSession, + } satisfies ProviderSessionRepositoryShape; +}); + +export const ProviderSessionRepositoryLive = Layer.effect( + ProviderSessionRepository, + makeProviderSessionRepository, +); diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index f8db8ee1be3957ad3f5a58af743b8a0938f9ad6e..05fb801eaa138c3dd883150bc3cfd5d97f7d02fa 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -15,6 +15,7 @@ import * as Layer from "effect/Layer"; import Migration0001 from "./Migrations/001_OrchestrationEvents.ts"; import Migration0002 from "./Migrations/002_Projects.ts"; import Migration0003 from "./Migrations/003_Checkpoints.ts"; +import Migration0004 from "./Migrations/004_ProviderSessions.ts"; /** * Migration loader with all migrations defined inline. @@ -30,6 +31,7 @@ const loader = Migrator.fromRecord({ "1_OrchestrationEvents": Migration0001, "2_Projects": Migration0002, "3_CheckpointCatalog": Migration0003, + "4_ProviderSessions": Migration0004, }); /** diff --git a/apps/server/src/persistence/Migrations/004_ProviderSessions.ts b/apps/server/src/persistence/Migrations/004_ProviderSessions.ts new file mode 100644 index 0000000000000000000000000000000000000000..72e308eba9af4a769505f996ff48f3571a7d3f74 --- /dev/null +++ b/apps/server/src/persistence/Migrations/004_ProviderSessions.ts @@ -0,0 +1,21 @@ +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as Effect from "effect/Effect"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE TABLE IF NOT EXISTS provider_sessions ( + session_id TEXT PRIMARY KEY, + provider TEXT NOT NULL, + thread_id TEXT, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP + ) + `; + + yield* sql` + CREATE INDEX IF NOT EXISTS idx_provider_sessions_provider + ON provider_sessions(provider) + `; +}); diff --git a/apps/server/src/persistence/Services/ProviderSessions.ts b/apps/server/src/persistence/Services/ProviderSessions.ts new file mode 100644 index 0000000000000000000000000000000000000000..673d343a9f7421b6509a447df8e1b89c40ad6c43 --- /dev/null +++ b/apps/server/src/persistence/Services/ProviderSessions.ts @@ -0,0 +1,51 @@ +import type { ProviderKind } from "@t3tools/contracts"; +import { Option, ServiceMap } from "effect"; +import type { Effect } from "effect"; + +import type { ProviderSessionRepositoryError } from "../Errors.ts"; + +export interface ProviderSessionEntry { + readonly sessionId: string; + readonly provider: ProviderKind; + readonly threadId?: string; + readonly createdAt: string; + readonly updatedAt: string; +} + +export interface UpsertProviderSessionInput { + readonly sessionId: string; + readonly provider: ProviderKind; + readonly threadId?: string; +} + +export interface DeleteProviderSessionInput { + readonly sessionId: string; +} + +export interface GetProviderSessionInput { + readonly sessionId: string; +} + +export interface ProviderSessionRepositoryShape { + readonly upsertSession: ( + input: UpsertProviderSessionInput, + ) => Effect.Effect; + + readonly getSession: ( + input: GetProviderSessionInput, + ) => Effect.Effect, ProviderSessionRepositoryError>; + + readonly listSessions: () => Effect.Effect< + ReadonlyArray, + ProviderSessionRepositoryError + >; + + readonly deleteSession: ( + input: DeleteProviderSessionInput, + ) => Effect.Effect; +} + +export class ProviderSessionRepository extends ServiceMap.Service< + ProviderSessionRepository, + ProviderSessionRepositoryShape +>()("persistence/ProviderSessionRepository") {} diff --git a/apps/server/src/provider/Errors.ts b/apps/server/src/provider/Errors.ts index 6339abc05e7f8e5f6b964412d4986d04e685d674..3ba38d995d0e1bf9fd807dee5dd515269ce4a6cd 100644 --- a/apps/server/src/provider/Errors.ts +++ b/apps/server/src/provider/Errors.ts @@ -148,6 +148,22 @@ export class ProviderSessionNotFoundError extends Schema.TaggedErrorClass()( + "ProviderSessionDirectoryPersistenceError", + { + operation: Schema.String, + detail: Schema.String, + cause: Schema.optional(Schema.Defect), + }, +) { + override get message(): string { + return `Provider session directory persistence error in ${this.operation}: ${this.detail}`; + } +} + /** * ProviderCheckpointUnavailableError - Checkpointing unavailable for this session. */ @@ -210,6 +226,7 @@ export type ProviderServiceError = | ProviderValidationError | ProviderUnsupportedError | ProviderSessionNotFoundError + | ProviderSessionDirectoryPersistenceError | ProviderCheckpointUnavailableError | ProviderCheckpointRangeError | ProviderFilesystemError diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index 46182941b8c01c0b467b0253be76aa4d487b1fa2..205d2e2f4d4924326cc8d6eda955ad05c899f7af 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -1,3 +1,7 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + import type { ProviderApprovalDecision, ProviderGetCheckpointDiffInput, @@ -29,9 +33,15 @@ import { 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 { ProviderServiceLive } from "./ProviderService.ts"; import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts"; import { NodeServices } from "@effect/platform-node"; +import { ProviderSessionRepositoryLive } from "../../persistence/Layers/ProviderSessions.ts"; +import { + makeSqlitePersistenceLive, + SqlitePersistenceMemory, +} from "../../persistence/Layers/Sqlite.ts"; function makeFakeCodexAdapter() { const sessions = new Map(); @@ -246,14 +256,19 @@ function makeProviderServiceLayer() { const checkpointLayer = Layer.succeed(CheckpointService, checkpoint.service); const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry); + const directoryLayer = ProviderSessionDirectoryLive.pipe( + Layer.provide(ProviderSessionRepositoryLive), + Layer.provide(SqlitePersistenceMemory), + ); const layer = it.layer( Layer.mergeAll( ProviderServiceLive.pipe( Layer.provide(providerAdapterLayer), - Layer.provide(ProviderSessionDirectoryLive), + Layer.provide(directoryLayer), Layer.provide(checkpointLayer), ), + directoryLayer, NodeServices.layer, ), ); @@ -266,6 +281,53 @@ function makeProviderServiceLayer() { } const routing = makeProviderServiceLayer(); +it.effect("ProviderServiceLive reconciles stale persisted 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 checkpoint = makeCheckpointServiceDouble(); + 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 directoryLayer = ProviderSessionDirectoryLive.pipe( + Layer.provide(ProviderSessionRepositoryLive), + Layer.provide(persistenceLayer), + ); + + yield* Effect.gen(function* () { + const directory = yield* ProviderSessionDirectory; + yield* directory.upsert({ + sessionId: "sess-stale", + provider: "codex", + threadId: "thread-stale", + }); + }).pipe(Effect.provide(directoryLayer)); + + const providerLayer = ProviderServiceLive.pipe( + Layer.provide(Layer.succeed(ProviderAdapterRegistry, registry)), + Layer.provide(directoryLayer), + Layer.provide(Layer.succeed(CheckpointService, checkpoint.service)), + ); + + checkpoint.releaseSession.mockClear(); + yield* Effect.gen(function* () { + yield* ProviderService; + }).pipe(Effect.provide(providerLayer)); + + assert.deepEqual(checkpoint.releaseSession.mock.calls, [[{ providerSessionId: "sess-stale" }]]); + + fs.rmSync(tempDir, { recursive: true, force: true }); + }).pipe(Effect.provide(NodeServices.layer)), +); + routing.layer("ProviderServiceLive routing", (it) => { it.effect("routes provider operations and delegates checkpoint workflows", () => Effect.gen(function* () { diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index e1f7d52d694e828f76d6cf68bab54b3712af9400..a088c6a81d128a1c2c4f044f6ad8d9b7a8d3b0a6 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -112,57 +112,64 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => const providers = yield* registry.listProviders(); const adapters = yield* Effect.forEach(providers, (provider) => registry.getByProvider(provider)); - const onTurnCompleted = (event: ProviderRuntimeTurnCompletedEvent): Effect.Effect => - checkpointService - .captureCurrentTurn({ - providerSessionId: event.sessionId, - ...(event.turnId !== undefined ? { turnId: event.turnId } : {}), - ...(event.status !== undefined ? { status: event.status } : {}), - }) - .pipe( - Effect.flatMap(() => checkpointService.listCheckpoints({ sessionId: event.sessionId })), - Effect.flatMap((result) => { - const currentCheckpoint = - result.checkpoints.find((checkpoint) => checkpoint.isCurrent) ?? - result.checkpoints[result.checkpoints.length - 1]; - if (!currentCheckpoint) { + const staleSessionIds = yield* directory.reconcileWithAdapters(adapters); + yield* Effect.forEach(staleSessionIds, (sessionId) => + checkpointService.releaseSession({ providerSessionId: sessionId }), + ).pipe(Effect.asVoid); + + const onTurnCompleted = (event: ProviderRuntimeTurnCompletedEvent): Effect.Effect => + checkpointService + .captureCurrentTurn({ + providerSessionId: event.sessionId, + ...(event.turnId !== undefined ? { turnId: event.turnId } : {}), + ...(event.status !== undefined ? { status: event.status } : {}), + }) + .pipe( + Effect.flatMap(() => checkpointService.listCheckpoints({ sessionId: event.sessionId })), + Effect.flatMap((result) => { + const currentCheckpoint = + result.checkpoints.find((checkpoint) => checkpoint.isCurrent) ?? + result.checkpoints[result.checkpoints.length - 1]; + if (!currentCheckpoint) { + return Effect.void; + } + + return publishRuntimeEvent( + toCheckpointCapturedEvent({ + event, + threadId: result.threadId, + turnCount: currentCheckpoint.turnCount, + }), + ); + }), + Effect.catch((error) => publishRuntimeEvent(toCheckpointCaptureErrorEvent(event, error))), + ); + + const processRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect => + publishRuntimeEvent(event).pipe( + Effect.flatMap(() => { + if (event.type !== "turn.completed") { return Effect.void; } - - return publishRuntimeEvent( - toCheckpointCapturedEvent({ - event, - threadId: result.threadId, - turnCount: currentCheckpoint.turnCount, - }), - ); + return onTurnCompleted(event); }), - Effect.catch((error) => publishRuntimeEvent(toCheckpointCaptureErrorEvent(event, error))), ); - const processRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect => - publishRuntimeEvent(event).pipe( - Effect.flatMap(() => { - if (event.type !== "turn.completed") { - return Effect.void; - } - return onTurnCompleted(event); - }), + const worker = Effect.forever( + Queue.take(runtimeEventQueue).pipe(Effect.flatMap(processRuntimeEvent)), ); + yield* Effect.forkScoped(worker); - const worker = Effect.forever(Queue.take(runtimeEventQueue).pipe(Effect.flatMap(processRuntimeEvent))); - yield* Effect.forkScoped(worker); + yield* Effect.forEach(adapters, (adapter) => + Stream.runForEach(adapter.streamEvents, (event) => + Queue.offer(runtimeEventQueue, event).pipe(Effect.asVoid), + ).pipe(Effect.forkScoped), + ).pipe(Effect.asVoid); - yield* Effect.forEach(adapters, (adapter) => - Stream.runForEach(adapter.streamEvents, (event) => - Queue.offer(runtimeEventQueue, event).pipe(Effect.asVoid), - ).pipe(Effect.forkScoped), - ).pipe(Effect.asVoid); - - const adapterForSession = (sessionId: string) => - directory - .getProvider(sessionId) - .pipe(Effect.flatMap((provider) => registry.getByProvider(provider))); + const adapterForSession = (sessionId: string) => + directory + .getProvider(sessionId) + .pipe(Effect.flatMap((provider) => registry.getByProvider(provider))); const startSession: ProviderServiceShape["startSession"] = (rawInput) => Effect.gen(function* () { @@ -361,12 +368,12 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => startSession, sendTurn, interruptTurn, - respondToRequest, - stopSession, - listSessions, - listCheckpoints, - getCheckpointDiff, - revertToCheckpoint, + respondToRequest, + stopSession, + listSessions, + listCheckpoints, + getCheckpointDiff, + revertToCheckpoint, stopAll, streamEvents: Stream.fromPubSub(runtimeEventPubSub), } satisfies ProviderServiceShape; diff --git a/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts b/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts index 1780609d37054c4c4130772f4322b653fb798902..db4dec63d99992b540bb354a5cd3e44ce43f347f 100644 --- a/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts @@ -1,12 +1,46 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import { NodeServices } from "@effect/platform-node"; import { it, assert } from "@effect/vitest"; import { assertFailure, assertSome } from "@effect/vitest/utils"; -import { Effect } from "effect"; +import { Effect, Layer, Stream } from "effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; -import { ProviderSessionNotFoundError, ProviderValidationError } from "../Errors.ts"; +import { makeSqlitePersistenceLive, SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; +import { ProviderSessionRepositoryLive } from "../../persistence/Layers/ProviderSessions.ts"; +import { ProviderSessionNotFoundError, ProviderValidationError, type ProviderAdapterError } from "../Errors.ts"; +import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts"; import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts"; -const layer = it.layer(ProviderSessionDirectoryLive); +function makeDirectoryLayer(persistenceLayer: Layer.Layer) { + return ProviderSessionDirectoryLive.pipe( + Layer.provide(ProviderSessionRepositoryLive), + Layer.provide(persistenceLayer), + ); +} + +function makeAdapter(activeSessionIds: ReadonlySet): ProviderAdapterShape { + const unsupported = () => Effect.die(new Error("unsupported test operation")) as never; + return { + provider: "codex", + startSession: () => unsupported(), + sendTurn: () => unsupported(), + interruptTurn: () => unsupported(), + respondToRequest: () => unsupported(), + stopSession: () => unsupported(), + listSessions: () => Effect.succeed([]), + hasSession: (sessionId) => Effect.succeed(activeSessionIds.has(sessionId)), + readThread: () => unsupported(), + rollbackThread: () => unsupported(), + stopAll: () => Effect.void, + streamEvents: Stream.empty, + }; +} + +const layer = it.layer(makeDirectoryLayer(SqlitePersistenceMemory)); layer("ProviderSessionDirectoryLive", (it) => { it("upserts, reads, and removes session bindings", () => @@ -57,4 +91,55 @@ layer("ProviderSessionDirectoryLive", (it) => { }), ); })); + + it("rehydrates persisted mappings across layer restart", () => + Effect.gen(function* () { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-directory-")); + const dbPath = path.join(tempDir, "orchestration.sqlite"); + const directoryLayer = makeDirectoryLayer(makeSqlitePersistenceLive(dbPath)); + + yield* Effect.gen(function* () { + const directory = yield* ProviderSessionDirectory; + yield* directory.upsert({ + sessionId: "sess-restart", + provider: "codex", + threadId: "thread-restart", + }); + }).pipe(Effect.provide(directoryLayer)); + + yield* Effect.gen(function* () { + const directory = yield* ProviderSessionDirectory; + const provider = yield* directory.getProvider("sess-restart"); + assert.equal(provider, "codex"); + + const threadId = yield* directory.getThreadId("sess-restart"); + assertSome(threadId, "thread-restart"); + }).pipe(Effect.provide(directoryLayer)); + + fs.rmSync(tempDir, { recursive: true, force: true }); + }).pipe(Effect.provide(NodeServices.layer))); + + it("reconciles stale sessions and keeps active mappings", () => + Effect.gen(function* () { + const directory = yield* ProviderSessionDirectory; + + yield* directory.upsert({ + sessionId: "sess-active", + provider: "codex", + threadId: "thread-active", + }); + yield* directory.upsert({ + sessionId: "sess-stale", + provider: "codex", + threadId: "thread-stale", + }); + + const pruned = yield* directory.reconcileWithAdapters([ + makeAdapter(new Set(["sess-active"])), + ]); + assert.deepEqual(pruned, ["sess-stale"]); + + const remaining = yield* directory.listSessionIds(); + assert.deepEqual(remaining, ["sess-active"]); + })); }); diff --git a/apps/server/src/provider/Layers/ProviderSessionDirectory.ts b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts index 4ff2f04955906f61d037354cc846ef50044e0757..98de7eaaca4bb4693ea467ae7b6d4ab01115ce69 100644 --- a/apps/server/src/provider/Layers/ProviderSessionDirectory.ts +++ b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts @@ -1,27 +1,78 @@ -/** - * ProviderSessionDirectoryLive - In-memory session ownership index layer. - * - * Tracks provider ownership and optional thread id by `sessionId` so - * `ProviderService` can route session-scoped API calls to the right adapter. - * This layer stores metadata only and does not call provider adapters directly. - * - * @module ProviderSessionDirectoryLive - */ -import { Effect, Layer, Option, Ref } from "effect"; +import { Effect, Layer, Option } from "effect"; -import { ProviderSessionNotFoundError, ProviderValidationError } from "../Errors.ts"; +import { ProviderSessionRepository } from "../../persistence/Services/ProviderSessions.ts"; +import { + ProviderSessionDirectoryPersistenceError, + ProviderSessionNotFoundError, + ProviderValidationError, +} from "../Errors.ts"; import { ProviderSessionDirectory, type ProviderSessionBinding, type ProviderSessionDirectoryShape, } from "../Services/ProviderSessionDirectory.ts"; +function toPersistenceError(operation: string) { + return (cause: unknown) => + new ProviderSessionDirectoryPersistenceError({ + operation, + detail: `Failed to execute ${operation}.`, + cause, + }); +} + +function normalizeBinding(binding: ProviderSessionBinding): ProviderSessionBinding { + return { + sessionId: binding.sessionId.trim(), + provider: binding.provider, + ...(binding.threadId?.trim() ? { threadId: binding.threadId.trim() } : {}), + }; +} + const makeProviderSessionDirectory = Effect.gen(function* () { - const bindingsRef = yield* Ref.make(new Map()); + const repository = yield* ProviderSessionRepository; + + const persistUpsert = (binding: ProviderSessionBinding) => + repository + .upsertSession({ + sessionId: binding.sessionId, + provider: binding.provider, + ...(binding.threadId !== undefined ? { threadId: binding.threadId } : {}), + }) + .pipe(Effect.mapError(toPersistenceError("ProviderSessionDirectory.upsert:upsertSession"))); + + const persistDelete = (sessionId: string) => + repository + .deleteSession({ sessionId }) + .pipe(Effect.mapError(toPersistenceError("ProviderSessionDirectory.remove:deleteSession"))); + + const getBinding = (sessionId: string) => + repository.getSession({ sessionId }).pipe( + Effect.mapError(toPersistenceError("ProviderSessionDirectory.getBinding:getSession")), + Effect.map((entry) => + Option.map(entry, (value) => ({ + sessionId: value.sessionId, + provider: value.provider, + ...(value.threadId !== undefined ? { threadId: value.threadId } : {}), + })), + ), + ); + + const listBindings = () => + repository.listSessions().pipe( + Effect.mapError(toPersistenceError("ProviderSessionDirectory.listSessionIds:listSessions")), + Effect.map((rows) => + rows.map((row) => ({ + sessionId: row.sessionId, + provider: row.provider, + ...(row.threadId !== undefined ? { threadId: row.threadId } : {}), + })), + ), + ); const upsert: ProviderSessionDirectoryShape["upsert"] = (binding) => { - const sessionId = binding.sessionId.trim(); - if (sessionId.length === 0) { + const normalized = normalizeBinding(binding); + if (normalized.sessionId.length === 0) { return Effect.fail( new ProviderValidationError({ operation: "ProviderSessionDirectory.upsert", @@ -30,53 +81,65 @@ const makeProviderSessionDirectory = Effect.gen(function* () { ); } - const normalized: ProviderSessionBinding = { - sessionId, - provider: binding.provider, - ...(binding.threadId !== undefined ? { threadId: binding.threadId } : {}), - }; - - return Ref.update(bindingsRef, (current) => { - const next = new Map(current); - next.set(sessionId, normalized); - return next; - }); + return persistUpsert(normalized); }; const getProvider: ProviderSessionDirectoryShape["getProvider"] = (sessionId) => - Ref.get(bindingsRef).pipe( - Effect.flatMap((bindings) => { - const binding = bindings.get(sessionId); - if (!binding) { - return Effect.fail(new ProviderSessionNotFoundError({ sessionId })); - } - return Effect.succeed(binding.provider); - }), + getBinding(sessionId).pipe( + Effect.flatMap((binding) => + Option.match(binding, { + onSome: (value) => Effect.succeed(value.provider), + onNone: () => Effect.fail(new ProviderSessionNotFoundError({ sessionId })), + }), + ), ); const getThreadId: ProviderSessionDirectoryShape["getThreadId"] = (sessionId) => - Ref.get(bindingsRef).pipe( - Effect.flatMap((bindings) => { - const binding = bindings.get(sessionId); - if (!binding) { - return Effect.fail(new ProviderSessionNotFoundError({ sessionId })); - } - return Effect.succeed(Option.fromNullishOr(binding.threadId)); - }), + getBinding(sessionId).pipe( + Effect.flatMap((binding) => + Option.match(binding, { + onSome: (value) => Effect.succeed(Option.fromNullishOr(value.threadId)), + onNone: () => Effect.fail(new ProviderSessionNotFoundError({ sessionId })), + }), + ), ); - const remove: ProviderSessionDirectoryShape["remove"] = (sessionId) => - Ref.update(bindingsRef, (current) => { - if (!current.has(sessionId)) { - return current; - } - const next = new Map(current); - next.delete(sessionId); - return next; - }); + const remove: ProviderSessionDirectoryShape["remove"] = (sessionId) => persistDelete(sessionId); const listSessionIds: ProviderSessionDirectoryShape["listSessionIds"] = () => - Ref.get(bindingsRef).pipe(Effect.map((bindings) => Array.from(bindings.keys()))); + listBindings().pipe(Effect.map((bindings) => bindings.map((binding) => binding.sessionId))); + + const reconcileWithAdapters: ProviderSessionDirectoryShape["reconcileWithAdapters"] = ( + adapters, + ) => + Effect.gen(function* () { + const byProvider = new Map(adapters.map((adapter) => [adapter.provider, adapter])); + const bindings = yield* listBindings(); + const staleSessionIds: string[] = []; + + for (const binding of bindings) { + const adapter = byProvider.get(binding.provider); + if (!adapter) { + staleSessionIds.push(binding.sessionId); + continue; + } + + const hasSession = yield* adapter.hasSession(binding.sessionId); + if (!hasSession) { + staleSessionIds.push(binding.sessionId); + } + } + + if (staleSessionIds.length === 0) { + return [] as ReadonlyArray; + } + + yield* Effect.forEach(staleSessionIds, (sessionId) => persistDelete(sessionId)).pipe( + Effect.asVoid, + ); + + return staleSessionIds as ReadonlyArray; + }); return { upsert, @@ -84,6 +147,7 @@ const makeProviderSessionDirectory = Effect.gen(function* () { getThreadId, remove, listSessionIds, + reconcileWithAdapters, } satisfies ProviderSessionDirectoryShape; }); @@ -91,3 +155,7 @@ export const ProviderSessionDirectoryLive = Layer.effect( ProviderSessionDirectory, makeProviderSessionDirectory, ); + +export function makeProviderSessionDirectoryLive() { + return Layer.effect(ProviderSessionDirectory, makeProviderSessionDirectory); +} diff --git a/apps/server/src/provider/Services/ProviderSessionDirectory.ts b/apps/server/src/provider/Services/ProviderSessionDirectory.ts index 70cb78b7c02af83f2512d8b37f17d2f3f81848ac..41009f45465495996645a76b9526380d7fc1cd1b 100644 --- a/apps/server/src/provider/Services/ProviderSessionDirectory.ts +++ b/apps/server/src/provider/Services/ProviderSessionDirectory.ts @@ -11,7 +11,13 @@ import type { ProviderKind } from "@t3tools/contracts"; import { Option, ServiceMap } from "effect"; import type { Effect } from "effect"; -import type { ProviderSessionNotFoundError, ProviderValidationError } from "../Errors.ts"; +import type { + ProviderAdapterError, + ProviderSessionDirectoryPersistenceError, + ProviderSessionNotFoundError, + ProviderValidationError, +} from "../Errors.ts"; +import type { ProviderAdapterShape } from "./ProviderAdapter.ts"; export interface ProviderSessionBinding { readonly sessionId: string; @@ -19,37 +25,62 @@ export interface ProviderSessionBinding { readonly threadId?: string; } +export type ProviderSessionDirectoryReadError = + | ProviderSessionNotFoundError + | ProviderSessionDirectoryPersistenceError; + +export type ProviderSessionDirectoryWriteError = + | ProviderValidationError + | ProviderSessionDirectoryPersistenceError; + export interface ProviderSessionDirectoryShape { /** * Record or update ownership for one provider session. */ readonly upsert: ( binding: ProviderSessionBinding, - ) => Effect.Effect; + ) => Effect.Effect; /** * Resolve the provider owner for a session id. */ readonly getProvider: ( sessionId: string, - ) => Effect.Effect; + ) => Effect.Effect; /** * Resolve the tracked thread id for a session, if known. */ readonly getThreadId: ( sessionId: string, - ) => Effect.Effect, ProviderSessionNotFoundError>; + ) => Effect.Effect, ProviderSessionDirectoryReadError>; /** * Remove a session binding. */ - readonly remove: (sessionId: string) => Effect.Effect; + readonly remove: ( + sessionId: string, + ) => Effect.Effect; /** * List tracked session ids. */ - readonly listSessionIds: () => Effect.Effect>; + readonly listSessionIds: () => Effect.Effect< + ReadonlyArray, + ProviderSessionDirectoryPersistenceError + >; + + /** + * Remove stale persisted sessions that are no longer active in adapters. + * + * Returns pruned session ids. + */ + readonly reconcileWithAdapters: ( + adapters: ReadonlyArray>, + ) => Effect.Effect< + ReadonlyArray, + ProviderSessionDirectoryPersistenceError | ProviderAdapterError + >; } /** diff --git a/apps/server/src/wsServer.ts b/apps/server/src/wsServer.ts index d600f960b28e7a473a3f571385954d7bb023c52e..55618c7c0cde4642d492b809018a1b01bdccd683 100644 --- a/apps/server/src/wsServer.ts +++ b/apps/server/src/wsServer.ts @@ -55,6 +55,7 @@ import { ProviderService, type ProviderServiceShape } from "./provider/Services/ import { CheckpointStoreLive } from "./checkpointing/Layers/CheckpointStore"; import { CheckpointServiceLive } from "./checkpointing/Layers/CheckpointService"; import { CheckpointRepositoryLive } from "./persistence/Layers/Checkpoints"; +import { ProviderSessionRepositoryLive } from "./persistence/Layers/ProviderSessions"; import { makeEventNdjsonLogger, type EventNdjsonLogger } from "./provider/Layers/EventNdjsonLogger"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import * as SqlError from "effect/unstable/sql/SqlError"; @@ -637,12 +638,15 @@ export function createServer(options: ServerOptions) { const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe( Layer.provide(codexAdapterLayer), ); + const providerSessionDirectoryLayer = ProviderSessionDirectoryLive.pipe( + Layer.provide(ProviderSessionRepositoryLive), + ); const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer)); const checkpointDependenciesLayer = Layer.mergeAll( checkpointStoreLayer, CheckpointRepositoryLive, adapterRegistryLayer, - ProviderSessionDirectoryLive, + providerSessionDirectoryLayer, ); const checkpointServiceLayer = CheckpointServiceLive.pipe( Layer.provideMerge(checkpointDependenciesLayer),