authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-23 11:40:23-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
logc0a218ca4858c2331ac52f91af1d633f61a1242d
treee3ec7964c9c061caae738a1dd68e312f9258979d
parentd4b6536a502e34e473cfa4f5d6b54be0f483cc1e

durable sessions

16 files changed, 859 insertions(+), 130 deletions(-)

apps/server/integration/providerService.integration.test.ts+8-3
...@@ -16,6 +16,7 @@ import {...@@ -16,6 +16,7 @@ import {
16} from "../src/provider/Services/ProviderService.ts";16} from "../src/provider/Services/ProviderService.ts";
17import { SqlitePersistenceMemory } from "../src/persistence/Layers/Sqlite.ts";17import { SqlitePersistenceMemory } from "../src/persistence/Layers/Sqlite.ts";
18import { CheckpointRepositoryLive } from "../src/persistence/Layers/Checkpoints.ts";18import { CheckpointRepositoryLive } from "../src/persistence/Layers/Checkpoints.ts";
19import { ProviderSessionRepositoryLive } from "../src/persistence/Layers/ProviderSessions.ts";
1920
20import {21import {
21 makeTestProviderAdapterHarness,22 makeTestProviderAdapterHarness,
...@@ -77,12 +78,16 @@ const makeIntegrationFixture = Effect.gen(function* () {...@@ -77,12 +78,16 @@ const makeIntegrationFixture = Effect.gen(function* () {
77 listProviders: () => Effect.succeed(["codex"]),78 listProviders: () => Effect.succeed(["codex"]),
78 };79 };
7980
81 const directoryLayer = ProviderSessionDirectoryLive.pipe(
82 Layer.provide(ProviderSessionRepositoryLive),
83 );
84
80 const shared = Layer.mergeAll(85 const shared = Layer.mergeAll(
81 ProviderSessionDirectoryLive,86 directoryLayer,
82 Layer.succeed(ProviderAdapterRegistry, registry),87 Layer.succeed(ProviderAdapterRegistry, registry),
83 Layer.provide(CheckpointStoreLive, NodeServices.layer),88 Layer.provide(CheckpointStoreLive, NodeServices.layer),
84 Layer.provide(CheckpointRepositoryLive, SqlitePersistenceMemory),89 CheckpointRepositoryLive,
85 );90 ).pipe(Layer.provide(SqlitePersistenceMemory));
8691
87 const checkpointLayer = CheckpointServiceLive.pipe(Layer.provide(shared));92 const checkpointLayer = CheckpointServiceLive.pipe(Layer.provide(shared));
8893
apps/server/src/checkpointing/Layers/CheckpointService.test.ts+8-3
...@@ -32,6 +32,7 @@ import { ProviderSessionDirectoryLive } from "../../provider/Layers/ProviderSess...@@ -32,6 +32,7 @@ import { ProviderSessionDirectoryLive } from "../../provider/Layers/ProviderSess
32import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";32import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";
33import { CheckpointRepository } from "../../persistence/Services/Checkpoints.ts";33import { CheckpointRepository } from "../../persistence/Services/Checkpoints.ts";
34import { CheckpointRepositoryLive } from "../../persistence/Layers/Checkpoints.ts";34import { CheckpointRepositoryLive } from "../../persistence/Layers/Checkpoints.ts";
35import { ProviderSessionRepositoryLive } from "../../persistence/Layers/ProviderSessions.ts";
3536
36const runGit = (cwd: string, args: readonly string[]) =>37const runGit = (cwd: string, args: readonly string[]) =>
37 runGitProcess({38 runGitProcess({
...@@ -211,13 +212,17 @@ function makeFixture(turns: ReadonlyArray<ProviderThreadTurnSnapshot>) {...@@ -211,13 +212,17 @@ function makeFixture(turns: ReadonlyArray<ProviderThreadTurnSnapshot>) {
211 listProviders: () => Effect.succeed(["codex"]),212 listProviders: () => Effect.succeed(["codex"]),
212 };213 };
213214
215 const directoryLayer = ProviderSessionDirectoryLive.pipe(
216 Layer.provide(ProviderSessionRepositoryLive),
217 );
218
214 const dependencies = Layer.mergeAll(219 const dependencies = Layer.mergeAll(
215 Layer.provide(CheckpointStoreLive, NodeServices.layer),220 Layer.provide(CheckpointStoreLive, NodeServices.layer),
216 Layer.provide(CheckpointRepositoryLive, SqlitePersistenceMemory),221 CheckpointRepositoryLive,
217 Layer.succeed(ProviderAdapterRegistry, registry),222 Layer.succeed(ProviderAdapterRegistry, registry),
218 ProviderSessionDirectoryLive,223 directoryLayer,
219 NodeServices.layer,224 NodeServices.layer,
220 );225 ).pipe(Layer.provide(SqlitePersistenceMemory));
221226
222 const layer = CheckpointServiceLive.pipe(Layer.provideMerge(dependencies));227 const layer = CheckpointServiceLive.pipe(Layer.provideMerge(dependencies));
223228
apps/server/src/checkpointing/Layers/CheckpointService.ts+36-12
...@@ -392,16 +392,40 @@ const makeCheckpointService = Effect.gen(function* () {...@@ -392,16 +392,40 @@ const makeCheckpointService = Effect.gen(function* () {
392 Effect.map((current) => Option.fromNullishOr(current.get(sessionId))),392 Effect.map((current) => Option.fromNullishOr(current.get(sessionId))),
393 );393 );
394394
395 const mapDirectoryLookupError = (
396 sessionId: string,
397 operation: string,
398 cause: unknown,
399 ): CheckpointSessionNotFoundError | CheckpointInvariantError => {
400 if (
401 typeof cause === "object" &&
402 cause !== null &&
403 "_tag" in cause &&
404 (cause as { _tag?: string })._tag === "ProviderSessionNotFoundError"
405 ) {
406 return new CheckpointSessionNotFoundError({
407 sessionId,
408 cause,
409 });
410 }
411
412 return invariantError(
413 operation,
414 `Failed to resolve provider session metadata for session ${sessionId}.`,
415 cause,
416 );
417 };
418
395 const resolveAdapterForSession = (419 const resolveAdapterForSession = (
396 sessionId: string,420 sessionId: string,
397 ): Effect.Effect<ProviderAdapterShape<ProviderAdapterError>, CheckpointServiceError> =>421 ): Effect.Effect<ProviderAdapterShape<ProviderAdapterError>, CheckpointServiceError> =>
398 directory.getProvider(sessionId).pipe(422 directory.getProvider(sessionId).pipe(
399 Effect.mapError(423 Effect.mapError((cause) =>
400 (cause) =>424 mapDirectoryLookupError(
401 new CheckpointSessionNotFoundError({425 sessionId,
402 sessionId,426 "CheckpointService.resolveAdapterForSession:getProvider",
403 cause,427 cause,
404 }),428 ),
405 ),429 ),
406 Effect.flatMap((provider) =>430 Effect.flatMap((provider) =>
407 registry431 registry
...@@ -456,12 +480,12 @@ const makeCheckpointService = Effect.gen(function* () {...@@ -456,12 +480,12 @@ const makeCheckpointService = Effect.gen(function* () {
456 operation: string,480 operation: string,
457 ): Effect.Effect<string, CheckpointServiceError> =>481 ): Effect.Effect<string, CheckpointServiceError> =>
458 directory.getThreadId(sessionId).pipe(482 directory.getThreadId(sessionId).pipe(
459 Effect.mapError(483 Effect.mapError((cause) =>
460 (cause) =>484 mapDirectoryLookupError(
461 new CheckpointSessionNotFoundError({485 sessionId,
462 sessionId,486 `${operation}:getThreadId`,
463 cause,487 cause,
464 }),488 ),
465 ),489 ),
466 Effect.flatMap((threadId) =>490 Effect.flatMap((threadId) =>
467 Option.match(threadId, {491 Option.match(threadId, {
apps/server/src/persistence/Errors.ts+34
...@@ -150,6 +150,36 @@ export class CheckpointRepositoryPersistenceError extends Schema.TaggedErrorClas...@@ -150,6 +150,36 @@ export class CheckpointRepositoryPersistenceError extends Schema.TaggedErrorClas
150 }150 }
151}151}
152152
153// ===============================
154// Provider Session Repository Errors
155// ===============================
156
157export class ProviderSessionRepositoryValidationError extends Schema.TaggedErrorClass<ProviderSessionRepositoryValidationError>()(
158 "ProviderSessionRepositoryValidationError",
159 {
160 operation: Schema.String,
161 issue: Schema.String,
162 cause: Schema.optional(Schema.Defect),
163 },
164) {
165 override get message(): string {
166 return `Provider session repository validation failed in ${this.operation}: ${this.issue}`;
167 }
168}
169
170export class ProviderSessionRepositoryPersistenceError extends Schema.TaggedErrorClass<ProviderSessionRepositoryPersistenceError>()(
171 "ProviderSessionRepositoryPersistenceError",
172 {
173 operation: Schema.String,
174 detail: Schema.String,
175 cause: Schema.optional(Schema.Defect),
176 },
177) {
178 override get message(): string {
179 return `Provider session repository persistence error in ${this.operation}: ${this.detail}`;
180 }
181}
182
153export type ProjectRepositoryError =183export type ProjectRepositoryError =
154 | PersistenceSqlError184 | PersistenceSqlError
155 | PersistenceDecodeError185 | PersistenceDecodeError
...@@ -166,3 +196,7 @@ export type OrchestrationEventStoreError =...@@ -166,3 +196,7 @@ export type OrchestrationEventStoreError =
166export type CheckpointRepositoryError =196export type CheckpointRepositoryError =
167 | CheckpointRepositoryValidationError197 | CheckpointRepositoryValidationError
168 | CheckpointRepositoryPersistenceError;198 | CheckpointRepositoryPersistenceError;
199
200export type ProviderSessionRepositoryError =
201 | ProviderSessionRepositoryValidationError
202 | ProviderSessionRepositoryPersistenceError;
apps/server/src/persistence/Layers/ProviderSessions.test.ts created+81
...@@ -0,0 +1,81 @@
1import fs from "node:fs";
2import os from "node:os";
3import path from "node:path";
4
5import { NodeServices } from "@effect/platform-node";
6import { it, assert } from "@effect/vitest";
7import { Effect, Layer, Option } from "effect";
8import * as SqlClient from "effect/unstable/sql/SqlClient";
9
10import { makeSqlitePersistenceLive, SqlitePersistenceMemory } from "./Sqlite.ts";
11import { ProviderSessionRepository } from "../Services/ProviderSessions.ts";
12import { ProviderSessionRepositoryLive } from "./ProviderSessions.ts";
13
14function makeRepositoryLayer<E, R>(
15 persistenceLayer: Layer.Layer<SqlClient.SqlClient, E, R>,
16) {
17 return ProviderSessionRepositoryLive.pipe(Layer.provide(persistenceLayer));
18}
19
20it.effect("ProviderSessionRepositoryLive upserts, lists, gets, and deletes sessions", () =>
21 Effect.gen(function* () {
22 const repository = yield* ProviderSessionRepository;
23
24 yield* repository.upsertSession({
25 sessionId: "sess-1",
26 provider: "codex",
27 threadId: "thread-1",
28 });
29 yield* repository.upsertSession({
30 sessionId: "sess-2",
31 provider: "claudeCode",
32 });
33
34 const all = yield* repository.listSessions();
35 assert.equal(all.length, 2);
36 assert.equal(all[0]?.sessionId, "sess-1");
37 assert.equal(all[1]?.sessionId, "sess-2");
38
39 const existing = yield* repository.getSession({ sessionId: "sess-1" });
40 assert.equal(Option.isSome(existing), true);
41 if (Option.isSome(existing)) {
42 assert.equal(existing.value.provider, "codex");
43 assert.equal(existing.value.threadId, "thread-1");
44 }
45
46 yield* repository.deleteSession({ sessionId: "sess-2" });
47 const afterDelete = yield* repository.listSessions();
48 assert.deepEqual(afterDelete.map((entry) => entry.sessionId), ["sess-1"]);
49 }).pipe(Effect.provide(makeRepositoryLayer(SqlitePersistenceMemory))),
50);
51
52it.effect("ProviderSessionRepositoryLive persists sessions across layer restarts", () =>
53 Effect.gen(function* () {
54 const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-sessions-"));
55 const dbPath = path.join(tempDir, "orchestration.sqlite");
56
57 const repositoryLayer = makeRepositoryLayer(makeSqlitePersistenceLive(dbPath));
58
59 yield* Effect.gen(function* () {
60 const repository = yield* ProviderSessionRepository;
61 yield* repository.upsertSession({
62 sessionId: "sess-persisted",
63 provider: "codex",
64 threadId: "thread-persisted",
65 });
66 }).pipe(Effect.provide(repositoryLayer));
67
68 const persisted = yield* Effect.gen(function* () {
69 const repository = yield* ProviderSessionRepository;
70 return yield* repository.getSession({ sessionId: "sess-persisted" });
71 }).pipe(Effect.provide(repositoryLayer));
72
73 assert.equal(Option.isSome(persisted), true);
74 if (Option.isSome(persisted)) {
75 assert.equal(persisted.value.provider, "codex");
76 assert.equal(persisted.value.threadId, "thread-persisted");
77 }
78
79 fs.rmSync(tempDir, { recursive: true, force: true });
80 }).pipe(Effect.provide(NodeServices.layer)),
81);
apps/server/src/persistence/Layers/ProviderSessions.ts created+232
...@@ -0,0 +1,232 @@
1import * as SqlClient from "effect/unstable/sql/SqlClient";
2import * as SqlSchema from "effect/unstable/sql/SqlSchema";
3import { Effect, Layer, Option, Schema } from "effect";
4
5import {
6 ProviderSessionRepositoryPersistenceError,
7 ProviderSessionRepositoryValidationError,
8} from "../Errors.ts";
9import {
10 ProviderSessionRepository,
11 type ProviderSessionEntry,
12 type ProviderSessionRepositoryShape,
13} from "../Services/ProviderSessions.ts";
14
15const NonEmptyString = Schema.NonEmptyString;
16
17const ProviderKindSchema = Schema.Literals(["codex", "claudeCode"]);
18
19const ProviderSessionRowSchema = Schema.Struct({
20 sessionId: Schema.String,
21 provider: ProviderKindSchema,
22 threadId: Schema.NullOr(Schema.String),
23 createdAt: Schema.String,
24 updatedAt: Schema.String,
25});
26
27const SessionIdRequestSchema = Schema.Struct({
28 sessionId: NonEmptyString,
29});
30
31const UpsertSessionRequestSchema = Schema.Struct({
32 sessionId: NonEmptyString,
33 provider: ProviderKindSchema,
34 threadId: Schema.NullOr(NonEmptyString),
35});
36
37function errorMessage(cause: unknown, fallback: string): string {
38 if (cause instanceof Error && cause.message.length > 0) {
39 return cause.message;
40 }
41 return fallback;
42}
43
44function toValidationError(
45 operation: string,
46 cause: unknown,
47): ProviderSessionRepositoryValidationError {
48 return new ProviderSessionRepositoryValidationError({
49 operation,
50 issue: errorMessage(cause, "Invalid provider session repository input."),
51 cause,
52 });
53}
54
55function decodeInput<A>(
56 schema: Schema.Schema<A>,
57 input: unknown,
58 operation: string,
59): Effect.Effect<A, ProviderSessionRepositoryValidationError> {
60 return Schema.decodeUnknownEffect(schema)(input).pipe(
61 Effect.mapError((cause) => toValidationError(operation, cause)),
62 ) as Effect.Effect<A, ProviderSessionRepositoryValidationError>;
63}
64
65function toPersistenceError(
66 operation: string,
67 cause: unknown,
68): ProviderSessionRepositoryPersistenceError {
69 return new ProviderSessionRepositoryPersistenceError({
70 operation,
71 detail: `Failed to execute ${operation}.`,
72 cause,
73 });
74}
75
76function toEntry(
77 row: Schema.Schema.Type<typeof ProviderSessionRowSchema>,
78): ProviderSessionEntry {
79 return {
80 sessionId: row.sessionId,
81 provider: row.provider,
82 ...(row.threadId !== null ? { threadId: row.threadId } : {}),
83 createdAt: row.createdAt,
84 updatedAt: row.updatedAt,
85 };
86}
87
88const makeProviderSessionRepository = Effect.gen(function* () {
89 const sql = yield* SqlClient.SqlClient;
90
91 const upsertSessionRow = SqlSchema.findOne({
92 Request: UpsertSessionRequestSchema,
93 Result: ProviderSessionRowSchema,
94 execute: (request) =>
95 sql`
96 INSERT INTO provider_sessions (
97 session_id,
98 provider,
99 thread_id,
100 created_at,
101 updated_at
102 )
103 VALUES (
104 ${request.sessionId},
105 ${request.provider},
106 ${request.threadId},
107 CURRENT_TIMESTAMP,
108 CURRENT_TIMESTAMP
109 )
110 ON CONFLICT (session_id)
111 DO UPDATE SET
112 provider = excluded.provider,
113 thread_id = excluded.thread_id,
114 updated_at = CURRENT_TIMESTAMP
115 RETURNING
116 session_id AS "sessionId",
117 provider,
118 thread_id AS "threadId",
119 created_at AS "createdAt",
120 updated_at AS "updatedAt"
121 `,
122 });
123
124 const findSessionRow = SqlSchema.findOneOption({
125 Request: SessionIdRequestSchema,
126 Result: ProviderSessionRowSchema,
127 execute: ({ sessionId }) =>
128 sql`
129 SELECT
130 session_id AS "sessionId",
131 provider,
132 thread_id AS "threadId",
133 created_at AS "createdAt",
134 updated_at AS "updatedAt"
135 FROM provider_sessions
136 WHERE session_id = ${sessionId}
137 `,
138 });
139
140 const listSessionRows = SqlSchema.findAll({
141 Request: Schema.Void,
142 Result: ProviderSessionRowSchema,
143 execute: () =>
144 sql`
145 SELECT
146 session_id AS "sessionId",
147 provider,
148 thread_id AS "threadId",
149 created_at AS "createdAt",
150 updated_at AS "updatedAt"
151 FROM provider_sessions
152 ORDER BY created_at ASC, session_id ASC
153 `,
154 });
155
156 const deleteSessionRow = SqlSchema.void({
157 Request: SessionIdRequestSchema,
158 execute: ({ sessionId }) => sql`DELETE FROM provider_sessions WHERE session_id = ${sessionId}`,
159 });
160
161 const upsertSession: ProviderSessionRepositoryShape["upsertSession"] = (input) =>
162 Effect.gen(function* () {
163 const parsed = yield* decodeInput(
164 UpsertSessionRequestSchema,
165 {
166 sessionId: input.sessionId.trim(),
167 provider: input.provider,
168 threadId: input.threadId?.trim() || null,
169 },
170 "ProviderSessionRepository.upsertSession",
171 );
172
173 yield* upsertSessionRow(parsed).pipe(
174 Effect.mapError((cause) =>
175 toPersistenceError("ProviderSessionRepository.upsertSession:query", cause),
176 ),
177 Effect.asVoid,
178 );
179 });
180
181 const getSession: ProviderSessionRepositoryShape["getSession"] = (input) =>
182 Effect.gen(function* () {
183 const parsed = yield* decodeInput(
184 SessionIdRequestSchema,
185 { sessionId: input.sessionId.trim() },
186 "ProviderSessionRepository.getSession",
187 );
188
189 const row = yield* findSessionRow(parsed).pipe(
190 Effect.mapError((cause) =>
191 toPersistenceError("ProviderSessionRepository.getSession:query", cause),
192 ),
193 );
194
195 return Option.map(row, toEntry);
196 });
197
198 const listSessions: ProviderSessionRepositoryShape["listSessions"] = () =>
199 listSessionRows(undefined).pipe(
200 Effect.mapError((cause) =>
201 toPersistenceError("ProviderSessionRepository.listSessions:query", cause),
202 ),
203 Effect.map((rows) => rows.map(toEntry)),
204 );
205
206 const deleteSession: ProviderSessionRepositoryShape["deleteSession"] = (input) =>
207 Effect.gen(function* () {
208 const parsed = yield* decodeInput(
209 SessionIdRequestSchema,
210 { sessionId: input.sessionId.trim() },
211 "ProviderSessionRepository.deleteSession",
212 );
213
214 yield* deleteSessionRow(parsed).pipe(
215 Effect.mapError((cause) =>
216 toPersistenceError("ProviderSessionRepository.deleteSession:query", cause),
217 ),
218 );
219 });
220
221 return {
222 upsertSession,
223 getSession,
224 listSessions,
225 deleteSession,
226 } satisfies ProviderSessionRepositoryShape;
227});
228
229export const ProviderSessionRepositoryLive = Layer.effect(
230 ProviderSessionRepository,
231 makeProviderSessionRepository,
232);
apps/server/src/persistence/Migrations.ts+2
...@@ -15,6 +15,7 @@ import * as Layer from "effect/Layer";...@@ -15,6 +15,7 @@ import * as Layer from "effect/Layer";
15import Migration0001 from "./Migrations/001_OrchestrationEvents.ts";15import Migration0001 from "./Migrations/001_OrchestrationEvents.ts";
16import Migration0002 from "./Migrations/002_Projects.ts";16import Migration0002 from "./Migrations/002_Projects.ts";
17import Migration0003 from "./Migrations/003_Checkpoints.ts";17import Migration0003 from "./Migrations/003_Checkpoints.ts";
18import Migration0004 from "./Migrations/004_ProviderSessions.ts";
1819
19/**20/**
20 * Migration loader with all migrations defined inline.21 * Migration loader with all migrations defined inline.
...@@ -30,6 +31,7 @@ const loader = Migrator.fromRecord({...@@ -30,6 +31,7 @@ const loader = Migrator.fromRecord({
30 "1_OrchestrationEvents": Migration0001,31 "1_OrchestrationEvents": Migration0001,
31 "2_Projects": Migration0002,32 "2_Projects": Migration0002,
32 "3_CheckpointCatalog": Migration0003,33 "3_CheckpointCatalog": Migration0003,
34 "4_ProviderSessions": Migration0004,
33});35});
3436
35/**37/**
apps/server/src/persistence/Migrations/004_ProviderSessions.ts created+21
...@@ -0,0 +1,21 @@
1import * as SqlClient from "effect/unstable/sql/SqlClient";
2import * as Effect from "effect/Effect";
3
4export default Effect.gen(function* () {
5 const sql = yield* SqlClient.SqlClient;
6
7 yield* sql`
8 CREATE TABLE IF NOT EXISTS provider_sessions (
9 session_id TEXT PRIMARY KEY,
10 provider TEXT NOT NULL,
11 thread_id TEXT,
12 created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
13 updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
14 )
15 `;
16
17 yield* sql`
18 CREATE INDEX IF NOT EXISTS idx_provider_sessions_provider
19 ON provider_sessions(provider)
20 `;
21});
apps/server/src/persistence/Services/ProviderSessions.ts created+51
...@@ -0,0 +1,51 @@
1import type { ProviderKind } from "@t3tools/contracts";
2import { Option, ServiceMap } from "effect";
3import type { Effect } from "effect";
4
5import type { ProviderSessionRepositoryError } from "../Errors.ts";
6
7export interface ProviderSessionEntry {
8 readonly sessionId: string;
9 readonly provider: ProviderKind;
10 readonly threadId?: string;
11 readonly createdAt: string;
12 readonly updatedAt: string;
13}
14
15export interface UpsertProviderSessionInput {
16 readonly sessionId: string;
17 readonly provider: ProviderKind;
18 readonly threadId?: string;
19}
20
21export interface DeleteProviderSessionInput {
22 readonly sessionId: string;
23}
24
25export interface GetProviderSessionInput {
26 readonly sessionId: string;
27}
28
29export interface ProviderSessionRepositoryShape {
30 readonly upsertSession: (
31 input: UpsertProviderSessionInput,
32 ) => Effect.Effect<void, ProviderSessionRepositoryError>;
33
34 readonly getSession: (
35 input: GetProviderSessionInput,
36 ) => Effect.Effect<Option.Option<ProviderSessionEntry>, ProviderSessionRepositoryError>;
37
38 readonly listSessions: () => Effect.Effect<
39 ReadonlyArray<ProviderSessionEntry>,
40 ProviderSessionRepositoryError
41 >;
42
43 readonly deleteSession: (
44 input: DeleteProviderSessionInput,
45 ) => Effect.Effect<void, ProviderSessionRepositoryError>;
46}
47
48export class ProviderSessionRepository extends ServiceMap.Service<
49 ProviderSessionRepository,
50 ProviderSessionRepositoryShape
51>()("persistence/ProviderSessionRepository") {}
apps/server/src/provider/Errors.ts+17
...@@ -148,6 +148,22 @@ export class ProviderSessionNotFoundError extends Schema.TaggedErrorClass<Provid...@@ -148,6 +148,22 @@ export class ProviderSessionNotFoundError extends Schema.TaggedErrorClass<Provid
148 }148 }
149}149}
150150
151/**
152 * ProviderSessionDirectoryPersistenceError - Session directory persistence failure.
153 */
154export class ProviderSessionDirectoryPersistenceError extends Schema.TaggedErrorClass<ProviderSessionDirectoryPersistenceError>()(
155 "ProviderSessionDirectoryPersistenceError",
156 {
157 operation: Schema.String,
158 detail: Schema.String,
159 cause: Schema.optional(Schema.Defect),
160 },
161) {
162 override get message(): string {
163 return `Provider session directory persistence error in ${this.operation}: ${this.detail}`;
164 }
165}
166
151/**167/**
152 * ProviderCheckpointUnavailableError - Checkpointing unavailable for this session.168 * ProviderCheckpointUnavailableError - Checkpointing unavailable for this session.
153 */169 */
...@@ -210,6 +226,7 @@ export type ProviderServiceError =...@@ -210,6 +226,7 @@ export type ProviderServiceError =
210 | ProviderValidationError226 | ProviderValidationError
211 | ProviderUnsupportedError227 | ProviderUnsupportedError
212 | ProviderSessionNotFoundError228 | ProviderSessionNotFoundError
229 | ProviderSessionDirectoryPersistenceError
213 | ProviderCheckpointUnavailableError230 | ProviderCheckpointUnavailableError
214 | ProviderCheckpointRangeError231 | ProviderCheckpointRangeError
215 | ProviderFilesystemError232 | ProviderFilesystemError
apps/server/src/provider/Layers/ProviderService.test.ts+63-1
...@@ -1,3 +1,7 @@...@@ -1,3 +1,7 @@
1import fs from "node:fs";
2import os from "node:os";
3import path from "node:path";
4
1import type {5import type {
2 ProviderApprovalDecision,6 ProviderApprovalDecision,
3 ProviderGetCheckpointDiffInput,7 ProviderGetCheckpointDiffInput,
...@@ -29,9 +33,15 @@ import {...@@ -29,9 +33,15 @@ import {
29import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts";33import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts";
30import { ProviderAdapterRegistry } from "../Services/ProviderAdapterRegistry.ts";34import { ProviderAdapterRegistry } from "../Services/ProviderAdapterRegistry.ts";
31import { ProviderService } from "../Services/ProviderService.ts";35import { ProviderService } from "../Services/ProviderService.ts";
36import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts";
32import { ProviderServiceLive } from "./ProviderService.ts";37import { ProviderServiceLive } from "./ProviderService.ts";
33import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts";38import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts";
34import { NodeServices } from "@effect/platform-node";39import { NodeServices } from "@effect/platform-node";
40import { ProviderSessionRepositoryLive } from "../../persistence/Layers/ProviderSessions.ts";
41import {
42 makeSqlitePersistenceLive,
43 SqlitePersistenceMemory,
44} from "../../persistence/Layers/Sqlite.ts";
3545
36function makeFakeCodexAdapter() {46function makeFakeCodexAdapter() {
37 const sessions = new Map<string, ProviderSession>();47 const sessions = new Map<string, ProviderSession>();
...@@ -246,14 +256,19 @@ function makeProviderServiceLayer() {...@@ -246,14 +256,19 @@ function makeProviderServiceLayer() {
246256
247 const checkpointLayer = Layer.succeed(CheckpointService, checkpoint.service);257 const checkpointLayer = Layer.succeed(CheckpointService, checkpoint.service);
248 const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry);258 const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry);
259 const directoryLayer = ProviderSessionDirectoryLive.pipe(
260 Layer.provide(ProviderSessionRepositoryLive),
261 Layer.provide(SqlitePersistenceMemory),
262 );
249263
250 const layer = it.layer(264 const layer = it.layer(
251 Layer.mergeAll(265 Layer.mergeAll(
252 ProviderServiceLive.pipe(266 ProviderServiceLive.pipe(
253 Layer.provide(providerAdapterLayer),267 Layer.provide(providerAdapterLayer),
254 Layer.provide(ProviderSessionDirectoryLive),268 Layer.provide(directoryLayer),
255 Layer.provide(checkpointLayer),269 Layer.provide(checkpointLayer),
256 ),270 ),
271 directoryLayer,
257 NodeServices.layer,272 NodeServices.layer,
258 ),273 ),
259 );274 );
...@@ -266,6 +281,53 @@ function makeProviderServiceLayer() {...@@ -266,6 +281,53 @@ function makeProviderServiceLayer() {
266}281}
267282
268const routing = makeProviderServiceLayer();283const routing = makeProviderServiceLayer();
284it.effect("ProviderServiceLive reconciles stale persisted sessions on startup", () =>
285 Effect.gen(function* () {
286 const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-service-"));
287 const dbPath = path.join(tempDir, "orchestration.sqlite");
288
289 const codex = makeFakeCodexAdapter();
290 const checkpoint = makeCheckpointServiceDouble();
291 const registry: typeof ProviderAdapterRegistry.Service = {
292 getByProvider: (provider) =>
293 provider === "codex"
294 ? Effect.succeed(codex.adapter)
295 : Effect.fail(new ProviderUnsupportedError({ provider })),
296 listProviders: () => Effect.succeed(["codex"]),
297 };
298
299 const persistenceLayer = makeSqlitePersistenceLive(dbPath);
300 const directoryLayer = ProviderSessionDirectoryLive.pipe(
301 Layer.provide(ProviderSessionRepositoryLive),
302 Layer.provide(persistenceLayer),
303 );
304
305 yield* Effect.gen(function* () {
306 const directory = yield* ProviderSessionDirectory;
307 yield* directory.upsert({
308 sessionId: "sess-stale",
309 provider: "codex",
310 threadId: "thread-stale",
311 });
312 }).pipe(Effect.provide(directoryLayer));
313
314 const providerLayer = ProviderServiceLive.pipe(
315 Layer.provide(Layer.succeed(ProviderAdapterRegistry, registry)),
316 Layer.provide(directoryLayer),
317 Layer.provide(Layer.succeed(CheckpointService, checkpoint.service)),
318 );
319
320 checkpoint.releaseSession.mockClear();
321 yield* Effect.gen(function* () {
322 yield* ProviderService;
323 }).pipe(Effect.provide(providerLayer));
324
325 assert.deepEqual(checkpoint.releaseSession.mock.calls, [[{ providerSessionId: "sess-stale" }]]);
326
327 fs.rmSync(tempDir, { recursive: true, force: true });
328 }).pipe(Effect.provide(NodeServices.layer)),
329);
330
269routing.layer("ProviderServiceLive routing", (it) => {331routing.layer("ProviderServiceLive routing", (it) => {
270 it.effect("routes provider operations and delegates checkpoint workflows", () =>332 it.effect("routes provider operations and delegates checkpoint workflows", () =>
271 Effect.gen(function* () {333 Effect.gen(function* () {
apps/server/src/provider/Layers/ProviderService.ts+56-49
...@@ -112,57 +112,64 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>...@@ -112,57 +112,64 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
112 const providers = yield* registry.listProviders();112 const providers = yield* registry.listProviders();
113 const adapters = yield* Effect.forEach(providers, (provider) => registry.getByProvider(provider));113 const adapters = yield* Effect.forEach(providers, (provider) => registry.getByProvider(provider));
114114
115 const onTurnCompleted = (event: ProviderRuntimeTurnCompletedEvent): Effect.Effect<void> =>115 const staleSessionIds = yield* directory.reconcileWithAdapters(adapters);
116 checkpointService116 yield* Effect.forEach(staleSessionIds, (sessionId) =>
117 .captureCurrentTurn({117 checkpointService.releaseSession({ providerSessionId: sessionId }),
118 providerSessionId: event.sessionId,118 ).pipe(Effect.asVoid);
119 ...(event.turnId !== undefined ? { turnId: event.turnId } : {}),119
120 ...(event.status !== undefined ? { status: event.status } : {}),120 const onTurnCompleted = (event: ProviderRuntimeTurnCompletedEvent): Effect.Effect<void> =>
121 })121 checkpointService
122 .pipe(122 .captureCurrentTurn({
123 Effect.flatMap(() => checkpointService.listCheckpoints({ sessionId: event.sessionId })),123 providerSessionId: event.sessionId,
124 Effect.flatMap((result) => {124 ...(event.turnId !== undefined ? { turnId: event.turnId } : {}),
125 const currentCheckpoint =125 ...(event.status !== undefined ? { status: event.status } : {}),
126 result.checkpoints.find((checkpoint) => checkpoint.isCurrent) ??126 })
127 result.checkpoints[result.checkpoints.length - 1];127 .pipe(
128 if (!currentCheckpoint) {128 Effect.flatMap(() => checkpointService.listCheckpoints({ sessionId: event.sessionId })),
129 Effect.flatMap((result) => {
130 const currentCheckpoint =
131 result.checkpoints.find((checkpoint) => checkpoint.isCurrent) ??
132 result.checkpoints[result.checkpoints.length - 1];
133 if (!currentCheckpoint) {
134 return Effect.void;
135 }
136
137 return publishRuntimeEvent(
138 toCheckpointCapturedEvent({
139 event,
140 threadId: result.threadId,
141 turnCount: currentCheckpoint.turnCount,
142 }),
143 );
144 }),
145 Effect.catch((error) => publishRuntimeEvent(toCheckpointCaptureErrorEvent(event, error))),
146 );
147
148 const processRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> =>
149 publishRuntimeEvent(event).pipe(
150 Effect.flatMap(() => {
151 if (event.type !== "turn.completed") {
129 return Effect.void;152 return Effect.void;
130 }153 }
131154 return onTurnCompleted(event);
132 return publishRuntimeEvent(
133 toCheckpointCapturedEvent({
134 event,
135 threadId: result.threadId,
136 turnCount: currentCheckpoint.turnCount,
137 }),
138 );
139 }),155 }),
140 Effect.catch((error) => publishRuntimeEvent(toCheckpointCaptureErrorEvent(event, error))),
141 );156 );
142157
143 const processRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> =>158 const worker = Effect.forever(
144 publishRuntimeEvent(event).pipe(159 Queue.take(runtimeEventQueue).pipe(Effect.flatMap(processRuntimeEvent)),
145 Effect.flatMap(() => {
146 if (event.type !== "turn.completed") {
147 return Effect.void;
148 }
149 return onTurnCompleted(event);
150 }),
151 );160 );
161 yield* Effect.forkScoped(worker);
152162
153 const worker = Effect.forever(Queue.take(runtimeEventQueue).pipe(Effect.flatMap(processRuntimeEvent)));163 yield* Effect.forEach(adapters, (adapter) =>
154 yield* Effect.forkScoped(worker);164 Stream.runForEach(adapter.streamEvents, (event) =>
155165 Queue.offer(runtimeEventQueue, event).pipe(Effect.asVoid),
156 yield* Effect.forEach(adapters, (adapter) =>166 ).pipe(Effect.forkScoped),
157 Stream.runForEach(adapter.streamEvents, (event) =>167 ).pipe(Effect.asVoid);
158 Queue.offer(runtimeEventQueue, event).pipe(Effect.asVoid),
159 ).pipe(Effect.forkScoped),
160 ).pipe(Effect.asVoid);
161168
162 const adapterForSession = (sessionId: string) =>169 const adapterForSession = (sessionId: string) =>
163 directory170 directory
164 .getProvider(sessionId)171 .getProvider(sessionId)
165 .pipe(Effect.flatMap((provider) => registry.getByProvider(provider)));172 .pipe(Effect.flatMap((provider) => registry.getByProvider(provider)));
166173
167 const startSession: ProviderServiceShape["startSession"] = (rawInput) =>174 const startSession: ProviderServiceShape["startSession"] = (rawInput) =>
168 Effect.gen(function* () {175 Effect.gen(function* () {
...@@ -361,12 +368,12 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>...@@ -361,12 +368,12 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
361 startSession,368 startSession,
362 sendTurn,369 sendTurn,
363 interruptTurn,370 interruptTurn,
364 respondToRequest,371 respondToRequest,
365 stopSession,372 stopSession,
366 listSessions,373 listSessions,
367 listCheckpoints,374 listCheckpoints,
368 getCheckpointDiff,375 getCheckpointDiff,
369 revertToCheckpoint,376 revertToCheckpoint,
370 stopAll,377 stopAll,
371 streamEvents: Stream.fromPubSub(runtimeEventPubSub),378 streamEvents: Stream.fromPubSub(runtimeEventPubSub),
372 } satisfies ProviderServiceShape;379 } satisfies ProviderServiceShape;
apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts+88-3
...@@ -1,12 +1,46 @@...@@ -1,12 +1,46 @@
1import fs from "node:fs";
2import os from "node:os";
3import path from "node:path";
4
5import { NodeServices } from "@effect/platform-node";
1import { it, assert } from "@effect/vitest";6import { it, assert } from "@effect/vitest";
2import { assertFailure, assertSome } from "@effect/vitest/utils";7import { assertFailure, assertSome } from "@effect/vitest/utils";
3import { Effect } from "effect";8import { Effect, Layer, Stream } from "effect";
9import * as SqlClient from "effect/unstable/sql/SqlClient";
410
5import { ProviderSessionNotFoundError, ProviderValidationError } from "../Errors.ts";11import { makeSqlitePersistenceLive, SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";
12import { ProviderSessionRepositoryLive } from "../../persistence/Layers/ProviderSessions.ts";
13import { ProviderSessionNotFoundError, ProviderValidationError, type ProviderAdapterError } from "../Errors.ts";
14import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts";
6import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts";15import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts";
7import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts";16import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts";
817
9const layer = it.layer(ProviderSessionDirectoryLive);18function makeDirectoryLayer<E, R>(persistenceLayer: Layer.Layer<SqlClient.SqlClient, E, R>) {
19 return ProviderSessionDirectoryLive.pipe(
20 Layer.provide(ProviderSessionRepositoryLive),
21 Layer.provide(persistenceLayer),
22 );
23}
24
25function makeAdapter(activeSessionIds: ReadonlySet<string>): ProviderAdapterShape<ProviderAdapterError> {
26 const unsupported = () => Effect.die(new Error("unsupported test operation")) as never;
27 return {
28 provider: "codex",
29 startSession: () => unsupported(),
30 sendTurn: () => unsupported(),
31 interruptTurn: () => unsupported(),
32 respondToRequest: () => unsupported(),
33 stopSession: () => unsupported(),
34 listSessions: () => Effect.succeed([]),
35 hasSession: (sessionId) => Effect.succeed(activeSessionIds.has(sessionId)),
36 readThread: () => unsupported(),
37 rollbackThread: () => unsupported(),
38 stopAll: () => Effect.void,
39 streamEvents: Stream.empty,
40 };
41}
42
43const layer = it.layer(makeDirectoryLayer(SqlitePersistenceMemory));
1044
11layer("ProviderSessionDirectoryLive", (it) => {45layer("ProviderSessionDirectoryLive", (it) => {
12 it("upserts, reads, and removes session bindings", () =>46 it("upserts, reads, and removes session bindings", () =>
...@@ -57,4 +91,55 @@ layer("ProviderSessionDirectoryLive", (it) => {...@@ -57,4 +91,55 @@ layer("ProviderSessionDirectoryLive", (it) => {
57 }),91 }),
58 );92 );
59 }));93 }));
94
95 it("rehydrates persisted mappings across layer restart", () =>
96 Effect.gen(function* () {
97 const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-directory-"));
98 const dbPath = path.join(tempDir, "orchestration.sqlite");
99 const directoryLayer = makeDirectoryLayer(makeSqlitePersistenceLive(dbPath));
100
101 yield* Effect.gen(function* () {
102 const directory = yield* ProviderSessionDirectory;
103 yield* directory.upsert({
104 sessionId: "sess-restart",
105 provider: "codex",
106 threadId: "thread-restart",
107 });
108 }).pipe(Effect.provide(directoryLayer));
109
110 yield* Effect.gen(function* () {
111 const directory = yield* ProviderSessionDirectory;
112 const provider = yield* directory.getProvider("sess-restart");
113 assert.equal(provider, "codex");
114
115 const threadId = yield* directory.getThreadId("sess-restart");
116 assertSome(threadId, "thread-restart");
117 }).pipe(Effect.provide(directoryLayer));
118
119 fs.rmSync(tempDir, { recursive: true, force: true });
120 }).pipe(Effect.provide(NodeServices.layer)));
121
122 it("reconciles stale sessions and keeps active mappings", () =>
123 Effect.gen(function* () {
124 const directory = yield* ProviderSessionDirectory;
125
126 yield* directory.upsert({
127 sessionId: "sess-active",
128 provider: "codex",
129 threadId: "thread-active",
130 });
131 yield* directory.upsert({
132 sessionId: "sess-stale",
133 provider: "codex",
134 threadId: "thread-stale",
135 });
136
137 const pruned = yield* directory.reconcileWithAdapters([
138 makeAdapter(new Set(["sess-active"])),
139 ]);
140 assert.deepEqual(pruned, ["sess-stale"]);
141
142 const remaining = yield* directory.listSessionIds();
143 assert.deepEqual(remaining, ["sess-active"]);
144 }));
60});145});
apps/server/src/provider/Layers/ProviderSessionDirectory.ts+120-52
...@@ -1,27 +1,78 @@...@@ -1,27 +1,78 @@
1/**1import { Effect, Layer, Option } from "effect";
2 * ProviderSessionDirectoryLive - In-memory session ownership index layer.2
3 *3import { ProviderSessionRepository } from "../../persistence/Services/ProviderSessions.ts";
4 * Tracks provider ownership and optional thread id by `sessionId` so4import {
5 * `ProviderService` can route session-scoped API calls to the right adapter.5 ProviderSessionDirectoryPersistenceError,
6 * This layer stores metadata only and does not call provider adapters directly.6 ProviderSessionNotFoundError,
7 *7 ProviderValidationError,
8 * @module ProviderSessionDirectoryLive8} from "../Errors.ts";
9 */
10import { Effect, Layer, Option, Ref } from "effect";
11
12import { ProviderSessionNotFoundError, ProviderValidationError } from "../Errors.ts";
13import {9import {
14 ProviderSessionDirectory,10 ProviderSessionDirectory,
15 type ProviderSessionBinding,11 type ProviderSessionBinding,
16 type ProviderSessionDirectoryShape,12 type ProviderSessionDirectoryShape,
17} from "../Services/ProviderSessionDirectory.ts";13} from "../Services/ProviderSessionDirectory.ts";
1814
15function toPersistenceError(operation: string) {
16 return (cause: unknown) =>
17 new ProviderSessionDirectoryPersistenceError({
18 operation,
19 detail: `Failed to execute ${operation}.`,
20 cause,
21 });
22}
23
24function normalizeBinding(binding: ProviderSessionBinding): ProviderSessionBinding {
25 return {
26 sessionId: binding.sessionId.trim(),
27 provider: binding.provider,
28 ...(binding.threadId?.trim() ? { threadId: binding.threadId.trim() } : {}),
29 };
30}
31
19const makeProviderSessionDirectory = Effect.gen(function* () {32const makeProviderSessionDirectory = Effect.gen(function* () {
20 const bindingsRef = yield* Ref.make(new Map<string, ProviderSessionBinding>());33 const repository = yield* ProviderSessionRepository;
34
35 const persistUpsert = (binding: ProviderSessionBinding) =>
36 repository
37 .upsertSession({
38 sessionId: binding.sessionId,
39 provider: binding.provider,
40 ...(binding.threadId !== undefined ? { threadId: binding.threadId } : {}),
41 })
42 .pipe(Effect.mapError(toPersistenceError("ProviderSessionDirectory.upsert:upsertSession")));
43
44 const persistDelete = (sessionId: string) =>
45 repository
46 .deleteSession({ sessionId })
47 .pipe(Effect.mapError(toPersistenceError("ProviderSessionDirectory.remove:deleteSession")));
48
49 const getBinding = (sessionId: string) =>
50 repository.getSession({ sessionId }).pipe(
51 Effect.mapError(toPersistenceError("ProviderSessionDirectory.getBinding:getSession")),
52 Effect.map((entry) =>
53 Option.map(entry, (value) => ({
54 sessionId: value.sessionId,
55 provider: value.provider,
56 ...(value.threadId !== undefined ? { threadId: value.threadId } : {}),
57 })),
58 ),
59 );
60
61 const listBindings = () =>
62 repository.listSessions().pipe(
63 Effect.mapError(toPersistenceError("ProviderSessionDirectory.listSessionIds:listSessions")),
64 Effect.map((rows) =>
65 rows.map((row) => ({
66 sessionId: row.sessionId,
67 provider: row.provider,
68 ...(row.threadId !== undefined ? { threadId: row.threadId } : {}),
69 })),
70 ),
71 );
2172
22 const upsert: ProviderSessionDirectoryShape["upsert"] = (binding) => {73 const upsert: ProviderSessionDirectoryShape["upsert"] = (binding) => {
23 const sessionId = binding.sessionId.trim();74 const normalized = normalizeBinding(binding);
24 if (sessionId.length === 0) {75 if (normalized.sessionId.length === 0) {
25 return Effect.fail(76 return Effect.fail(
26 new ProviderValidationError({77 new ProviderValidationError({
27 operation: "ProviderSessionDirectory.upsert",78 operation: "ProviderSessionDirectory.upsert",
...@@ -30,53 +81,65 @@ const makeProviderSessionDirectory = Effect.gen(function* () {...@@ -30,53 +81,65 @@ const makeProviderSessionDirectory = Effect.gen(function* () {
30 );81 );
31 }82 }
3283
33 const normalized: ProviderSessionBinding = {84 return persistUpsert(normalized);
34 sessionId,
35 provider: binding.provider,
36 ...(binding.threadId !== undefined ? { threadId: binding.threadId } : {}),
37 };
38
39 return Ref.update(bindingsRef, (current) => {
40 const next = new Map(current);
41 next.set(sessionId, normalized);
42 return next;
43 });
44 };85 };
4586
46 const getProvider: ProviderSessionDirectoryShape["getProvider"] = (sessionId) =>87 const getProvider: ProviderSessionDirectoryShape["getProvider"] = (sessionId) =>
47 Ref.get(bindingsRef).pipe(88 getBinding(sessionId).pipe(
48 Effect.flatMap((bindings) => {89 Effect.flatMap((binding) =>
49 const binding = bindings.get(sessionId);90 Option.match(binding, {
50 if (!binding) {91 onSome: (value) => Effect.succeed(value.provider),
51 return Effect.fail(new ProviderSessionNotFoundError({ sessionId }));92 onNone: () => Effect.fail(new ProviderSessionNotFoundError({ sessionId })),
52 }93 }),
53 return Effect.succeed(binding.provider);94 ),
54 }),
55 );95 );
5696
57 const getThreadId: ProviderSessionDirectoryShape["getThreadId"] = (sessionId) =>97 const getThreadId: ProviderSessionDirectoryShape["getThreadId"] = (sessionId) =>
58 Ref.get(bindingsRef).pipe(98 getBinding(sessionId).pipe(
59 Effect.flatMap((bindings) => {99 Effect.flatMap((binding) =>
60 const binding = bindings.get(sessionId);100 Option.match(binding, {
61 if (!binding) {101 onSome: (value) => Effect.succeed(Option.fromNullishOr(value.threadId)),
62 return Effect.fail(new ProviderSessionNotFoundError({ sessionId }));102 onNone: () => Effect.fail(new ProviderSessionNotFoundError({ sessionId })),
63 }103 }),
64 return Effect.succeed(Option.fromNullishOr(binding.threadId));104 ),
65 }),
66 );105 );
67106
68 const remove: ProviderSessionDirectoryShape["remove"] = (sessionId) =>107 const remove: ProviderSessionDirectoryShape["remove"] = (sessionId) => persistDelete(sessionId);
69 Ref.update(bindingsRef, (current) => {
70 if (!current.has(sessionId)) {
71 return current;
72 }
73 const next = new Map(current);
74 next.delete(sessionId);
75 return next;
76 });
77108
78 const listSessionIds: ProviderSessionDirectoryShape["listSessionIds"] = () =>109 const listSessionIds: ProviderSessionDirectoryShape["listSessionIds"] = () =>
79 Ref.get(bindingsRef).pipe(Effect.map((bindings) => Array.from(bindings.keys())));110 listBindings().pipe(Effect.map((bindings) => bindings.map((binding) => binding.sessionId)));
111
112 const reconcileWithAdapters: ProviderSessionDirectoryShape["reconcileWithAdapters"] = (
113 adapters,
114 ) =>
115 Effect.gen(function* () {
116 const byProvider = new Map(adapters.map((adapter) => [adapter.provider, adapter]));
117 const bindings = yield* listBindings();
118 const staleSessionIds: string[] = [];
119
120 for (const binding of bindings) {
121 const adapter = byProvider.get(binding.provider);
122 if (!adapter) {
123 staleSessionIds.push(binding.sessionId);
124 continue;
125 }
126
127 const hasSession = yield* adapter.hasSession(binding.sessionId);
128 if (!hasSession) {
129 staleSessionIds.push(binding.sessionId);
130 }
131 }
132
133 if (staleSessionIds.length === 0) {
134 return [] as ReadonlyArray<string>;
135 }
136
137 yield* Effect.forEach(staleSessionIds, (sessionId) => persistDelete(sessionId)).pipe(
138 Effect.asVoid,
139 );
140
141 return staleSessionIds as ReadonlyArray<string>;
142 });
80143
81 return {144 return {
82 upsert,145 upsert,
...@@ -84,6 +147,7 @@ const makeProviderSessionDirectory = Effect.gen(function* () {...@@ -84,6 +147,7 @@ const makeProviderSessionDirectory = Effect.gen(function* () {
84 getThreadId,147 getThreadId,
85 remove,148 remove,
86 listSessionIds,149 listSessionIds,
150 reconcileWithAdapters,
87 } satisfies ProviderSessionDirectoryShape;151 } satisfies ProviderSessionDirectoryShape;
88});152});
89153
...@@ -91,3 +155,7 @@ export const ProviderSessionDirectoryLive = Layer.effect(...@@ -91,3 +155,7 @@ export const ProviderSessionDirectoryLive = Layer.effect(
91 ProviderSessionDirectory,155 ProviderSessionDirectory,
92 makeProviderSessionDirectory,156 makeProviderSessionDirectory,
93);157);
158
159export function makeProviderSessionDirectoryLive() {
160 return Layer.effect(ProviderSessionDirectory, makeProviderSessionDirectory);
161}
apps/server/src/provider/Services/ProviderSessionDirectory.ts+37-6
...@@ -11,7 +11,13 @@ import type { ProviderKind } from "@t3tools/contracts";...@@ -11,7 +11,13 @@ import type { ProviderKind } from "@t3tools/contracts";
11import { Option, ServiceMap } from "effect";11import { Option, ServiceMap } from "effect";
12import type { Effect } from "effect";12import type { Effect } from "effect";
1313
14import type { ProviderSessionNotFoundError, ProviderValidationError } from "../Errors.ts";14import type {
15 ProviderAdapterError,
16 ProviderSessionDirectoryPersistenceError,
17 ProviderSessionNotFoundError,
18 ProviderValidationError,
19} from "../Errors.ts";
20import type { ProviderAdapterShape } from "./ProviderAdapter.ts";
1521
16export interface ProviderSessionBinding {22export interface ProviderSessionBinding {
17 readonly sessionId: string;23 readonly sessionId: string;
...@@ -19,37 +25,62 @@ export interface ProviderSessionBinding {...@@ -19,37 +25,62 @@ export interface ProviderSessionBinding {
19 readonly threadId?: string;25 readonly threadId?: string;
20}26}
2127
28export type ProviderSessionDirectoryReadError =
29 | ProviderSessionNotFoundError
30 | ProviderSessionDirectoryPersistenceError;
31
32export type ProviderSessionDirectoryWriteError =
33 | ProviderValidationError
34 | ProviderSessionDirectoryPersistenceError;
35
22export interface ProviderSessionDirectoryShape {36export interface ProviderSessionDirectoryShape {
23 /**37 /**
24 * Record or update ownership for one provider session.38 * Record or update ownership for one provider session.
25 */39 */
26 readonly upsert: (40 readonly upsert: (
27 binding: ProviderSessionBinding,41 binding: ProviderSessionBinding,
28 ) => Effect.Effect<void, ProviderValidationError>;42 ) => Effect.Effect<void, ProviderSessionDirectoryWriteError>;
2943
30 /**44 /**
31 * Resolve the provider owner for a session id.45 * Resolve the provider owner for a session id.
32 */46 */
33 readonly getProvider: (47 readonly getProvider: (
34 sessionId: string,48 sessionId: string,
35 ) => Effect.Effect<ProviderKind, ProviderSessionNotFoundError>;49 ) => Effect.Effect<ProviderKind, ProviderSessionDirectoryReadError>;
3650
37 /**51 /**
38 * Resolve the tracked thread id for a session, if known.52 * Resolve the tracked thread id for a session, if known.
39 */53 */
40 readonly getThreadId: (54 readonly getThreadId: (
41 sessionId: string,55 sessionId: string,
42 ) => Effect.Effect<Option.Option<string>, ProviderSessionNotFoundError>;56 ) => Effect.Effect<Option.Option<string>, ProviderSessionDirectoryReadError>;
4357
44 /**58 /**
45 * Remove a session binding.59 * Remove a session binding.
46 */60 */
47 readonly remove: (sessionId: string) => Effect.Effect<void>;61 readonly remove: (
62 sessionId: string,
63 ) => Effect.Effect<void, ProviderSessionDirectoryPersistenceError>;
4864
49 /**65 /**
50 * List tracked session ids.66 * List tracked session ids.
51 */67 */
52 readonly listSessionIds: () => Effect.Effect<ReadonlyArray<string>>;68 readonly listSessionIds: () => Effect.Effect<
69 ReadonlyArray<string>,
70 ProviderSessionDirectoryPersistenceError
71 >;
72
73 /**
74 * Remove stale persisted sessions that are no longer active in adapters.
75 *
76 * Returns pruned session ids.
77 */
78 readonly reconcileWithAdapters: (
79 adapters: ReadonlyArray<ProviderAdapterShape<ProviderAdapterError>>,
80 ) => Effect.Effect<
81 ReadonlyArray<string>,
82 ProviderSessionDirectoryPersistenceError | ProviderAdapterError
83 >;
53}84}
5485
55/**86/**
apps/server/src/wsServer.ts+5-1
...@@ -55,6 +55,7 @@ import { ProviderService, type ProviderServiceShape } from "./provider/Services/...@@ -55,6 +55,7 @@ import { ProviderService, type ProviderServiceShape } from "./provider/Services/
55import { CheckpointStoreLive } from "./checkpointing/Layers/CheckpointStore";55import { CheckpointStoreLive } from "./checkpointing/Layers/CheckpointStore";
56import { CheckpointServiceLive } from "./checkpointing/Layers/CheckpointService";56import { CheckpointServiceLive } from "./checkpointing/Layers/CheckpointService";
57import { CheckpointRepositoryLive } from "./persistence/Layers/Checkpoints";57import { CheckpointRepositoryLive } from "./persistence/Layers/Checkpoints";
58import { ProviderSessionRepositoryLive } from "./persistence/Layers/ProviderSessions";
58import { makeEventNdjsonLogger, type EventNdjsonLogger } from "./provider/Layers/EventNdjsonLogger";59import { makeEventNdjsonLogger, type EventNdjsonLogger } from "./provider/Layers/EventNdjsonLogger";
59import * as SqlClient from "effect/unstable/sql/SqlClient";60import * as SqlClient from "effect/unstable/sql/SqlClient";
60import * as SqlError from "effect/unstable/sql/SqlError";61import * as SqlError from "effect/unstable/sql/SqlError";
...@@ -637,12 +638,15 @@ export function createServer(options: ServerOptions) {...@@ -637,12 +638,15 @@ export function createServer(options: ServerOptions) {
637 const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe(638 const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe(
638 Layer.provide(codexAdapterLayer),639 Layer.provide(codexAdapterLayer),
639 );640 );
641 const providerSessionDirectoryLayer = ProviderSessionDirectoryLive.pipe(
642 Layer.provide(ProviderSessionRepositoryLive),
643 );
640 const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer));644 const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer));
641 const checkpointDependenciesLayer = Layer.mergeAll(645 const checkpointDependenciesLayer = Layer.mergeAll(
642 checkpointStoreLayer,646 checkpointStoreLayer,
643 CheckpointRepositoryLive,647 CheckpointRepositoryLive,
644 adapterRegistryLayer,648 adapterRegistryLayer,
645 ProviderSessionDirectoryLive,649 providerSessionDirectoryLayer,
646 );650 );
647 const checkpointServiceLayer = CheckpointServiceLive.pipe(651 const checkpointServiceLayer = CheckpointServiceLive.pipe(
648 Layer.provideMerge(checkpointDependenciesLayer),652 Layer.provideMerge(checkpointDependenciesLayer),