authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-23 14:32:30-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
log7307079d4063a5c546a4c277dcc3b6594b41a846
tree5aaf1103261db502f6b8f0136afb54dd65989ff6
parent23861cf3002a7b15a30968de6edd80ebefdefa86

spec


30 files changed, 1901 insertions(+), 808 deletions(-)

.plans/15-effect-server.md+10-1
......@@ -1 +1,10 @@
1Rewrite `createServer` and `index.ts` to be Effect native.
\ No newline at end of file
1Rewrite `createServer` and `index.ts` to be Effect native.
2
3Maybe use `effect/unstable/Socket` for the web socket server
4- https://github.com/Effect-TS/effect-smol/blob/main/packages/effect/src/unstable/socket/SocketServer.ts
5- https://github.com/Effect-TS/effect-smol/blob/main/packages/platform-node/test/NodeSocket.test.ts
6
7- Migrate remaining runtime code to Effect
8 - `gitManager` -> `src/git`
9 - `terminalManager` -> `src/terminal` (Manager + PTY)
10 - ...
\ No newline at end of file
.vscode/settings.json+2-1
......@@ -4,5 +4,6 @@
44 "editor.codeActionsOnSave": {
55 "source.fixAll.oxc": "always"
66 },
7 "oxc.unusedDisableDirectives": "warn"
7 "oxc.unusedDisableDirectives": "warn",
8 "typescript.tsdk": "node_modules/typescript/lib"
89}
AGENTS.md+2
......@@ -41,3 +41,5 @@ Docs:
4141- Codex-Monitor (Tauri, feature-complete, strong reference implementation): https://github.com/Dimillian/CodexMonitor
4242
4343Use these as implementation references when designing protocol handling, UX flows, and operational safeguards.
44
45<!-- Dummy edit marker for workflow testing. -->
SPEC.md created+569
......@@ -0,0 +1,569 @@
1# Orchestration Interface Spec (Required State)
2
3Status: Required target architecture (not a description of current behavior).
4
5## 1. Identifier Schema (Branded)
6
7All identifiers are branded and never plain `string` in schema definitions.
8
9```ts
10const ThreadId = Schema.String.pipe(Schema.brand("ThreadId"));
11const ProjectId = Schema.String.pipe(Schema.brand("ProjectId"));
12const CommandId = Schema.String.pipe(Schema.brand("CommandId"));
13const EventId = Schema.String.pipe(Schema.brand("EventId"));
14const MessageId = Schema.String.pipe(Schema.brand("MessageId"));
15const TurnId = Schema.String.pipe(Schema.brand("TurnId"));
16
17const ProviderSessionId = Schema.String.pipe(Schema.brand("ProviderSessionId"));
18const ProviderThreadId = Schema.String.pipe(Schema.brand("ProviderThreadId"));
19const ProviderTurnId = Schema.String.pipe(Schema.brand("ProviderTurnId"));
20const ProviderItemId = Schema.String.pipe(Schema.brand("ProviderItemId"));
21const ApprovalRequestId = Schema.String.pipe(Schema.brand("ApprovalRequestId"));
22
23const CheckpointRef = Schema.String.pipe(Schema.brand("CheckpointRef"));
24```
25
26Rules:
27
28- `ThreadId` is app/orchestration thread identity.
29- `ProviderThreadId` is provider runtime thread identity.
30- They are never assigned to each other.
31- `ProviderSessionId` is internal server/provider routing identity; clients do not send it.
32- use e.g. `ThreadId.make(randomUUID())` to create new entity ids at the source (e.g. client create thread id and send to server, provider adapter creates the provider thread id from the codex app server etc)
33
34## 2. Client Commands (Domain Commands)
35
36All commands are sent through one RPC: `orchestration.dispatchCommand`.
37
38Output for every command: `DispatchResultSchema = { sequence: number }`.
39
40### 2.1 Client-Dispatchable Commands
41
421. `project.create`
43
44- Input:
45 - `commandId: CommandId`
46 - `projectId: ProjectId`
47 - `title: string`
48 - `workspaceRoot: string`
49 - `defaultModel?: string`
50 - `createdAt: IsoDateTime`
51- Output: `DispatchResult`
52
532. `project.meta.update`
54
55- Input:
56 - `commandId: CommandId`
57 - `projectId: ProjectId`
58 - optional `{ title, workspaceRoot, defaultModel }`
59- Output: `DispatchResult`
60
613. `project.delete`
62
63- Input: `commandId`, `projectId`
64- Output: `DispatchResult`
65
664. `thread.create`
67
68- Input:
69 - `commandId: CommandId`
70 - `threadId: ThreadId`
71 - `projectId: ProjectId`
72 - `title: string`
73 - `model: string`
74 - `branch: string | null`
75 - `worktreePath: string | null`
76 - `createdAt: IsoDateTime`
77- Output: `DispatchResult`
78
795. `thread.delete`
80
81- Input: `commandId`, `threadId`
82- Output: `DispatchResult`
83
846. `thread.meta.update`
85
86- Input: `commandId`, `threadId`, optional `{ title, model, branch, worktreePath }`
87- Output: `DispatchResult`
88
897. `thread.turn.start`
90
91- Input:
92 - `commandId`
93 - `threadId`
94 - `message`:
95 - `messageId: MessageId`
96 - `role: "user"`
97 - `text: string`
98 - `attachments: ChatAttachment[]` (schema-shared)
99 - `model?: string`
100 - `effort?: string`
101 - `createdAt`
102- Output: `DispatchResult`
103
1048. `thread.turn.interrupt`
105
106- Input: `commandId`, `threadId`, optional `turnId: TurnId`, `createdAt`
107- Output: `DispatchResult`
108
1099. `thread.approval.respond`
110
111- Input:
112 - `commandId`
113 - `threadId`
114 - `requestId: ApprovalRequestId`
115 - `decision: "accept" | "acceptForSession" | "decline" | "cancel"`
116 - `createdAt`
117- Output: `DispatchResult`
118
11910. `thread.checkpoint.revert`
120
121- Input: `commandId`, `threadId`, `turnCount: number`, `createdAt`
122- Output: `DispatchResult`
123
12411. `thread.session.stop`
125
126- Input: `commandId`, `threadId`, `createdAt`
127- Output: `DispatchResult`
128
129### 2.2 Internal-Only Commands (Not Client Dispatchable)
130
1311. `thread.session.set`
132
133- Server-owned projection update for session lifecycle.
134
1352. `thread.message.assistant.delta`
136
137- Server-owned incremental assistant content append.
138
1393. `thread.message.assistant.complete`
140
141- Server-owned assistant message completion marker.
142
1434. `thread.turn.diff.complete`
144
145- Server-owned checkpoint diff summary write.
146
1475. `thread.activity.append`
148
149- Server-owned activity feed append.
150
151## 3. Client RPC (Required)
152
153These are the required client-facing RPC methods.
154
1551. `orchestration.getSnapshot`
156
157- Input: `{}`
158- Output: `OrchestrationReadModel`
159
1602. `orchestration.dispatchCommand`
161
162- Input: `ClientOrchestrationCommand` (`Schema.Union` of all client-dispatchable commands)
163- Output: `DispatchResult`
164
1653. `orchestration.getTurnDiff`
166
167- Input:
168 - `threadId: ThreadId`
169 - `fromTurnCount: number`
170 - `toTurnCount: number`
171- Output:
172 - `threadId: ThreadId`
173 - `fromTurnCount: number`
174 - `toTurnCount: number`
175 - `diff: string`
176
1774. `orchestration.replayEvents` (optional operational/debug API)
178
179- Input: `{ fromSequenceExclusive: number }`
180- Output: `OrchestrationEvent[]`
181
182### 3.1 RPC To Remove
183
184Remove from client boundary:
185
186- `providers.startSession`
187- `providers.sendTurn`
188- `providers.interruptTurn`
189- `providers.respondToRequest`
190- `providers.stopSession`
191- `providers.listCheckpoints`
192- `providers.getCheckpointDiff`
193- `providers.revertToCheckpoint`
194
195Rationale:
196
197- Session/turn/checkpoint control belongs to orchestrator command path.
198- Snapshot already includes checkpoint summaries (`checkpoints`); only full textual diff needs explicit query RPC.
199
200## 4. Server Orchestrator Command <-> Event Schemas
201
202Each command deterministically yields one or more domain events that are persisted to the event store.
203
2041. `project.create` -> `project.created`
2052. `project.meta.update` -> `project.meta-updated`
2063. `project.delete` -> `project.deleted`
2074. `thread.create` -> `thread.created`
2085. `thread.delete` -> `thread.deleted`
2096. `thread.meta.update` -> `thread.meta-updated`
2107. `thread.turn.start` ->
211
212- `thread.message-sent` (user message append)
213- `thread.turn-start-requested`
214- then (async side effect success) `thread.session-set` + provider-driven message/turn events
215
2168. `thread.turn.interrupt` -> `thread.turn-interrupt-requested`
2179. `thread.approval.respond` -> `thread.approval-response-requested`
21810. `thread.checkpoint.revert` -> `thread.checkpoint-revert-requested` then `thread.reverted`
21911. `thread.session.stop` -> `thread.session-stop-requested` then `thread.session-set`
220
221Internal command/event path:
222
223- `thread.session.set` -> `thread.session-set`
224- `thread.message.assistant.delta` -> `thread.message-sent` (assistant streaming)
225- `thread.message.assistant.complete` -> `thread.message-sent` (streaming false)
226- `thread.turn.diff.complete` -> `thread.turn-diff-completed`
227- `thread.activity.append` -> `thread.activity-appended`
228
229## 5. Orchestration Events <-> Generic Provider Events <-> Raw Provider Events
230
231Required mapping pipeline:
232
233`Raw Provider Event (adapter-specific)`
234-> `GenericProviderRuntimeEvent` (provider-agnostic schema)
235-> `Internal Orchestration Command`
236-> `Orchestration Domain Event`
237
238### 5.1 Mapping Table for Codex Adapter
239
2401. Raw `thread/started`
241
242- Generic: `thread.started`
243- Internal command: `thread.session.set` (attach `ProviderThreadId`, status `ready|running`)
244- Domain event: `thread.session-set`
245
2462. Raw `turn/started`
247
248- Generic: `turn.started`
249- Internal command: `thread.session.set` (status `running`, active turn)
250- Domain event: `thread.session-set`
251
2523. Raw `item/agentMessage/delta`
253
254- Generic: `message.delta`
255- Internal command: `thread.message.assistant.delta`
256- Domain event: `thread.message-sent` (assistant, streaming)
257
2584. Raw `item/completed` (agent message)
259
260- Generic: `message.completed`
261- Internal command: `thread.message.assistant.complete`
262- Domain event: `thread.message-sent` (assistant, completed)
263
2645. Raw `turn/completed`
265
266- Generic: `turn.completed`
267- Internal commands:
268 - `thread.session.set` (ready/error)
269 - `thread.activity.append` (optional summary)
270- Domain events: `thread.session-set`, `thread.activity-appended`
271
2726. Raw approval request
273
274- Generic: `approval.requested`
275- Internal command: `thread.activity.append` (typed `requestId`, `requestKind`)
276- Domain event: `thread.activity-appended`
277
2787. Raw approval decision ack
279
280- Generic: `approval.resolved`
281- Internal command: `thread.activity.append`
282- Domain event: `thread.activity-appended`
283
2848. Raw tool start/complete
285
286- Generic: `tool.started` / `tool.completed`
287- Internal command: `thread.activity.append`
288- Domain event: `thread.activity-appended`
289
2909. Checkpoint capture (server-produced)
291
292- Generic: `checkpoint.captured`
293- Internal command: `thread.turn.diff.complete`
294- Domain event: `thread.turn-diff-completed`
295
29610. Raw/runtime error
297
298- Generic: `runtime.error`
299- Internal commands:
300 - `thread.session.set` (status `error`)
301 - `thread.activity.append` (`tone=error`)
302- Domain events: `thread.session-set`, `thread.activity-appended`
303
304## 6. Schema Section (Boundary + Reuse)
305
306### 6.1 Shared Schemas (contracts package)
307
308Expose to both web and server:
309
310- `OrchestrationReadModelSchema`
311- `ClientOrchestrationCommandSchema`
312- `OrchestrationEventSchema`
313- `OrchestrationRpcSchemas` (`getSnapshot`, `dispatchCommand`, `getTurnDiff`, optional replay)
314- `GenericProviderRuntimeEventSchema` (used for diagnostics/ingestion tests)
315- Shared message/activity/checkpoint summary schemas used in snapshot
316
317### 6.2 Server-Internal Schemas
318
319Not exposed to web:
320
321- `InternalOrchestrationCommandSchema`
322- Adapter raw-event schemas and provider-specific payload schemas
323- Provider session directory records and persistence-only row schemas
324- Git service cache/query schemas (non-orchestration read model)
325
326### 6.3 Reuse Requirements
327
328- Snapshot checkpoint summary schema is source of truth for checkpoint list UI.
329- `getTurnDiff` response reuses `ThreadId`/turn count brands and textual `diff` only.
330- Activity schemas are reused for approval/tool/runtime annotations; no separate UI-only structure.
331- Session schema must include both `threadId: ThreadId` and optional `providerThreadId: ProviderThreadId` (explicitly distinct).
332
333## 7. Persistence Model (Required)
334
335This section defines required persisted tables, the canonical persisted event envelope, and required projected/read tables.
336
337### 7.1 Write-Side Persisted Tables
338
3391. `orchestration_events` (append-only event store)
340
341- `sequence: number` (global monotonic sequence, primary key)
342- `eventId: EventId` (unique)
343- `aggregateKind: "project" | "thread"`
344- `streamId: ProjectId | ThreadId` (aggregate stream id)
345- `streamVersion: number` (per-aggregate monotonic version)
346- `eventType: OrchestrationEventType`
347- `occurredAt: IsoDateTime`
348- `commandId: CommandId | null`
349- `causationEventId: EventId | null`
350- `correlationId: CommandId | null`
351- `actorKind: "client" | "server" | "provider"`
352- `payload: OrchestrationEventPayload` (type-validated by `eventType`)
353- `metadata: OrchestrationEventMetadata`
354
3552. `orchestration_command_receipts` (idempotency + ack replay)
356
357- `commandId: CommandId` (primary key)
358- `aggregateKind: "project" | "thread"`
359- `aggregateId: ProjectId | ThreadId`
360- `acceptedAt: IsoDateTime`
361- `resultSequence: number`
362- `status: "accepted" | "rejected"`
363- `error: string | null`
364
3653. `checkpoint_diff_blobs` (large plaintext diffs; separate from summaries)
366
367- `threadId: ThreadId`
368- `fromTurnCount: number`
369- `toTurnCount: number`
370- `diff: string`
371- `createdAt: IsoDateTime`
372- unique key: `(threadId, fromTurnCount, toTurnCount)`
373
3744. `provider_session_runtime` (server-internal adapter resume state)
375
376- `providerSessionId: ProviderSessionId` (primary key)
377- `threadId: ThreadId`
378- `providerName: string`
379- `adapterKey: string`
380- `providerThreadId: ProviderThreadId | null`
381- `status: "starting" | "running" | "stopped" | "error"`
382- `lastSeenAt: IsoDateTime`
383- `resumeCursor: JsonValue | null` (adapter-specific opaque state)
384- `runtimePayload: JsonValue | null` (adapter-specific opaque state)
385
386### 7.2 Canonical Persisted Event Schema
387
388`OrchestrationPersistedEventSchema` (full envelope):
389
390```ts
391type OrchestrationPersistedEvent = {
392 sequence: number;
393 eventId: EventId;
394 aggregateKind: "project" | "thread";
395 streamId: ProjectId | ThreadId;
396 streamVersion: number;
397 eventType: OrchestrationEventType;
398 occurredAt: IsoDateTime;
399 commandId: CommandId | null;
400 causationEventId: EventId | null;
401 correlationId: CommandId | null;
402 actorKind: "client" | "server" | "provider";
403 payload: OrchestrationEventPayload;
404 metadata: {
405 providerSessionId?: ProviderSessionId;
406 providerThreadId?: ProviderThreadId;
407 providerTurnId?: ProviderTurnId;
408 providerItemId?: ProviderItemId;
409 adapterKey?: string;
410 requestId?: ApprovalRequestId;
411 ingestedAt?: IsoDateTime;
412 };
413};
414```
415
416Rules:
417
418- `payload` must be schema-discriminated by `eventType`.
419- provider identifiers only appear in `metadata` and provider-specific payload sub-shapes; they do not replace `ThreadId`.
420- `streamVersion` is concurrency guard for aggregate writes.
421
422### 7.3 Required Projected Tables (Read Models)
423
4241. `projection_projects`
425
426- `projectId: ProjectId` (primary key)
427- `title: string`
428- `workspaceRoot: string`
429- `defaultModel: string | null`
430- `createdAt: IsoDateTime`
431- `updatedAt: IsoDateTime`
432- `deletedAt: IsoDateTime | null`
433
4342. `projection_threads`
435
436- `threadId: ThreadId` (primary key)
437- `projectId: ProjectId`
438- `title: string`
439- `model: string`
440- `branch: string | null`
441- `worktreePath: string | null`
442- `latestTurnId: TurnId | null`
443- `createdAt: IsoDateTime`
444- `updatedAt: IsoDateTime`
445- `deletedAt: IsoDateTime | null`
446
4473. `projection_thread_messages`
448
449- `messageId: MessageId` (primary key)
450- `threadId: ThreadId`
451- `turnId: TurnId | null`
452- `role: "user" | "assistant" | "system"`
453- `text: string`
454- `isStreaming: boolean`
455- `createdAt: IsoDateTime`
456- `updatedAt: IsoDateTime`
457
4584. `projection_thread_activities`
459
460- `activityId: EventId` (primary key; derived from source event)
461- `threadId: ThreadId`
462- `turnId: TurnId | null`
463- `tone: "info" | "tool" | "approval" | "error"`
464- `kind: string`
465- `summary: string`
466- `payload: JsonValue`
467- `createdAt: IsoDateTime`
468
4695. `projection_thread_sessions`
470
471- `threadId: ThreadId` (primary key)
472- `status: "idle" | "starting" | "running" | "ready" | "interrupted" | "stopped" | "error"`
473- `providerName: string | null`
474- `providerSessionId: ProviderSessionId | null`
475- `providerThreadId: ProviderThreadId | null`
476- `activeTurnId: TurnId | null`
477- `lastError: string | null`
478- `updatedAt: IsoDateTime`
479
4806. `projection_thread_turns`
481
482- `turnId: TurnId` (primary key)
483- `threadId: ThreadId`
484- `turnCount: number`
485- `status: "running" | "completed" | "interrupted" | "error"`
486- `userMessageId: MessageId | null`
487- `assistantMessageId: MessageId | null`
488- `startedAt: IsoDateTime`
489- `completedAt: IsoDateTime | null`
490
4917. `projection_checkpoints`
492
493- `threadId: ThreadId`
494- `turnId: TurnId`
495- `checkpointTurnCount: number`
496- `checkpointRef: CheckpointRef`
497- `status: "ready" | "missing" | "error"`
498- `files: JsonArray` (typed as file diff summary schema)
499- `assistantMessageId: MessageId | null`
500- `completedAt: IsoDateTime`
501- unique key: `(threadId, checkpointTurnCount)`
502
5038. `projection_pending_approvals`
504
505- `requestId: ApprovalRequestId` (primary key)
506- `threadId: ThreadId`
507- `turnId: TurnId | null`
508- `status: "pending" | "resolved"`
509- `decision: "accept" | "acceptForSession" | "decline" | "cancel" | null`
510- `createdAt: IsoDateTime`
511- `resolvedAt: IsoDateTime | null`
512
5139. `projection_state`
514
515- `projector: string` (primary key; e.g. `threads`, `messages`, `sessions`)
516- `lastAppliedSequence: number`
517- `updatedAt: IsoDateTime`
518
519Projection consistency rules:
520
521- Every projector applies read-model row updates and `projection_state.lastAppliedSequence` in the same database transaction.
522- Optional debug field on projection rows: `lastEventSequence: number` (not required for correctness).
523
524### 7.4 Snapshot and RPC Requirements
525
5261. `orchestration.getSnapshot` is fully served from projection tables and returns `snapshotSequence: number`.
5272. Snapshot must include `projects[]` from `projection_projects`.
5283. Thread snapshot must include `checkpoints[]` from `projection_checkpoints`:
529- `turnId: TurnId`
530- `completedAt: IsoDateTime`
531- `status`
532- `files[]` (`path`, `kind`, `additions`, `deletions`)
533- `checkpointRef: CheckpointRef`
534- `assistantMessageId?: MessageId`
535- `checkpointTurnCount: number`
5364. Client does not require `listCheckpoints` RPC:
537- checkpoint list comes from snapshot projections
538- full diff text comes from `orchestration.getTurnDiff` backed by `checkpoint_diff_blobs`
5395. Provider session identity is not a client routing key:
540- client acts on `ThreadId`
541- server resolves provider session internally via `projection_thread_sessions`
5426. `snapshotSequence` is derived from `projection_state`:
543- if snapshot depends on multiple projectors, use the minimum `lastAppliedSequence` across those projectors.
5447. Event subscription handoff contract:
545- client performs `getSnapshot` and reads `snapshotSequence`
546- client subscribes/replays with `fromSequenceExclusive = snapshotSequence`
547- server guarantees no gap between snapshot visibility and subsequent event stream from that sequence.
548
549### 7.5 External Derived State (Non-Orchestration)
550
5511. Current git/worktree state is not projected from orchestration events.
5522. If persisted, it belongs in a separate git cache/read model (for example `git_state_cache`) owned by a git service.
5533. Orchestration may embed git metadata only when captured as a domain fact (for example checkpoint metadata at turn completion).
5544. Any RPC for current git status should be outside orchestration RPC (for example `git.getThreadState`).
555
556### 7.6 Existing Repository Placement
557
5581. `ProjectsRepository`
559- Fits as projection/query access on top of `projection_projects` plus orchestration command dispatch for writes.
560- Must not bypass command->event append path for mutations.
561
5622. `CheckpointsRepository`
563- Fits as projection/query access on top of `projection_checkpoints` plus `checkpoint_diff_blobs` (or git-service on-demand diff implementation).
564- Represents the same concept as prior `turn_diff_summary`; canonical naming is `checkpoint`.
565- Must not be an independent source of truth outside orchestration events.
566
5673. `ProviderSessionsRepository`
568- Fits as server-internal runtime persistence on top of `provider_session_runtime`, with domain-visible state projected into `projection_thread_sessions`.
569- Not part of client RPC/domain aggregate boundary.
apps/server/integration/TestProviderAdapter.integration.ts+10-14
......@@ -70,13 +70,11 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () {
7070 const startSession: ProviderAdapterShape<ProviderAdapterError>["startSession"] = (input) =>
7171 Effect.gen(function* () {
7272 if (input.provider !== undefined && input.provider !== PROVIDER) {
73 return yield* Effect.fail(
74 new ProviderAdapterValidationError({
75 provider: PROVIDER,
76 operation: "startSession",
77 issue: `Expected provider '${PROVIDER}' but received '${input.provider}'.`,
78 }),
79 );
73 return yield* new ProviderAdapterValidationError({
74 provider: PROVIDER,
75 operation: "startSession",
76 issue: `Expected provider '${PROVIDER}' but received '${input.provider}'.`,
77 });
8078 }
8179
8280 sessionCount += 1;
......@@ -121,13 +119,11 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () {
121119
122120 const response = state.queuedResponses.shift();
123121 if (!response) {
124 return yield* Effect.fail(
125 new ProviderAdapterValidationError({
126 provider: PROVIDER,
127 operation: "sendTurn",
128 issue: `No queued turn response for session ${input.sessionId}.`,
129 }),
130 );
122 return yield* new ProviderAdapterValidationError({
123 provider: PROVIDER,
124 operation: "sendTurn",
125 issue: `No queued turn response for session ${input.sessionId}.`,
126 });
131127 }
132128
133129 const assistantDeltas: string[] = [];
apps/server/package.json+3
......@@ -13,11 +13,13 @@
1313 "dev": "VITE_DEV_SERVER_URL=${VITE_DEV_SERVER_URL:-http://localhost:5173} bun run src/index.ts",
1414 "build": "tsdown && node scripts/bundle-client.ts",
1515 "start": "node dist/index.mjs",
16 "prepare": "effect-language-service patch",
1617 "typecheck": "tsc --noEmit",
1718 "test": "vitest run"
1819 },
1920 "dependencies": {
2021 "@effect/platform-node": "4.0.0-beta.10",
22 "@effect/sql-sqlite-bun": "4.0.0-beta.10",
2123 "@pierre/diffs": "^1.1.0-beta.16",
2224 "effect": "4.0.0-beta.10",
2325 "node-pty": "^1.1.0",
......@@ -25,6 +27,7 @@
2527 "ws": "^8.18.0"
2628 },
2729 "devDependencies": {
30 "@effect/language-service": "0.75.1",
2831 "@effect/vitest": "4.0.0-beta.10",
2932 "@t3tools/contracts": "workspace:*",
3033 "@t3tools/web": "workspace:*",
apps/server/src/checkpointing/Layers/CheckpointService.test.ts+136-28
......@@ -60,41 +60,56 @@ const makeGitRepository = Effect.gen(function* () {
6060class InMemoryProviderAdapter implements ProviderAdapterShape<ProviderAdapterError> {
6161 readonly provider = "codex" as const;
6262
63 private active = true;
6463 private readThreadFailure: ProviderAdapterError | undefined;
65 private snapshot: ProviderThreadSnapshot;
66 private readonly session: ProviderSession;
64 private readonly sessions = new Map<string, ProviderSession>();
65 private readonly snapshots = new Map<string, ProviderThreadSnapshot>();
6766
6867 constructor(
6968 readonly sessionId: string,
7069 readonly cwd: string,
7170 turns: ReadonlyArray<ProviderThreadTurnSnapshot>,
7271 ) {
73 const now = new Date().toISOString();
74 this.session = {
72 this.addSession({
7573 sessionId,
74 threadId: "thread-1",
75 turns,
76 });
77 }
78
79 addSession(input: {
80 readonly sessionId: string;
81 readonly threadId: string;
82 readonly turns: ReadonlyArray<ProviderThreadTurnSnapshot>;
83 }): void {
84 const now = new Date().toISOString();
85 this.sessions.set(input.sessionId, {
86 sessionId: input.sessionId,
7687 provider: "codex",
7788 status: "ready",
78 threadId: "thread-1",
79 cwd,
89 threadId: input.threadId,
90 cwd: this.cwd,
8091 createdAt: now,
8192 updatedAt: now,
82 };
83 this.snapshot = {
84 threadId: this.session.threadId ?? "thread-1",
85 turns,
86 };
93 });
94 this.snapshots.set(input.sessionId, {
95 threadId: input.threadId,
96 turns: input.turns,
97 });
8798 }
8899
89 setTurns(turns: ReadonlyArray<ProviderThreadTurnSnapshot>): void {
90 this.snapshot = {
91 threadId: this.snapshot.threadId,
100 setTurns(turns: ReadonlyArray<ProviderThreadTurnSnapshot>, sessionId = this.sessionId): void {
101 const snapshot = this.snapshots.get(sessionId);
102 if (!snapshot) {
103 return;
104 }
105 this.snapshots.set(sessionId, {
106 threadId: snapshot.threadId,
92107 turns,
93 };
108 });
94109 }
95110
96 getTurns(): ReadonlyArray<ProviderThreadTurnSnapshot> {
97 return this.snapshot.turns;
111 getTurns(sessionId = this.sessionId): ReadonlyArray<ProviderThreadTurnSnapshot> {
112 return this.snapshots.get(sessionId)?.turns ?? [];
98113 }
99114
100115 setReadThreadFailure(error: ProviderAdapterError | undefined): void {
......@@ -136,16 +151,15 @@ class InMemoryProviderAdapter implements ProviderAdapterShape<ProviderAdapterErr
136151
137152 readonly stopSession = (sessionId: string): Effect.Effect<void, ProviderAdapterError> =>
138153 Effect.sync(() => {
139 if (sessionId === this.sessionId) {
140 this.active = false;
141 }
154 this.sessions.delete(sessionId);
155 this.snapshots.delete(sessionId);
142156 });
143157
144158 readonly listSessions = (): Effect.Effect<ReadonlyArray<ProviderSession>> =>
145 Effect.sync(() => (this.active ? [this.session] : []));
159 Effect.sync(() => Array.from(this.sessions.values()));
146160
147161 readonly hasSession = (sessionId: string): Effect.Effect<boolean> =>
148 Effect.succeed(this.active && sessionId === this.sessionId);
162 Effect.succeed(this.sessions.has(sessionId));
149163
150164 readonly readThread = (
151165 sessionId: string,
......@@ -155,7 +169,18 @@ class InMemoryProviderAdapter implements ProviderAdapterShape<ProviderAdapterErr
155169 hasSession
156170 ? this.readThreadFailure
157171 ? Effect.fail(this.readThreadFailure)
158 : Effect.succeed(this.snapshot)
172 : (() => {
173 const snapshot = this.snapshots.get(sessionId);
174 if (!snapshot) {
175 return Effect.fail(
176 new ProviderAdapterSessionNotFoundError({
177 provider: this.provider,
178 sessionId,
179 }),
180 );
181 }
182 return Effect.succeed(snapshot);
183 })()
159184 : Effect.fail(
160185 new ProviderAdapterSessionNotFoundError({
161186 provider: this.provider,
......@@ -184,14 +209,15 @@ class InMemoryProviderAdapter implements ProviderAdapterShape<ProviderAdapterErr
184209 threadId: snapshot.threadId,
185210 turns: snapshot.turns.slice(0, snapshot.turns.length - numTurns),
186211 } satisfies ProviderThreadSnapshot;
187 this.snapshot = next;
212 this.snapshots.set(sessionId, next);
188213 return Effect.succeed(next);
189214 }),
190215 );
191216
192217 readonly stopAll = (): Effect.Effect<void, ProviderAdapterError> =>
193218 Effect.sync(() => {
194 this.active = false;
219 this.sessions.clear();
220 this.snapshots.clear();
195221 });
196222
197223 readonly streamEvents = Stream.empty;
......@@ -315,6 +341,88 @@ it.effect("captures checkpoints, diffs refs, and reverts workspace + turns", ()
315341 }).pipe(Effect.provide(NodeServices.layer)),
316342);
317343
344it.effect(
345 "CheckpointServiceLive reuses checkpoint history after session churn on the same thread",
346 () =>
347 Effect.gen(function* () {
348 const firstTurn: ProviderThreadTurnSnapshot = {
349 id: "turn-1",
350 items: [{ type: "userMessage", content: [{ type: "text", text: "first" }] }],
351 };
352 const secondTurn: ProviderThreadTurnSnapshot = {
353 id: "turn-2",
354 items: [{ type: "agentMessage", text: "second" }],
355 };
356
357 const fixture = yield* makeFixture([]);
358
359 yield* Effect.gen(function* () {
360 const fs = yield* FileSystem.FileSystem;
361 const path = yield* Path.Path;
362 const service = yield* CheckpointService;
363 const directory = yield* ProviderSessionDirectory;
364
365 yield* directory.upsert({
366 sessionId: fixture.sessionId,
367 provider: "codex",
368 threadId: "thread-1",
369 });
370 yield* service.initializeForSession({
371 providerSessionId: fixture.sessionId,
372 cwd: fixture.cwd,
373 });
374
375 yield* fs.writeFileString(path.join(fixture.cwd, "README.md"), "v2\n");
376 fixture.adapter.setTurns([firstTurn]);
377 yield* service.captureCurrentTurn({
378 providerSessionId: fixture.sessionId,
379 });
380
381 const resumedSessionId = "sess-2";
382 fixture.adapter.addSession({
383 sessionId: resumedSessionId,
384 threadId: "thread-1",
385 turns: [firstTurn, secondTurn],
386 });
387 yield* directory.upsert({
388 sessionId: resumedSessionId,
389 provider: "codex",
390 threadId: "thread-1",
391 });
392
393 yield* fs.writeFileString(path.join(fixture.cwd, "README.md"), "v3\n");
394 yield* service.captureCurrentTurn({
395 providerSessionId: resumedSessionId,
396 });
397
398 const listed = yield* service.listCheckpoints({ sessionId: resumedSessionId });
399 assert.equal(listed.threadId, "thread-1");
400 assert.equal(listed.checkpoints.length, 3);
401 assert.equal(listed.checkpoints[2]?.id, "turn-2");
402 assert.equal(listed.checkpoints[2]?.isCurrent, true);
403
404 const diff = yield* service.getCheckpointDiff({
405 sessionId: resumedSessionId,
406 fromTurnCount: 1,
407 toTurnCount: 2,
408 });
409 assert.equal(diff.threadId, "thread-1");
410 assert.equal(diff.diff.includes("-v2"), true);
411 assert.equal(diff.diff.includes("+v3"), true);
412
413 const reverted = yield* service.revertToCheckpoint({
414 sessionId: resumedSessionId,
415 turnCount: 1,
416 });
417 const afterRevert = yield* fs.readFileString(path.join(fixture.cwd, "README.md"));
418 assert.equal(afterRevert, "v2\n");
419 assert.equal(reverted.turnCount, 1);
420 assert.equal(reverted.rolledBackTurns, 1);
421 assert.equal(fixture.adapter.getTurns(resumedSessionId).length, 1);
422 }).pipe(Effect.provide(fixture.layer));
423 }).pipe(Effect.provide(NodeServices.layer)),
424);
425
318426it.effect(
319427 "CheckpointServiceLive fails with CheckpointSessionNotFoundError for unknown sessions",
320428 () =>
......@@ -374,7 +482,7 @@ it.effect(
374482 });
375483
376484 const checkpoint = yield* checkpointRepository.getCheckpoint({
377 providerSessionId: fixture.sessionId,
485 threadId: "thread-1",
378486 turnCount: 1,
379487 });
380488 if (checkpoint._tag !== "Some") {
......@@ -429,7 +537,7 @@ it.effect("CheckpointServiceLive initializes root checkpoints without requiring
429537 });
430538
431539 const root = yield* checkpointRepository.getCheckpoint({
432 providerSessionId: fixture.sessionId,
540 threadId: "thread-1",
433541 turnCount: 0,
434542 });
435543 assert.equal(root._tag, "Some");
apps/server/src/checkpointing/Layers/CheckpointService.ts+129-103
......@@ -481,11 +481,7 @@ const makeCheckpointService = Effect.gen(function* () {
481481 ): Effect.Effect<string, CheckpointServiceError> =>
482482 directory.getThreadId(sessionId).pipe(
483483 Effect.mapError((cause) =>
484 mapDirectoryLookupError(
485 sessionId,
486 `${operation}:getThreadId`,
487 cause,
488 ),
484 mapDirectoryLookupError(sessionId, `${operation}:getThreadId`, cause),
489485 ),
490486 Effect.flatMap((threadId) =>
491487 Option.match(threadId, {
......@@ -495,10 +491,7 @@ const makeCheckpointService = Effect.gen(function* () {
495491 return Effect.succeed(normalizedThreadId);
496492 }
497493 return Effect.fail(
498 invariantError(
499 operation,
500 `Resolved empty thread id for session ${sessionId}.`,
501 ),
494 invariantError(operation, `Resolved empty thread id for session ${sessionId}.`),
502495 );
503496 },
504497 onNone: () =>
......@@ -512,6 +505,30 @@ const makeCheckpointService = Effect.gen(function* () {
512505 ),
513506 );
514507
508 const ensureThreadMatchesSnapshot = (
509 resolvedThreadId: string,
510 snapshot: ProviderThreadSnapshot,
511 operation: string,
512 ): Effect.Effect<void, CheckpointInvariantError> => {
513 const snapshotThreadId = snapshot.threadId.trim();
514 if (snapshotThreadId.length === 0) {
515 return Effect.fail(
516 invariantError(operation, "Provider thread snapshot is missing threadId."),
517 );
518 }
519
520 if (snapshotThreadId !== resolvedThreadId) {
521 return Effect.fail(
522 invariantError(
523 operation,
524 `Resolved thread ${resolvedThreadId} does not match provider snapshot thread ${snapshotThreadId}.`,
525 ),
526 );
527 }
528
529 return Effect.void;
530 };
531
515532 const rollbackThreadSnapshot = (
516533 sessionId: string,
517534 adapter: ProviderAdapterShape<ProviderAdapterError>,
......@@ -601,29 +618,30 @@ const makeCheckpointService = Effect.gen(function* () {
601618 };
602619
603620 const ensureRootCheckpointMetadata = (
604 providerSessionId: string,
621 threadId: string,
605622 operation: string,
606623 ): Effect.Effect<void, CheckpointServiceError> =>
607 checkpointRepository.getCheckpoint({ providerSessionId, turnCount: 0 }).pipe(
608 Effect.flatMap((rootEntry) =>
609 Option.isSome(rootEntry)
610 ? Effect.void
611 : Effect.fail(
612 invariantError(
613 operation,
614 `Missing root checkpoint metadata for session ${providerSessionId}.`,
624 checkpointRepository
625 .getCheckpoint({ threadId, turnCount: 0 })
626 .pipe(
627 Effect.flatMap((rootEntry) =>
628 Option.isSome(rootEntry)
629 ? Effect.void
630 : Effect.fail(
631 invariantError(
632 operation,
633 `Missing root checkpoint metadata for thread ${threadId}.`,
634 ),
615635 ),
616 ),
617 ),
618 );
636 ),
637 );
619638
620639 const getCheckpointRefOrFail = (
621 providerSessionId: string,
622640 threadId: string,
623641 turnCount: number,
624642 operation: string,
625643 ): Effect.Effect<string, CheckpointServiceError> =>
626 checkpointRepository.getCheckpoint({ providerSessionId, turnCount }).pipe(
644 checkpointRepository.getCheckpoint({ threadId, turnCount }).pipe(
627645 Effect.mapError((error): CheckpointServiceError => error),
628646 Effect.flatMap((checkpoint): Effect.Effect<string, CheckpointServiceError> => {
629647 if (Option.isNone(checkpoint)) {
......@@ -716,17 +734,15 @@ const makeCheckpointService = Effect.gen(function* () {
716734 }
717735
718736 const existingRoot = yield* checkpointRepository.getCheckpoint({
719 providerSessionId: input.providerSessionId,
737 threadId,
720738 turnCount: 0,
721739 });
722740 if (Option.isSome(existingRoot)) {
723741 const existingRef = existingRoot.value.checkpointRef.trim();
724742 if (existingRef.length === 0) {
725 return yield* Effect.fail(
726 invariantError(
727 "CheckpointService.initializeForSession",
728 "Existing root checkpoint metadata is missing checkpointRef.",
729 ),
743 return yield* invariantError(
744 "CheckpointService.initializeForSession",
745 "Existing root checkpoint metadata is missing checkpointRef.",
730746 );
731747 }
732748 const exists = yield* store.hasCheckpointRef({
......@@ -734,11 +750,9 @@ const makeCheckpointService = Effect.gen(function* () {
734750 checkpointRef: existingRef,
735751 });
736752 if (!exists) {
737 return yield* Effect.fail(
738 invariantError(
739 "CheckpointService.initializeForSession",
740 "Root checkpoint metadata exists but filesystem ref is missing.",
741 ),
753 return yield* invariantError(
754 "CheckpointService.initializeForSession",
755 "Root checkpoint metadata exists but filesystem ref is missing.",
742756 );
743757 }
744758 yield* setTrackedCheckpointCwd(input.providerSessionId, input.cwd);
......@@ -770,6 +784,15 @@ const makeCheckpointService = Effect.gen(function* () {
770784 adapter,
771785 "CheckpointService.captureCurrentTurn:readThread",
772786 );
787 const threadId = yield* resolveThreadIdForSession(
788 input.providerSessionId,
789 "CheckpointService.captureCurrentTurn",
790 );
791 yield* ensureThreadMatchesSnapshot(
792 threadId,
793 snapshot,
794 "CheckpointService.captureCurrentTurn",
795 );
773796 if (snapshot.turns.length === 0) {
774797 return;
775798 }
......@@ -784,23 +807,18 @@ const makeCheckpointService = Effect.gen(function* () {
784807 "CheckpointService.captureCurrentTurn",
785808 );
786809
787 yield* ensureRootCheckpointMetadata(
788 input.providerSessionId,
789 "CheckpointService.captureCurrentTurn",
790 );
810 yield* ensureRootCheckpointMetadata(threadId, "CheckpointService.captureCurrentTurn");
791811
792812 const existing = yield* checkpointRepository.getCheckpoint({
793 providerSessionId: input.providerSessionId,
813 threadId,
794814 turnCount: checkpoint.turnCount,
795815 });
796816 if (Option.isSome(existing)) {
797817 const checkpointRef = existing.value.checkpointRef.trim();
798818 if (checkpointRef.length === 0) {
799 return yield* Effect.fail(
800 invariantError(
801 "CheckpointService.captureCurrentTurn",
802 `Checkpoint metadata for turn ${checkpoint.turnCount} is missing checkpointRef.`,
803 ),
819 return yield* invariantError(
820 "CheckpointService.captureCurrentTurn",
821 `Checkpoint metadata for turn ${checkpoint.turnCount} is missing checkpointRef.`,
804822 );
805823 }
806824 const exists = yield* store.hasCheckpointRef({
......@@ -808,11 +826,9 @@ const makeCheckpointService = Effect.gen(function* () {
808826 checkpointRef,
809827 });
810828 if (!exists) {
811 return yield* Effect.fail(
812 invariantError(
813 "CheckpointService.captureCurrentTurn",
814 `Checkpoint metadata exists for turn ${checkpoint.turnCount} but filesystem ref is missing.`,
815 ),
829 return yield* invariantError(
830 "CheckpointService.captureCurrentTurn",
831 `Checkpoint metadata exists for turn ${checkpoint.turnCount} but filesystem ref is missing.`,
816832 );
817833 }
818834 return;
......@@ -821,7 +837,7 @@ const makeCheckpointService = Effect.gen(function* () {
821837 yield* captureAndPersistCheckpoint({
822838 providerSessionId: input.providerSessionId,
823839 cwd: resolution.checkpointCwd.value,
824 threadId: snapshot.threadId,
840 threadId,
825841 checkpoint,
826842 });
827843 }),
......@@ -842,31 +858,36 @@ const makeCheckpointService = Effect.gen(function* () {
842858 adapter,
843859 "CheckpointService.listCheckpoints:readThread",
844860 );
861 const threadId = yield* resolveThreadIdForSession(
862 input.sessionId,
863 "CheckpointService.listCheckpoints",
864 );
865 yield* ensureThreadMatchesSnapshot(
866 threadId,
867 snapshot,
868 "CheckpointService.listCheckpoints",
869 );
845870
846871 const resolution = yield* resolveCheckpointCwd(input.sessionId, adapter);
847872 if (Option.isNone(resolution.checkpointCwd)) {
848 return yield* Effect.fail(
849 new CheckpointRepositoryError({
850 cwd: resolution.candidateCwd,
851 detail: "Filesystem checkpoints are unavailable for this session.",
852 }),
853 );
873 return yield* new CheckpointRepositoryError({
874 cwd: resolution.candidateCwd,
875 detail: "Filesystem checkpoints are unavailable for this session.",
876 });
854877 }
855878
856879 const checkpoints = yield* checkpointRepository.listCheckpoints({
857 providerSessionId: input.sessionId,
880 threadId,
858881 });
859882 if (checkpoints.length === 0) {
860 return yield* Effect.fail(
861 invariantError(
862 "CheckpointService.listCheckpoints",
863 `Missing checkpoint metadata for active session ${input.sessionId}.`,
864 ),
883 return yield* invariantError(
884 "CheckpointService.listCheckpoints",
885 `Missing checkpoint metadata for active session ${input.sessionId}.`,
865886 );
866887 }
867888
868889 return {
869 threadId: snapshot.threadId,
890 threadId,
870891 checkpoints: Array.from(checkpoints),
871892 };
872893 }),
......@@ -887,35 +908,38 @@ const makeCheckpointService = Effect.gen(function* () {
887908 adapter,
888909 "CheckpointService.getCheckpointDiff:readThread",
889910 );
911 const threadId = yield* resolveThreadIdForSession(
912 input.sessionId,
913 "CheckpointService.getCheckpointDiff",
914 );
915 yield* ensureThreadMatchesSnapshot(
916 threadId,
917 snapshot,
918 "CheckpointService.getCheckpointDiff",
919 );
890920
891921 if (input.toTurnCount > snapshot.turns.length) {
892 return yield* Effect.fail(
893 validationError(
894 "CheckpointService.getCheckpointDiff",
895 `Checkpoint turn count ${input.toTurnCount} exceeds current turn count ${snapshot.turns.length}.`,
896 ),
922 return yield* validationError(
923 "CheckpointService.getCheckpointDiff",
924 `Checkpoint turn count ${input.toTurnCount} exceeds current turn count ${snapshot.turns.length}.`,
897925 );
898926 }
899927
900928 const resolution = yield* resolveCheckpointCwd(input.sessionId, adapter);
901929 if (Option.isNone(resolution.checkpointCwd)) {
902 return yield* Effect.fail(
903 new CheckpointRepositoryError({
904 cwd: resolution.candidateCwd,
905 detail: "Filesystem checkpoints are unavailable for this session.",
906 }),
907 );
930 return yield* new CheckpointRepositoryError({
931 cwd: resolution.candidateCwd,
932 detail: "Filesystem checkpoints are unavailable for this session.",
933 });
908934 }
909935
910936 const fromCheckpointRef = yield* getCheckpointRefOrFail(
911 input.sessionId,
912 snapshot.threadId,
937 threadId,
913938 input.fromTurnCount,
914939 "CheckpointService.getCheckpointDiff:fromCheckpoint",
915940 );
916941 const toCheckpointRef = yield* getCheckpointRefOrFail(
917 input.sessionId,
918 snapshot.threadId,
942 threadId,
919943 input.toTurnCount,
920944 "CheckpointService.getCheckpointDiff:toCheckpoint",
921945 );
......@@ -928,7 +952,7 @@ const makeCheckpointService = Effect.gen(function* () {
928952 });
929953
930954 return {
931 threadId: snapshot.threadId,
955 threadId,
932956 fromTurnCount: input.fromTurnCount,
933957 toTurnCount: input.toTurnCount,
934958 diff,
......@@ -951,31 +975,35 @@ const makeCheckpointService = Effect.gen(function* () {
951975 adapter,
952976 "CheckpointService.revertToCheckpoint:readThread",
953977 );
978 const threadId = yield* resolveThreadIdForSession(
979 input.sessionId,
980 "CheckpointService.revertToCheckpoint",
981 );
982 yield* ensureThreadMatchesSnapshot(
983 threadId,
984 beforeSnapshot,
985 "CheckpointService.revertToCheckpoint",
986 );
954987 const currentTurnCount = beforeSnapshot.turns.length;
955988
956989 if (input.turnCount > currentTurnCount) {
957 return yield* Effect.fail(
958 validationError(
959 "CheckpointService.revertToCheckpoint",
960 `Checkpoint turn count ${input.turnCount} exceeds current turn count ${currentTurnCount}.`,
961 ),
990 return yield* validationError(
991 "CheckpointService.revertToCheckpoint",
992 `Checkpoint turn count ${input.turnCount} exceeds current turn count ${currentTurnCount}.`,
962993 );
963994 }
964995
965996 const resolution = yield* resolveCheckpointCwd(input.sessionId, adapter);
966997 if (Option.isNone(resolution.checkpointCwd)) {
967 return yield* Effect.fail(
968 new CheckpointRepositoryError({
969 cwd: resolution.candidateCwd,
970 detail: "Filesystem checkpoints are unavailable for this session.",
971 }),
972 );
998 return yield* new CheckpointRepositoryError({
999 cwd: resolution.candidateCwd,
1000 detail: "Filesystem checkpoints are unavailable for this session.",
1001 });
9731002 }
9741003
9751004 const checkpointCwd = resolution.checkpointCwd.value;
9761005 const checkpointRef = yield* getCheckpointRefOrFail(
977 input.sessionId,
978 beforeSnapshot.threadId,
1006 threadId,
9791007 input.turnCount,
9801008 "CheckpointService.revertToCheckpoint:checkpointLookup",
9811009 );
......@@ -986,13 +1014,11 @@ const makeCheckpointService = Effect.gen(function* () {
9861014 fallbackToHead: input.turnCount === 0,
9871015 });
9881016 if (!restored) {
989 return yield* Effect.fail(
990 new CheckpointUnavailableError({
991 threadId: beforeSnapshot.threadId,
992 turnCount: input.turnCount,
993 detail: `Filesystem checkpoint is unavailable for turn ${input.turnCount} in thread ${beforeSnapshot.threadId}.`,
994 }),
995 );
1017 return yield* new CheckpointUnavailableError({
1018 threadId,
1019 turnCount: input.turnCount,
1020 detail: `Filesystem checkpoint is unavailable for turn ${input.turnCount} in thread ${threadId}.`,
1021 });
9961022 }
9971023
9981024 const requestedRollbackTurns = currentTurnCount - input.turnCount;
......@@ -1015,7 +1041,7 @@ const makeCheckpointService = Effect.gen(function* () {
10151041 staleTurnCounts,
10161042 (turnCount) =>
10171043 checkpointRepository.getCheckpoint({
1018 providerSessionId: input.sessionId,
1044 threadId,
10191045 turnCount,
10201046 }),
10211047 { discard: false },
......@@ -1034,11 +1060,11 @@ const makeCheckpointService = Effect.gen(function* () {
10341060 checkpointRefs: staleCheckpointRefs,
10351061 });
10361062 yield* checkpointRepository.deleteAfterTurn({
1037 providerSessionId: input.sessionId,
1063 threadId,
10381064 maxTurnCount: input.turnCount,
10391065 });
10401066 const checkpoints = yield* checkpointRepository.listCheckpoints({
1041 providerSessionId: input.sessionId,
1067 threadId,
10421068 });
10431069 const currentCheckpoint =
10441070 checkpoints.find((checkpoint) => checkpoint.isCurrent) ??
......@@ -1047,7 +1073,7 @@ const makeCheckpointService = Effect.gen(function* () {
10471073 const rolledBackTurns = Math.max(0, currentTurnCount - afterSnapshot.turns.length);
10481074
10491075 return {
1050 threadId: afterSnapshot.threadId,
1076 threadId,
10511077 turnCount: currentCheckpoint?.turnCount ?? 0,
10521078 messageCount: currentCheckpoint?.messageCount ?? 0,
10531079 rolledBackTurns,
apps/server/src/index.ts+2
......@@ -11,6 +11,8 @@ import { createServer } from "./wsServer";
1111fixPath();
1212
1313const DEFAULT_PORT = 3773;
14// Dummy comment for a no-op change request.
15// Additional dummy comment requested in chat.
1416const cwd = process.cwd();
1517const logger = createLogger("server");
1618
apps/server/src/persistence/Layers/Checkpoints.test.ts+101-68
......@@ -11,72 +11,105 @@ function makeCheckpointRepositoryLayer() {
1111
1212it.effect("CheckpointRepositoryLive upserts, lists, reads, and prunes checkpoint metadata", () =>
1313 Effect.gen(function* () {
14 yield* Effect.gen(function* () {
15 const repository = yield* CheckpointRepository;
16
17 yield* repository.upsertCheckpoint({
18 providerSessionId: "sess-1",
19 threadId: "thread-1",
20 checkpointId: "root",
21 checkpointRef: "refs/t3/checkpoints/thread-1/turn/0",
22 turnCount: 0,
23 messageCount: 0,
24 label: "Start of conversation",
25 createdAt: new Date().toISOString(),
26 });
27
28 yield* repository.upsertCheckpoint({
29 providerSessionId: "sess-1",
30 threadId: "thread-1",
31 checkpointId: "turn-1",
32 checkpointRef: "refs/t3/checkpoints/thread-1/turn/1",
33 turnCount: 1,
34 messageCount: 2,
35 label: "Turn 1",
36 preview: "hello",
37 createdAt: new Date().toISOString(),
38 });
39
40 yield* repository.upsertCheckpoint({
41 providerSessionId: "sess-1",
42 threadId: "thread-1",
43 checkpointId: "turn-2",
44 checkpointRef: "refs/t3/checkpoints/thread-1/turn/2",
45 turnCount: 2,
46 messageCount: 4,
47 label: "Turn 2",
48 preview: "done",
49 createdAt: new Date().toISOString(),
50 });
51
52 const checkpoints = yield* repository.listCheckpoints({ providerSessionId: "sess-1" });
53 assert.equal(checkpoints.length, 3);
54 assert.equal(checkpoints[0]?.id, "root");
55 assert.equal(checkpoints[0]?.isCurrent, false);
56 assert.equal(checkpoints[2]?.id, "turn-2");
57 assert.equal(checkpoints[2]?.isCurrent, true);
58
59 const one = yield* repository.getCheckpoint({ providerSessionId: "sess-1", turnCount: 1 });
60 assert.equal(Option.isSome(one), true);
61 if (Option.isSome(one)) {
62 assert.equal(one.value.id, "turn-1");
63 assert.equal(one.value.isCurrent, false);
64 assert.equal(one.value.checkpointRef, "refs/t3/checkpoints/thread-1/turn/1");
65 }
66
67 yield* repository.deleteAfterTurn({
68 providerSessionId: "sess-1",
69 maxTurnCount: 1,
70 });
71
72 const pruned = yield* repository.listCheckpoints({ providerSessionId: "sess-1" });
73 assert.equal(pruned.length, 2);
74 assert.equal(pruned[1]?.turnCount, 1);
75 assert.equal(pruned[1]?.isCurrent, true);
76
77 yield* repository.deleteAllForSession({ providerSessionId: "sess-1" });
78 const empty = yield* repository.listCheckpoints({ providerSessionId: "sess-1" });
79 assert.equal(empty.length, 0);
80 }).pipe(Effect.provide(makeCheckpointRepositoryLayer()));
81 }),
14 const repository = yield* CheckpointRepository;
15
16 yield* repository.upsertCheckpoint({
17 providerSessionId: "sess-1",
18 threadId: "thread-1",
19 checkpointId: "root",
20 checkpointRef: "refs/t3/checkpoints/thread-1/turn/0",
21 turnCount: 0,
22 messageCount: 0,
23 label: "Start of conversation",
24 createdAt: new Date().toISOString(),
25 });
26
27 yield* repository.upsertCheckpoint({
28 providerSessionId: "sess-1",
29 threadId: "thread-1",
30 checkpointId: "turn-1",
31 checkpointRef: "refs/t3/checkpoints/thread-1/turn/1",
32 turnCount: 1,
33 messageCount: 2,
34 label: "Turn 1",
35 preview: "hello",
36 createdAt: new Date().toISOString(),
37 });
38
39 yield* repository.upsertCheckpoint({
40 providerSessionId: "sess-1",
41 threadId: "thread-1",
42 checkpointId: "turn-2",
43 checkpointRef: "refs/t3/checkpoints/thread-1/turn/2",
44 turnCount: 2,
45 messageCount: 4,
46 label: "Turn 2",
47 preview: "done",
48 createdAt: new Date().toISOString(),
49 });
50
51 const checkpoints = yield* repository.listCheckpoints({ threadId: "thread-1" });
52 assert.equal(checkpoints.length, 3);
53 assert.equal(checkpoints[0]?.id, "root");
54 assert.equal(checkpoints[0]?.isCurrent, false);
55 assert.equal(checkpoints[2]?.id, "turn-2");
56 assert.equal(checkpoints[2]?.isCurrent, true);
57
58 const one = yield* repository.getCheckpoint({ threadId: "thread-1", turnCount: 1 });
59 assert.equal(Option.isSome(one), true);
60 if (Option.isSome(one)) {
61 assert.equal(one.value.id, "turn-1");
62 assert.equal(one.value.isCurrent, false);
63 assert.equal(one.value.checkpointRef, "refs/t3/checkpoints/thread-1/turn/1");
64 }
65
66 yield* repository.upsertCheckpoint({
67 providerSessionId: "sess-2",
68 threadId: "thread-1",
69 checkpointId: "turn-3",
70 checkpointRef: "refs/t3/checkpoints/thread-1/turn/3",
71 turnCount: 3,
72 messageCount: 6,
73 label: "Turn 3",
74 preview: "new session",
75 createdAt: new Date().toISOString(),
76 });
77
78 const resumed = yield* repository.listCheckpoints({ threadId: "thread-1" });
79 assert.equal(resumed.length, 4);
80 assert.equal(resumed[3]?.id, "turn-3");
81 assert.equal(resumed[3]?.isCurrent, true);
82
83 yield* repository.upsertCheckpoint({
84 providerSessionId: "sess-3",
85 threadId: "thread-2",
86 checkpointId: "root",
87 checkpointRef: "refs/t3/checkpoints/thread-2/turn/0",
88 turnCount: 0,
89 messageCount: 0,
90 label: "Start of conversation",
91 createdAt: new Date().toISOString(),
92 });
93
94 const otherThread = yield* repository.listCheckpoints({ threadId: "thread-2" });
95 assert.equal(otherThread.length, 1);
96 assert.equal(otherThread[0]?.id, "root");
97
98 yield* repository.deleteAfterTurn({
99 threadId: "thread-1",
100 maxTurnCount: 1,
101 });
102
103 const pruned = yield* repository.listCheckpoints({ threadId: "thread-1" });
104 assert.equal(pruned.length, 2);
105 assert.equal(pruned[1]?.turnCount, 1);
106 assert.equal(pruned[1]?.isCurrent, true);
107
108 yield* repository.deleteAllForThread({ threadId: "thread-1" });
109 const empty = yield* repository.listCheckpoints({ threadId: "thread-1" });
110 assert.equal(empty.length, 0);
111
112 const stillPresent = yield* repository.listCheckpoints({ threadId: "thread-2" });
113 assert.equal(stillPresent.length, 1);
114 }).pipe(Effect.provide(makeCheckpointRepositoryLayer())),
82115);
apps/server/src/persistence/Layers/Checkpoints.ts+29-28
......@@ -29,7 +29,7 @@ import {
2929const NonEmptyString = Schema.NonEmptyString;
3030
3131const CheckpointRowSchema = Schema.Struct({
32 providerSessionId: Schema.String,
32 providerSessionId: Schema.NullOr(Schema.String),
3333 threadId: Schema.String,
3434 checkpointId: Schema.String,
3535 checkpointRef: Schema.String,
......@@ -40,12 +40,12 @@ const CheckpointRowSchema = Schema.Struct({
4040 createdAt: Schema.String,
4141});
4242
43const SessionRequestSchema = Schema.Struct({
44 providerSessionId: NonEmptyString,
43const ThreadRequestSchema = Schema.Struct({
44 threadId: NonEmptyString,
4545});
4646
4747const UpsertRequestSchema = Schema.Struct({
48 providerSessionId: NonEmptyString,
48 providerSessionId: Schema.NullOr(NonEmptyString),
4949 threadId: NonEmptyString,
5050 checkpointId: NonEmptyString,
5151 checkpointRef: NonEmptyString,
......@@ -57,12 +57,12 @@ const UpsertRequestSchema = Schema.Struct({
5757});
5858
5959const GetCheckpointRequestSchema = Schema.Struct({
60 providerSessionId: NonEmptyString,
60 threadId: NonEmptyString,
6161 turnCount: Schema.Int,
6262});
6363
6464const DeleteAfterTurnRequestSchema = Schema.Struct({
65 providerSessionId: NonEmptyString,
65 threadId: NonEmptyString,
6666 maxTurnCount: Schema.Int,
6767});
6868
......@@ -176,8 +176,9 @@ const makeCheckpointRepository = Effect.gen(function* () {
176176 ${request.preview},
177177 ${request.createdAt}
178178 )
179 ON CONFLICT (provider_session_id, turn_count)
179 ON CONFLICT (thread_id, turn_count)
180180 DO UPDATE SET
181 provider_session_id = excluded.provider_session_id,
181182 thread_id = excluded.thread_id,
182183 checkpoint_id = excluded.checkpoint_id,
183184 checkpoint_ref = excluded.checkpoint_ref,
......@@ -199,9 +200,9 @@ const makeCheckpointRepository = Effect.gen(function* () {
199200 });
200201
201202 const listCheckpointRows = SqlSchema.findAll({
202 Request: SessionRequestSchema,
203 Request: ThreadRequestSchema,
203204 Result: CheckpointRowSchema,
204 execute: ({ providerSessionId }) =>
205 execute: ({ threadId }) =>
205206 sql`
206207 SELECT
207208 provider_session_id AS "providerSessionId",
......@@ -214,15 +215,15 @@ const makeCheckpointRepository = Effect.gen(function* () {
214215 preview,
215216 created_at AS "createdAt"
216217 FROM provider_checkpoints
217 WHERE provider_session_id = ${providerSessionId}
218 WHERE thread_id = ${threadId}
218219 ORDER BY turn_count ASC
219220 `,
220221 });
221222
222 const listForSession = (
223 providerSessionId: string,
223 const listForThread = (
224 threadId: string,
224225 ): Effect.Effect<ReadonlyArray<CheckpointRepositoryEntry>, CheckpointRepositoryError> =>
225 listCheckpointRows({ providerSessionId }).pipe(
226 listCheckpointRows({ threadId }).pipe(
226227 Effect.mapError((cause) =>
227228 toPersistenceError("CheckpointRepository.listCheckpoints:query", cause),
228229 ),
......@@ -238,7 +239,7 @@ const makeCheckpointRepository = Effect.gen(function* () {
238239 const parsed = yield* decodeInput(
239240 UpsertRequestSchema,
240241 {
241 providerSessionId: input.providerSessionId.trim(),
242 providerSessionId: input.providerSessionId?.trim() || null,
242243 threadId: input.threadId.trim(),
243244 checkpointId: input.checkpointId.trim(),
244245 checkpointRef: input.checkpointRef.trim(),
......@@ -264,14 +265,14 @@ const makeCheckpointRepository = Effect.gen(function* () {
264265 const listCheckpoints: CheckpointRepositoryShape["listCheckpoints"] = (input) =>
265266 Effect.gen(function* () {
266267 const parsed = yield* decodeInput(
267 SessionRequestSchema,
268 ThreadRequestSchema,
268269 {
269 providerSessionId: input.providerSessionId.trim(),
270 threadId: input.threadId.trim(),
270271 },
271272 "CheckpointRepository.listCheckpoints",
272273 );
273274
274 const entries = yield* listForSession(parsed.providerSessionId);
275 const entries = yield* listForThread(parsed.threadId);
275276 return entries.map(toProviderCheckpoint);
276277 });
277278
......@@ -280,7 +281,7 @@ const makeCheckpointRepository = Effect.gen(function* () {
280281 const parsed = yield* decodeInput(
281282 GetCheckpointRequestSchema,
282283 {
283 providerSessionId: input.providerSessionId.trim(),
284 threadId: input.threadId.trim(),
284285 turnCount: input.turnCount,
285286 },
286287 "CheckpointRepository.getCheckpoint",
......@@ -291,7 +292,7 @@ const makeCheckpointRepository = Effect.gen(function* () {
291292 parsed.turnCount,
292293 );
293294
294 const entries = yield* listForSession(parsed.providerSessionId);
295 const entries = yield* listForThread(parsed.threadId);
295296 const checkpoint = entries.find((entry) => entry.turnCount === parsed.turnCount);
296297 return Option.fromNullishOr(checkpoint);
297298 });
......@@ -301,7 +302,7 @@ const makeCheckpointRepository = Effect.gen(function* () {
301302 const parsed = yield* decodeInput(
302303 DeleteAfterTurnRequestSchema,
303304 {
304 providerSessionId: input.providerSessionId.trim(),
305 threadId: input.threadId.trim(),
305306 maxTurnCount: input.maxTurnCount,
306307 },
307308 "CheckpointRepository.deleteAfterTurn",
......@@ -314,7 +315,7 @@ const makeCheckpointRepository = Effect.gen(function* () {
314315
315316 yield* sql`
316317 DELETE FROM provider_checkpoints
317 WHERE provider_session_id = ${parsed.providerSessionId}
318 WHERE thread_id = ${parsed.threadId}
318319 AND turn_count > ${parsed.maxTurnCount}
319320 `.pipe(
320321 Effect.mapError((cause) =>
......@@ -323,22 +324,22 @@ const makeCheckpointRepository = Effect.gen(function* () {
323324 );
324325 });
325326
326 const deleteAllForSession: CheckpointRepositoryShape["deleteAllForSession"] = (input) =>
327 const deleteAllForThread: CheckpointRepositoryShape["deleteAllForThread"] = (input) =>
327328 Effect.gen(function* () {
328329 const parsed = yield* decodeInput(
329 SessionRequestSchema,
330 ThreadRequestSchema,
330331 {
331 providerSessionId: input.providerSessionId.trim(),
332 threadId: input.threadId.trim(),
332333 },
333 "CheckpointRepository.deleteAllForSession",
334 "CheckpointRepository.deleteAllForThread",
334335 );
335336
336337 yield* sql`
337338 DELETE FROM provider_checkpoints
338 WHERE provider_session_id = ${parsed.providerSessionId}
339 WHERE thread_id = ${parsed.threadId}
339340 `.pipe(
340341 Effect.mapError((cause) =>
341 toPersistenceError("CheckpointRepository.deleteAllForSession:query", cause),
342 toPersistenceError("CheckpointRepository.deleteAllForThread:query", cause),
342343 ),
343344 );
344345 });
......@@ -348,7 +349,7 @@ const makeCheckpointRepository = Effect.gen(function* () {
348349 listCheckpoints,
349350 getCheckpoint,
350351 deleteAfterTurn,
351 deleteAllForSession,
352 deleteAllForThread,
352353 } satisfies CheckpointRepositoryShape;
353354});
354355
apps/server/src/persistence/Layers/Projects.ts+1-1
......@@ -333,7 +333,7 @@ const makeRepository = Effect.gen(function* () {
333333 const input = yield* decodeAddInput(rawInput);
334334 const normalizedCwd = normalizeCwd(input.cwd);
335335 if (!isDirectory(normalizedCwd)) {
336 return yield* Effect.fail(new ProjectPathMissingError({ cwd: normalizedCwd }));
336 return yield* new ProjectPathMissingError({ cwd: normalizedCwd });
337337 }
338338
339339 const existing = yield* findProjectByCwd({ cwd: normalizedCwd }).pipe(
apps/server/src/persistence/Layers/Sqlite.ts+33-4
......@@ -1,11 +1,34 @@
11import { Effect, Layer, FileSystem, Path } from "effect";
2import * as SqlClient from "effect/unstable/sql/SqlClient";
23
34import { runMigrations } from "../Migrations.ts";
4import * as SqliteClient from "../SqliteClient.ts";
5
6type RuntimeSqliteLayerConfig = {
7 readonly filename: string;
8 readonly disableWAL?: boolean | undefined;
9};
10
11type Loader = {
12 layer: (config: RuntimeSqliteLayerConfig) => Layer.Layer<SqlClient.SqlClient>;
13};
14const defaultSqliteClientLoaders = {
15 bun: () => import("@effect/sql-sqlite-bun/SqliteClient"),
16 node: () => import("../NodeSqliteClient.ts"),
17} satisfies Record<string, () => Promise<Loader>>;
18
19const makeRuntimeSqliteLayer = (
20 config: RuntimeSqliteLayerConfig,
21): Layer.Layer<SqlClient.SqlClient> =>
22 Effect.gen(function* () {
23 const runtime = process.versions.bun !== undefined ? "bun" : "node";
24 const loader = defaultSqliteClientLoaders[runtime];
25 const clientModule = yield* Effect.promise<Loader>(loader);
26 return clientModule.layer(config);
27 }).pipe(Layer.unwrap);
528
629const setup = Layer.effectDiscard(
730 Effect.gen(function* () {
8 const sql = yield* SqliteClient.SqliteClient;
31 const sql = yield* SqlClient.SqlClient;
932 yield* sql`PRAGMA journal_mode = WAL;`;
1033 yield* sql`PRAGMA foreign_keys = ON;`;
1134 yield* runMigrations;
......@@ -18,7 +41,13 @@ export const makeSqlitePersistenceLive = (dbPath: string) =>
1841 const path = yield* Path.Path;
1942 yield* fs.makeDirectory(path.dirname(dbPath), { recursive: true });
2043
21 return Layer.provideMerge(setup, SqliteClient.layer({ filename: dbPath }));
44 return Layer.provideMerge(setup, makeRuntimeSqliteLayer({ filename: dbPath }));
2245 }).pipe(Layer.unwrap);
2346
24export const SqlitePersistenceMemory = Layer.provideMerge(setup, SqliteClient.layerMemory());
47export const SqlitePersistenceMemory = Layer.provideMerge(
48 setup,
49 makeRuntimeSqliteLayer({
50 filename: ":memory:",
51 disableWAL: true,
52 }),
53);
apps/server/src/persistence/Migrations/003_Checkpoints.ts+9-4
......@@ -6,7 +6,7 @@ export default Effect.gen(function* () {
66
77 yield* sql`
88 CREATE TABLE IF NOT EXISTS provider_checkpoints (
9 provider_session_id TEXT NOT NULL,
9 provider_session_id TEXT,
1010 thread_id TEXT NOT NULL,
1111 checkpoint_id TEXT NOT NULL,
1212 checkpoint_ref TEXT NOT NULL,
......@@ -15,12 +15,17 @@ export default Effect.gen(function* () {
1515 label TEXT NOT NULL,
1616 preview TEXT,
1717 created_at TEXT NOT NULL,
18 PRIMARY KEY (provider_session_id, turn_count)
18 PRIMARY KEY (thread_id, turn_count)
1919 )
2020 `;
2121
2222 yield* sql`
23 CREATE INDEX IF NOT EXISTS idx_provider_checkpoints_session_turn
24 ON provider_checkpoints(provider_session_id, turn_count ASC)
23 CREATE INDEX IF NOT EXISTS idx_provider_checkpoints_thread_turn
24 ON provider_checkpoints(thread_id, turn_count ASC)
25 `;
26
27 yield* sql`
28 CREATE INDEX IF NOT EXISTS idx_provider_checkpoints_session
29 ON provider_checkpoints(provider_session_id)
2530 `;
2631});
apps/server/src/persistence/NodeSqliteClient.test.ts created+26
......@@ -0,0 +1,26 @@
1import { assert, it } from "@effect/vitest";
2import { Effect } from "effect";
3import * as SqlClient from "effect/unstable/sql/SqlClient";
4
5import * as SqliteClient from "./NodeSqliteClient.ts";
6
7it.effect("runs prepared queries and returns positional values", () =>
8 Effect.gen(function* () {
9 const sql = yield* SqlClient.SqlClient;
10
11 yield* sql`CREATE TABLE entries(id INTEGER PRIMARY KEY, name TEXT NOT NULL)`;
12 yield* sql`INSERT INTO entries(name) VALUES (${"alpha"}), (${"beta"})`;
13
14 const rows = yield* sql<{ readonly id: number; readonly name: string }>`
15 SELECT id, name FROM entries ORDER BY id
16 `;
17 assert.equal(rows.length, 2);
18 assert.equal(rows[0]?.name, "alpha");
19 assert.equal(rows[1]?.name, "beta");
20
21 const values = yield* sql`SELECT id, name FROM entries ORDER BY id`.values;
22 assert.equal(values.length, 2);
23 assert.equal(values[0]?.[1], "alpha");
24 assert.equal(values[1]?.[1], "beta");
25 }).pipe(Effect.provide(SqliteClient.layerMemory())),
26);
apps/server/src/persistence/NodeSqliteClient.ts created+242
......@@ -0,0 +1,242 @@
1/**
2 * Port of `@effect/sql-sqlite-node` that uses the native `node:sqlite`
3 * bindings instead of `better-sqlite3`.
4 */
5import { DatabaseSync, type StatementSync } from "node:sqlite";
6
7import * as Cache from "effect/Cache";
8import * as Config from "effect/Config";
9import * as Duration from "effect/Duration";
10import * as Effect from "effect/Effect";
11import * as Fiber from "effect/Fiber";
12import { identity } from "effect/Function";
13import * as Layer from "effect/Layer";
14import * as Scope from "effect/Scope";
15import * as Semaphore from "effect/Semaphore";
16import * as ServiceMap from "effect/ServiceMap";
17import * as Stream from "effect/Stream";
18import * as Reactivity from "effect/unstable/reactivity/Reactivity";
19import * as Client from "effect/unstable/sql/SqlClient";
20import type { Connection } from "effect/unstable/sql/SqlConnection";
21import { SqlError } from "effect/unstable/sql/SqlError";
22import * as Statement from "effect/unstable/sql/Statement";
23
24const ATTR_DB_SYSTEM_NAME = "db.system.name";
25
26export const TypeId: TypeId = "~local/sqlite-node/SqliteClient";
27
28export type TypeId = "~local/sqlite-node/SqliteClient";
29
30export const SqliteClient = ServiceMap.Service<Client.SqlClient>("@local/sqlite-node/SqliteClient");
31
32export interface SqliteClientConfig {
33 readonly filename: string;
34 readonly readonly?: boolean | undefined;
35 readonly allowExtension?: boolean | undefined;
36 readonly prepareCacheSize?: number | undefined;
37 readonly prepareCacheTTL?: Duration.Input | undefined;
38 readonly disableWAL?: boolean | undefined;
39 readonly spanAttributes?: Record<string, unknown> | undefined;
40 readonly transformResultNames?: ((str: string) => string) | undefined;
41 readonly transformQueryNames?: ((str: string) => string) | undefined;
42}
43
44export interface SqliteMemoryClientConfig extends Omit<
45 SqliteClientConfig,
46 "filename" | "readonly"
47> {}
48
49const makeWithDatabase = (
50 options: SqliteClientConfig,
51 openDatabase: () => DatabaseSync,
52): Effect.Effect<Client.SqlClient, never, Scope.Scope | Reactivity.Reactivity> =>
53 Effect.gen(function* () {
54 const compiler = Statement.makeCompilerSqlite(options.transformQueryNames);
55 const transformRows = options.transformResultNames
56 ? Statement.defaultTransforms(options.transformResultNames).array
57 : undefined;
58
59 const makeConnection = Effect.gen(function* () {
60 const scope = yield* Effect.scope;
61 const db = openDatabase();
62 yield* Scope.addFinalizer(
63 scope,
64 Effect.sync(() => db.close()),
65 );
66
67 if (options.disableWAL !== true) {
68 db.exec("PRAGMA journal_mode = WAL");
69 }
70
71 const statementReaderCache = new WeakMap<StatementSync, boolean>();
72 const hasRows = (statement: StatementSync): boolean => {
73 const cached = statementReaderCache.get(statement);
74 if (cached !== undefined) {
75 return cached;
76 }
77 const value = statement.columns().length > 0;
78 statementReaderCache.set(statement, value);
79 return value;
80 };
81
82 const prepareCache = yield* Cache.make({
83 capacity: options.prepareCacheSize ?? 200,
84 timeToLive: options.prepareCacheTTL ?? Duration.minutes(10),
85 lookup: (sql: string) =>
86 Effect.try({
87 try: () => db.prepare(sql),
88 catch: (cause) => new SqlError({ cause, message: "Failed to prepare statement" }),
89 }),
90 });
91
92 const runStatement = (
93 statement: StatementSync,
94 params: ReadonlyArray<unknown>,
95 raw: boolean,
96 ) =>
97 Effect.withFiber<ReadonlyArray<any>, SqlError>((fiber) => {
98 statement.setReadBigInts(Boolean(ServiceMap.get(fiber.services, Client.SafeIntegers)));
99 try {
100 if (hasRows(statement)) {
101 return Effect.succeed(statement.all(...(params as any)));
102 }
103 const result = statement.run(...(params as any));
104 return Effect.succeed(raw ? (result as unknown as ReadonlyArray<any>) : []);
105 } catch (cause) {
106 return Effect.fail(new SqlError({ cause, message: "Failed to execute statement" }));
107 }
108 });
109
110 const run = (sql: string, params: ReadonlyArray<unknown>, raw = false) =>
111 Effect.flatMap(Cache.get(prepareCache, sql), (s) => runStatement(s, params, raw));
112
113 const runValues = (sql: string, params: ReadonlyArray<unknown>) =>
114 Effect.acquireUseRelease(
115 Cache.get(prepareCache, sql),
116 (statement) =>
117 Effect.try({
118 try: () => {
119 if (hasRows(statement)) {
120 statement.setReturnArrays(true);
121 // Safe to cast to array after we've setReturnArrays(true)
122 return statement.all(...(params as any)) as unknown as ReadonlyArray<
123 ReadonlyArray<unknown>
124 >;
125 }
126 statement.run(...(params as any));
127 return [];
128 },
129 catch: (cause) => new SqlError({ cause, message: "Failed to execute statement" }),
130 }),
131 (statement) =>
132 Effect.sync(() => {
133 if (hasRows(statement)) {
134 statement.setReturnArrays(false);
135 }
136 }),
137 );
138
139 return identity<Connection>({
140 execute(sql, params, rowTransform) {
141 return rowTransform ? Effect.map(run(sql, params), rowTransform) : run(sql, params);
142 },
143 executeRaw(sql, params) {
144 return run(sql, params, true);
145 },
146 executeValues(sql, params) {
147 return runValues(sql, params);
148 },
149 executeUnprepared(sql, params, rowTransform) {
150 const effect = runStatement(db.prepare(sql), params ?? [], false);
151 return rowTransform ? Effect.map(effect, rowTransform) : effect;
152 },
153 executeStream(_sql, _params) {
154 return Stream.die("executeStream not implemented");
155 },
156 });
157 });
158
159 const semaphore = yield* Semaphore.make(1);
160 const connection = yield* makeConnection;
161
162 const acquirer = semaphore.withPermits(1)(Effect.succeed(connection));
163 const transactionAcquirer = Effect.uninterruptibleMask((restore) => {
164 const fiber = Fiber.getCurrent()!;
165 const scope = ServiceMap.getUnsafe(fiber.services, Scope.Scope);
166 return Effect.as(
167 Effect.tap(restore(semaphore.take(1)), () =>
168 Scope.addFinalizer(scope, semaphore.release(1)),
169 ),
170 connection,
171 );
172 });
173
174 return yield* Client.make({
175 acquirer,
176 compiler,
177 transactionAcquirer,
178 spanAttributes: [
179 ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []),
180 [ATTR_DB_SYSTEM_NAME, "sqlite"],
181 ],
182 transformRows,
183 });
184 });
185
186const make = (
187 options: SqliteClientConfig,
188): Effect.Effect<Client.SqlClient, never, Scope.Scope | Reactivity.Reactivity> =>
189 makeWithDatabase(
190 options,
191 () =>
192 new DatabaseSync(options.filename, {
193 readOnly: options.readonly ?? false,
194 allowExtension: options.allowExtension ?? false,
195 }),
196 );
197
198const makeMemory = (
199 config: SqliteMemoryClientConfig = {},
200): Effect.Effect<Client.SqlClient, never, Scope.Scope | Reactivity.Reactivity> =>
201 makeWithDatabase(
202 {
203 ...config,
204 filename: ":memory:",
205 readonly: false,
206 disableWAL: config.disableWAL ?? true,
207 },
208 () => {
209 const database = new DatabaseSync(":memory:", {
210 allowExtension: config.allowExtension ?? false,
211 });
212 return database;
213 },
214 );
215
216export const layerConfig = (
217 config: Config.Wrap<SqliteClientConfig>,
218): Layer.Layer<Client.SqlClient, Config.ConfigError> =>
219 Layer.effectServices(
220 Config.unwrap(config)
221 .asEffect()
222 .pipe(
223 Effect.flatMap(make),
224 Effect.map((client) =>
225 ServiceMap.make(SqliteClient, client).pipe(ServiceMap.add(Client.SqlClient, client)),
226 ),
227 ),
228 ).pipe(Layer.provide(Reactivity.layer));
229
230export const layer = (config: SqliteClientConfig): Layer.Layer<Client.SqlClient> =>
231 Layer.effectServices(
232 Effect.map(make(config), (client) =>
233 ServiceMap.make(SqliteClient, client).pipe(ServiceMap.add(Client.SqlClient, client)),
234 ),
235 ).pipe(Layer.provide(Reactivity.layer));
236
237export const layerMemory = (config: SqliteMemoryClientConfig = {}): Layer.Layer<Client.SqlClient> =>
238 Layer.effectServices(
239 Effect.map(makeMemory(config), (client) =>
240 ServiceMap.make(SqliteClient, client).pipe(ServiceMap.add(Client.SqlClient, client)),
241 ),
242 ).pipe(Layer.provide(Reactivity.layer));
apps/server/src/persistence/Services/Checkpoints.ts+13-13
......@@ -2,7 +2,7 @@
22 * CheckpointRepository - Repository interface for checkpoint metadata persistence.
33 *
44 * Stores and queries checkpoint metadata and checkpoint Git refs keyed by
5 * provider session + turn count. It does not read/write Git refs itself and it
5 * thread + turn count. It does not read/write Git refs itself and it
66 * does not coordinate provider rollback workflows.
77 *
88 * @module CheckpointRepository
......@@ -14,7 +14,7 @@ import type { Effect } from "effect";
1414import type { CheckpointRepositoryError } from "../Errors.ts";
1515
1616export interface UpsertCheckpointInput {
17 readonly providerSessionId: string;
17 readonly providerSessionId?: string;
1818 readonly threadId: string;
1919 readonly checkpointId: string;
2020 readonly checkpointRef: string;
......@@ -31,22 +31,22 @@ export interface CheckpointRepositoryEntry extends ProviderCheckpoint {
3131 readonly createdAt: string;
3232}
3333
34export interface ListSessionCheckpointsInput {
35 readonly providerSessionId: string;
34export interface ListThreadCheckpointsInput {
35 readonly threadId: string;
3636}
3737
3838export interface GetCheckpointInput {
39 readonly providerSessionId: string;
39 readonly threadId: string;
4040 readonly turnCount: number;
4141}
4242
4343export interface DeleteAfterTurnInput {
44 readonly providerSessionId: string;
44 readonly threadId: string;
4545 readonly maxTurnCount: number;
4646}
4747
48export interface DeleteAllForSessionInput {
49 readonly providerSessionId: string;
48export interface DeleteAllForThreadInput {
49 readonly threadId: string;
5050}
5151
5252export interface CheckpointRepositoryShape {
......@@ -58,10 +58,10 @@ export interface CheckpointRepositoryShape {
5858 ) => Effect.Effect<void, CheckpointRepositoryError>;
5959
6060 /**
61 * List user-facing checkpoints for a provider session.
61 * List user-facing checkpoints for a thread.
6262 */
6363 readonly listCheckpoints: (
64 input: ListSessionCheckpointsInput,
64 input: ListThreadCheckpointsInput,
6565 ) => Effect.Effect<ReadonlyArray<ProviderCheckpoint>, CheckpointRepositoryError>;
6666
6767 /**
......@@ -79,10 +79,10 @@ export interface CheckpointRepositoryShape {
7979 ) => Effect.Effect<void, CheckpointRepositoryError>;
8080
8181 /**
82 * Delete all checkpoint metadata for a provider session.
82 * Delete all checkpoint metadata for a thread.
8383 */
84 readonly deleteAllForSession: (
85 input: DeleteAllForSessionInput,
84 readonly deleteAllForThread: (
85 input: DeleteAllForThreadInput,
8686 ) => Effect.Effect<void, CheckpointRepositoryError>;
8787}
8888
apps/server/src/persistence/SqliteClient.test.ts deleted-26
......@@ -1,26 +0,0 @@
1import { assert, it } from "@effect/vitest";
2import { Effect } from "effect";
3import * as SqlClient from "effect/unstable/sql/SqlClient";
4
5import * as SqliteClient from "./SqliteClient.ts";
6
7it.effect("runs prepared queries and returns positional values", () =>
8 Effect.gen(function* () {
9 const sql = yield* SqlClient.SqlClient;
10
11 yield* sql`CREATE TABLE entries(id INTEGER PRIMARY KEY, name TEXT NOT NULL)`;
12 yield* sql`INSERT INTO entries(name) VALUES (${"alpha"}), (${"beta"})`;
13
14 const rows = yield* sql<{ readonly id: number; readonly name: string }>`
15 SELECT id, name FROM entries ORDER BY id
16 `;
17 assert.equal(rows.length, 2);
18 assert.equal(rows[0]?.name, "alpha");
19 assert.equal(rows[1]?.name, "beta");
20
21 const values = yield* sql`SELECT id, name FROM entries ORDER BY id`.values;
22 assert.equal(values.length, 2);
23 assert.equal(values[0]?.[1], "alpha");
24 assert.equal(values[1]?.[1], "beta");
25 }).pipe(Effect.provide(SqliteClient.layerMemory())),
26);
apps/server/src/persistence/SqliteClient.ts deleted-242
......@@ -1,242 +0,0 @@
1/**
2 * Port of `@effect/sql-sqlite-node` that uses the native `node:sqlite`
3 * bindings instead of `better-sqlite3`.
4 */
5import { DatabaseSync, type StatementSync } from "node:sqlite";
6
7import * as Cache from "effect/Cache";
8import * as Config from "effect/Config";
9import * as Duration from "effect/Duration";
10import * as Effect from "effect/Effect";
11import * as Fiber from "effect/Fiber";
12import { identity } from "effect/Function";
13import * as Layer from "effect/Layer";
14import * as Scope from "effect/Scope";
15import * as Semaphore from "effect/Semaphore";
16import * as ServiceMap from "effect/ServiceMap";
17import * as Stream from "effect/Stream";
18import * as Reactivity from "effect/unstable/reactivity/Reactivity";
19import * as Client from "effect/unstable/sql/SqlClient";
20import type { Connection } from "effect/unstable/sql/SqlConnection";
21import { SqlError } from "effect/unstable/sql/SqlError";
22import * as Statement from "effect/unstable/sql/Statement";
23
24const ATTR_DB_SYSTEM_NAME = "db.system.name";
25
26export const TypeId: TypeId = "~local/sqlite-node/SqliteClient";
27
28export type TypeId = "~local/sqlite-node/SqliteClient";
29
30export const SqliteClient = ServiceMap.Service<Client.SqlClient>("@local/sqlite-node/SqliteClient");
31
32export interface SqliteClientConfig {
33 readonly filename: string;
34 readonly readonly?: boolean | undefined;
35 readonly allowExtension?: boolean | undefined;
36 readonly prepareCacheSize?: number | undefined;
37 readonly prepareCacheTTL?: Duration.Input | undefined;
38 readonly disableWAL?: boolean | undefined;
39 readonly spanAttributes?: Record<string, unknown> | undefined;
40 readonly transformResultNames?: ((str: string) => string) | undefined;
41 readonly transformQueryNames?: ((str: string) => string) | undefined;
42}
43
44export interface SqliteMemoryClientConfig extends Omit<
45 SqliteClientConfig,
46 "filename" | "readonly"
47> {}
48
49const makeWithDatabase = (
50 options: SqliteClientConfig,
51 openDatabase: () => DatabaseSync,
52): Effect.Effect<Client.SqlClient, never, Scope.Scope | Reactivity.Reactivity> =>
53 Effect.gen(function* () {
54 const compiler = Statement.makeCompilerSqlite(options.transformQueryNames);
55 const transformRows = options.transformResultNames
56 ? Statement.defaultTransforms(options.transformResultNames).array
57 : undefined;
58
59 const makeConnection = Effect.gen(function* () {
60 const scope = yield* Effect.scope;
61 const db = openDatabase();
62 yield* Scope.addFinalizer(
63 scope,
64 Effect.sync(() => db.close()),
65 );
66
67 if (options.disableWAL !== true) {
68 db.exec("PRAGMA journal_mode = WAL");
69 }
70
71 const statementReaderCache = new WeakMap<StatementSync, boolean>();
72 const hasRows = (statement: StatementSync): boolean => {
73 const cached = statementReaderCache.get(statement);
74 if (cached !== undefined) {
75 return cached;
76 }
77 const value = statement.columns().length > 0;
78 statementReaderCache.set(statement, value);
79 return value;
80 };
81
82 const prepareCache = yield* Cache.make({
83 capacity: options.prepareCacheSize ?? 200,
84 timeToLive: options.prepareCacheTTL ?? Duration.minutes(10),
85 lookup: (sql: string) =>
86 Effect.try({
87 try: () => db.prepare(sql),
88 catch: (cause) => new SqlError({ cause, message: "Failed to prepare statement" }),
89 }),
90 });
91
92 const runStatement = (
93 statement: StatementSync,
94 params: ReadonlyArray<unknown>,
95 raw: boolean,
96 ) =>
97 Effect.withFiber<ReadonlyArray<any>, SqlError>((fiber) => {
98 statement.setReadBigInts(Boolean(ServiceMap.get(fiber.services, Client.SafeIntegers)));
99 try {
100 if (hasRows(statement)) {
101 return Effect.succeed(statement.all(...(params as any)));
102 }
103 const result = statement.run(...(params as any));
104 return Effect.succeed(raw ? (result as unknown as ReadonlyArray<any>) : []);
105 } catch (cause) {
106 return Effect.fail(new SqlError({ cause, message: "Failed to execute statement" }));
107 }
108 });
109
110 const run = (sql: string, params: ReadonlyArray<unknown>, raw = false) =>
111 Effect.flatMap(Cache.get(prepareCache, sql), (s) => runStatement(s, params, raw));
112
113 const runValues = (sql: string, params: ReadonlyArray<unknown>) =>
114 Effect.acquireUseRelease(
115 Cache.get(prepareCache, sql),
116 (statement) =>
117 Effect.try({
118 try: () => {
119 if (hasRows(statement)) {
120 statement.setReturnArrays(true);
121 // Safe to cast to array after we've setReturnArrays(true)
122 return statement.all(...(params as any)) as unknown as ReadonlyArray<
123 ReadonlyArray<unknown>
124 >;
125 }
126 statement.run(...(params as any));
127 return [];
128 },
129 catch: (cause) => new SqlError({ cause, message: "Failed to execute statement" }),
130 }),
131 (statement) =>
132 Effect.sync(() => {
133 if (hasRows(statement)) {
134 statement.setReturnArrays(false);
135 }
136 }),
137 );
138
139 return identity<Connection>({
140 execute(sql, params, rowTransform) {
141 return rowTransform ? Effect.map(run(sql, params), rowTransform) : run(sql, params);
142 },
143 executeRaw(sql, params) {
144 return run(sql, params, true);
145 },
146 executeValues(sql, params) {
147 return runValues(sql, params);
148 },
149 executeUnprepared(sql, params, rowTransform) {
150 const effect = runStatement(db.prepare(sql), params ?? [], false);
151 return rowTransform ? Effect.map(effect, rowTransform) : effect;
152 },
153 executeStream(_sql, _params) {
154 return Stream.die("executeStream not implemented");
155 },
156 });
157 });
158
159 const semaphore = yield* Semaphore.make(1);
160 const connection = yield* makeConnection;
161
162 const acquirer = semaphore.withPermits(1)(Effect.succeed(connection));
163 const transactionAcquirer = Effect.uninterruptibleMask((restore) => {
164 const fiber = Fiber.getCurrent()!;
165 const scope = ServiceMap.getUnsafe(fiber.services, Scope.Scope);
166 return Effect.as(
167 Effect.tap(restore(semaphore.take(1)), () =>
168 Scope.addFinalizer(scope, semaphore.release(1)),
169 ),
170 connection,
171 );
172 });
173
174 return yield* Client.make({
175 acquirer,
176 compiler,
177 transactionAcquirer,
178 spanAttributes: [
179 ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []),
180 [ATTR_DB_SYSTEM_NAME, "sqlite"],
181 ],
182 transformRows,
183 });
184 });
185
186const make = (
187 options: SqliteClientConfig,
188): Effect.Effect<Client.SqlClient, never, Scope.Scope | Reactivity.Reactivity> =>
189 makeWithDatabase(
190 options,
191 () =>
192 new DatabaseSync(options.filename, {
193 readOnly: options.readonly ?? false,
194 allowExtension: options.allowExtension ?? false,
195 }),
196 );
197
198const makeMemory = (
199 config: SqliteMemoryClientConfig = {},
200): Effect.Effect<Client.SqlClient, never, Scope.Scope | Reactivity.Reactivity> =>
201 makeWithDatabase(
202 {
203 ...config,
204 filename: ":memory:",
205 readonly: false,
206 disableWAL: config.disableWAL ?? true,
207 },
208 () => {
209 const database = new DatabaseSync(":memory:", {
210 allowExtension: config.allowExtension ?? false,
211 });
212 return database;
213 },
214 );
215
216export const layerConfig = (
217 config: Config.Wrap<SqliteClientConfig>,
218): Layer.Layer<Client.SqlClient, Config.ConfigError> =>
219 Layer.effectServices(
220 Config.unwrap(config)
221 .asEffect()
222 .pipe(
223 Effect.flatMap(make),
224 Effect.map((client) =>
225 ServiceMap.make(SqliteClient, client).pipe(ServiceMap.add(Client.SqlClient, client)),
226 ),
227 ),
228 ).pipe(Layer.provide(Reactivity.layer));
229
230export const layer = (config: SqliteClientConfig): Layer.Layer<Client.SqlClient> =>
231 Layer.effectServices(
232 Effect.map(make(config), (client) =>
233 ServiceMap.make(SqliteClient, client).pipe(ServiceMap.add(Client.SqlClient, client)),
234 ),
235 ).pipe(Layer.provide(Reactivity.layer));
236
237export const layerMemory = (config: SqliteMemoryClientConfig = {}): Layer.Layer<Client.SqlClient> =>
238 Layer.effectServices(
239 Effect.map(makeMemory(config), (client) =>
240 ServiceMap.make(SqliteClient, client).pipe(ServiceMap.add(Client.SqlClient, client)),
241 ),
242 ).pipe(Layer.provide(Reactivity.layer));
apps/server/src/provider/Layers/ProviderService.test.ts+141-3
......@@ -55,7 +55,7 @@ function makeFakeCodexAdapter() {
5555 sessionId: `sess-${nextSession}`,
5656 provider: "codex",
5757 status: "ready",
58 threadId: `thread-${nextSession}`,
58 threadId: input.resumeThreadId ?? `thread-${nextSession}`,
5959 cwd: input.cwd ?? process.cwd(),
6060 createdAt: now,
6161 updatedAt: now,
......@@ -281,7 +281,7 @@ function makeProviderServiceLayer() {
281281}
282282
283283const routing = makeProviderServiceLayer();
284it.effect("ProviderServiceLive reconciles stale persisted sessions on startup", () =>
284it.effect("ProviderServiceLive keeps persisted resumable sessions on startup", () =>
285285 Effect.gen(function* () {
286286 const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-service-"));
287287 const dbPath = path.join(tempDir, "orchestration.sqlite");
......@@ -322,7 +322,100 @@ it.effect("ProviderServiceLive reconciles stale persisted sessions on startup",
322322 yield* ProviderService;
323323 }).pipe(Effect.provide(providerLayer));
324324
325 assert.deepEqual(checkpoint.releaseSession.mock.calls, [[{ providerSessionId: "sess-stale" }]]);
325 assert.deepEqual(checkpoint.releaseSession.mock.calls, []);
326 const persistedProvider = yield* Effect.gen(function* () {
327 const directory = yield* ProviderSessionDirectory;
328 return yield* directory.getProvider("sess-stale");
329 }).pipe(Effect.provide(directoryLayer));
330 assert.equal(persistedProvider, "codex");
331
332 fs.rmSync(tempDir, { recursive: true, force: true });
333 }).pipe(Effect.provide(NodeServices.layer)),
334);
335
336it.effect("ProviderServiceLive restores checkpoint RPC routing after restart using persisted thread mapping", () =>
337 Effect.gen(function* () {
338 const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-service-restart-"));
339 const dbPath = path.join(tempDir, "orchestration.sqlite");
340 const persistenceLayer = makeSqlitePersistenceLive(dbPath);
341
342 const checkpoint = makeCheckpointServiceDouble();
343 const firstCodex = makeFakeCodexAdapter();
344 const firstRegistry: typeof ProviderAdapterRegistry.Service = {
345 getByProvider: (provider) =>
346 provider === "codex"
347 ? Effect.succeed(firstCodex.adapter)
348 : Effect.fail(new ProviderUnsupportedError({ provider })),
349 listProviders: () => Effect.succeed(["codex"]),
350 };
351
352 const firstDirectoryLayer = ProviderSessionDirectoryLive.pipe(
353 Layer.provide(ProviderSessionRepositoryLive),
354 Layer.provide(persistenceLayer),
355 );
356 const firstProviderLayer = ProviderServiceLive.pipe(
357 Layer.provide(Layer.succeed(ProviderAdapterRegistry, firstRegistry)),
358 Layer.provide(firstDirectoryLayer),
359 Layer.provide(Layer.succeed(CheckpointService, checkpoint.service)),
360 );
361
362 const startedSession = yield* Effect.gen(function* () {
363 const provider = yield* ProviderService;
364 return yield* provider.startSession({
365 provider: "codex",
366 cwd: "/tmp/project",
367 });
368 }).pipe(Effect.provide(firstProviderLayer));
369
370 yield* Effect.gen(function* () {
371 const provider = yield* ProviderService;
372 yield* provider.stopAll();
373 }).pipe(Effect.provide(firstProviderLayer));
374
375 const secondCodex = makeFakeCodexAdapter();
376 const secondRegistry: typeof ProviderAdapterRegistry.Service = {
377 getByProvider: (provider) =>
378 provider === "codex"
379 ? Effect.succeed(secondCodex.adapter)
380 : Effect.fail(new ProviderUnsupportedError({ provider })),
381 listProviders: () => Effect.succeed(["codex"]),
382 };
383 const secondDirectoryLayer = ProviderSessionDirectoryLive.pipe(
384 Layer.provide(ProviderSessionRepositoryLive),
385 Layer.provide(persistenceLayer),
386 );
387 const secondProviderLayer = ProviderServiceLive.pipe(
388 Layer.provide(Layer.succeed(ProviderAdapterRegistry, secondRegistry)),
389 Layer.provide(secondDirectoryLayer),
390 Layer.provide(Layer.succeed(CheckpointService, checkpoint.service)),
391 );
392
393 checkpoint.getCheckpointDiff.mockClear();
394 secondCodex.startSession.mockClear();
395
396 yield* Effect.gen(function* () {
397 const provider = yield* ProviderService;
398 yield* provider.getCheckpointDiff({
399 sessionId: startedSession.sessionId,
400 fromTurnCount: 0,
401 toTurnCount: 0,
402 });
403 }).pipe(Effect.provide(secondProviderLayer));
404
405 assert.deepEqual(secondCodex.startSession.mock.calls, [
406 [
407 {
408 provider: "codex",
409 resumeThreadId: startedSession.threadId,
410 },
411 ],
412 ]);
413 assert.equal(checkpoint.getCheckpointDiff.mock.calls.length, 1);
414 const recoveredDiffInput = checkpoint.getCheckpointDiff.mock.calls[0]?.[0];
415 assert.equal(typeof recoveredDiffInput?.sessionId, "string");
416 assert.equal((recoveredDiffInput?.sessionId?.length ?? 0) > 0, true);
417 assert.equal(recoveredDiffInput?.fromTurnCount, 0);
418 assert.equal(recoveredDiffInput?.toTurnCount, 0);
326419
327420 fs.rmSync(tempDir, { recursive: true, force: true });
328421 }).pipe(Effect.provide(NodeServices.layer)),
......@@ -403,6 +496,51 @@ routing.layer("ProviderServiceLive routing", (it) => {
403496 }),
404497 );
405498
499 it.effect("recovers stale persisted sessions for checkpoint RPCs by resuming thread identity", () =>
500 Effect.gen(function* () {
501 const provider = yield* ProviderService;
502
503 const initial = yield* provider.startSession({
504 provider: "codex",
505 cwd: "/tmp/project",
506 });
507 yield* routing.codex.stopSession(initial.sessionId);
508
509 routing.checkpoint.getCheckpointDiff.mockClear();
510 routing.checkpoint.revertToCheckpoint.mockClear();
511 routing.codex.startSession.mockClear();
512
513 yield* provider.getCheckpointDiff({
514 sessionId: initial.sessionId,
515 fromTurnCount: 0,
516 toTurnCount: 0,
517 });
518 yield* provider.revertToCheckpoint({
519 sessionId: initial.sessionId,
520 turnCount: 0,
521 });
522
523 assert.deepEqual(routing.codex.startSession.mock.calls, [
524 [
525 {
526 provider: "codex",
527 resumeThreadId: initial.threadId,
528 },
529 ],
530 ]);
531 assert.equal(routing.checkpoint.getCheckpointDiff.mock.calls.length, 1);
532 assert.equal(routing.checkpoint.revertToCheckpoint.mock.calls.length, 1);
533
534 const diffInput = routing.checkpoint.getCheckpointDiff.mock.calls[0]?.[0];
535 const revertInput = routing.checkpoint.revertToCheckpoint.mock.calls[0]?.[0];
536 assert.equal(diffInput?.fromTurnCount, 0);
537 assert.equal(diffInput?.toTurnCount, 0);
538 assert.equal(diffInput?.sessionId === initial.sessionId, false);
539 assert.equal(revertInput?.sessionId, diffInput?.sessionId);
540 assert.equal(revertInput?.turnCount, 0);
541 }),
542 );
543
406544 it.effect("releases checkpoint session state when stopping all sessions", () =>
407545 Effect.gen(function* () {
408546 const provider = yield* ProviderService;
apps/server/src/provider/Layers/ProviderService.ts+413-179
......@@ -23,8 +23,9 @@ import {
2323 providerStopSessionInputSchema,
2424 type ProviderRuntimeEvent,
2525 type ProviderRuntimeTurnCompletedEvent,
26 type ProviderSession,
2627} from "@t3tools/contracts";
27import { Effect, Layer, PubSub, Queue, Stream } from "effect";
28import { Effect, Layer, Option, PubSub, Queue, Ref, Stream } from "effect";
2829
2930import { CheckpointService } from "../../checkpointing/Services/CheckpointService.ts";
3031import { ProviderValidationError } from "../Errors.ts";
......@@ -97,25 +98,100 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
9798
9899 const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
99100 const runtimeEventPubSub = yield* PubSub.unbounded<ProviderRuntimeEvent>();
101 const routedSessionAliasesRef = yield* Ref.make<Map<string, string>>(new Map());
102
103 const canonicalizeRuntimeEventSession = (
104 event: ProviderRuntimeEvent,
105 ): Effect.Effect<ProviderRuntimeEvent> =>
106 Ref.get(routedSessionAliasesRef).pipe(
107 Effect.map((aliases) => {
108 for (const [staleSessionId, liveSessionId] of aliases) {
109 if (liveSessionId === event.sessionId) {
110 return {
111 ...event,
112 sessionId: staleSessionId,
113 } satisfies ProviderRuntimeEvent;
114 }
115 }
116 return event;
117 }),
118 );
100119
101120 const publishRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> =>
102 Effect.sync(() => {
103 canonicalEventLogger?.write({
104 observedAt: new Date().toISOString(),
105 event,
106 });
107 }).pipe(
108 Effect.flatMap(() => PubSub.publish(runtimeEventPubSub, event)),
121 canonicalizeRuntimeEventSession(event).pipe(
122 Effect.tap((canonicalEvent) =>
123 Effect.sync(() => {
124 canonicalEventLogger?.write({
125 observedAt: new Date().toISOString(),
126 event: canonicalEvent,
127 });
128 }),
129 ),
130 Effect.flatMap((canonicalEvent) => PubSub.publish(runtimeEventPubSub, canonicalEvent)),
109131 Effect.asVoid,
110132 );
111133
112 const providers = yield* registry.listProviders();
113 const adapters = yield* Effect.forEach(providers, (provider) => registry.getByProvider(provider));
134 const upsertSessionBinding = (
135 session: ProviderSession,
136 operation: string,
137 ) =>
138 Effect.gen(function* () {
139 const threadId = session.threadId?.trim();
140 if (!threadId) {
141 return yield* toValidationError(
142 operation,
143 `Provider '${session.provider}' returned a session without threadId. threadId is required for checkpoint initialization.`,
144 );
145 }
146
147 yield* directory.upsert({
148 sessionId: session.sessionId,
149 provider: session.provider,
150 threadId,
151 });
114152
115 const staleSessionIds = yield* directory.reconcileWithAdapters(adapters);
116 yield* Effect.forEach(staleSessionIds, (sessionId) =>
117 checkpointService.releaseSession({ providerSessionId: sessionId }),
118 ).pipe(Effect.asVoid);
153 return threadId;
154 });
155
156 const clearAliasKey = (staleSessionId: string) =>
157 Ref.update(routedSessionAliasesRef, (current) => {
158 if (!current.has(staleSessionId)) {
159 return current;
160 }
161 const next = new Map(current);
162 next.delete(staleSessionId);
163 return next;
164 });
165
166 const clearAliasesReferencing = (sessionId: string) =>
167 Ref.update(routedSessionAliasesRef, (current) => {
168 let changed = false;
169 const next = new Map<string, string>();
170 for (const [key, value] of current) {
171 if (key === sessionId || value === sessionId) {
172 changed = true;
173 continue;
174 }
175 next.set(key, value);
176 }
177 return changed ? next : current;
178 });
179
180 const setAlias = (staleSessionId: string, liveSessionId: string) =>
181 Ref.update(routedSessionAliasesRef, (current) => {
182 const existing = current.get(staleSessionId);
183 if (existing === liveSessionId) {
184 return current;
185 }
186 const next = new Map(current);
187 next.set(staleSessionId, liveSessionId);
188 return next;
189 });
190
191 const providers = yield* registry.listProviders();
192 const adapters = yield* Effect.forEach(providers, (provider) =>
193 registry.getByProvider(provider),
194 );
119195
120196 const onTurnCompleted = (event: ProviderRuntimeTurnCompletedEvent): Effect.Effect<void> =>
121197 checkpointService
......@@ -166,203 +242,361 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
166242 ).pipe(Effect.forkScoped),
167243 ).pipe(Effect.asVoid);
168244
169 const adapterForSession = (sessionId: string) =>
170 directory
171 .getProvider(sessionId)
172 .pipe(Effect.flatMap((provider) => registry.getByProvider(provider)));
173
174 const startSession: ProviderServiceShape["startSession"] = (rawInput) =>
175 Effect.gen(function* () {
176 const parsed = providerSessionStartInputSchema.safeParse(rawInput);
177 if (!parsed.success) {
178 return yield* Effect.fail(
179 toValidationError(
245 const recoverSessionForThread = (input: {
246 readonly staleSessionId: string;
247 readonly provider: ProviderSession["provider"];
248 readonly threadId: string;
249 readonly operation: string;
250 }) =>
251 Effect.gen(function* () {
252 const adapter = yield* registry.getByProvider(input.provider);
253 const activeSessions = yield* adapter.listSessions();
254 const existing = activeSessions.find(
255 (session) => session.threadId?.trim() === input.threadId,
256 );
257 if (existing) {
258 const existingThreadId = yield* upsertSessionBinding(
259 existing,
260 `${input.operation}:upsertExistingSession`,
261 );
262 yield* directory.upsert({
263 sessionId: input.staleSessionId,
264 provider: existing.provider,
265 threadId: existingThreadId,
266 });
267 if (existing.sessionId !== input.staleSessionId) {
268 yield* setAlias(input.staleSessionId, existing.sessionId);
269 } else {
270 yield* clearAliasKey(input.staleSessionId);
271 }
272 return {
273 adapter,
274 sessionId: existing.sessionId,
275 } as const;
276 }
277
278 const resumed = yield* adapter.startSession({
279 provider: input.provider,
280 resumeThreadId: input.threadId,
281 });
282 if (resumed.provider !== adapter.provider) {
283 return yield* toValidationError(
284 input.operation,
285 `Adapter/provider mismatch while recovering stale session '${input.staleSessionId}'. Expected '${adapter.provider}', received '${resumed.provider}'.`,
286 );
287 }
288
289 const resumedThreadId = yield* upsertSessionBinding(
290 resumed,
291 `${input.operation}:upsertRecoveredSession`,
292 );
293 if (resumedThreadId !== input.threadId) {
294 return yield* toValidationError(
295 input.operation,
296 `Recovered session thread '${resumedThreadId}' does not match expected thread '${input.threadId}'.`,
297 );
298 }
299
300 yield* directory.upsert({
301 sessionId: input.staleSessionId,
302 provider: resumed.provider,
303 threadId: resumedThreadId,
304 });
305
306 const checkpointCwd = resumed.cwd ?? process.cwd();
307 yield* checkpointService.initializeForSession({
308 providerSessionId: resumed.sessionId,
309 cwd: checkpointCwd,
310 });
311
312 if (resumed.sessionId !== input.staleSessionId) {
313 yield* setAlias(input.staleSessionId, resumed.sessionId);
314 } else {
315 yield* clearAliasKey(input.staleSessionId);
316 }
317
318 return {
319 adapter,
320 sessionId: resumed.sessionId,
321 } as const;
322 });
323
324 const resolveRoutableSession = (input: {
325 readonly sessionId: string;
326 readonly operation: string;
327 readonly allowRecovery: boolean;
328 }) =>
329 Effect.gen(function* () {
330 const provider = yield* directory.getProvider(input.sessionId);
331 const adapter = yield* registry.getByProvider(provider);
332
333 const hasRequestedSession = yield* adapter.hasSession(input.sessionId);
334 if (hasRequestedSession) {
335 yield* clearAliasKey(input.sessionId);
336 return {
337 adapter,
338 sessionId: input.sessionId,
339 isActive: true,
340 } as const;
341 }
342
343 const alias = yield* Ref.get(routedSessionAliasesRef).pipe(
344 Effect.map((aliases) => aliases.get(input.sessionId)),
345 );
346 if (alias) {
347 const aliasIsActive = yield* adapter.hasSession(alias);
348 if (aliasIsActive) {
349 return {
350 adapter,
351 sessionId: alias,
352 isActive: true,
353 } as const;
354 }
355 yield* clearAliasKey(input.sessionId);
356 }
357
358 if (!input.allowRecovery) {
359 return {
360 adapter,
361 sessionId: input.sessionId,
362 isActive: false,
363 } as const;
364 }
365
366 const threadIdOption = yield* directory.getThreadId(input.sessionId);
367 const threadId = Option.getOrUndefined(threadIdOption)?.trim();
368 if (!threadId) {
369 return yield* toValidationError(
370 input.operation,
371 `Cannot recover stale session '${input.sessionId}' because no thread id is persisted.`,
372 );
373 }
374
375 const recovered = yield* recoverSessionForThread({
376 staleSessionId: input.sessionId,
377 provider,
378 threadId,
379 operation: input.operation,
380 });
381
382 return {
383 adapter: recovered.adapter,
384 sessionId: recovered.sessionId,
385 isActive: true,
386 } as const;
387 });
388
389 const startSession: ProviderServiceShape["startSession"] = (rawInput) =>
390 Effect.gen(function* () {
391 const parsed = providerSessionStartInputSchema.safeParse(rawInput);
392 if (!parsed.success) {
393 return yield* toValidationError(
180394 "ProviderService.startSession",
181395 parsed.error.message,
182396 parsed.error.cause,
183 ),
184 );
185 }
397 );
398 }
186399
187 const input = parsed.data;
188 const adapter = yield* registry.getByProvider(input.provider);
189 const session = yield* adapter.startSession(input);
400 const input = parsed.data;
401 const adapter = yield* registry.getByProvider(input.provider);
402 const session = yield* adapter.startSession(input);
190403
191 if (session.provider !== adapter.provider) {
192 return yield* Effect.fail(
193 toValidationError(
404 if (session.provider !== adapter.provider) {
405 return yield* toValidationError(
194406 "ProviderService.startSession",
195407 `Adapter/provider mismatch: requested '${adapter.provider}', received '${session.provider}'.`,
196 ),
197 );
198 }
408 );
409 }
199410
200 const threadId = session.threadId?.trim();
201 if (!threadId) {
202 return yield* Effect.fail(
203 toValidationError(
204 "ProviderService.startSession",
205 `Provider '${adapter.provider}' returned a session without threadId. threadId is required for checkpoint initialization.`,
206 ),
207 );
208 }
411 yield* upsertSessionBinding(session, "ProviderService.startSession");
209412
210 yield* directory.upsert({
211 sessionId: session.sessionId,
212 provider: session.provider,
213 threadId,
214 });
413 const checkpointCwd = session.cwd ?? input.cwd ?? process.cwd();
414 yield* checkpointService.initializeForSession({
415 providerSessionId: session.sessionId,
416 cwd: checkpointCwd,
417 });
215418
216 const checkpointCwd = session.cwd ?? input.cwd ?? process.cwd();
217 yield* checkpointService.initializeForSession({
218 providerSessionId: session.sessionId,
219 cwd: checkpointCwd,
419 return session;
220420 });
221421
222 return session;
223 });
422 const sendTurn: ProviderServiceShape["sendTurn"] = (rawInput) =>
423 Effect.gen(function* () {
424 const parsed = providerSendTurnInputSchema.safeParse(rawInput);
425 if (!parsed.success) {
426 return yield* toValidationError(
427 "ProviderService.sendTurn",
428 parsed.error.message,
429 parsed.error.cause,
430 );
431 }
432
433 const input = parsed.data;
434 const routed = yield* resolveRoutableSession({
435 sessionId: input.sessionId,
436 operation: "ProviderService.sendTurn",
437 allowRecovery: true,
438 });
439 return yield* routed.adapter.sendTurn({
440 ...input,
441 sessionId: routed.sessionId,
442 });
443 });
224444
225 const sendTurn: ProviderServiceShape["sendTurn"] = (rawInput) =>
226 Effect.gen(function* () {
227 const parsed = providerSendTurnInputSchema.safeParse(rawInput);
228 if (!parsed.success) {
229 return yield* Effect.fail(
230 toValidationError("ProviderService.sendTurn", parsed.error.message, parsed.error.cause),
231 );
232 }
233
234 const input = parsed.data;
235 const adapter = yield* adapterForSession(input.sessionId);
236 return yield* adapter.sendTurn(input);
237 });
238
239 const interruptTurn: ProviderServiceShape["interruptTurn"] = (rawInput) =>
240 Effect.gen(function* () {
241 const parsed = providerInterruptTurnInputSchema.safeParse(rawInput);
242 if (!parsed.success) {
243 return yield* Effect.fail(
244 toValidationError(
445 const interruptTurn: ProviderServiceShape["interruptTurn"] = (rawInput) =>
446 Effect.gen(function* () {
447 const parsed = providerInterruptTurnInputSchema.safeParse(rawInput);
448 if (!parsed.success) {
449 return yield* toValidationError(
245450 "ProviderService.interruptTurn",
246451 parsed.error.message,
247452 parsed.error.cause,
248 ),
249 );
250 }
251
252 const input = parsed.data;
253 const adapter = yield* adapterForSession(input.sessionId);
254 yield* adapter.interruptTurn(input.sessionId, input.turnId);
255 });
256
257 const respondToRequest: ProviderServiceShape["respondToRequest"] = (rawInput) =>
258 Effect.gen(function* () {
259 const parsed = providerRespondToRequestInputSchema.safeParse(rawInput);
260 if (!parsed.success) {
261 return yield* Effect.fail(
262 toValidationError(
453 );
454 }
455
456 const input = parsed.data;
457 const routed = yield* resolveRoutableSession({
458 sessionId: input.sessionId,
459 operation: "ProviderService.interruptTurn",
460 allowRecovery: true,
461 });
462 yield* routed.adapter.interruptTurn(routed.sessionId, input.turnId);
463 });
464
465 const respondToRequest: ProviderServiceShape["respondToRequest"] = (rawInput) =>
466 Effect.gen(function* () {
467 const parsed = providerRespondToRequestInputSchema.safeParse(rawInput);
468 if (!parsed.success) {
469 return yield* toValidationError(
263470 "ProviderService.respondToRequest",
264471 parsed.error.message,
265472 parsed.error.cause,
266 ),
267 );
268 }
269
270 const input = parsed.data;
271 const adapter = yield* adapterForSession(input.sessionId);
272 yield* adapter.respondToRequest(input.sessionId, input.requestId, input.decision);
273 });
274
275 const stopSession: ProviderServiceShape["stopSession"] = (rawInput) =>
276 Effect.gen(function* () {
277 const parsed = providerStopSessionInputSchema.safeParse(rawInput);
278 if (!parsed.success) {
279 return yield* Effect.fail(
280 toValidationError(
473 );
474 }
475
476 const input = parsed.data;
477 const routed = yield* resolveRoutableSession({
478 sessionId: input.sessionId,
479 operation: "ProviderService.respondToRequest",
480 allowRecovery: true,
481 });
482 yield* routed.adapter.respondToRequest(routed.sessionId, input.requestId, input.decision);
483 });
484
485 const stopSession: ProviderServiceShape["stopSession"] = (rawInput) =>
486 Effect.gen(function* () {
487 const parsed = providerStopSessionInputSchema.safeParse(rawInput);
488 if (!parsed.success) {
489 return yield* toValidationError(
281490 "ProviderService.stopSession",
282491 parsed.error.message,
283492 parsed.error.cause,
284 ),
285 );
286 }
287
288 const input = parsed.data;
289 const adapter = yield* adapterForSession(input.sessionId);
290 yield* adapter.stopSession(input.sessionId);
291 yield* checkpointService.releaseSession({ providerSessionId: input.sessionId });
292 yield* directory.remove(input.sessionId);
293 });
294
295 const listSessions: ProviderServiceShape["listSessions"] = () =>
296 Effect.forEach(adapters, (adapter) => adapter.listSessions()).pipe(
297 Effect.map((sessionsByProvider) => sessionsByProvider.flatMap((sessions) => sessions)),
298 );
493 );
494 }
495
496 const input = parsed.data;
497 const routed = yield* resolveRoutableSession({
498 sessionId: input.sessionId,
499 operation: "ProviderService.stopSession",
500 allowRecovery: false,
501 });
502 if (routed.isActive) {
503 yield* routed.adapter.stopSession(routed.sessionId);
504 }
505 if (routed.sessionId !== input.sessionId) {
506 yield* checkpointService.releaseSession({ providerSessionId: routed.sessionId });
507 yield* directory.remove(routed.sessionId);
508 yield* clearAliasesReferencing(routed.sessionId);
509 }
510 yield* checkpointService.releaseSession({ providerSessionId: input.sessionId });
511 yield* directory.remove(input.sessionId);
512 yield* clearAliasesReferencing(input.sessionId);
513 });
514
515 const listSessions: ProviderServiceShape["listSessions"] = () =>
516 Effect.forEach(adapters, (adapter) => adapter.listSessions()).pipe(
517 Effect.map((sessionsByProvider) => sessionsByProvider.flatMap((sessions) => sessions)),
518 );
299519
300 const listCheckpoints: ProviderServiceShape["listCheckpoints"] = (rawInput) =>
301 Effect.gen(function* () {
302 const parsed = providerListCheckpointsInputSchema.safeParse(rawInput);
303 if (!parsed.success) {
304 return yield* Effect.fail(
305 toValidationError(
520 const listCheckpoints: ProviderServiceShape["listCheckpoints"] = (rawInput) =>
521 Effect.gen(function* () {
522 const parsed = providerListCheckpointsInputSchema.safeParse(rawInput);
523 if (!parsed.success) {
524 return yield* toValidationError(
306525 "ProviderService.listCheckpoints",
307526 parsed.error.message,
308527 parsed.error.cause,
309 ),
310 );
311 }
312
313 const input = parsed.data;
314 yield* directory.getProvider(input.sessionId);
315 return yield* checkpointService.listCheckpoints(input);
316 });
317
318 const getCheckpointDiff: ProviderServiceShape["getCheckpointDiff"] = (rawInput) =>
319 Effect.gen(function* () {
320 const parsed = providerGetCheckpointDiffInputSchema.safeParse(rawInput);
321 if (!parsed.success) {
322 return yield* Effect.fail(
323 toValidationError(
528 );
529 }
530
531 const input = parsed.data;
532 const routed = yield* resolveRoutableSession({
533 sessionId: input.sessionId,
534 operation: "ProviderService.listCheckpoints",
535 allowRecovery: true,
536 });
537 return yield* checkpointService.listCheckpoints({
538 ...input,
539 sessionId: routed.sessionId,
540 });
541 });
542
543 const getCheckpointDiff: ProviderServiceShape["getCheckpointDiff"] = (rawInput) =>
544 Effect.gen(function* () {
545 const parsed = providerGetCheckpointDiffInputSchema.safeParse(rawInput);
546 if (!parsed.success) {
547 return yield* toValidationError(
324548 "ProviderService.getCheckpointDiff",
325549 parsed.error.message,
326550 parsed.error.cause,
327 ),
328 );
329 }
330
331 const input = parsed.data;
332 yield* directory.getProvider(input.sessionId);
333 return yield* checkpointService.getCheckpointDiff(input);
334 });
335
336 const revertToCheckpoint: ProviderServiceShape["revertToCheckpoint"] = (rawInput) =>
337 Effect.gen(function* () {
338 const parsed = providerRevertToCheckpointInputSchema.safeParse(rawInput);
339 if (!parsed.success) {
340 return yield* Effect.fail(
341 toValidationError(
551 );
552 }
553
554 const input = parsed.data;
555 const routed = yield* resolveRoutableSession({
556 sessionId: input.sessionId,
557 operation: "ProviderService.getCheckpointDiff",
558 allowRecovery: true,
559 });
560 return yield* checkpointService.getCheckpointDiff({
561 ...input,
562 sessionId: routed.sessionId,
563 });
564 });
565
566 const revertToCheckpoint: ProviderServiceShape["revertToCheckpoint"] = (rawInput) =>
567 Effect.gen(function* () {
568 const parsed = providerRevertToCheckpointInputSchema.safeParse(rawInput);
569 if (!parsed.success) {
570 return yield* toValidationError(
342571 "ProviderService.revertToCheckpoint",
343572 parsed.error.message,
344573 parsed.error.cause,
345 ),
346 );
347 }
348
349 const input = parsed.data;
350 yield* directory.getProvider(input.sessionId);
351 return yield* checkpointService.revertToCheckpoint(input);
352 });
353
354 const stopAll: ProviderServiceShape["stopAll"] = () =>
355 Effect.gen(function* () {
356 const sessionIds = yield* directory.listSessionIds();
357 yield* Effect.forEach(adapters, (adapter) => adapter.stopAll()).pipe(Effect.asVoid);
358 yield* Effect.forEach(
359 sessionIds,
360 (sessionId) =>
361 checkpointService.releaseSession({ providerSessionId: sessionId }).pipe(
362 Effect.flatMap(() => directory.remove(sessionId)),
363 ),
364 ).pipe(Effect.asVoid);
365 });
574 );
575 }
576
577 const input = parsed.data;
578 const routed = yield* resolveRoutableSession({
579 sessionId: input.sessionId,
580 operation: "ProviderService.revertToCheckpoint",
581 allowRecovery: true,
582 });
583 return yield* checkpointService.revertToCheckpoint({
584 ...input,
585 sessionId: routed.sessionId,
586 });
587 });
588
589 const stopAll: ProviderServiceShape["stopAll"] = () =>
590 Effect.gen(function* () {
591 const sessionIds = yield* directory.listSessionIds();
592 yield* Effect.forEach(adapters, (adapter) => adapter.stopAll()).pipe(Effect.asVoid);
593 // Keep persisted session bindings so stale sessions can be resumed after
594 // process restart via resumeThreadId.
595 yield* Effect.forEach(sessionIds, (sessionId) =>
596 checkpointService.releaseSession({ providerSessionId: sessionId }),
597 ).pipe(Effect.asVoid);
598 yield* Ref.set(routedSessionAliasesRef, new Map());
599 });
366600
367601 return {
368602 startSession,
apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts+2-45
......@@ -5,13 +5,12 @@ import path from "node:path";
55import { NodeServices } from "@effect/platform-node";
66import { it, assert } from "@effect/vitest";
77import { assertFailure, assertSome } from "@effect/vitest/utils";
8import { Effect, Layer, Stream } from "effect";
8import { Effect, Layer } from "effect";
99import * as SqlClient from "effect/unstable/sql/SqlClient";
1010
1111import { makeSqlitePersistenceLive, SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";
1212import { ProviderSessionRepositoryLive } from "../../persistence/Layers/ProviderSessions.ts";
13import { ProviderSessionNotFoundError, ProviderValidationError, type ProviderAdapterError } from "../Errors.ts";
14import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts";
13import { ProviderSessionNotFoundError, ProviderValidationError } from "../Errors.ts";
1514import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts";
1615import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts";
1716
......@@ -22,24 +21,6 @@ function makeDirectoryLayer<E, R>(persistenceLayer: Layer.Layer<SqlClient.SqlCli
2221 );
2322}
2423
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
4324const layer = it.layer(makeDirectoryLayer(SqlitePersistenceMemory));
4425
4526layer("ProviderSessionDirectoryLive", (it) => {
......@@ -118,28 +99,4 @@ layer("ProviderSessionDirectoryLive", (it) => {
11899
119100 fs.rmSync(tempDir, { recursive: true, force: true });
120101 }).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 }));
145102});
apps/server/src/provider/Layers/ProviderSessionDirectory.ts-33
......@@ -109,45 +109,12 @@ const makeProviderSessionDirectory = Effect.gen(function* () {
109109 const listSessionIds: ProviderSessionDirectoryShape["listSessionIds"] = () =>
110110 listBindings().pipe(Effect.map((bindings) => bindings.map((binding) => binding.sessionId)));
111111
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 });
143
144112 return {
145113 upsert,
146114 getProvider,
147115 getThreadId,
148116 remove,
149117 listSessionIds,
150 reconcileWithAdapters,
151118 } satisfies ProviderSessionDirectoryShape;
152119});
153120
apps/server/src/provider/Services/ProviderAdapterRegistry.ts+2
......@@ -35,3 +35,5 @@ export class ProviderAdapterRegistry extends ServiceMap.Service<
3535 ProviderAdapterRegistry,
3636 ProviderAdapterRegistryShape
3737>()("provider/ProviderAdapterRegistry") {}
38
39// Dummy comment for workflow testing.
apps/server/src/provider/Services/ProviderService.ts+4-1
......@@ -95,7 +95,10 @@ export interface ProviderServiceShape {
9595 ) => Effect.Effect<ProviderRevertToCheckpointResult, ProviderServiceError>;
9696
9797 /**
98 * Stop all provider sessions.
98 * Stop all active provider sessions.
99 *
100 * Persisted session bindings remain so stale sessions can be resumed after
101 * process restart.
99102 */
100103 readonly stopAll: () => Effect.Effect<void, ProviderServiceError>;
101104
apps/server/src/provider/Services/ProviderSessionDirectory.ts-13
......@@ -12,12 +12,10 @@ import { Option, ServiceMap } from "effect";
1212import type { Effect } from "effect";
1313
1414import type {
15 ProviderAdapterError,
1615 ProviderSessionDirectoryPersistenceError,
1716 ProviderSessionNotFoundError,
1817 ProviderValidationError,
1918} from "../Errors.ts";
20import type { ProviderAdapterShape } from "./ProviderAdapter.ts";
2119
2220export interface ProviderSessionBinding {
2321 readonly sessionId: string;
......@@ -70,17 +68,6 @@ export interface ProviderSessionDirectoryShape {
7068 ProviderSessionDirectoryPersistenceError
7169 >;
7270
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 >;
8471}
8572
8673/**
apps/server/tsconfig.json+6-1
......@@ -5,7 +5,12 @@
55 "types": ["node"],
66 "lib": ["ES2023", "esnext.disposable"],
77 "noEmit": true,
8 "allowImportingTsExtensions": true
8 "allowImportingTsExtensions": true,
9 "plugins": [
10 {
11 "name": "@effect/language-service"
12 }
13 ]
914 },
1015 "include": ["src", "integration", "tsdown.config.ts", "scripts"]
1116}
bun.lock+7
......@@ -35,6 +35,7 @@
3535 },
3636 "dependencies": {
3737 "@effect/platform-node": "4.0.0-beta.10",
38 "@effect/sql-sqlite-bun": "4.0.0-beta.10",
3839 "@pierre/diffs": "^1.1.0-beta.16",
3940 "effect": "4.0.0-beta.10",
4041 "node-pty": "^1.1.0",
......@@ -42,6 +43,7 @@
4243 "ws": "^8.18.0",
4344 },
4445 "devDependencies": {
46 "@effect/language-service": "0.75.1",
4547 "@effect/vitest": "4.0.0-beta.10",
4648 "@t3tools/contracts": "workspace:*",
4749 "@t3tools/web": "workspace:*",
......@@ -96,6 +98,7 @@
9698 "zod": "^3.24.1",
9799 },
98100 "devDependencies": {
101 "@effect/language-service": "0.75.1",
99102 "tsdown": "^0.20.3",
100103 "typescript": "^5.7.3",
101104 },
......@@ -150,10 +153,14 @@
150153
151154 "@base-ui/utils": ["@base-ui/utils@0.2.5", "", { "dependencies": { "@babel/runtime": "^7.28.6", "@floating-ui/utils": "^0.2.10", "reselect": "^5.1.1", "use-sync-external-store": "^1.6.0" }, "peerDependencies": { "@types/react": "^17 || ^18 || ^19", "react": "^17 || ^18 || ^19", "react-dom": "^17 || ^18 || ^19" }, "optionalPeers": ["@types/react"] }, "sha512-oYC7w0gp76RI5MxprlGLV0wze0SErZaRl3AAkeP3OnNB/UBMb6RqNf6ZSIlxOc9Qp68Ab3C2VOcJQyRs7Xc7Vw=="],
152155
156 "@effect/language-service": ["@effect/language-service@0.75.1", "", { "bin": { "effect-language-service": "cli.js" } }, "sha512-g9xD2tAQgRFpYC2YgpZq02VeSL5fBbFJ0B/g1o+14NuNmwtaYJc7SjiLWAA9eyhJHosNrn6h1Ye+Kx6j5mN0AA=="],
157
153158 "@effect/platform-node": ["@effect/platform-node@4.0.0-beta.10", "", { "dependencies": { "@effect/platform-node-shared": "^4.0.0-beta.10", "mime": "^4.1.0", "undici": "^7.20.0" }, "peerDependencies": { "effect": "^4.0.0-beta.10", "ioredis": "^5.7.0" } }, "sha512-ikZ/uyos5okYAHE+wWOwZkHgriFUTY3aY8Kdqbj/ro8xw0z9YXWN4fbyQAZUWkBcieDKS36N+QfHRMPG4fEWRA=="],
154159
155160 "@effect/platform-node-shared": ["@effect/platform-node-shared@4.0.0-beta.10", "", { "dependencies": { "@types/ws": "^8.18.1", "ws": "^8.19.0" }, "peerDependencies": { "effect": "^4.0.0-beta.10" } }, "sha512-gBSeOG2bfchXFdgdi5oipHKynOzSd7BlzTy5QcjFOoCUTtLP9igqohPB9hTkV3a6seWE2JJqsgsQ79VuLb7fAg=="],
156161
162 "@effect/sql-sqlite-bun": ["@effect/sql-sqlite-bun@4.0.0-beta.10", "", { "peerDependencies": { "effect": "^4.0.0-beta.10" } }, "sha512-LY1jNu2uqGYsJCVjVU73nzcZ+kAYBDsG7O8x5ddmkIXj10RmnIhjYMStWBa0NZOFg4Hsl+CadxtsDy3edaOVeA=="],
163
157164 "@effect/vitest": ["@effect/vitest@4.0.0-beta.10", "", { "peerDependencies": { "effect": "^4.0.0-beta.10", "vitest": "^3.0.0" } }, "sha512-WmE8vCNw+eDoAq9CwwymrAdIBw1f+8hDP3AUF7jyC7/Cj/kyAYOFwkbtKmDXIFaTlf7G/2fmulEXeCO5W8j0kQ=="],
158165
159166 "@electron/get": ["@electron/get@2.0.3", "", { "dependencies": { "debug": "^4.1.1", "env-paths": "^2.2.0", "fs-extra": "^8.1.0", "got": "^11.8.5", "progress": "^2.0.3", "semver": "^6.2.0", "sumchecker": "^3.0.1" }, "optionalDependencies": { "global-agent": "^3.0.0" } }, "sha512-Qkzpg2s9GnVV2I2BjRksUi43U5e6+zaQMcjoJy0C+C5oxaKl+fmckGDQFtRpZpZV0NQekuZZ+tGz7EA9TVnQtQ=="],
packages/contracts/package.json+2
......@@ -19,6 +19,7 @@
1919 "scripts": {
2020 "dev": "tsdown src/index.ts --format esm,cjs --dts --watch --clean",
2121 "build": "tsdown src/index.ts --format esm,cjs --dts --clean",
22 "prepare": "effect-language-service patch",
2223 "typecheck": "tsc --noEmit",
2324 "test": "vitest run"
2425 },
......@@ -27,6 +28,7 @@
2728 "zod": "^3.24.1"
2829 },
2930 "devDependencies": {
31 "@effect/language-service": "0.75.1",
3032 "tsdown": "^0.20.3",
3133 "typescript": "^5.7.3"
3234 }
packages/contracts/tsconfig.json+7
......@@ -1,4 +1,11 @@
11{
22 "extends": "../../tsconfig.base.json",
3 "compilerOptions": {
4 "plugins": [
5 {
6 "name": "@effect/language-service"
7 }
8 ]
9 },
310 "include": ["src"]
411}