authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-22 16:12:48-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
log54dd17e7b9f576f3219d2512497c492180b76fd1
treeb7cd4b4eb0710f54134efd537de2f90ae330fe7b
parent32942a5f20246ad6a0bd8f0e72e73c084676d45b

more effect:checkpointing

18 files changed, 2832 insertions(+), 183 deletions(-)

.plans/11-effect.md created+42
......@@ -0,0 +1,42 @@
1
2
3PR 1: Service contracts + error taxonomy
4Add ProviderService, CodexService, CheckpointStore as Context.Tag service defs.
5Add typed Schema.TaggedError hierarchies for all 3 services (cause: Schema.optional(Schema.Defect) on each).
6No behavior change yet, just interfaces and compile-time wiring points.
7PR 2: CheckpointStore Effect adapter
8Wrap current filesystemCheckpointStore behind CheckpointStoreLive (adapter).
9Map all thrown/Promise errors to tagged errors.
10Add service tests proving parity for isGitRepository, capture, restore, diff, prune.
11PR 3: CodexService Effect adapter
12Wrap current CodexAppServerManager behind CodexServiceLive (adapter).
13Convert public API to Effect return types with typed errors.
14Preserve existing EventEmitter internally for now, but expose Effect-friendly subscribe API.
15PR 4: ProviderService Effect adapter
16Wrap current ProviderManager behind ProviderServiceLive (adapter).
17Provider methods become Effect methods with typed errors.
18Route emitted provider events through an Effect PubSub surface.
19PR 5: wsServer migration to Effect services
20Stop instantiating provider/codex classes directly in wsServer.
21Resolve ProviderService (and related services) from one runtime/layer graph.
22Keep WS contract behavior identical.
23PR 6: Native CheckpointStore implementation
24Refactor checkpoint internals from Promise/throws to native Effect.
25Replace ad-hoc locking with Effect concurrency primitive (keyed lock/semaphore/queue).
26Keep adapter tests plus new failure-path tests.
27PR 7: Codex transport/RPC core as native Effect
28Split codex into scoped process layer + RPC request/response layer + session registry.
29Replace timeout/pending maps with Deferred + Effect timeout/finalizer semantics.
30Keep protocol behavior and ordering guarantees.
31PR 8: Codex protocol decoding hardening
32Replace ad-hoc unknown parsing with runtime schema decoding for inbound/outbound protocol shapes.
33Map decode failures to typed tagged errors (with root cause).
34Add regression tests for malformed/partial protocol messages.
35PR 9: Native ProviderService orchestration
36Rebuild provider logic in Effect using CodexService + CheckpointStore dependencies.
37Move event fanout, checkpoint capture/revert orchestration, thread-log routing to Effect state/services.
38Remove throw-based flow entirely from provider path.
39PR 10: Cleanup + deprecation removal
40Remove legacy class implementations/adapters once parity is proven.
41Finalize layer composition and startup graph docs.
42Add architecture notes for service boundaries and error model.
\ No newline at end of file
apps/server/src/checkpointing/Errors.ts+2-22
......@@ -1,4 +1,5 @@
11import { Schema } from "effect";
2import type { CheckpointRepositoryError as CheckpointMetadataRepositoryError } from "../persistence/Errors.ts";
23
34/**
45 * CheckpointValidationError - Invalid checkpoint API input.
......@@ -67,22 +68,6 @@ export class CheckpointRepositoryError extends Schema.TaggedErrorClass<Checkpoin
6768 }
6869}
6970
70/**
71 * CheckpointCatalogPersistenceError - Metadata persistence/read failure.
72 */
73export class CheckpointCatalogPersistenceError extends Schema.TaggedErrorClass<CheckpointCatalogPersistenceError>()(
74 "CheckpointCatalogPersistenceError",
75 {
76 operation: Schema.String,
77 detail: Schema.String,
78 cause: Schema.optional(Schema.Defect),
79 },
80) {
81 override get message(): string {
82 return `Checkpoint catalog persistence error in ${this.operation}: ${this.detail}`;
83 }
84}
85
8671/**
8772 * CheckpointServiceValidationError - Invalid application-level checkpoint request.
8873 */
......@@ -131,18 +116,13 @@ export class CheckpointInvariantError extends Schema.TaggedErrorClass<Checkpoint
131116}
132117
133118export type CheckpointStoreError =
134 | CheckpointValidationError
135119 | CheckpointGitCommandError
136120 | CheckpointUnavailableError
137121 | CheckpointRepositoryError;
138122
139export type CheckpointCatalogError =
140 | CheckpointValidationError
141 | CheckpointCatalogPersistenceError;
142
143123export type CheckpointServiceError =
144124 | CheckpointStoreError
145 | CheckpointCatalogError
125 | CheckpointMetadataRepositoryError
146126 | CheckpointServiceValidationError
147127 | CheckpointSessionNotFoundError
148128 | CheckpointInvariantError;
apps/server/src/checkpointing/Layers/CheckpointService.test.ts created+404
......@@ -0,0 +1,404 @@
1import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
2import os from "node:os";
3import path from "node:path";
4
5import type {
6 ProviderApprovalDecision,
7 ProviderEvent,
8 ProviderKind,
9 ProviderSendTurnInput,
10 ProviderSession,
11 ProviderSessionStartInput,
12 ProviderTurnStartResult,
13} from "@t3tools/contracts";
14import { NodeServices } from "@effect/platform-node";
15import { it, assert } from "@effect/vitest";
16import { Effect, Layer } from "effect";
17
18import { runGit as runGitProcess } from "../../git/Process.ts";
19import { CheckpointServiceLive } from "./CheckpointService.ts";
20import { CheckpointStoreLive } from "./CheckpointStore.ts";
21import { CheckpointService } from "../Services/CheckpointService.ts";
22import { CheckpointStore } from "../Services/CheckpointStore.ts";
23import {
24 ProviderAdapterSessionNotFoundError,
25 ProviderAdapterValidationError,
26 ProviderUnsupportedError,
27 type ProviderAdapterError,
28} from "../../provider/Errors.ts";
29import { ProviderAdapterRegistry } from "../../provider/Services/ProviderAdapterRegistry.ts";
30import type {
31 ProviderAdapterShape,
32 ProviderThreadSnapshot,
33 ProviderThreadTurnSnapshot,
34} from "../../provider/Services/ProviderAdapter.ts";
35import { ProviderSessionDirectory } from "../../provider/Services/ProviderSessionDirectory.ts";
36import { ProviderSessionDirectoryLive } from "../../provider/Layers/ProviderSessionDirectory.ts";
37import { makeSqliteCheckpointRepositoryLive } from "../../persistence/Layers/Checkpoints.ts";
38import { CheckpointRepository } from "../../persistence/Services/Checkpoints.ts";
39
40const makeTemporaryDirectoryScoped = (prefix: string) =>
41 Effect.acquireRelease(
42 Effect.promise(() => mkdtemp(path.join(os.tmpdir(), prefix))),
43 (directory) => Effect.promise(() => rm(directory, { recursive: true, force: true })),
44 );
45
46const runGit = (cwd: string, args: readonly string[]) =>
47 runGitProcess({
48 operation: "CheckpointService.test.git",
49 cwd,
50 args,
51 }).pipe(
52 Effect.map((result) => result.stdout.trim()),
53 Effect.provide(NodeServices.layer),
54 );
55
56function createRepo() {
57 return Effect.gen(function* () {
58 const cwd = yield* makeTemporaryDirectoryScoped("t3-checkpoint-service-repo-");
59 yield* runGit(cwd, ["init", "--initial-branch=main"]);
60 yield* runGit(cwd, ["config", "user.email", "test@example.com"]);
61 yield* runGit(cwd, ["config", "user.name", "Test User"]);
62 yield* Effect.promise(() => writeFile(path.join(cwd, "README.md"), "v1\n"));
63 yield* runGit(cwd, ["add", "."]);
64 yield* runGit(cwd, ["commit", "-m", "Initial"]);
65 return cwd;
66 });
67}
68
69class InMemoryProviderAdapter implements ProviderAdapterShape<ProviderAdapterError> {
70 readonly provider = "codex" as const;
71
72 private active = true;
73 private snapshot: ProviderThreadSnapshot;
74 private readonly session: ProviderSession;
75
76 constructor(
77 readonly sessionId: string,
78 readonly cwd: string,
79 turns: ReadonlyArray<ProviderThreadTurnSnapshot>,
80 ) {
81 const now = new Date().toISOString();
82 this.session = {
83 sessionId,
84 provider: "codex",
85 status: "ready",
86 threadId: "thread-1",
87 cwd,
88 createdAt: now,
89 updatedAt: now,
90 };
91 this.snapshot = {
92 threadId: this.session.threadId ?? "thread-1",
93 turns,
94 };
95 }
96
97 setTurns(turns: ReadonlyArray<ProviderThreadTurnSnapshot>): void {
98 this.snapshot = {
99 threadId: this.snapshot.threadId,
100 turns,
101 };
102 }
103
104 getTurns(): ReadonlyArray<ProviderThreadTurnSnapshot> {
105 return this.snapshot.turns;
106 }
107
108 readonly startSession = (
109 _input: ProviderSessionStartInput,
110 ): Effect.Effect<ProviderSession, ProviderAdapterError> =>
111 Effect.fail(
112 new ProviderAdapterValidationError({
113 provider: this.provider,
114 operation: "startSession",
115 issue: "not implemented in test adapter",
116 }),
117 );
118
119 readonly sendTurn = (
120 _input: ProviderSendTurnInput,
121 ): Effect.Effect<ProviderTurnStartResult, ProviderAdapterError> =>
122 Effect.fail(
123 new ProviderAdapterValidationError({
124 provider: this.provider,
125 operation: "sendTurn",
126 issue: "not implemented in test adapter",
127 }),
128 );
129
130 readonly interruptTurn = (
131 _sessionId: string,
132 _turnId?: string,
133 ): Effect.Effect<void, ProviderAdapterError> => Effect.void;
134
135 readonly respondToRequest = (
136 _sessionId: string,
137 _requestId: string,
138 _decision: ProviderApprovalDecision,
139 ): Effect.Effect<void, ProviderAdapterError> => Effect.void;
140
141 readonly stopSession = (sessionId: string): Effect.Effect<void, ProviderAdapterError> =>
142 Effect.sync(() => {
143 if (sessionId === this.sessionId) {
144 this.active = false;
145 }
146 });
147
148 readonly listSessions = (): Effect.Effect<ReadonlyArray<ProviderSession>> =>
149 Effect.sync(() => (this.active ? [this.session] : []));
150
151 readonly hasSession = (sessionId: string): Effect.Effect<boolean> =>
152 Effect.succeed(this.active && sessionId === this.sessionId);
153
154 readonly readThread = (
155 sessionId: string,
156 ): Effect.Effect<ProviderThreadSnapshot, ProviderAdapterError> =>
157 this.hasSession(sessionId).pipe(
158 Effect.flatMap((hasSession) =>
159 hasSession
160 ? Effect.succeed(this.snapshot)
161 : Effect.fail(
162 new ProviderAdapterSessionNotFoundError({
163 provider: this.provider,
164 sessionId,
165 }),
166 ),
167 ),
168 );
169
170 readonly rollbackThread = (
171 sessionId: string,
172 numTurns: number,
173 ): Effect.Effect<ProviderThreadSnapshot, ProviderAdapterError> =>
174 this.readThread(sessionId).pipe(
175 Effect.flatMap((snapshot) => {
176 if (!Number.isInteger(numTurns) || numTurns < 0 || numTurns > snapshot.turns.length) {
177 return Effect.fail(
178 new ProviderAdapterValidationError({
179 provider: this.provider,
180 operation: "rollbackThread",
181 issue: "invalid rollback input",
182 }),
183 );
184 }
185 const next = {
186 threadId: snapshot.threadId,
187 turns: snapshot.turns.slice(0, snapshot.turns.length - numTurns),
188 } satisfies ProviderThreadSnapshot;
189 this.snapshot = next;
190 return Effect.succeed(next);
191 }),
192 );
193
194 readonly stopAll = (): Effect.Effect<void, ProviderAdapterError> =>
195 Effect.sync(() => {
196 this.active = false;
197 });
198
199 readonly subscribeToEvents = (
200 _callback: (event: ProviderEvent) => void,
201 ): Effect.Effect<() => void, ProviderAdapterError> => Effect.succeed(() => undefined);
202}
203
204function makeFixture(turns: ReadonlyArray<ProviderThreadTurnSnapshot>) {
205 return Effect.gen(function* () {
206 const cwd = yield* createRepo();
207 const stateDir = yield* makeTemporaryDirectoryScoped("t3-checkpoint-service-state-");
208 const dbPath = path.join(stateDir, "orchestration.sqlite");
209
210 const sessionId = "sess-1";
211 const adapter = new InMemoryProviderAdapter(sessionId, cwd, turns);
212
213 const registry: typeof ProviderAdapterRegistry.Service = {
214 getByProvider: (provider: ProviderKind) =>
215 provider === "codex"
216 ? Effect.succeed(adapter)
217 : Effect.fail(new ProviderUnsupportedError({ provider })),
218 listProviders: () => Effect.succeed(["codex"]),
219 };
220
221 const dependencies = Layer.mergeAll(
222 Layer.provide(CheckpointStoreLive, NodeServices.layer),
223 makeSqliteCheckpointRepositoryLive(dbPath),
224 Layer.succeed(ProviderAdapterRegistry, registry),
225 ProviderSessionDirectoryLive,
226 );
227
228 const layer = CheckpointServiceLive.pipe(Layer.provideMerge(dependencies));
229
230 return {
231 layer,
232 adapter,
233 sessionId,
234 cwd,
235 };
236 });
237}
238
239it.effect(
240 "CheckpointServiceLive captures checkpoints, diffs refs, and reverts workspace + turns",
241 () =>
242 Effect.gen(function* () {
243 const firstTurn: ProviderThreadTurnSnapshot = {
244 id: "turn-1",
245 items: [{ type: "userMessage", content: [{ type: "text", text: "first" }] }],
246 };
247 const secondTurn: ProviderThreadTurnSnapshot = {
248 id: "turn-2",
249 items: [{ type: "agentMessage", text: "second" }],
250 };
251
252 const fixture = yield* makeFixture([]);
253
254 yield* Effect.gen(function* () {
255 const service = yield* CheckpointService;
256 const directory = yield* ProviderSessionDirectory;
257
258 yield* directory.upsert({
259 sessionId: fixture.sessionId,
260 provider: "codex",
261 threadId: "thread-1",
262 });
263
264 yield* service.initializeForSession({
265 providerSessionId: fixture.sessionId,
266 cwd: fixture.cwd,
267 });
268
269 yield* Effect.promise(() => writeFile(path.join(fixture.cwd, "README.md"), "v2\n"));
270 fixture.adapter.setTurns([firstTurn]);
271
272 yield* service.captureCurrentTurn({
273 providerSessionId: fixture.sessionId,
274 });
275
276 const listed = yield* service.listCheckpoints({ sessionId: fixture.sessionId });
277 assert.equal(listed.checkpoints.length, 2);
278 assert.equal(listed.checkpoints[1]?.id, "turn-1");
279 assert.equal(listed.checkpoints[1]?.isCurrent, true);
280
281 const diff = yield* service.getCheckpointDiff({
282 sessionId: fixture.sessionId,
283 fromTurnCount: 0,
284 toTurnCount: 1,
285 });
286 assert.equal(diff.threadId, "thread-1");
287 assert.equal(diff.diff.includes("diff --git a/README.md b/README.md"), true);
288 assert.equal(diff.diff.includes("-v1"), true);
289 assert.equal(diff.diff.includes("+v2"), true);
290
291 yield* Effect.promise(() => writeFile(path.join(fixture.cwd, "README.md"), "v3\n"));
292 fixture.adapter.setTurns([firstTurn, secondTurn]);
293
294 yield* service.captureCurrentTurn({
295 providerSessionId: fixture.sessionId,
296 });
297
298 const beforeRevert = yield* Effect.promise(() =>
299 readFile(path.join(fixture.cwd, "README.md"), "utf8"),
300 );
301 assert.equal(beforeRevert, "v3\n");
302
303 const reverted = yield* service.revertToCheckpoint({
304 sessionId: fixture.sessionId,
305 turnCount: 1,
306 });
307
308 const afterRevert = yield* Effect.promise(() =>
309 readFile(path.join(fixture.cwd, "README.md"), "utf8"),
310 );
311 assert.equal(afterRevert, "v2\n");
312 assert.equal(reverted.turnCount, 1);
313 assert.equal(reverted.rolledBackTurns, 1);
314 assert.equal(reverted.checkpoints.length, 2);
315 assert.equal(reverted.checkpoints[1]?.id, "turn-1");
316 assert.equal(reverted.checkpoints[1]?.isCurrent, true);
317 assert.equal(fixture.adapter.getTurns().length, 1);
318 }).pipe(Effect.provide(fixture.layer));
319 }),
320);
321
322it.effect(
323 "CheckpointServiceLive fails with CheckpointSessionNotFoundError for unknown sessions",
324 () =>
325 Effect.gen(function* () {
326 const fixture = yield* makeFixture([]);
327
328 yield* Effect.gen(function* () {
329 const service = yield* CheckpointService;
330 const result = yield* service
331 .listCheckpoints({
332 sessionId: "missing",
333 })
334 .pipe(Effect.result);
335
336 assert.equal(result._tag, "Failure");
337 if (result._tag === "Failure") {
338 assert.equal(result.failure._tag, "CheckpointSessionNotFoundError");
339 }
340 }).pipe(Effect.provide(fixture.layer));
341 }),
342);
343
344it.effect(
345 "CheckpointServiceLive fails with CheckpointUnavailableError when a checkpoint ref is missing",
346 () =>
347 Effect.gen(function* () {
348 const firstTurn: ProviderThreadTurnSnapshot = {
349 id: "turn-1",
350 items: [{ type: "userMessage", content: [{ type: "text", text: "first" }] }],
351 };
352
353 const fixture = yield* makeFixture([]);
354
355 yield* Effect.gen(function* () {
356 const service = yield* CheckpointService;
357 const store = yield* CheckpointStore;
358 const checkpointRepository = yield* CheckpointRepository;
359 const directory = yield* ProviderSessionDirectory;
360
361 yield* directory.upsert({
362 sessionId: fixture.sessionId,
363 provider: "codex",
364 threadId: "thread-1",
365 });
366
367 yield* service.initializeForSession({
368 providerSessionId: fixture.sessionId,
369 cwd: fixture.cwd,
370 });
371
372 yield* Effect.promise(() => writeFile(path.join(fixture.cwd, "README.md"), "v2\n"));
373 fixture.adapter.setTurns([firstTurn]);
374 yield* service.captureCurrentTurn({
375 providerSessionId: fixture.sessionId,
376 });
377
378 const checkpoint = yield* checkpointRepository.getCheckpoint({
379 providerSessionId: fixture.sessionId,
380 turnCount: 1,
381 });
382 if (checkpoint._tag !== "Some") {
383 assert.fail("Expected turn-1 checkpoint row to exist.");
384 }
385
386 yield* store.deleteCheckpointRefs({
387 cwd: fixture.cwd,
388 checkpointRefs: [checkpoint.value.checkpointRef],
389 });
390
391 const result = yield* service
392 .revertToCheckpoint({
393 sessionId: fixture.sessionId,
394 turnCount: 1,
395 })
396 .pipe(Effect.result);
397
398 assert.equal(result._tag, "Failure");
399 if (result._tag === "Failure") {
400 assert.equal(result.failure._tag, "CheckpointUnavailableError");
401 }
402 }).pipe(Effect.provide(fixture.layer));
403 }),
404);
apps/server/src/checkpointing/Layers/CheckpointService.ts created+1008
......@@ -0,0 +1,1008 @@
1/**
2 * CheckpointServiceLive - Checkpoint workflow orchestration layer.
3 *
4 * Coordinates provider thread snapshots, filesystem checkpoint refs
5 * (`CheckpointStore`), and checkpoint metadata persistence
6 * (`CheckpointRepository`) for initialize/capture/list/diff/revert workflows.
7 *
8 * This layer owns checkpoint workflow rules and invariants. It does not execute
9 * provider protocol details directly (adapter concern), and it does not perform
10 * raw SQL or Git operations directly (catalog/store concerns).
11 *
12 * @module CheckpointServiceLive
13 */
14import type {
15 ProviderCheckpoint,
16 ProviderGetCheckpointDiffInput,
17 ProviderListCheckpointsInput,
18 ProviderRevertToCheckpointInput,
19} from "@t3tools/contracts";
20import {
21 providerGetCheckpointDiffInputSchema,
22 providerListCheckpointsInputSchema,
23 providerRevertToCheckpointInputSchema,
24} from "@t3tools/contracts";
25import { Effect, Layer, Option, Ref, Semaphore } from "effect";
26
27import {
28 CheckpointInvariantError,
29 CheckpointRepositoryError,
30 CheckpointServiceValidationError,
31 CheckpointSessionNotFoundError,
32 CheckpointUnavailableError,
33 type CheckpointServiceError,
34} from "../Errors.ts";
35import {
36 CheckpointService,
37 type CaptureCurrentTurnInput,
38 type CheckpointServiceShape,
39 type InitializeCheckpointSessionInput,
40} from "../Services/CheckpointService.ts";
41import { CheckpointStore } from "../Services/CheckpointStore.ts";
42import { ProviderAdapterRegistry } from "../../provider/Services/ProviderAdapterRegistry.ts";
43import { ProviderSessionDirectory } from "../../provider/Services/ProviderSessionDirectory.ts";
44import { CheckpointRepository } from "../../persistence/Services/Checkpoints.ts";
45import type {
46 ProviderAdapterShape,
47 ProviderThreadSnapshot,
48 ProviderThreadTurnSnapshot,
49} from "../../provider/Services/ProviderAdapter.ts";
50import type { ProviderAdapterError } from "../../provider/Errors.ts";
51
52interface CheckpointCwdResolution {
53 readonly candidateCwd: string;
54 readonly checkpointCwd: Option.Option<string>;
55}
56
57const CHECKPOINT_REFS_PREFIX = "refs/t3/checkpoints";
58
59function checkpointRefThreadSegment(threadId: string): string {
60 return Buffer.from(threadId, "utf8").toString("base64url");
61}
62
63function checkpointRefForThreadTurn(threadId: string, turnCount: number): string {
64 return `${CHECKPOINT_REFS_PREFIX}/${checkpointRefThreadSegment(threadId)}/turn/${turnCount}`;
65}
66
67function asObject(value: unknown): Record<string, unknown> | undefined {
68 if (!value || typeof value !== "object") {
69 return undefined;
70 }
71 return value as Record<string, unknown>;
72}
73
74function asString(value: unknown): string | undefined {
75 return typeof value === "string" ? value : undefined;
76}
77
78function asArray(value: unknown): unknown[] {
79 return Array.isArray(value) ? value : [];
80}
81
82function trimToPreview(value: string): string {
83 const normalized = value.replace(/\s+/g, " ").trim();
84 if (normalized.length <= 120) {
85 return normalized;
86 }
87 return `${normalized.slice(0, 117)}...`;
88}
89
90function summarizeUserMessageContent(content: unknown[]): string | undefined {
91 const segments: string[] = [];
92
93 for (const part of content) {
94 const record = asObject(part);
95 const type = asString(record?.type);
96 if (!type) {
97 continue;
98 }
99
100 if (type === "text") {
101 const text = asString(record?.text);
102 if (text && text.trim().length > 0) {
103 segments.push(text.trim());
104 }
105 continue;
106 }
107
108 if (type === "image") {
109 segments.push("[Image attachment]");
110 continue;
111 }
112
113 if (type === "localImage") {
114 segments.push("[Local image attachment]");
115 }
116 }
117
118 if (segments.length === 0) {
119 return undefined;
120 }
121 return trimToPreview(segments.join(" "));
122}
123
124function summarizeTurn(turn: ProviderThreadTurnSnapshot): {
125 readonly messageCountDelta: number;
126 readonly preview?: string;
127} {
128 let messageCountDelta = 0;
129 let preview: string | undefined;
130
131 for (const item of turn.items) {
132 const record = asObject(item);
133 const type = asString(record?.type);
134 if (!type) {
135 continue;
136 }
137
138 if (type === "userMessage") {
139 messageCountDelta += 1;
140 if (!preview) {
141 const content = asArray(record?.content);
142 preview = summarizeUserMessageContent(content);
143 }
144 continue;
145 }
146
147 if (type === "agentMessage") {
148 messageCountDelta += 1;
149 if (!preview) {
150 const text = asString(record?.text);
151 if (text && text.trim().length > 0) {
152 preview = trimToPreview(text);
153 }
154 }
155 }
156 }
157
158 return {
159 messageCountDelta,
160 ...(preview ? { preview } : {}),
161 };
162}
163
164function buildCheckpoints(turns: ReadonlyArray<ProviderThreadTurnSnapshot>): ProviderCheckpoint[] {
165 const checkpoints: ProviderCheckpoint[] = [];
166 let messageCount = 0;
167 const isEmpty = turns.length === 0;
168
169 checkpoints.push({
170 id: "root",
171 turnCount: 0,
172 messageCount: 0,
173 label: "Start of conversation",
174 isCurrent: isEmpty,
175 });
176
177 for (let index = 0; index < turns.length; index += 1) {
178 const turn = turns[index];
179 if (!turn) {
180 continue;
181 }
182
183 const turnSummary = summarizeTurn(turn);
184 messageCount += turnSummary.messageCountDelta;
185 checkpoints.push({
186 id: turn.id,
187 turnCount: index + 1,
188 messageCount,
189 label: `Turn ${index + 1}`,
190 ...(turnSummary.preview ? { preview: turnSummary.preview } : {}),
191 isCurrent: index === turns.length - 1,
192 });
193 }
194
195 return checkpoints;
196}
197
198function rootCheckpoint(): ProviderCheckpoint {
199 return {
200 id: "root",
201 turnCount: 0,
202 messageCount: 0,
203 label: "Start of conversation",
204 isCurrent: true,
205 };
206}
207
208function validationError(
209 operation: string,
210 issue: string,
211 cause?: unknown,
212): CheckpointServiceValidationError {
213 return new CheckpointServiceValidationError({
214 operation,
215 issue,
216 ...(cause !== undefined ? { cause } : {}),
217 });
218}
219
220function invariantError(
221 operation: string,
222 detail: string,
223 cause?: unknown,
224): CheckpointInvariantError {
225 return new CheckpointInvariantError({
226 operation,
227 detail,
228 ...(cause !== undefined ? { cause } : {}),
229 });
230}
231
232function normalizeNonEmpty(
233 value: string,
234 operation: string,
235 field: string,
236): Effect.Effect<string, CheckpointServiceValidationError> {
237 const normalized = value.trim();
238 if (normalized.length === 0) {
239 return Effect.fail(validationError(operation, `${field} must be a non-empty string.`));
240 }
241 return Effect.succeed(normalized);
242}
243
244function decodeInitializeInput(
245 input: InitializeCheckpointSessionInput,
246): Effect.Effect<InitializeCheckpointSessionInput, CheckpointServiceValidationError> {
247 return Effect.gen(function* () {
248 const operation = "CheckpointService.initializeForSession";
249 const providerSessionId = yield* normalizeNonEmpty(
250 input.providerSessionId,
251 operation,
252 "providerSessionId",
253 );
254 const cwd = yield* normalizeNonEmpty(input.cwd, operation, "cwd");
255 return {
256 providerSessionId,
257 cwd,
258 } satisfies InitializeCheckpointSessionInput;
259 });
260}
261
262function decodeCaptureInput(
263 input: CaptureCurrentTurnInput,
264): Effect.Effect<CaptureCurrentTurnInput, CheckpointServiceValidationError> {
265 return Effect.gen(function* () {
266 const operation = "CheckpointService.captureCurrentTurn";
267 const providerSessionId = yield* normalizeNonEmpty(
268 input.providerSessionId,
269 operation,
270 "providerSessionId",
271 );
272 const turnId = input.turnId?.trim();
273 const status = input.status?.trim();
274
275 return {
276 providerSessionId,
277 ...(turnId ? { turnId } : {}),
278 ...(status ? { status } : {}),
279 } satisfies CaptureCurrentTurnInput;
280 });
281}
282
283function decodeListInput(
284 input: ProviderListCheckpointsInput,
285): Effect.Effect<ProviderListCheckpointsInput, CheckpointServiceValidationError> {
286 const parsed = providerListCheckpointsInputSchema.safeParse(input);
287 if (!parsed.success) {
288 return Effect.fail(
289 validationError(
290 "CheckpointService.listCheckpoints",
291 parsed.error.message,
292 parsed.error.cause,
293 ),
294 );
295 }
296 return Effect.succeed(parsed.data);
297}
298
299function decodeDiffInput(
300 input: ProviderGetCheckpointDiffInput,
301): Effect.Effect<ProviderGetCheckpointDiffInput, CheckpointServiceValidationError> {
302 const parsed = providerGetCheckpointDiffInputSchema.safeParse(input);
303 if (!parsed.success) {
304 return Effect.fail(
305 validationError(
306 "CheckpointService.getCheckpointDiff",
307 parsed.error.message,
308 parsed.error.cause,
309 ),
310 );
311 }
312 return Effect.succeed(parsed.data);
313}
314
315function decodeRevertInput(
316 input: ProviderRevertToCheckpointInput,
317): Effect.Effect<ProviderRevertToCheckpointInput, CheckpointServiceValidationError> {
318 const parsed = providerRevertToCheckpointInputSchema.safeParse(input);
319 if (!parsed.success) {
320 return Effect.fail(
321 validationError(
322 "CheckpointService.revertToCheckpoint",
323 parsed.error.message,
324 parsed.error.cause,
325 ),
326 );
327 }
328 return Effect.succeed(parsed.data);
329}
330
331const makeCheckpointService = Effect.gen(function* () {
332 const store = yield* CheckpointStore;
333 const checkpointRepository = yield* CheckpointRepository;
334 const registry = yield* ProviderAdapterRegistry;
335 const directory = yield* ProviderSessionDirectory;
336
337 const sessionCheckpointCwdsRef = yield* Ref.make<Map<string, string>>(new Map<string, string>());
338 const sessionSemaphoresRef = yield* Ref.make<Map<string, Semaphore.Semaphore>>(
339 new Map<string, Semaphore.Semaphore>(),
340 );
341
342 const getSessionSemaphore = (sessionId: string) =>
343 Ref.modify(sessionSemaphoresRef, (current) => {
344 const existing = current.get(sessionId);
345 if (existing) {
346 return [existing, current] as const;
347 }
348
349 const next = new Map(current);
350 const semaphore = Semaphore.makeUnsafe(1);
351 next.set(sessionId, semaphore);
352 return [semaphore, next] as const;
353 });
354
355 const withSessionLock = <A>(
356 sessionId: string,
357 effect: Effect.Effect<A, CheckpointServiceError>,
358 ) =>
359 getSessionSemaphore(sessionId).pipe(
360 Effect.flatMap((semaphore) => semaphore.withPermits(1)(effect)),
361 );
362
363 const setTrackedCheckpointCwd = (sessionId: string, cwd: string) =>
364 Ref.update(sessionCheckpointCwdsRef, (current) => {
365 const next = new Map(current);
366 next.set(sessionId, cwd);
367 return next;
368 });
369
370 const clearTrackedCheckpointCwd = (sessionId: string) =>
371 Ref.update(sessionCheckpointCwdsRef, (current) => {
372 if (!current.has(sessionId)) {
373 return current;
374 }
375 const next = new Map(current);
376 next.delete(sessionId);
377 return next;
378 });
379
380 const getTrackedCheckpointCwd = (sessionId: string) =>
381 Ref.get(sessionCheckpointCwdsRef).pipe(
382 Effect.map((current) => Option.fromNullishOr(current.get(sessionId))),
383 );
384
385 const resolveAdapterForSession = (
386 sessionId: string,
387 ): Effect.Effect<ProviderAdapterShape<ProviderAdapterError>, CheckpointServiceError> =>
388 directory.getProvider(sessionId).pipe(
389 Effect.mapError(
390 (cause) =>
391 new CheckpointSessionNotFoundError({
392 sessionId,
393 cause,
394 }),
395 ),
396 Effect.flatMap((provider) =>
397 registry
398 .getByProvider(provider)
399 .pipe(
400 Effect.mapError((cause) =>
401 invariantError(
402 "CheckpointService.resolveAdapterForSession",
403 `Provider '${provider}' is not registered for session ${sessionId}.`,
404 cause,
405 ),
406 ),
407 ),
408 ),
409 );
410
411 const ensureSessionIsActive = (
412 sessionId: string,
413 adapter: ProviderAdapterShape<ProviderAdapterError>,
414 ): Effect.Effect<void, CheckpointSessionNotFoundError> =>
415 adapter.hasSession(sessionId).pipe(
416 Effect.flatMap((hasSession) =>
417 hasSession
418 ? Effect.void
419 : Effect.fail(
420 new CheckpointSessionNotFoundError({
421 sessionId,
422 }),
423 ),
424 ),
425 );
426
427 const readThreadSnapshot = (
428 sessionId: string,
429 adapter: ProviderAdapterShape<ProviderAdapterError>,
430 operation: string,
431 ): Effect.Effect<ProviderThreadSnapshot, CheckpointInvariantError> =>
432 adapter
433 .readThread(sessionId)
434 .pipe(
435 Effect.mapError((cause) =>
436 invariantError(
437 operation,
438 `Failed to read thread snapshot for session ${sessionId}.`,
439 cause,
440 ),
441 ),
442 );
443
444 const rollbackThreadSnapshot = (
445 sessionId: string,
446 adapter: ProviderAdapterShape<ProviderAdapterError>,
447 numTurns: number,
448 operation: string,
449 ): Effect.Effect<ProviderThreadSnapshot, CheckpointInvariantError> =>
450 adapter
451 .rollbackThread(sessionId, numTurns)
452 .pipe(
453 Effect.mapError((cause) =>
454 invariantError(operation, `Failed to rollback thread for session ${sessionId}.`, cause),
455 ),
456 );
457
458 const persistCheckpointMetadata = (input: {
459 readonly providerSessionId: string;
460 readonly threadId: string;
461 readonly checkpoint: ProviderCheckpoint;
462 }): Effect.Effect<void, CheckpointServiceError> =>
463 checkpointRepository.upsertCheckpoint({
464 providerSessionId: input.providerSessionId,
465 threadId: input.threadId,
466 checkpointId: input.checkpoint.id,
467 checkpointRef: checkpointRefForThreadTurn(input.threadId, input.checkpoint.turnCount),
468 turnCount: input.checkpoint.turnCount,
469 messageCount: input.checkpoint.messageCount,
470 label: input.checkpoint.label,
471 ...(input.checkpoint.preview !== undefined ? { preview: input.checkpoint.preview } : {}),
472 createdAt: new Date().toISOString(),
473 });
474
475 const captureAndPersistCheckpoint = (input: {
476 readonly providerSessionId: string;
477 readonly cwd: string;
478 readonly threadId: string;
479 readonly checkpoint: ProviderCheckpoint;
480 }): Effect.Effect<void, CheckpointServiceError> => {
481 const checkpointRef = checkpointRefForThreadTurn(input.threadId, input.checkpoint.turnCount);
482 return Effect.gen(function* () {
483 yield* store.captureCheckpoint({
484 cwd: input.cwd,
485 checkpointRef,
486 });
487
488 yield* persistCheckpointMetadata({
489 providerSessionId: input.providerSessionId,
490 threadId: input.threadId,
491 checkpoint: input.checkpoint,
492 }).pipe(
493 Effect.catch((error) =>
494 store
495 .deleteCheckpointRefs({
496 cwd: input.cwd,
497 checkpointRefs: [checkpointRef],
498 })
499 .pipe(
500 Effect.catch(() => Effect.void),
501 Effect.flatMap(() => Effect.fail(error)),
502 ),
503 ),
504 );
505 });
506 };
507
508 const latestCheckpointFromSnapshot = (
509 snapshot: ProviderThreadSnapshot,
510 operation: string,
511 ): Effect.Effect<ProviderCheckpoint, CheckpointInvariantError> => {
512 if (snapshot.turns.length === 0) {
513 return Effect.fail(
514 invariantError(operation, "Cannot derive latest checkpoint from an empty thread snapshot."),
515 );
516 }
517
518 const checkpoints = buildCheckpoints(snapshot.turns);
519 const latest = checkpoints[checkpoints.length - 1];
520 if (!latest || latest.turnCount !== snapshot.turns.length) {
521 return Effect.fail(
522 invariantError(
523 operation,
524 `Unable to derive checkpoint metadata for turn ${snapshot.turns.length}.`,
525 ),
526 );
527 }
528
529 return Effect.succeed(latest);
530 };
531
532 const ensureRootCheckpointMetadata = (
533 providerSessionId: string,
534 operation: string,
535 ): Effect.Effect<void, CheckpointServiceError> =>
536 checkpointRepository.getCheckpoint({ providerSessionId, turnCount: 0 }).pipe(
537 Effect.flatMap((rootEntry) =>
538 Option.isSome(rootEntry)
539 ? Effect.void
540 : Effect.fail(
541 invariantError(
542 operation,
543 `Missing root checkpoint metadata for session ${providerSessionId}.`,
544 ),
545 ),
546 ),
547 );
548
549 const getCheckpointRefOrFail = (
550 providerSessionId: string,
551 threadId: string,
552 turnCount: number,
553 operation: string,
554 ): Effect.Effect<string, CheckpointServiceError> =>
555 checkpointRepository.getCheckpoint({ providerSessionId, turnCount }).pipe(
556 Effect.mapError((error): CheckpointServiceError => error),
557 Effect.flatMap((checkpoint): Effect.Effect<string, CheckpointServiceError> => {
558 if (Option.isNone(checkpoint)) {
559 return Effect.fail(
560 new CheckpointUnavailableError({
561 threadId,
562 turnCount,
563 detail: `Filesystem checkpoint is unavailable for turn ${turnCount} in thread ${threadId}.`,
564 }),
565 );
566 }
567
568 const checkpointRef = checkpoint.value.checkpointRef.trim();
569 if (checkpointRef.length === 0) {
570 return Effect.fail(
571 invariantError(
572 operation,
573 `Checkpoint metadata for turn ${turnCount} is missing checkpointRef.`,
574 ),
575 );
576 }
577
578 return Effect.succeed(checkpointRef);
579 }),
580 );
581
582 const resolveCandidateCwd = (
583 sessionId: string,
584 adapter: ProviderAdapterShape<ProviderAdapterError>,
585 ): Effect.Effect<string> =>
586 adapter.listSessions().pipe(
587 Effect.map((sessions) => sessions.find((session) => session.sessionId === sessionId)?.cwd),
588 Effect.map((cwd) => {
589 const candidate = cwd?.trim() || process.cwd();
590 if (candidate.length > 0) {
591 return candidate;
592 }
593 return process.cwd();
594 }),
595 );
596
597 const resolveCheckpointCwd = (
598 sessionId: string,
599 adapter: ProviderAdapterShape<ProviderAdapterError>,
600 ): Effect.Effect<CheckpointCwdResolution, CheckpointServiceError> =>
601 Effect.gen(function* () {
602 const tracked = yield* getTrackedCheckpointCwd(sessionId);
603 if (Option.isSome(tracked)) {
604 return {
605 candidateCwd: tracked.value,
606 checkpointCwd: tracked,
607 } satisfies CheckpointCwdResolution;
608 }
609
610 const candidateCwd = yield* resolveCandidateCwd(sessionId, adapter);
611 const supportsGit = yield* store.isGitRepository(candidateCwd);
612 if (!supportsGit) {
613 yield* clearTrackedCheckpointCwd(sessionId);
614 return {
615 candidateCwd,
616 checkpointCwd: Option.none(),
617 } satisfies CheckpointCwdResolution;
618 }
619
620 yield* setTrackedCheckpointCwd(sessionId, candidateCwd);
621
622 return {
623 candidateCwd,
624 checkpointCwd: Option.some(candidateCwd),
625 } satisfies CheckpointCwdResolution;
626 });
627
628 const initializeForSession: CheckpointServiceShape["initializeForSession"] = (rawInput) =>
629 decodeInitializeInput(rawInput).pipe(
630 Effect.flatMap((input) =>
631 withSessionLock(
632 input.providerSessionId,
633 Effect.gen(function* () {
634 const adapter = yield* resolveAdapterForSession(input.providerSessionId);
635 yield* ensureSessionIsActive(input.providerSessionId, adapter);
636 const snapshot = yield* readThreadSnapshot(
637 input.providerSessionId,
638 adapter,
639 "CheckpointService.initializeForSession:readThread",
640 );
641 if (snapshot.turns.length !== 0) {
642 return yield* Effect.fail(
643 invariantError(
644 "CheckpointService.initializeForSession",
645 `Expected empty thread snapshot during initialization, received ${snapshot.turns.length} turn(s).`,
646 ),
647 );
648 }
649
650 const supportsGit = yield* store.isGitRepository(input.cwd);
651 if (!supportsGit) {
652 yield* clearTrackedCheckpointCwd(input.providerSessionId);
653 return;
654 }
655
656 const existingRoot = yield* checkpointRepository.getCheckpoint({
657 providerSessionId: input.providerSessionId,
658 turnCount: 0,
659 });
660 if (Option.isSome(existingRoot)) {
661 const existingRef = existingRoot.value.checkpointRef.trim();
662 if (existingRef.length === 0) {
663 return yield* Effect.fail(
664 invariantError(
665 "CheckpointService.initializeForSession",
666 "Existing root checkpoint metadata is missing checkpointRef.",
667 ),
668 );
669 }
670 const exists = yield* store.hasCheckpointRef({
671 cwd: input.cwd,
672 checkpointRef: existingRef,
673 });
674 if (!exists) {
675 return yield* Effect.fail(
676 invariantError(
677 "CheckpointService.initializeForSession",
678 "Root checkpoint metadata exists but filesystem ref is missing.",
679 ),
680 );
681 }
682 yield* setTrackedCheckpointCwd(input.providerSessionId, input.cwd);
683 return;
684 }
685
686 yield* captureAndPersistCheckpoint({
687 providerSessionId: input.providerSessionId,
688 cwd: input.cwd,
689 threadId: snapshot.threadId,
690 checkpoint: rootCheckpoint(),
691 });
692 yield* setTrackedCheckpointCwd(input.providerSessionId, input.cwd);
693 }),
694 ),
695 ),
696 );
697
698 const captureCurrentTurn: CheckpointServiceShape["captureCurrentTurn"] = (rawInput) =>
699 decodeCaptureInput(rawInput).pipe(
700 Effect.flatMap((input) =>
701 withSessionLock(
702 input.providerSessionId,
703 Effect.gen(function* () {
704 const adapter = yield* resolveAdapterForSession(input.providerSessionId);
705 yield* ensureSessionIsActive(input.providerSessionId, adapter);
706 const snapshot = yield* readThreadSnapshot(
707 input.providerSessionId,
708 adapter,
709 "CheckpointService.captureCurrentTurn:readThread",
710 );
711 if (snapshot.turns.length === 0) {
712 return;
713 }
714
715 const resolution = yield* resolveCheckpointCwd(input.providerSessionId, adapter);
716 if (Option.isNone(resolution.checkpointCwd)) {
717 return;
718 }
719
720 const checkpoint = yield* latestCheckpointFromSnapshot(
721 snapshot,
722 "CheckpointService.captureCurrentTurn",
723 );
724
725 yield* ensureRootCheckpointMetadata(
726 input.providerSessionId,
727 "CheckpointService.captureCurrentTurn",
728 );
729
730 const existing = yield* checkpointRepository.getCheckpoint({
731 providerSessionId: input.providerSessionId,
732 turnCount: checkpoint.turnCount,
733 });
734 if (Option.isSome(existing)) {
735 const checkpointRef = existing.value.checkpointRef.trim();
736 if (checkpointRef.length === 0) {
737 return yield* Effect.fail(
738 invariantError(
739 "CheckpointService.captureCurrentTurn",
740 `Checkpoint metadata for turn ${checkpoint.turnCount} is missing checkpointRef.`,
741 ),
742 );
743 }
744 const exists = yield* store.hasCheckpointRef({
745 cwd: resolution.checkpointCwd.value,
746 checkpointRef,
747 });
748 if (!exists) {
749 return yield* Effect.fail(
750 invariantError(
751 "CheckpointService.captureCurrentTurn",
752 `Checkpoint metadata exists for turn ${checkpoint.turnCount} but filesystem ref is missing.`,
753 ),
754 );
755 }
756 return;
757 }
758
759 yield* captureAndPersistCheckpoint({
760 providerSessionId: input.providerSessionId,
761 cwd: resolution.checkpointCwd.value,
762 threadId: snapshot.threadId,
763 checkpoint,
764 });
765 }),
766 ),
767 ),
768 );
769
770 const listCheckpoints: CheckpointServiceShape["listCheckpoints"] = (rawInput) =>
771 decodeListInput(rawInput).pipe(
772 Effect.flatMap((input) =>
773 withSessionLock(
774 input.sessionId,
775 Effect.gen(function* () {
776 const adapter = yield* resolveAdapterForSession(input.sessionId);
777 yield* ensureSessionIsActive(input.sessionId, adapter);
778 const snapshot = yield* readThreadSnapshot(
779 input.sessionId,
780 adapter,
781 "CheckpointService.listCheckpoints:readThread",
782 );
783
784 const resolution = yield* resolveCheckpointCwd(input.sessionId, adapter);
785 if (Option.isNone(resolution.checkpointCwd)) {
786 return yield* Effect.fail(
787 new CheckpointRepositoryError({
788 cwd: resolution.candidateCwd,
789 detail: "Filesystem checkpoints are unavailable for this session.",
790 }),
791 );
792 }
793
794 const checkpoints = yield* checkpointRepository.listCheckpoints({
795 providerSessionId: input.sessionId,
796 });
797 if (checkpoints.length === 0) {
798 return yield* Effect.fail(
799 invariantError(
800 "CheckpointService.listCheckpoints",
801 `Missing checkpoint metadata for active session ${input.sessionId}.`,
802 ),
803 );
804 }
805
806 return {
807 threadId: snapshot.threadId,
808 checkpoints: Array.from(checkpoints),
809 };
810 }),
811 ),
812 ),
813 );
814
815 const getCheckpointDiff: CheckpointServiceShape["getCheckpointDiff"] = (rawInput) =>
816 decodeDiffInput(rawInput).pipe(
817 Effect.flatMap((input) =>
818 withSessionLock(
819 input.sessionId,
820 Effect.gen(function* () {
821 const adapter = yield* resolveAdapterForSession(input.sessionId);
822 yield* ensureSessionIsActive(input.sessionId, adapter);
823 const snapshot = yield* readThreadSnapshot(
824 input.sessionId,
825 adapter,
826 "CheckpointService.getCheckpointDiff:readThread",
827 );
828
829 if (input.toTurnCount > snapshot.turns.length) {
830 return yield* Effect.fail(
831 validationError(
832 "CheckpointService.getCheckpointDiff",
833 `Checkpoint turn count ${input.toTurnCount} exceeds current turn count ${snapshot.turns.length}.`,
834 ),
835 );
836 }
837
838 const resolution = yield* resolveCheckpointCwd(input.sessionId, adapter);
839 if (Option.isNone(resolution.checkpointCwd)) {
840 return yield* Effect.fail(
841 new CheckpointRepositoryError({
842 cwd: resolution.candidateCwd,
843 detail: "Filesystem checkpoints are unavailable for this session.",
844 }),
845 );
846 }
847
848 const fromCheckpointRef = yield* getCheckpointRefOrFail(
849 input.sessionId,
850 snapshot.threadId,
851 input.fromTurnCount,
852 "CheckpointService.getCheckpointDiff:fromCheckpoint",
853 );
854 const toCheckpointRef = yield* getCheckpointRefOrFail(
855 input.sessionId,
856 snapshot.threadId,
857 input.toTurnCount,
858 "CheckpointService.getCheckpointDiff:toCheckpoint",
859 );
860
861 const diff = yield* store.diffCheckpoints({
862 cwd: resolution.checkpointCwd.value,
863 fromCheckpointRef,
864 toCheckpointRef,
865 fallbackFromToHead: input.fromTurnCount === 0,
866 });
867
868 return {
869 threadId: snapshot.threadId,
870 fromTurnCount: input.fromTurnCount,
871 toTurnCount: input.toTurnCount,
872 diff,
873 };
874 }),
875 ),
876 ),
877 );
878
879 const revertToCheckpoint: CheckpointServiceShape["revertToCheckpoint"] = (rawInput) =>
880 decodeRevertInput(rawInput).pipe(
881 Effect.flatMap((input) =>
882 withSessionLock(
883 input.sessionId,
884 Effect.gen(function* () {
885 const adapter = yield* resolveAdapterForSession(input.sessionId);
886 yield* ensureSessionIsActive(input.sessionId, adapter);
887 const beforeSnapshot = yield* readThreadSnapshot(
888 input.sessionId,
889 adapter,
890 "CheckpointService.revertToCheckpoint:readThread",
891 );
892 const currentTurnCount = beforeSnapshot.turns.length;
893
894 if (input.turnCount > currentTurnCount) {
895 return yield* Effect.fail(
896 validationError(
897 "CheckpointService.revertToCheckpoint",
898 `Checkpoint turn count ${input.turnCount} exceeds current turn count ${currentTurnCount}.`,
899 ),
900 );
901 }
902
903 const resolution = yield* resolveCheckpointCwd(input.sessionId, adapter);
904 if (Option.isNone(resolution.checkpointCwd)) {
905 return yield* Effect.fail(
906 new CheckpointRepositoryError({
907 cwd: resolution.candidateCwd,
908 detail: "Filesystem checkpoints are unavailable for this session.",
909 }),
910 );
911 }
912
913 const checkpointCwd = resolution.checkpointCwd.value;
914 const checkpointRef = yield* getCheckpointRefOrFail(
915 input.sessionId,
916 beforeSnapshot.threadId,
917 input.turnCount,
918 "CheckpointService.revertToCheckpoint:checkpointLookup",
919 );
920
921 const restored = yield* store.restoreCheckpoint({
922 cwd: checkpointCwd,
923 checkpointRef,
924 fallbackToHead: input.turnCount === 0,
925 });
926 if (!restored) {
927 return yield* Effect.fail(
928 new CheckpointUnavailableError({
929 threadId: beforeSnapshot.threadId,
930 turnCount: input.turnCount,
931 detail: `Filesystem checkpoint is unavailable for turn ${input.turnCount} in thread ${beforeSnapshot.threadId}.`,
932 }),
933 );
934 }
935
936 const requestedRollbackTurns = currentTurnCount - input.turnCount;
937 const afterSnapshot =
938 requestedRollbackTurns > 0
939 ? yield* rollbackThreadSnapshot(
940 input.sessionId,
941 adapter,
942 requestedRollbackTurns,
943 "CheckpointService.revertToCheckpoint:rollbackThread",
944 )
945 : beforeSnapshot;
946
947 const staleTurnCounts: number[] = [];
948 for (let turn = input.turnCount + 1; turn <= currentTurnCount; turn += 1) {
949 staleTurnCounts.push(turn);
950 }
951
952 const staleCheckpointRefs = yield* Effect.forEach(
953 staleTurnCounts,
954 (turnCount) =>
955 checkpointRepository.getCheckpoint({
956 providerSessionId: input.sessionId,
957 turnCount,
958 }),
959 { discard: false },
960 ).pipe(
961 Effect.map((entries) =>
962 entries
963 .flatMap((entry) =>
964 Option.isSome(entry) ? [entry.value.checkpointRef.trim()] : [],
965 )
966 .filter((ref) => ref.length > 0),
967 ),
968 );
969
970 yield* store.deleteCheckpointRefs({
971 cwd: checkpointCwd,
972 checkpointRefs: staleCheckpointRefs,
973 });
974 yield* checkpointRepository.deleteAfterTurn({
975 providerSessionId: input.sessionId,
976 maxTurnCount: input.turnCount,
977 });
978 const checkpoints = yield* checkpointRepository.listCheckpoints({
979 providerSessionId: input.sessionId,
980 });
981 const currentCheckpoint =
982 checkpoints.find((checkpoint) => checkpoint.isCurrent) ??
983 checkpoints[checkpoints.length - 1] ??
984 checkpoints[0];
985 const rolledBackTurns = Math.max(0, currentTurnCount - afterSnapshot.turns.length);
986
987 return {
988 threadId: afterSnapshot.threadId,
989 turnCount: currentCheckpoint?.turnCount ?? 0,
990 messageCount: currentCheckpoint?.messageCount ?? 0,
991 rolledBackTurns,
992 checkpoints: Array.from(checkpoints),
993 };
994 }),
995 ),
996 ),
997 );
998
999 return {
1000 initializeForSession,
1001 captureCurrentTurn,
1002 listCheckpoints,
1003 getCheckpointDiff,
1004 revertToCheckpoint,
1005 } satisfies CheckpointServiceShape;
1006});
1007
1008export const CheckpointServiceLive = Layer.effect(CheckpointService, makeCheckpointService);
apps/server/src/checkpointing/Layers/CheckpointStore.ts created+418
......@@ -0,0 +1,418 @@
1/**
2 * CheckpointStoreLive - Filesystem checkpoint store adapter layer.
3 *
4 * Implements hidden Git-ref checkpoint capture/restore directly with
5 * Effect-native child process execution (`effect/unstable/process`).
6 *
7 * This layer owns filesystem/Git interactions only; it does not persist
8 * checkpoint metadata and does not coordinate provider rollback semantics.
9 *
10 * @module CheckpointStoreLive
11 */
12import { mkdtemp, rm } from "node:fs/promises";
13import { randomUUID } from "node:crypto";
14import os from "node:os";
15import path from "node:path";
16
17import { Effect, Layer } from "effect";
18import { ChildProcessSpawner } from "effect/unstable/process";
19
20import {
21 CheckpointGitCommandError,
22 CheckpointRepositoryError,
23 CheckpointUnavailableError,
24 type CheckpointStoreError,
25} from "../Errors.ts";
26import { GitCommandError } from "../../git/Errors.ts";
27import { runGit, RunGitInput } from "../../git/Process.ts";
28import { CheckpointStore, type CheckpointStoreShape } from "../Services/CheckpointStore.ts";
29
30function causeDetail(cause: unknown, fallback: string): string {
31 if (cause instanceof GitCommandError) {
32 return cause.detail;
33 }
34 if (cause instanceof Error && cause.message.length > 0) {
35 return cause.message;
36 }
37 return fallback;
38}
39
40function mapFilesystemFailure(
41 operation: string,
42 cwd: string,
43 cause: unknown,
44): CheckpointStoreError {
45 const detail = causeDetail(cause, `${operation} failed.`);
46
47 const unavailable =
48 /^Filesystem checkpoint is unavailable for turn (\d+) in thread (.+)\.?$/i.exec(detail);
49 if (unavailable) {
50 const [, rawTurnCount, threadId] = unavailable;
51 const turnCount = Number.parseInt(rawTurnCount ?? "", 10);
52 if (threadId && Number.isInteger(turnCount) && turnCount >= 0) {
53 return new CheckpointUnavailableError({
54 threadId,
55 turnCount,
56 detail,
57 cause,
58 });
59 }
60 }
61
62 const lowered = detail.toLowerCase();
63 if (lowered.includes("not a git repository")) {
64 return new CheckpointRepositoryError({
65 cwd,
66 detail,
67 cause,
68 });
69 }
70
71 return new CheckpointGitCommandError({
72 operation,
73 command: cause instanceof GitCommandError ? cause.command : "git",
74 cwd,
75 detail,
76 cause,
77 });
78}
79
80function resolveHeadCommit(
81 cwd: string,
82): Effect.Effect<string | null, GitCommandError, ChildProcessSpawner.ChildProcessSpawner> {
83 return runGit({
84 operation: "CheckpointStore.resolveHeadCommit",
85 cwd,
86 args: ["rev-parse", "--verify", "--quiet", "HEAD^{commit}"],
87 allowNonZeroExit: true,
88 }).pipe(
89 Effect.map((result) => {
90 if (result.code !== 0) {
91 return null;
92 }
93 const commit = result.stdout.trim();
94 return commit.length > 0 ? commit : null;
95 }),
96 );
97}
98
99function hasHeadCommit(
100 cwd: string,
101): Effect.Effect<boolean, GitCommandError, ChildProcessSpawner.ChildProcessSpawner> {
102 return runGit({
103 operation: "CheckpointStore.hasHeadCommit",
104 cwd,
105 args: ["rev-parse", "--verify", "HEAD"],
106 allowNonZeroExit: true,
107 }).pipe(Effect.map((result) => result.code === 0));
108}
109
110function resolveCheckpointCommit(
111 cwd: string,
112 checkpointRef: string,
113): Effect.Effect<string | null, GitCommandError, ChildProcessSpawner.ChildProcessSpawner> {
114 return runGit({
115 operation: "CheckpointStore.resolveCheckpointCommit",
116 cwd,
117 args: ["rev-parse", "--verify", "--quiet", `${checkpointRef}^{commit}`],
118 allowNonZeroExit: true,
119 }).pipe(
120 Effect.map((result) => {
121 if (result.code !== 0) {
122 return null;
123 }
124 const commit = result.stdout.trim();
125 return commit.length > 0 ? commit : null;
126 }),
127 );
128}
129
130function removeDirectory(pathname: string): Effect.Effect<void, never> {
131 return Effect.tryPromise({
132 try: () => rm(pathname, { recursive: true, force: true }),
133 catch: (cause) => cause,
134 }).pipe(Effect.orDie);
135}
136
137const makeCheckpointStore = Effect.gen(function* () {
138 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
139 const runGitWithSpawner = (input: RunGitInput) =>
140 runGit(input).pipe(Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner));
141 const resolveHeadCommitWithSpawner = (cwd: string) =>
142 resolveHeadCommit(cwd).pipe(
143 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
144 );
145 const hasHeadCommitWithSpawner = (cwd: string) =>
146 hasHeadCommit(cwd).pipe(
147 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
148 );
149 const resolveCheckpointCommitWithSpawner = (cwd: string, checkpointRef: string) =>
150 resolveCheckpointCommit(cwd, checkpointRef).pipe(
151 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
152 );
153
154 const isGitRepository: CheckpointStoreShape["isGitRepository"] = (cwd) =>
155 Effect.gen(function* () {
156 const normalizedCwd = cwd.trim();
157
158 return yield* runGitWithSpawner({
159 operation: "CheckpointStore.isGitRepository",
160 cwd: normalizedCwd,
161 args: ["rev-parse", "--is-inside-work-tree"],
162 allowNonZeroExit: true,
163 }).pipe(
164 Effect.map((result) => result.code === 0 && result.stdout.trim() === "true"),
165 Effect.catch(() => Effect.succeed(false)),
166 );
167 });
168
169 const captureCheckpoint: CheckpointStoreShape["captureCheckpoint"] = (input) =>
170 Effect.gen(function* () {
171 const operation = "CheckpointStore.captureCheckpoint";
172 const normalizedCwd = input.cwd.trim();
173 const checkpointRef = input.checkpointRef.trim();
174
175 yield* Effect.acquireUseRelease(
176 Effect.tryPromise({
177 try: () => mkdtemp(path.join(os.tmpdir(), "t3-fs-checkpoint-")),
178 catch: (cause) => cause,
179 }),
180 (tempDir) =>
181 Effect.gen(function* () {
182 const tempIndexPath = path.join(tempDir, `index-${randomUUID()}`);
183 const commitEnv: NodeJS.ProcessEnv = {
184 ...process.env,
185 GIT_INDEX_FILE: tempIndexPath,
186 GIT_AUTHOR_NAME: "T3 Code",
187 GIT_AUTHOR_EMAIL: "codex@users.noreply.github.com",
188 GIT_COMMITTER_NAME: "T3 Code",
189 GIT_COMMITTER_EMAIL: "codex@users.noreply.github.com",
190 };
191
192 const headExists = yield* hasHeadCommitWithSpawner(normalizedCwd);
193 if (headExists) {
194 yield* runGitWithSpawner({
195 operation,
196 cwd: normalizedCwd,
197 args: ["read-tree", "HEAD"],
198 env: commitEnv,
199 });
200 }
201
202 yield* runGitWithSpawner({
203 operation,
204 cwd: normalizedCwd,
205 args: ["add", "-A", "--", "."],
206 env: commitEnv,
207 });
208
209 const writeTreeResult = yield* runGitWithSpawner({
210 operation,
211 cwd: normalizedCwd,
212 args: ["write-tree"],
213 env: commitEnv,
214 });
215 const treeOid = writeTreeResult.stdout.trim();
216 if (treeOid.length === 0) {
217 return yield* Effect.fail(
218 new CheckpointGitCommandError({
219 operation,
220 command: "git write-tree",
221 cwd: normalizedCwd,
222 detail: "git write-tree returned an empty tree oid.",
223 }),
224 );
225 }
226
227 const message = `t3 checkpoint ref=${checkpointRef}`;
228 const commitTreeResult = yield* runGitWithSpawner({
229 operation,
230 cwd: normalizedCwd,
231 args: ["commit-tree", treeOid, "-m", message],
232 env: commitEnv,
233 });
234 const commitOid = commitTreeResult.stdout.trim();
235 if (commitOid.length === 0) {
236 return yield* Effect.fail(
237 new CheckpointGitCommandError({
238 operation,
239 command: "git commit-tree",
240 cwd: normalizedCwd,
241 detail: "git commit-tree returned an empty commit oid.",
242 }),
243 );
244 }
245
246 yield* runGitWithSpawner({
247 operation,
248 cwd: normalizedCwd,
249 args: ["update-ref", checkpointRef, commitOid],
250 });
251 }),
252 removeDirectory,
253 ).pipe(Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)));
254 });
255
256 const hasCheckpointRef: CheckpointStoreShape["hasCheckpointRef"] = (input) =>
257 Effect.gen(function* () {
258 const operation = "CheckpointStore.hasCheckpointRef";
259 const normalizedCwd = input.cwd.trim();
260 const checkpointRef = input.checkpointRef.trim();
261
262 return yield* resolveCheckpointCommitWithSpawner(normalizedCwd, checkpointRef).pipe(
263 Effect.map((commit) => commit !== null),
264 Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)),
265 );
266 });
267
268 const ensureRootCheckpoint: CheckpointStoreShape["ensureRootCheckpoint"] = (input) =>
269 Effect.gen(function* () {
270 const operation = "CheckpointStore.ensureRootCheckpoint";
271 const normalizedCwd = input.cwd.trim();
272 const checkpointRef = input.checkpointRef.trim();
273
274 const existing = yield* resolveCheckpointCommitWithSpawner(normalizedCwd, checkpointRef).pipe(
275 Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)),
276 );
277 if (existing) {
278 return true;
279 }
280
281 const headCommit = yield* resolveHeadCommitWithSpawner(normalizedCwd).pipe(
282 Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)),
283 );
284 if (!headCommit) {
285 return false;
286 }
287
288 yield* runGitWithSpawner({
289 operation,
290 cwd: normalizedCwd,
291 args: ["update-ref", checkpointRef, headCommit],
292 }).pipe(Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)));
293 return true;
294 });
295
296 const restoreCheckpoint: CheckpointStoreShape["restoreCheckpoint"] = (input) =>
297 Effect.gen(function* () {
298 const operation = "CheckpointStore.restoreCheckpoint";
299 const normalizedCwd = input.cwd.trim();
300 const checkpointRef = input.checkpointRef.trim();
301
302 let commitOid = yield* resolveCheckpointCommitWithSpawner(normalizedCwd, checkpointRef).pipe(
303 Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)),
304 );
305
306 if (!commitOid && input.fallbackToHead === true) {
307 commitOid = yield* resolveHeadCommitWithSpawner(normalizedCwd).pipe(
308 Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)),
309 );
310 }
311
312 if (!commitOid) {
313 return false;
314 }
315
316 yield* runGitWithSpawner({
317 operation,
318 cwd: normalizedCwd,
319 args: ["restore", "--source", commitOid, "--worktree", "--staged", "--", "."],
320 }).pipe(Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)));
321 yield* runGitWithSpawner({
322 operation,
323 cwd: normalizedCwd,
324 args: ["clean", "-fd", "--", "."],
325 }).pipe(Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)));
326
327 const headExists = yield* hasHeadCommitWithSpawner(normalizedCwd).pipe(
328 Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)),
329 );
330 if (headExists) {
331 yield* runGitWithSpawner({
332 operation,
333 cwd: normalizedCwd,
334 args: ["reset", "--quiet", "--", "."],
335 }).pipe(Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)));
336 }
337
338 return true;
339 });
340
341 const diffCheckpoints: CheckpointStoreShape["diffCheckpoints"] = (input) =>
342 Effect.gen(function* () {
343 const operation = "CheckpointStore.diffCheckpoints";
344 const normalizedCwd = input.cwd.trim();
345 const fromCheckpointRef = input.fromCheckpointRef.trim();
346 const toCheckpointRef = input.toCheckpointRef.trim();
347
348 let fromCommitOid = yield* resolveCheckpointCommitWithSpawner(
349 normalizedCwd,
350 fromCheckpointRef,
351 ).pipe(Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)));
352 const toCommitOid = yield* resolveCheckpointCommitWithSpawner(
353 normalizedCwd,
354 toCheckpointRef,
355 ).pipe(Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)));
356
357 if (!fromCommitOid && input.fallbackFromToHead === true) {
358 const headCommit = yield* resolveHeadCommitWithSpawner(normalizedCwd).pipe(
359 Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)),
360 );
361 if (headCommit) {
362 fromCommitOid = headCommit;
363 }
364 }
365
366 if (!fromCommitOid || !toCommitOid) {
367 return yield* Effect.fail(
368 new CheckpointGitCommandError({
369 operation,
370 command: "git diff",
371 cwd: normalizedCwd,
372 detail: "Checkpoint ref is unavailable for diff operation.",
373 }),
374 );
375 }
376
377 const result = yield* runGitWithSpawner({
378 operation,
379 cwd: normalizedCwd,
380 args: ["diff", "--patch", "--minimal", "--no-color", fromCommitOid, toCommitOid],
381 }).pipe(Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)));
382
383 return result.stdout;
384 });
385
386 const deleteCheckpointRefs: CheckpointStoreShape["deleteCheckpointRefs"] = (input) =>
387 Effect.gen(function* () {
388 const operation = "CheckpointStore.deleteCheckpointRefs";
389 const normalizedCwd = input.cwd.trim();
390 const checkpointRefs = input.checkpointRefs.map((ref) => ref.trim());
391
392 yield* Effect.forEach(
393 checkpointRefs,
394 (checkpointRef) =>
395 runGitWithSpawner({
396 operation,
397 cwd: normalizedCwd,
398 args: ["update-ref", "-d", checkpointRef],
399 allowNonZeroExit: true,
400 }).pipe(
401 Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)),
402 ),
403 { discard: true },
404 );
405 });
406
407 return {
408 isGitRepository,
409 captureCheckpoint,
410 hasCheckpointRef,
411 ensureRootCheckpoint,
412 restoreCheckpoint,
413 diffCheckpoints,
414 deleteCheckpointRefs,
415 } satisfies CheckpointStoreShape;
416});
417
418export const CheckpointStoreLive = Layer.effect(CheckpointStore, makeCheckpointStore);
apps/server/src/checkpointing/Services/CheckpointCatalog.ts deleted-83
......@@ -1,83 +0,0 @@
1/**
2 * CheckpointCatalog - Metadata repository for checkpoints.
3 *
4 * Stores user-facing checkpoint metadata used by list/revert APIs. It does not
5 * touch the filesystem or hidden Git refs; `CheckpointStore` owns that state.
6 * It also does not drive provider conversation rollback.
7 *
8 * @module CheckpointCatalog
9 */
10import type { ProviderCheckpoint } from "@t3tools/contracts";
11import { Option, ServiceMap } from "effect";
12import type { Effect } from "effect";
13
14import type { CheckpointCatalogError } from "../Errors.ts";
15
16export interface UpsertCheckpointInput {
17 readonly providerSessionId: string;
18 readonly threadId: string;
19 readonly turnCount: number;
20 readonly messageCount: number;
21 readonly label: string;
22 readonly preview?: string;
23 readonly createdAt: string;
24}
25
26export interface ListSessionCheckpointsInput {
27 readonly providerSessionId: string;
28}
29
30export interface GetCheckpointInput {
31 readonly providerSessionId: string;
32 readonly turnCount: number;
33}
34
35export interface DeleteAfterTurnInput {
36 readonly providerSessionId: string;
37 readonly maxTurnCount: number;
38}
39
40export interface DeleteAllForSessionInput {
41 readonly providerSessionId: string;
42}
43
44export interface CheckpointCatalogShape {
45 /**
46 * Insert or update one checkpoint metadata row.
47 */
48 readonly upsertCheckpoint: (input: UpsertCheckpointInput) => Effect.Effect<void, CheckpointCatalogError>;
49
50 /**
51 * List checkpoints for a provider session.
52 */
53 readonly listCheckpoints: (
54 input: ListSessionCheckpointsInput,
55 ) => Effect.Effect<ReadonlyArray<ProviderCheckpoint>, CheckpointCatalogError>;
56
57 /**
58 * Read one checkpoint by turn count.
59 */
60 readonly getCheckpoint: (
61 input: GetCheckpointInput,
62 ) => Effect.Effect<Option.Option<ProviderCheckpoint>, CheckpointCatalogError>;
63
64 /**
65 * Delete checkpoint metadata newer than the provided turn count.
66 */
67 readonly deleteAfterTurn: (input: DeleteAfterTurnInput) => Effect.Effect<void, CheckpointCatalogError>;
68
69 /**
70 * Delete all checkpoint metadata for a provider session.
71 */
72 readonly deleteAllForSession: (
73 input: DeleteAllForSessionInput,
74 ) => Effect.Effect<void, CheckpointCatalogError>;
75}
76
77/**
78 * CheckpointCatalog - Service tag for checkpoint metadata persistence.
79 */
80export class CheckpointCatalog extends ServiceMap.Service<
81 CheckpointCatalog,
82 CheckpointCatalogShape
83>()("checkpointing/CheckpointCatalog") {}
apps/server/src/checkpointing/Services/CheckpointService.ts+4-2
......@@ -1,7 +1,7 @@
11/**
22 * CheckpointService - Application orchestrator for checkpoint workflows.
33 *
4 * Coordinates `CheckpointStore` (Git refs), `CheckpointCatalog` (metadata), and
4 * Coordinates `CheckpointStore` (Git refs), `CheckpointRepository` (metadata), and
55 * provider timeline operations resolved through `ProviderSessionDirectory` and
66 * `ProviderAdapterRegistry` (read/rollback/hasSession on adapters).
77 *
......@@ -45,7 +45,9 @@ export interface CheckpointServiceShape {
4545 /**
4646 * Capture the current thread turn into checkpoint refs/metadata.
4747 */
48 readonly captureCurrentTurn: (input: CaptureCurrentTurnInput) => Effect.Effect<void, CheckpointServiceError>;
48 readonly captureCurrentTurn: (
49 input: CaptureCurrentTurnInput,
50 ) => Effect.Effect<void, CheckpointServiceError>;
4951
5052 /**
5153 * List checkpoints for a provider session.
apps/server/src/checkpointing/Services/CheckpointStore.ts+28-23
......@@ -17,32 +17,30 @@ import type { CheckpointStoreError } from "../Errors.ts";
1717
1818export interface CaptureCheckpointInput {
1919 readonly cwd: string;
20 readonly threadId: string;
21 readonly turnCount: number;
20 readonly checkpointRef: string;
2221}
2322
2423export interface EnsureRootCheckpointInput {
2524 readonly cwd: string;
26 readonly threadId: string;
25 readonly checkpointRef: string;
2726}
2827
2928export interface RestoreCheckpointInput {
3029 readonly cwd: string;
31 readonly threadId: string;
32 readonly turnCount: number;
30 readonly checkpointRef: string;
31 readonly fallbackToHead?: boolean;
3332}
3433
3534export interface DiffCheckpointsInput {
3635 readonly cwd: string;
37 readonly threadId: string;
38 readonly fromTurnCount: number;
39 readonly toTurnCount: number;
36 readonly fromCheckpointRef: string;
37 readonly toCheckpointRef: string;
38 readonly fallbackFromToHead?: boolean;
4039}
4140
42export interface PruneAfterTurnInput {
41export interface DeleteCheckpointRefsInput {
4342 readonly cwd: string;
44 readonly threadId: string;
45 readonly maxTurnCount: number;
43 readonly checkpointRefs: ReadonlyArray<string>;
4644}
4745
4846export interface CheckpointStoreShape {
......@@ -52,17 +50,21 @@ export interface CheckpointStoreShape {
5250 readonly isGitRepository: (cwd: string) => Effect.Effect<boolean, CheckpointStoreError>;
5351
5452 /**
55 * Capture a checkpoint for a thread at a turn count.
53 * Capture a checkpoint commit and store it at the provided checkpoint ref.
5654 */
57 readonly captureCheckpoint: (input: CaptureCheckpointInput) => Effect.Effect<void, CheckpointStoreError>;
55 readonly captureCheckpoint: (
56 input: CaptureCheckpointInput,
57 ) => Effect.Effect<void, CheckpointStoreError>;
5858
5959 /**
6060 * Check whether a checkpoint ref exists.
6161 */
62 readonly hasCheckpoint: (input: RestoreCheckpointInput) => Effect.Effect<boolean, CheckpointStoreError>;
62 readonly hasCheckpointRef: (
63 input: Omit<RestoreCheckpointInput, "fallbackToHead">,
64 ) => Effect.Effect<boolean, CheckpointStoreError>;
6365
6466 /**
65 * Ensure a turn-0 checkpoint exists for a thread.
67 * Ensure a root checkpoint ref exists.
6668 */
6769 readonly ensureRootCheckpoint: (
6870 input: EnsureRootCheckpointInput,
......@@ -76,20 +78,23 @@ export interface CheckpointStoreShape {
7678 ) => Effect.Effect<boolean, CheckpointStoreError>;
7779
7880 /**
79 * Compute patch diff between two checkpoint turn counts.
81 * Compute patch diff between two checkpoint refs.
8082 */
81 readonly diffCheckpoints: (input: DiffCheckpointsInput) => Effect.Effect<string, CheckpointStoreError>;
83 readonly diffCheckpoints: (
84 input: DiffCheckpointsInput,
85 ) => Effect.Effect<string, CheckpointStoreError>;
8286
8387 /**
84 * Delete checkpoints newer than the provided turn count.
88 * Delete the provided checkpoint refs.
8589 */
86 readonly pruneAfterTurn: (input: PruneAfterTurnInput) => Effect.Effect<void, CheckpointStoreError>;
90 readonly deleteCheckpointRefs: (
91 input: DeleteCheckpointRefsInput,
92 ) => Effect.Effect<void, CheckpointStoreError>;
8793}
8894
8995/**
9096 * CheckpointStore - Service tag for checkpoint persistence and restore operations.
9197 */
92export class CheckpointStore extends ServiceMap.Service<
93 CheckpointStore,
94 CheckpointStoreShape
95>()("checkpointing/CheckpointStore") {}
98export class CheckpointStore extends ServiceMap.Service<CheckpointStore, CheckpointStoreShape>()(
99 "checkpointing/CheckpointStore",
100) {}
apps/server/src/git/Errors.ts created+16
......@@ -0,0 +1,16 @@
1import { Schema } from "effect";
2
3/**
4 * GitCommandError - Git command execution failed.
5 */
6export class GitCommandError extends Schema.TaggedErrorClass<GitCommandError>()("GitCommandError", {
7 operation: Schema.String,
8 command: Schema.String,
9 cwd: Schema.String,
10 detail: Schema.String,
11 cause: Schema.optional(Schema.Defect),
12}) {
13 override get message(): string {
14 return `Git command failed in ${this.operation}: ${this.command} (${this.cwd}) - ${this.detail}`;
15 }
16}
apps/server/src/git/process.test.ts created+51
......@@ -0,0 +1,51 @@
1import { NodeServices } from "@effect/platform-node";
2import { it, assert } from "@effect/vitest";
3import { Effect } from "effect";
4
5import { GitCommandError } from "./Errors.ts";
6import { runGit } from "./Process.ts";
7
8it.effect("runGit executes successful git commands", () =>
9 Effect.gen(function* () {
10 const result = yield* runGit({
11 operation: "GitProcess.test.version",
12 cwd: process.cwd(),
13 args: ["--version"],
14 });
15
16 assert.equal(result.code, 0);
17 assert.ok(result.stdout.toLowerCase().includes("git version"));
18 }).pipe(Effect.provide(NodeServices.layer)),
19);
20
21it.effect("runGit can return non-zero exit codes when allowed", () =>
22 Effect.gen(function* () {
23 const result = yield* runGit({
24 operation: "GitProcess.test.allowNonZero",
25 cwd: process.cwd(),
26 args: ["rev-parse", "--verify", "__definitely_missing_ref__"],
27 allowNonZeroExit: true,
28 });
29
30 assert.notEqual(result.code, 0);
31 }).pipe(Effect.provide(NodeServices.layer)),
32);
33
34it.effect("runGit fails with GitCommandError when non-zero exits are not allowed", () =>
35 Effect.gen(function* () {
36 const result = yield* Effect.result(
37 runGit({
38 operation: "GitProcess.test.failOnNonZero",
39 cwd: process.cwd(),
40 args: ["rev-parse", "--verify", "__definitely_missing_ref__"],
41 }),
42 );
43
44 assert.equal(result._tag, "Failure");
45 if (result._tag === "Failure") {
46 assert.ok(result.failure instanceof GitCommandError);
47 assert.equal(result.failure.operation, "GitProcess.test.failOnNonZero");
48 assert.equal(result.failure.command, "git rev-parse --verify __definitely_missing_ref__");
49 }
50 }).pipe(Effect.provide(NodeServices.layer)),
51);
apps/server/src/git/process.ts created+174
......@@ -0,0 +1,174 @@
1/**
2 * Git process helpers - Effect-native git execution with typed errors.
3 *
4 * Centralizes child-process git invocation for server modules. This module
5 * only executes git commands and reports structured failures.
6 *
7 * @module GitProcess
8 */
9import { Effect, Option, Stream } from "effect";
10import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
11import { GitCommandError } from "./Errors.ts";
12
13const DEFAULT_TIMEOUT_MS = 30_000;
14const DEFAULT_MAX_OUTPUT_BYTES = 1_000_000;
15
16export interface RunGitInput {
17 readonly operation: string;
18 readonly cwd: string;
19 readonly args: ReadonlyArray<string>;
20 readonly env?: NodeJS.ProcessEnv;
21 readonly allowNonZeroExit?: boolean;
22 readonly timeoutMs?: number;
23 readonly maxOutputBytes?: number;
24}
25
26export interface GitRunResult {
27 readonly code: number;
28 readonly stdout: string;
29 readonly stderr: string;
30}
31
32function quoteGitCommand(args: ReadonlyArray<string>): string {
33 return `git ${args.join(" ")}`;
34}
35
36function causeMessage(cause: unknown, fallback: string): string {
37 if (cause instanceof Error && cause.message.length > 0) {
38 return cause.message;
39 }
40 return fallback;
41}
42
43function gitError(
44 input: Pick<RunGitInput, "operation" | "cwd" | "args">,
45 detail: string,
46 cause?: unknown,
47): GitCommandError {
48 return new GitCommandError({
49 operation: input.operation,
50 command: quoteGitCommand(input.args),
51 cwd: input.cwd,
52 detail,
53 ...(cause !== undefined ? { cause } : {}),
54 });
55}
56
57function collectOutput(
58 input: Pick<RunGitInput, "operation" | "cwd" | "args">,
59 stream: Stream.Stream<Uint8Array, unknown>,
60 maxOutputBytes: number,
61): Effect.Effect<string, GitCommandError> {
62 return Effect.gen(function* () {
63 const decoder = new TextDecoder();
64 let bytes = 0;
65 let text = "";
66
67 yield* Stream.runForEach(stream, (chunk) =>
68 Effect.gen(function* () {
69 bytes += chunk.byteLength;
70 if (bytes > maxOutputBytes) {
71 return yield* Effect.fail(
72 gitError(
73 input,
74 `${quoteGitCommand(input.args)} output exceeded ${maxOutputBytes} bytes and was truncated.`,
75 ),
76 );
77 }
78 text += decoder.decode(chunk, { stream: true });
79 }),
80 );
81
82 text += decoder.decode();
83 return text;
84 }).pipe(
85 Effect.mapError((cause) =>
86 gitError(
87 input,
88 causeMessage(cause, `${quoteGitCommand(input.args)} output stream failed.`),
89 cause,
90 ),
91 ),
92 );
93}
94
95export function runGit(
96 input: RunGitInput,
97): Effect.Effect<GitRunResult, GitCommandError, ChildProcessSpawner.ChildProcessSpawner> {
98 const normalizedCwd = input.cwd.trim();
99 const commandInput = {
100 ...input,
101 cwd: normalizedCwd,
102 args: [...input.args],
103 } as const;
104 const timeoutMs = input.timeoutMs ?? DEFAULT_TIMEOUT_MS;
105 const maxOutputBytes = input.maxOutputBytes ?? DEFAULT_MAX_OUTPUT_BYTES;
106
107 const commandEffect = Effect.gen(function* () {
108 const child = yield* ChildProcess.spawn(
109 ChildProcess.make("git", commandInput.args, {
110 cwd: commandInput.cwd,
111 ...(input.env ? { env: input.env } : {}),
112 }),
113 ).pipe(
114 Effect.mapError((cause) =>
115 gitError(
116 commandInput,
117 causeMessage(cause, `Failed to spawn ${quoteGitCommand(commandInput.args)}.`),
118 cause,
119 ),
120 ),
121 );
122
123 const [stdout, stderr, exitCode] = yield* Effect.all(
124 [
125 collectOutput(commandInput, child.stdout, maxOutputBytes),
126 collectOutput(commandInput, child.stderr, maxOutputBytes),
127 child.exitCode.pipe(
128 Effect.map((value) => Number(value)),
129 Effect.mapError((cause) =>
130 gitError(
131 commandInput,
132 causeMessage(
133 cause,
134 `${quoteGitCommand(commandInput.args)} failed to report exit code.`,
135 ),
136 cause,
137 ),
138 ),
139 ),
140 ],
141 { concurrency: "unbounded" },
142 );
143
144 if (!input.allowNonZeroExit && exitCode !== 0) {
145 const trimmedStderr = stderr.trim();
146 return yield* Effect.fail(
147 gitError(
148 commandInput,
149 trimmedStderr.length > 0
150 ? `${quoteGitCommand(commandInput.args)} failed: ${trimmedStderr}`
151 : `${quoteGitCommand(commandInput.args)} failed with code ${exitCode}.`,
152 ),
153 );
154 }
155
156 return {
157 code: exitCode,
158 stdout,
159 stderr,
160 } satisfies GitRunResult;
161 });
162
163 return commandEffect.pipe(
164 Effect.scoped,
165 Effect.timeoutOption(timeoutMs),
166 Effect.flatMap((result) =>
167 Option.match(result, {
168 onNone: () =>
169 Effect.fail(gitError(commandInput, `${quoteGitCommand(commandInput.args)} timed out.`)),
170 onSome: Effect.succeed,
171 }),
172 ),
173 );
174}
apps/server/src/persistence/Errors.ts+77-35
......@@ -1,5 +1,9 @@
11import { Schema, SchemaIssue } from "effect";
22
3// ===============================
4// Core Persistence Errors
5// ===============================
6
37export class PersistenceSqlError extends Schema.TaggedErrorClass<PersistenceSqlError>()(
48 "PersistenceSqlError",
59 {
......@@ -39,6 +43,46 @@ export class PersistenceSerializationError extends Schema.TaggedErrorClass<Persi
3943 }
4044}
4145
46export function toPersistenceSqlError(operation: string) {
47 return (cause: unknown): PersistenceSqlError =>
48 new PersistenceSqlError({
49 operation,
50 detail: `Failed to execute ${operation}`,
51 cause,
52 });
53}
54
55export function toPersistenceDecodeError(operation: string) {
56 return (error: Schema.SchemaError): PersistenceDecodeError =>
57 new PersistenceDecodeError({
58 operation,
59 issue: SchemaIssue.makeFormatterDefault()(error.issue),
60 cause: error,
61 });
62}
63
64export function toPersistenceDecodeCauseError(operation: string) {
65 return (cause: unknown): PersistenceDecodeError =>
66 new PersistenceDecodeError({
67 operation,
68 issue: `Failed to execute ${operation}`,
69 cause,
70 });
71}
72
73export function toPersistenceSerializationError(operation: string) {
74 return (cause: unknown): PersistenceSerializationError =>
75 new PersistenceSerializationError({
76 operation,
77 detail: `Failed to execute ${operation}`,
78 cause,
79 });
80}
81
82// ===============================
83// Project Repository Errors
84// ===============================
85
4286export class ProjectValidationError extends Schema.TaggedErrorClass<ProjectValidationError>()(
4387 "ProjectValidationError",
4488 {
......@@ -76,6 +120,36 @@ export class ProjectNotFoundError extends Schema.TaggedErrorClass<ProjectNotFoun
76120 }
77121}
78122
123// ===============================
124// Checkpoint Repository Errors
125// ===============================
126
127export class CheckpointRepositoryValidationError extends Schema.TaggedErrorClass<CheckpointRepositoryValidationError>()(
128 "CheckpointRepositoryValidationError",
129 {
130 operation: Schema.String,
131 issue: Schema.String,
132 cause: Schema.optional(Schema.Defect),
133 },
134) {
135 override get message(): string {
136 return `Checkpoint repository validation failed in ${this.operation}: ${this.issue}`;
137 }
138}
139
140export class CheckpointRepositoryPersistenceError extends Schema.TaggedErrorClass<CheckpointRepositoryPersistenceError>()(
141 "CheckpointRepositoryPersistenceError",
142 {
143 operation: Schema.String,
144 detail: Schema.String,
145 cause: Schema.optional(Schema.Defect),
146 },
147) {
148 override get message(): string {
149 return `Checkpoint repository persistence error in ${this.operation}: ${this.detail}`;
150 }
151}
152
79153export type ProjectRepositoryError =
80154 | PersistenceSqlError
81155 | PersistenceDecodeError
......@@ -89,38 +163,6 @@ export type OrchestrationEventRepositoryError =
89163 | PersistenceDecodeError
90164 | PersistenceSerializationError;
91165
92export function toPersistenceSqlError(operation: string) {
93 return (cause: unknown): PersistenceSqlError =>
94 new PersistenceSqlError({
95 operation,
96 detail: `Failed to execute ${operation}`,
97 cause,
98 });
99}
100
101export function toPersistenceDecodeError(operation: string) {
102 return (error: Schema.SchemaError): PersistenceDecodeError =>
103 new PersistenceDecodeError({
104 operation,
105 issue: SchemaIssue.makeFormatterDefault()(error.issue),
106 cause: error,
107 });
108}
109
110export function toPersistenceDecodeCauseError(operation: string) {
111 return (cause: unknown): PersistenceDecodeError =>
112 new PersistenceDecodeError({
113 operation,
114 issue: `Failed to execute ${operation}`,
115 cause,
116 });
117}
118
119export function toPersistenceSerializationError(operation: string) {
120 return (cause: unknown): PersistenceSerializationError =>
121 new PersistenceSerializationError({
122 operation,
123 detail: `Failed to execute ${operation}`,
124 cause,
125 });
126}
166export type CheckpointRepositoryError =
167 | CheckpointRepositoryValidationError
168 | CheckpointRepositoryPersistenceError;
apps/server/src/persistence/Layers/Checkpoints.test.ts created+95
......@@ -0,0 +1,95 @@
1import { mkdtemp, rm } from "node:fs/promises";
2import os from "node:os";
3import path from "node:path";
4
5import { it, assert } from "@effect/vitest";
6import { Effect, Layer, Option } from "effect";
7
8import { CheckpointRepository } from "../Services/Checkpoints.ts";
9import { CheckpointRepositoryLive } from "./Checkpoints.ts";
10import { makeSqlitePersistenceLive } from "./Sqlite.ts";
11
12const makeTemporaryDirectoryScoped = (prefix: string) =>
13 Effect.acquireRelease(
14 Effect.promise(() => mkdtemp(path.join(os.tmpdir(), prefix))),
15 (directory) => Effect.promise(() => rm(directory, { recursive: true, force: true })),
16 );
17
18function makeCheckpointRepositoryLayer(dbPath: string) {
19 return CheckpointRepositoryLive.pipe(Layer.provide(makeSqlitePersistenceLive(dbPath)));
20}
21
22it.effect("CheckpointRepositoryLive upserts, lists, reads, and prunes checkpoint metadata", () =>
23 Effect.gen(function* () {
24 const stateDir = yield* makeTemporaryDirectoryScoped("t3code-checkpoint-repository-");
25 const dbPath = path.join(stateDir, "orchestration.sqlite");
26
27 yield* Effect.gen(function* () {
28 const repository = yield* CheckpointRepository;
29
30 yield* repository.upsertCheckpoint({
31 providerSessionId: "sess-1",
32 threadId: "thread-1",
33 checkpointId: "root",
34 checkpointRef: "refs/t3/checkpoints/thread-1/turn/0",
35 turnCount: 0,
36 messageCount: 0,
37 label: "Start of conversation",
38 createdAt: new Date().toISOString(),
39 });
40
41 yield* repository.upsertCheckpoint({
42 providerSessionId: "sess-1",
43 threadId: "thread-1",
44 checkpointId: "turn-1",
45 checkpointRef: "refs/t3/checkpoints/thread-1/turn/1",
46 turnCount: 1,
47 messageCount: 2,
48 label: "Turn 1",
49 preview: "hello",
50 createdAt: new Date().toISOString(),
51 });
52
53 yield* repository.upsertCheckpoint({
54 providerSessionId: "sess-1",
55 threadId: "thread-1",
56 checkpointId: "turn-2",
57 checkpointRef: "refs/t3/checkpoints/thread-1/turn/2",
58 turnCount: 2,
59 messageCount: 4,
60 label: "Turn 2",
61 preview: "done",
62 createdAt: new Date().toISOString(),
63 });
64
65 const checkpoints = yield* repository.listCheckpoints({ providerSessionId: "sess-1" });
66 assert.equal(checkpoints.length, 3);
67 assert.equal(checkpoints[0]?.id, "root");
68 assert.equal(checkpoints[0]?.isCurrent, false);
69 assert.equal(checkpoints[2]?.id, "turn-2");
70 assert.equal(checkpoints[2]?.isCurrent, true);
71
72 const one = yield* repository.getCheckpoint({ providerSessionId: "sess-1", turnCount: 1 });
73 assert.equal(Option.isSome(one), true);
74 if (Option.isSome(one)) {
75 assert.equal(one.value.id, "turn-1");
76 assert.equal(one.value.isCurrent, false);
77 assert.equal(one.value.checkpointRef, "refs/t3/checkpoints/thread-1/turn/1");
78 }
79
80 yield* repository.deleteAfterTurn({
81 providerSessionId: "sess-1",
82 maxTurnCount: 1,
83 });
84
85 const pruned = yield* repository.listCheckpoints({ providerSessionId: "sess-1" });
86 assert.equal(pruned.length, 2);
87 assert.equal(pruned[1]?.turnCount, 1);
88 assert.equal(pruned[1]?.isCurrent, true);
89
90 yield* repository.deleteAllForSession({ providerSessionId: "sess-1" });
91 const empty = yield* repository.listCheckpoints({ providerSessionId: "sess-1" });
92 assert.equal(empty.length, 0);
93 }).pipe(Effect.provide(makeCheckpointRepositoryLayer(dbPath)));
94 }),
95);
apps/server/src/persistence/Layers/Checkpoints.ts created+363
......@@ -0,0 +1,363 @@
1/**
2 * CheckpointRepositoryLive - SQLite-backed checkpoint metadata repository layer.
3 *
4 * Persists and queries checkpoint metadata (`turnCount`, labels, previews,
5 * message counts) and the associated hidden Git ref for each turn.
6 *
7 * This layer does not read/write Git refs and does not perform provider-thread
8 * rollback. Those responsibilities remain in `CheckpointStore` and
9 * `CheckpointService`.
10 *
11 * @module CheckpointRepositoryLive
12 */
13import type { ProviderCheckpoint } from "@t3tools/contracts";
14import * as SqlClient from "effect/unstable/sql/SqlClient";
15import * as SqlSchema from "effect/unstable/sql/SqlSchema";
16import { Effect, Layer, Option, Schema } from "effect";
17
18import {
19 CheckpointRepositoryPersistenceError,
20 CheckpointRepositoryValidationError,
21 type CheckpointRepositoryError,
22} from "../Errors.ts";
23import { makeSqlitePersistenceLive } from "./Sqlite.ts";
24import {
25 CheckpointRepository,
26 type CheckpointRepositoryEntry,
27 type CheckpointRepositoryShape,
28} from "../Services/Checkpoints.ts";
29
30const NonEmptyString = Schema.NonEmptyString;
31
32const CheckpointRowSchema = Schema.Struct({
33 providerSessionId: Schema.String,
34 threadId: Schema.String,
35 checkpointId: Schema.String,
36 checkpointRef: Schema.String,
37 turnCount: Schema.Number,
38 messageCount: Schema.Number,
39 label: Schema.String,
40 preview: Schema.NullOr(Schema.String),
41 createdAt: Schema.String,
42});
43
44const SessionRequestSchema = Schema.Struct({
45 providerSessionId: NonEmptyString,
46});
47
48const UpsertRequestSchema = Schema.Struct({
49 providerSessionId: NonEmptyString,
50 threadId: NonEmptyString,
51 checkpointId: NonEmptyString,
52 checkpointRef: NonEmptyString,
53 turnCount: Schema.Int,
54 messageCount: Schema.Int,
55 label: NonEmptyString,
56 preview: Schema.NullOr(Schema.String),
57 createdAt: NonEmptyString,
58});
59
60const GetCheckpointRequestSchema = Schema.Struct({
61 providerSessionId: NonEmptyString,
62 turnCount: Schema.Int,
63});
64
65const DeleteAfterTurnRequestSchema = Schema.Struct({
66 providerSessionId: NonEmptyString,
67 maxTurnCount: Schema.Int,
68});
69
70function errorMessage(cause: unknown, fallback: string): string {
71 if (cause instanceof Error && cause.message.length > 0) {
72 return cause.message;
73 }
74 return fallback;
75}
76
77function toValidationError(operation: string, cause: unknown): CheckpointRepositoryValidationError {
78 return new CheckpointRepositoryValidationError({
79 operation,
80 issue: errorMessage(cause, "Invalid checkpoint repository input."),
81 cause,
82 });
83}
84
85function decodeInput<A>(
86 schema: Schema.Schema<A>,
87 input: unknown,
88 operation: string,
89): Effect.Effect<A, CheckpointRepositoryValidationError> {
90 return Schema.decodeUnknownEffect(schema)(input).pipe(
91 Effect.mapError((cause) => toValidationError(operation, cause)),
92 ) as Effect.Effect<A, CheckpointRepositoryValidationError>;
93}
94
95function ensureNonNegativeInt(
96 operation: string,
97 field: string,
98 value: number,
99): Effect.Effect<number, CheckpointRepositoryValidationError> {
100 if (value < 0) {
101 return Effect.fail(
102 new CheckpointRepositoryValidationError({
103 operation,
104 issue: `${field} must be an integer >= 0.`,
105 }),
106 );
107 }
108 return Effect.succeed(value);
109}
110
111function toPersistenceError(
112 operation: string,
113 cause: unknown,
114): CheckpointRepositoryPersistenceError {
115 return new CheckpointRepositoryPersistenceError({
116 operation,
117 detail: `Failed to execute ${operation}.`,
118 cause,
119 });
120}
121
122function toCheckpointEntry(
123 row: Schema.Schema.Type<typeof CheckpointRowSchema>,
124 isCurrent: boolean,
125): CheckpointRepositoryEntry {
126 return {
127 id: row.checkpointId,
128 threadId: row.threadId,
129 checkpointRef: row.checkpointRef,
130 turnCount: row.turnCount,
131 messageCount: row.messageCount,
132 label: row.label,
133 ...(row.preview !== null ? { preview: row.preview } : {}),
134 createdAt: row.createdAt,
135 isCurrent,
136 };
137}
138
139function toProviderCheckpoint(entry: CheckpointRepositoryEntry): ProviderCheckpoint {
140 return {
141 id: entry.id,
142 turnCount: entry.turnCount,
143 messageCount: entry.messageCount,
144 label: entry.label,
145 ...(entry.preview !== undefined ? { preview: entry.preview } : {}),
146 isCurrent: entry.isCurrent,
147 };
148}
149
150const makeCheckpointRepository = Effect.gen(function* () {
151 const sql = yield* SqlClient.SqlClient;
152
153 const upsertCheckpointRow = SqlSchema.findOne({
154 Request: UpsertRequestSchema,
155 Result: CheckpointRowSchema,
156 execute: (request) =>
157 sql`
158 INSERT INTO provider_checkpoints (
159 provider_session_id,
160 thread_id,
161 checkpoint_id,
162 checkpoint_ref,
163 turn_count,
164 message_count,
165 label,
166 preview,
167 created_at
168 )
169 VALUES (
170 ${request.providerSessionId},
171 ${request.threadId},
172 ${request.checkpointId},
173 ${request.checkpointRef},
174 ${request.turnCount},
175 ${request.messageCount},
176 ${request.label},
177 ${request.preview},
178 ${request.createdAt}
179 )
180 ON CONFLICT (provider_session_id, turn_count)
181 DO UPDATE SET
182 thread_id = excluded.thread_id,
183 checkpoint_id = excluded.checkpoint_id,
184 checkpoint_ref = excluded.checkpoint_ref,
185 message_count = excluded.message_count,
186 label = excluded.label,
187 preview = excluded.preview,
188 created_at = excluded.created_at
189 RETURNING
190 provider_session_id AS "providerSessionId",
191 thread_id AS "threadId",
192 checkpoint_id AS "checkpointId",
193 checkpoint_ref AS "checkpointRef",
194 turn_count AS "turnCount",
195 message_count AS "messageCount",
196 label,
197 preview,
198 created_at AS "createdAt"
199 `,
200 });
201
202 const listCheckpointRows = SqlSchema.findAll({
203 Request: SessionRequestSchema,
204 Result: CheckpointRowSchema,
205 execute: ({ providerSessionId }) =>
206 sql`
207 SELECT
208 provider_session_id AS "providerSessionId",
209 thread_id AS "threadId",
210 checkpoint_id AS "checkpointId",
211 checkpoint_ref AS "checkpointRef",
212 turn_count AS "turnCount",
213 message_count AS "messageCount",
214 label,
215 preview,
216 created_at AS "createdAt"
217 FROM provider_checkpoints
218 WHERE provider_session_id = ${providerSessionId}
219 ORDER BY turn_count ASC
220 `,
221 });
222
223 const listForSession = (
224 providerSessionId: string,
225 ): Effect.Effect<ReadonlyArray<CheckpointRepositoryEntry>, CheckpointRepositoryError> =>
226 listCheckpointRows({ providerSessionId }).pipe(
227 Effect.mapError((cause) =>
228 toPersistenceError("CheckpointRepository.listCheckpoints:query", cause),
229 ),
230 Effect.map((rows) => {
231 const currentTurnCount = rows[rows.length - 1]?.turnCount;
232 return rows.map((row) => toCheckpointEntry(row, row.turnCount === currentTurnCount));
233 }),
234 );
235
236 const upsertCheckpoint: CheckpointRepositoryShape["upsertCheckpoint"] = (input) =>
237 Effect.gen(function* () {
238 const operation = "CheckpointRepository.upsertCheckpoint";
239 const parsed = yield* decodeInput(
240 UpsertRequestSchema,
241 {
242 providerSessionId: input.providerSessionId.trim(),
243 threadId: input.threadId.trim(),
244 checkpointId: input.checkpointId.trim(),
245 checkpointRef: input.checkpointRef.trim(),
246 turnCount: input.turnCount,
247 messageCount: input.messageCount,
248 label: input.label.trim(),
249 preview: input.preview?.trim() || null,
250 createdAt: input.createdAt.trim(),
251 },
252 operation,
253 );
254 yield* ensureNonNegativeInt(operation, "turnCount", parsed.turnCount);
255 yield* ensureNonNegativeInt(operation, "messageCount", parsed.messageCount);
256
257 yield* upsertCheckpointRow(parsed).pipe(
258 Effect.mapError((cause) =>
259 toPersistenceError("CheckpointRepository.upsertCheckpoint:query", cause),
260 ),
261 Effect.asVoid,
262 );
263 });
264
265 const listCheckpoints: CheckpointRepositoryShape["listCheckpoints"] = (input) =>
266 Effect.gen(function* () {
267 const parsed = yield* decodeInput(
268 SessionRequestSchema,
269 {
270 providerSessionId: input.providerSessionId.trim(),
271 },
272 "CheckpointRepository.listCheckpoints",
273 );
274
275 const entries = yield* listForSession(parsed.providerSessionId);
276 return entries.map(toProviderCheckpoint);
277 });
278
279 const getCheckpoint: CheckpointRepositoryShape["getCheckpoint"] = (input) =>
280 Effect.gen(function* () {
281 const parsed = yield* decodeInput(
282 GetCheckpointRequestSchema,
283 {
284 providerSessionId: input.providerSessionId.trim(),
285 turnCount: input.turnCount,
286 },
287 "CheckpointRepository.getCheckpoint",
288 );
289 yield* ensureNonNegativeInt(
290 "CheckpointRepository.getCheckpoint",
291 "turnCount",
292 parsed.turnCount,
293 );
294
295 const entries = yield* listForSession(parsed.providerSessionId);
296 const checkpoint = entries.find((entry) => entry.turnCount === parsed.turnCount);
297 return Option.fromNullishOr(checkpoint);
298 });
299
300 const deleteAfterTurn: CheckpointRepositoryShape["deleteAfterTurn"] = (input) =>
301 Effect.gen(function* () {
302 const parsed = yield* decodeInput(
303 DeleteAfterTurnRequestSchema,
304 {
305 providerSessionId: input.providerSessionId.trim(),
306 maxTurnCount: input.maxTurnCount,
307 },
308 "CheckpointRepository.deleteAfterTurn",
309 );
310 yield* ensureNonNegativeInt(
311 "CheckpointRepository.deleteAfterTurn",
312 "maxTurnCount",
313 parsed.maxTurnCount,
314 );
315
316 yield* sql`
317 DELETE FROM provider_checkpoints
318 WHERE provider_session_id = ${parsed.providerSessionId}
319 AND turn_count > ${parsed.maxTurnCount}
320 `.pipe(
321 Effect.mapError((cause) =>
322 toPersistenceError("CheckpointRepository.deleteAfterTurn:query", cause),
323 ),
324 );
325 });
326
327 const deleteAllForSession: CheckpointRepositoryShape["deleteAllForSession"] = (input) =>
328 Effect.gen(function* () {
329 const parsed = yield* decodeInput(
330 SessionRequestSchema,
331 {
332 providerSessionId: input.providerSessionId.trim(),
333 },
334 "CheckpointRepository.deleteAllForSession",
335 );
336
337 yield* sql`
338 DELETE FROM provider_checkpoints
339 WHERE provider_session_id = ${parsed.providerSessionId}
340 `.pipe(
341 Effect.mapError((cause) =>
342 toPersistenceError("CheckpointRepository.deleteAllForSession:query", cause),
343 ),
344 );
345 });
346
347 return {
348 upsertCheckpoint,
349 listCheckpoints,
350 getCheckpoint,
351 deleteAfterTurn,
352 deleteAllForSession,
353 } satisfies CheckpointRepositoryShape;
354});
355
356export const CheckpointRepositoryLive = Layer.effect(
357 CheckpointRepository,
358 makeCheckpointRepository,
359);
360
361export function makeSqliteCheckpointRepositoryLive(dbPath: string) {
362 return CheckpointRepositoryLive.pipe(Layer.provide(makeSqlitePersistenceLive(dbPath)));
363}
apps/server/src/persistence/Migrations.ts+2
......@@ -14,6 +14,7 @@ import * as Layer from "effect/Layer";
1414// Import all migrations statically
1515import Migration0001 from "./Migrations/001_OrchestrationEvents.ts";
1616import Migration0002 from "./Migrations/002_Projects.ts";
17import Migration0003 from "./Migrations/003_Checkpoints.ts";
1718
1819/**
1920 * Migration loader with all migrations defined inline.
......@@ -28,6 +29,7 @@ import Migration0002 from "./Migrations/002_Projects.ts";
2829const loader = Migrator.fromRecord({
2930 "1_OrchestrationEvents": Migration0001,
3031 "2_Projects": Migration0002,
32 "3_CheckpointCatalog": Migration0003,
3133});
3234
3335/**
apps/server/src/persistence/Migrations/003_Checkpoints.ts created+26
......@@ -0,0 +1,26 @@
1import * as SqlClient from "effect/unstable/sql/SqlClient";
2import * as Effect from "effect/Effect";
3
4export default Effect.gen(function* () {
5 const sql = yield* SqlClient.SqlClient;
6
7 yield* sql`
8 CREATE TABLE IF NOT EXISTS provider_checkpoints (
9 provider_session_id TEXT NOT NULL,
10 thread_id TEXT NOT NULL,
11 checkpoint_id TEXT NOT NULL,
12 checkpoint_ref TEXT NOT NULL,
13 turn_count INTEGER NOT NULL,
14 message_count INTEGER NOT NULL,
15 label TEXT NOT NULL,
16 preview TEXT,
17 created_at TEXT NOT NULL,
18 PRIMARY KEY (provider_session_id, turn_count)
19 )
20 `;
21
22 yield* sql`
23 CREATE INDEX IF NOT EXISTS idx_provider_checkpoints_session_turn
24 ON provider_checkpoints(provider_session_id, turn_count ASC)
25 `;
26});
apps/server/src/persistence/Services/Checkpoints.ts created+95
......@@ -0,0 +1,95 @@
1/**
2 * CheckpointRepository - Repository interface for checkpoint metadata persistence.
3 *
4 * 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
6 * does not coordinate provider rollback workflows.
7 *
8 * @module CheckpointRepository
9 */
10import type { ProviderCheckpoint } from "@t3tools/contracts";
11import { Option, ServiceMap } from "effect";
12import type { Effect } from "effect";
13
14import type { CheckpointRepositoryError } from "../Errors.ts";
15
16export interface UpsertCheckpointInput {
17 readonly providerSessionId: string;
18 readonly threadId: string;
19 readonly checkpointId: string;
20 readonly checkpointRef: string;
21 readonly turnCount: number;
22 readonly messageCount: number;
23 readonly label: string;
24 readonly preview?: string;
25 readonly createdAt: string;
26}
27
28export interface CheckpointRepositoryEntry extends ProviderCheckpoint {
29 readonly threadId: string;
30 readonly checkpointRef: string;
31 readonly createdAt: string;
32}
33
34export interface ListSessionCheckpointsInput {
35 readonly providerSessionId: string;
36}
37
38export interface GetCheckpointInput {
39 readonly providerSessionId: string;
40 readonly turnCount: number;
41}
42
43export interface DeleteAfterTurnInput {
44 readonly providerSessionId: string;
45 readonly maxTurnCount: number;
46}
47
48export interface DeleteAllForSessionInput {
49 readonly providerSessionId: string;
50}
51
52export interface CheckpointRepositoryShape {
53 /**
54 * Insert or update one checkpoint metadata row.
55 */
56 readonly upsertCheckpoint: (
57 input: UpsertCheckpointInput,
58 ) => Effect.Effect<void, CheckpointRepositoryError>;
59
60 /**
61 * List user-facing checkpoints for a provider session.
62 */
63 readonly listCheckpoints: (
64 input: ListSessionCheckpointsInput,
65 ) => Effect.Effect<ReadonlyArray<ProviderCheckpoint>, CheckpointRepositoryError>;
66
67 /**
68 * Read one checkpoint record (including checkpoint ref) by turn count.
69 */
70 readonly getCheckpoint: (
71 input: GetCheckpointInput,
72 ) => Effect.Effect<Option.Option<CheckpointRepositoryEntry>, CheckpointRepositoryError>;
73
74 /**
75 * Delete checkpoint metadata newer than the provided turn count.
76 */
77 readonly deleteAfterTurn: (
78 input: DeleteAfterTurnInput,
79 ) => Effect.Effect<void, CheckpointRepositoryError>;
80
81 /**
82 * Delete all checkpoint metadata for a provider session.
83 */
84 readonly deleteAllForSession: (
85 input: DeleteAllForSessionInput,
86 ) => Effect.Effect<void, CheckpointRepositoryError>;
87}
88
89/**
90 * CheckpointRepository - Service tag for checkpoint metadata persistence.
91 */
92export class CheckpointRepository extends ServiceMap.Service<
93 CheckpointRepository,
94 CheckpointRepositoryShape
95>()("persistence/CheckpointRepository") {}
apps/server/src/provider/contracts.test.ts+27-18
......@@ -11,14 +11,18 @@ import {
1111 type ProviderServiceError,
1212} from "./Errors.ts";
1313import {
14 CheckpointValidationError,
15 type CheckpointCatalogError,
14 CheckpointGitCommandError,
15 CheckpointServiceValidationError,
1616 type CheckpointServiceError,
1717 type CheckpointStoreError,
1818} from "../checkpointing/Errors.ts";
19import { CheckpointCatalog } from "../checkpointing/Services/CheckpointCatalog.ts";
19import {
20 CheckpointRepositoryValidationError,
21 type CheckpointRepositoryError,
22} from "../persistence/Errors.ts";
2023import { CheckpointService } from "../checkpointing/Services/CheckpointService.ts";
2124import { CheckpointStore } from "../checkpointing/Services/CheckpointStore.ts";
25import { CheckpointRepository } from "../persistence/Services/Checkpoints.ts";
2226import { CodexAdapter } from "./Services/CodexAdapter.ts";
2327import { ProviderAdapterRegistry } from "./Services/ProviderAdapterRegistry.ts";
2428import { ProviderService } from "./Services/ProviderService.ts";
......@@ -107,20 +111,22 @@ describe("provider/checkpoint service contracts", () => {
107111 const checkpointStoreLive: typeof CheckpointStore.Service = {
108112 isGitRepository: () => Effect.succeed(false),
109113 captureCheckpoint: () => Effect.void,
110 hasCheckpoint: () => Effect.succeed(false),
114 hasCheckpointRef: () => Effect.succeed(false),
111115 ensureRootCheckpoint: () => Effect.succeed(false),
112116 restoreCheckpoint: () => Effect.succeed(false),
113117 diffCheckpoints: () =>
114118 Effect.fail<CheckpointStoreError>(
115 new CheckpointValidationError({
119 new CheckpointGitCommandError({
116120 operation: "diffCheckpoints",
117 issue: "not implemented",
121 command: "git diff",
122 cwd: "/tmp",
123 detail: "not implemented",
118124 }),
119125 ),
120 pruneAfterTurn: () => Effect.void,
126 deleteCheckpointRefs: () => Effect.void,
121127 };
122128
123 const checkpointCatalogLive: typeof CheckpointCatalog.Service = {
129 const checkpointRepositoryLive: typeof CheckpointRepository.Service = {
124130 upsertCheckpoint: () => Effect.void,
125131 listCheckpoints: () => Effect.succeed([]),
126132 getCheckpoint: () => Effect.succeed(Option.none()),
......@@ -133,18 +139,21 @@ describe("provider/checkpoint service contracts", () => {
133139 captureCurrentTurn: () => Effect.void,
134140 listCheckpoints: () =>
135141 Effect.fail<CheckpointServiceError>(
136 new CheckpointValidationError({ operation: "listCheckpoints", issue: "not implemented" }),
142 new CheckpointServiceValidationError({
143 operation: "listCheckpoints",
144 issue: "not implemented",
145 }),
137146 ),
138147 getCheckpointDiff: () =>
139148 Effect.fail<CheckpointServiceError>(
140 new CheckpointValidationError({
149 new CheckpointServiceValidationError({
141150 operation: "getCheckpointDiff",
142151 issue: "not implemented",
143152 }),
144153 ),
145154 revertToCheckpoint: () =>
146155 Effect.fail<CheckpointServiceError>(
147 new CheckpointValidationError({
156 new CheckpointServiceValidationError({
148157 operation: "revertToCheckpoint",
149158 issue: "not implemented",
150159 }),
......@@ -174,7 +183,7 @@ describe("provider/checkpoint service contracts", () => {
174183 Layer.succeed(ProviderService, providerLive),
175184 Layer.succeed(CodexAdapter, codexLive),
176185 Layer.succeed(CheckpointStore, checkpointStoreLive),
177 Layer.succeed(CheckpointCatalog, checkpointCatalogLive),
186 Layer.succeed(CheckpointRepository, checkpointRepositoryLive),
178187 Layer.succeed(CheckpointService, checkpointServiceLive),
179188 Layer.succeed(ProviderAdapterRegistry, providerAdapterRegistryLive),
180189 Layer.succeed(ProviderSessionDirectory, providerSessionDirectoryLive),
......@@ -185,7 +194,7 @@ describe("provider/checkpoint service contracts", () => {
185194 provider,
186195 codex,
187196 checkpointStore,
188 checkpointCatalog,
197 checkpointRepository,
189198 checkpointService,
190199 providerAdapterRegistry,
191200 providerSessionDirectory,
......@@ -193,7 +202,7 @@ describe("provider/checkpoint service contracts", () => {
193202 runtime.runPromise(Effect.service(ProviderService)),
194203 runtime.runPromise(Effect.service(CodexAdapter)),
195204 runtime.runPromise(Effect.service(CheckpointStore)),
196 runtime.runPromise(Effect.service(CheckpointCatalog)),
205 runtime.runPromise(Effect.service(CheckpointRepository)),
197206 runtime.runPromise(Effect.service(CheckpointService)),
198207 runtime.runPromise(Effect.service(ProviderAdapterRegistry)),
199208 runtime.runPromise(Effect.service(ProviderSessionDirectory)),
......@@ -202,7 +211,7 @@ describe("provider/checkpoint service contracts", () => {
202211 expect(provider).toBe(providerLive);
203212 expect(codex).toBe(codexLive);
204213 expect(checkpointStore).toBe(checkpointStoreLive);
205 expect(checkpointCatalog).toBe(checkpointCatalogLive);
214 expect(checkpointRepository).toBe(checkpointRepositoryLive);
206215 expect(checkpointService).toBe(checkpointServiceLive);
207216 expect(providerAdapterRegistry).toBe(providerAdapterRegistryLive);
208217 expect(providerSessionDirectory).toBe(providerSessionDirectoryLive);
......@@ -225,12 +234,12 @@ describe("provider/checkpoint errors", () => {
225234 expect(fsError.cause).toBe(fsCause);
226235 });
227236
228 it("accepts checkpoint catalog error channel type", () => {
229 const error: CheckpointCatalogError = new CheckpointValidationError({
237 it("accepts checkpoint repository error channel type", () => {
238 const error: CheckpointRepositoryError = new CheckpointRepositoryValidationError({
230239 operation: "upsertCheckpoint",
231240 issue: "turnCount must be >= 0",
232241 });
233242
234 expect(error._tag).toBe("CheckpointValidationError");
243 expect(error._tag).toBe("CheckpointRepositoryValidationError");
235244 });
236245});