1import * as Arr from "effect/Array";
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Option from "effect/Option";
6import * as Result from "effect/Result";
7import * as Schema from "effect/Schema";
8import * as Struct from "effect/Struct";
9import * as SqlClient from "effect/sql/SqlClient";
10import * as SqlSchema from "effect/sql/SqlSchema";
11
12import {
13 AgentSessionImportSource,
14 IsoDateTime,
15 ProviderInstanceId,
16 RuntimeMode,
17 ThreadId,
18} from "@t3tools/contracts";
19
20import {
21 PersistenceDecodeError,
22 type PersistenceErrorCorrelation,
23 PersistenceSqlError,
24 type ProviderSessionRuntimeRepositoryError,
25} from "./Errors.ts";
26
27/**
28 * ProviderSessionRuntimeRepository - Repository interface for provider runtime sessions.
29 *
30 * Owns persistence operations for provider runtime metadata and resume cursors.
31 *
32 * @module ProviderSessionRuntimeRepository
33 */
34
35const ProviderSessionRuntimeStatus = Schema.Literals(["starting", "running", "stopped", "error"]);
36
37export const ProviderSessionRuntime = Schema.Struct({
38 threadId: ThreadId,
39 providerName: Schema.String,
40 /**
41 * User-defined routing key for the configured provider instance that
42 * owns this session. Nullable only at the storage/migration boundary:
43 * rows persisted before the driver/instance split carry only
44 * `providerName`. Repository consumers must materialize a concrete
45 * instance id before routing.
46 */
47 providerInstanceId: Schema.NullOr(ProviderInstanceId),
48 adapterKey: Schema.String,
49 runtimeMode: RuntimeMode,
50 status: ProviderSessionRuntimeStatus,
51 lastSeenAt: IsoDateTime,
52 resumeCursor: Schema.NullOr(Schema.Unknown),
53 runtimePayload: Schema.NullOr(Schema.Unknown),
54});
55export type ProviderSessionRuntime = typeof ProviderSessionRuntime.Type;
56
57export const GetProviderSessionRuntimeInput = Schema.Struct({ threadId: ThreadId });
58export type GetProviderSessionRuntimeInput = typeof GetProviderSessionRuntimeInput.Type;
59
60export const DeleteProviderSessionRuntimeInput = Schema.Struct({ threadId: ThreadId });
61export type DeleteProviderSessionRuntimeInput = typeof DeleteProviderSessionRuntimeInput.Type;
62
63export const RecordImportedTranscriptInput = Schema.Struct({
64 threadId: ThreadId,
65 source: AgentSessionImportSource,
66});
67export type RecordImportedTranscriptInput = typeof RecordImportedTranscriptInput.Type;
68
69export interface ProviderSessionRuntimeUpsertOptions {
70 readonly onConflict?: "update" | "ignore";
71}
72
73/**
74 * ProviderSessionRuntimeRepository - Service tag for provider runtime persistence.
75 */
76export class ProviderSessionRuntimeRepository extends Context.Service<
77 ProviderSessionRuntimeRepository,
78 {
79 /**
80 * Insert or replace a provider runtime row.
81 *
82 * Upserts by canonical `threadId`, retaining imported transcript records
83 * from the current database row.
84 */
85 readonly upsert: (
86 runtime: ProviderSessionRuntime,
87 options?: ProviderSessionRuntimeUpsertOptions,
88 ) => Effect.Effect<void, ProviderSessionRuntimeRepositoryError>;
89
90 /** Record one source file without replacing the current session state. */
91 readonly recordImportedTranscript: (
92 input: RecordImportedTranscriptInput,
93 ) => Effect.Effect<void, ProviderSessionRuntimeRepositoryError>;
94
95 /**
96 * Read provider runtime state by canonical thread id.
97 */
98 readonly getByThreadId: (
99 input: GetProviderSessionRuntimeInput,
100 ) => Effect.Effect<
101 Option.Option<ProviderSessionRuntime>,
102 ProviderSessionRuntimeRepositoryError
103 >;
104
105 /**
106 * List provider runtime rows.
107 *
108 * Returned in ascending last-seen order. `excludeStopped` filters stopped
109 * rows in SQL. Long-lived installs keep thousands for their resume cursors.
110 */
111 readonly list: (options?: {
112 readonly excludeStopped?: boolean;
113 }) => Effect.Effect<
114 ReadonlyArray<ProviderSessionRuntime>,
115 ProviderSessionRuntimeRepositoryError
116 >;
117
118 /**
119 * Delete provider runtime state by canonical thread id.
120 */
121 readonly deleteByThreadId: (
122 input: DeleteProviderSessionRuntimeInput,
123 ) => Effect.Effect<void, ProviderSessionRuntimeRepositoryError>;
124 }
125>()("t3/persistence/ProviderSessionRuntime/ProviderSessionRuntimeRepository") {}
126
127const ProviderSessionRuntimeDbRowSchema = ProviderSessionRuntime.mapFields(
128 Struct.assign({
129 resumeCursor: Schema.NullOr(Schema.fromJsonString(Schema.Unknown)),
130 runtimePayload: Schema.NullOr(Schema.fromJsonString(Schema.Unknown)),
131 }),
132);
133
134const ProviderSessionRuntimeRawDbRowSchema = Schema.Struct({
135 threadId: Schema.String,
136 providerName: Schema.Unknown,
137 providerInstanceId: Schema.Unknown,
138 adapterKey: Schema.Unknown,
139 runtimeMode: Schema.Unknown,
140 status: Schema.Unknown,
141 lastSeenAt: Schema.Unknown,
142 resumeCursor: Schema.Unknown,
143 runtimePayload: Schema.Unknown,
144});
145
146const decodeRuntimeRow = Schema.decodeUnknownEffect(ProviderSessionRuntimeDbRowSchema);
147
148const GetRuntimeRequestSchema = Schema.Struct({
149 threadId: ThreadId,
150});
151
152const DeleteRuntimeRequestSchema = GetRuntimeRequestSchema;
153
154const RecordImportedTranscriptRequestSchema = RecordImportedTranscriptInput.mapFields(
155 Struct.assign({ source: Schema.fromJsonString(AgentSessionImportSource) }),
156);
157
158function toPersistenceSqlOrDecodeError(
159 sqlOperation: string,
160 decodeOperation: string,
161 correlation?: PersistenceErrorCorrelation,
162) {
163 return (cause: unknown): ProviderSessionRuntimeRepositoryError =>
164 Schema.isSchemaError(cause)
165 ? PersistenceDecodeError.fromSchemaError(decodeOperation, cause, correlation)
166 : new PersistenceSqlError({
167 operation: sqlOperation,
168 ...(correlation === undefined ? {} : { correlation }),
169 cause,
170 });
171}
172
173/** @public Service construction is part of the canonical Effect module API. */
174export const make = Effect.gen(function* () {
175 const sql = yield* SqlClient.SqlClient;
176
177 // Runtime writes can carry stale payloads. Only recordImportedTranscript may
178 // change source records, so restore that field from the row being updated.
179 const upsertRuntimeRow = SqlSchema.void({
180 Request: ProviderSessionRuntimeDbRowSchema,
181 execute: (runtime) =>
182 sql`
183 INSERT INTO provider_session_runtime (
184 thread_id,
185 provider_name,
186 provider_instance_id,
187 adapter_key,
188 runtime_mode,
189 status,
190 last_seen_at,
191 resume_cursor_json,
192 runtime_payload_json
193 )
194 VALUES (
195 ${runtime.threadId},
196 ${runtime.providerName},
197 ${runtime.providerInstanceId},
198 ${runtime.adapterKey},
199 ${runtime.runtimeMode},
200 ${runtime.status},
201 ${runtime.lastSeenAt},
202 ${runtime.resumeCursor},
203 CASE
204 WHEN json_type(${runtime.runtimePayload}) = 'object'
205 THEN json_remove(${runtime.runtimePayload}, '$.importedTranscripts')
206 ELSE ${runtime.runtimePayload}
207 END
208 )
209 ON CONFLICT (thread_id)
210 DO UPDATE SET
211 provider_name = excluded.provider_name,
212 provider_instance_id = excluded.provider_instance_id,
213 adapter_key = excluded.adapter_key,
214 runtime_mode = excluded.runtime_mode,
215 status = excluded.status,
216 last_seen_at = excluded.last_seen_at,
217 resume_cursor_json = excluded.resume_cursor_json,
218 runtime_payload_json = CASE
219 WHEN json_type(
220 CASE
221 WHEN json_valid(provider_session_runtime.runtime_payload_json)
222 THEN provider_session_runtime.runtime_payload_json
223 ELSE '{}'
224 END,
225 '$.importedTranscripts'
226 ) IS NOT NULL
227 THEN json_set(
228 CASE
229 WHEN json_type(excluded.runtime_payload_json) = 'object'
230 THEN excluded.runtime_payload_json
231 ELSE '{}'
232 END,
233 '$.importedTranscripts',
234 json_extract(provider_session_runtime.runtime_payload_json, '$.importedTranscripts')
235 )
236 ELSE excluded.runtime_payload_json
237 END
238 `,
239 });
240
241 const insertRuntimeRow = SqlSchema.void({
242 Request: ProviderSessionRuntimeDbRowSchema,
243 execute: (runtime) =>
244 sql`
245 INSERT INTO provider_session_runtime (
246 thread_id,
247 provider_name,
248 provider_instance_id,
249 adapter_key,
250 runtime_mode,
251 status,
252 last_seen_at,
253 resume_cursor_json,
254 runtime_payload_json
255 )
256 VALUES (
257 ${runtime.threadId},
258 ${runtime.providerName},
259 ${runtime.providerInstanceId},
260 ${runtime.adapterKey},
261 ${runtime.runtimeMode},
262 ${runtime.status},
263 ${runtime.lastSeenAt},
264 ${runtime.resumeCursor},
265 CASE
266 WHEN json_type(${runtime.runtimePayload}) = 'object'
267 THEN json_remove(${runtime.runtimePayload}, '$.importedTranscripts')
268 ELSE ${runtime.runtimePayload}
269 END
270 )
271 ON CONFLICT (thread_id) DO NOTHING
272 `,
273 });
274
275 const recordImportedTranscriptRow = SqlSchema.void({
276 Request: RecordImportedTranscriptRequestSchema,
277 execute: ({ threadId, source }) =>
278 sql`
279 WITH current_runtime AS (
280 SELECT CASE
281 WHEN json_valid(runtime_payload_json) THEN CASE
282 WHEN json_type(runtime_payload_json) = 'object' THEN runtime_payload_json
283 ELSE '{}'
284 END
285 ELSE '{}'
286 END AS payload
287 FROM provider_session_runtime
288 WHERE thread_id = ${threadId}
289 )
290 UPDATE provider_session_runtime
291 SET runtime_payload_json = (
292 SELECT json_set(
293 payload,
294 '$.importedTranscripts',
295 json((
296 SELECT json_group_array(json(value))
297 FROM (
298 SELECT value
299 FROM json_each(CASE
300 WHEN json_type(payload, '$.importedTranscripts') = 'array'
301 THEN json_extract(payload, '$.importedTranscripts')
302 ELSE '[]'
303 END)
304 WHERE CASE
305 WHEN type = 'object' THEN
306 json_extract(value, '$.providerInstanceId')
307 IS NOT json_extract(${source}, '$.providerInstanceId')
308 OR json_extract(value, '$.filePath') IS NOT json_extract(${source}, '$.filePath')
309 ELSE 0
310 END
311 UNION ALL
312 SELECT ${source} AS value
313 )
314 ))
315 )
316 FROM current_runtime
317 )
318 WHERE thread_id = ${threadId}
319 `,
320 });
321
322 const getRuntimeRowByThreadId = SqlSchema.findOneOption({
323 Request: GetRuntimeRequestSchema,
324 Result: ProviderSessionRuntimeRawDbRowSchema,
325 execute: ({ threadId }) =>
326 sql`
327 SELECT
328 thread_id AS "threadId",
329 provider_name AS "providerName",
330 provider_instance_id AS "providerInstanceId",
331 adapter_key AS "adapterKey",
332 runtime_mode AS "runtimeMode",
333 status,
334 last_seen_at AS "lastSeenAt",
335 resume_cursor_json AS "resumeCursor",
336 runtime_payload_json AS "runtimePayload"
337 FROM provider_session_runtime
338 WHERE thread_id = ${threadId}
339 `,
340 });
341
342 const listRuntimeRows = SqlSchema.findAll({
343 Request: Schema.Struct({ excludeStopped: Schema.Boolean }),
344 Result: ProviderSessionRuntimeRawDbRowSchema,
345 execute: ({ excludeStopped }) =>
346 sql`
347 SELECT
348 thread_id AS "threadId",
349 provider_name AS "providerName",
350 provider_instance_id AS "providerInstanceId",
351 adapter_key AS "adapterKey",
352 runtime_mode AS "runtimeMode",
353 status,
354 last_seen_at AS "lastSeenAt",
355 resume_cursor_json AS "resumeCursor",
356 runtime_payload_json AS "runtimePayload"
357 FROM provider_session_runtime
358 ${excludeStopped ? sql`WHERE status != 'stopped'` : sql``}
359 ORDER BY last_seen_at ASC, thread_id ASC
360 `,
361 });
362
363 const deleteRuntimeByThreadId = SqlSchema.void({
364 Request: DeleteRuntimeRequestSchema,
365 execute: ({ threadId }) =>
366 sql`
367 DELETE FROM provider_session_runtime
368 WHERE thread_id = ${threadId}
369 `,
370 });
371
372 const upsert: ProviderSessionRuntimeRepository["Service"]["upsert"] = (runtime, options) =>
373 (options?.onConflict === "ignore" ? insertRuntimeRow(runtime) : upsertRuntimeRow(runtime)).pipe(
374 Effect.mapError(
375 toPersistenceSqlOrDecodeError(
376 "ProviderSessionRuntimeRepository.upsert:query",
377 "ProviderSessionRuntimeRepository.upsert:encodeRequest",
378 { threadId: runtime.threadId },
379 ),
380 ),
381 );
382
383 const recordImportedTranscript: ProviderSessionRuntimeRepository["Service"]["recordImportedTranscript"] =
384 (input) =>
385 recordImportedTranscriptRow(input).pipe(
386 Effect.mapError(
387 toPersistenceSqlOrDecodeError(
388 "ProviderSessionRuntimeRepository.recordImportedTranscript:query",
389 "ProviderSessionRuntimeRepository.recordImportedTranscript:encodeRequest",
390 { threadId: input.threadId },
391 ),
392 ),
393 );
394
395 const getByThreadId: ProviderSessionRuntimeRepository["Service"]["getByThreadId"] = (input) =>
396 getRuntimeRowByThreadId(input).pipe(
397 Effect.mapError(
398 toPersistenceSqlOrDecodeError(
399 "ProviderSessionRuntimeRepository.getByThreadId:query",
400 "ProviderSessionRuntimeRepository.getByThreadId:decodeRow",
401 { threadId: input.threadId },
402 ),
403 ),
404 Effect.flatMap((runtimeRowOption) =>
405 Option.match(runtimeRowOption, {
406 onNone: () => Effect.succeedNone,
407 onSome: (row) =>
408 decodeRuntimeRow(row).pipe(
409 Effect.mapError((cause) =>
410 PersistenceDecodeError.fromSchemaError(
411 "ProviderSessionRuntimeRepository.getByThreadId:decodeRow",
412 cause,
413 { threadId: input.threadId },
414 ),
415 ),
416 Effect.asSome,
417 ),
418 }),
419 ),
420 );
421
422 const list: ProviderSessionRuntimeRepository["Service"]["list"] = (options) =>
423 listRuntimeRows({ excludeStopped: options?.excludeStopped === true }).pipe(
424 Effect.mapError(
425 toPersistenceSqlOrDecodeError(
426 "ProviderSessionRuntimeRepository.list:query",
427 "ProviderSessionRuntimeRepository.list:decodeRows",
428 ),
429 ),
430 Effect.flatMap((rows) =>
431 // Skip rows that no longer decode (e.g. written by an older build)
432 // instead of failing the whole list — one stale row must not disable
433 // every consumer that enumerates sessions, such as the reaper.
434 Effect.forEach(rows, (row) =>
435 decodeRuntimeRow(row).pipe(
436 Effect.asSome,
437 Effect.catch((cause) =>
438 Effect.logWarning("provider.session.runtime.row-skipped", {
439 threadId: row.threadId,
440 error: PersistenceDecodeError.fromSchemaError(
441 "ProviderSessionRuntimeRepository.list:decodeRows",
442 cause,
443 { threadId: row.threadId },
444 ).message,
445 }).pipe(Effect.as(Option.none<ProviderSessionRuntime>())),
446 ),
447 ),
448 ),
449 ),
450 Effect.map((decoded) =>
451 Arr.filterMap(decoded, (row) =>
452 Option.isSome(row) ? Result.succeed(row.value) : Result.failVoid,
453 ),
454 ),
455 );
456
457 const deleteByThreadId: ProviderSessionRuntimeRepository["Service"]["deleteByThreadId"] = (
458 input,
459 ) =>
460 deleteRuntimeByThreadId(input).pipe(
461 Effect.mapError(
462 (cause) =>
463 new PersistenceSqlError({
464 operation: "ProviderSessionRuntimeRepository.deleteByThreadId:query",
465 correlation: { threadId: input.threadId },
466 cause,
467 }),
468 ),
469 );
470
471 return {
472 upsert,
473 recordImportedTranscript,
474 getByThreadId,
475 list,
476 deleteByThreadId,
477 } satisfies ProviderSessionRuntimeRepository["Service"];
478});
479
480export const layer = Layer.effect(ProviderSessionRuntimeRepository, make);