diff --git a/.plans/11-effect.md b/.plans/11-effect.md new file mode 100644 index 0000000000000000000000000000000000000000..05f118dd77333779a8070f32cc462b86202ed331 --- /dev/null +++ b/.plans/11-effect.md @@ -0,0 +1,42 @@ + + +PR 1: Service contracts + error taxonomy +Add ProviderService, CodexService, CheckpointStore as Context.Tag service defs. +Add typed Schema.TaggedError hierarchies for all 3 services (cause: Schema.optional(Schema.Defect) on each). +No behavior change yet, just interfaces and compile-time wiring points. +PR 2: CheckpointStore Effect adapter +Wrap current filesystemCheckpointStore behind CheckpointStoreLive (adapter). +Map all thrown/Promise errors to tagged errors. +Add service tests proving parity for isGitRepository, capture, restore, diff, prune. +PR 3: CodexService Effect adapter +Wrap current CodexAppServerManager behind CodexServiceLive (adapter). +Convert public API to Effect return types with typed errors. +Preserve existing EventEmitter internally for now, but expose Effect-friendly subscribe API. +PR 4: ProviderService Effect adapter +Wrap current ProviderManager behind ProviderServiceLive (adapter). +Provider methods become Effect methods with typed errors. +Route emitted provider events through an Effect PubSub surface. +PR 5: wsServer migration to Effect services +Stop instantiating provider/codex classes directly in wsServer. +Resolve ProviderService (and related services) from one runtime/layer graph. +Keep WS contract behavior identical. +PR 6: Native CheckpointStore implementation +Refactor checkpoint internals from Promise/throws to native Effect. +Replace ad-hoc locking with Effect concurrency primitive (keyed lock/semaphore/queue). +Keep adapter tests plus new failure-path tests. +PR 7: Codex transport/RPC core as native Effect +Split codex into scoped process layer + RPC request/response layer + session registry. +Replace timeout/pending maps with Deferred + Effect timeout/finalizer semantics. +Keep protocol behavior and ordering guarantees. +PR 8: Codex protocol decoding hardening +Replace ad-hoc unknown parsing with runtime schema decoding for inbound/outbound protocol shapes. +Map decode failures to typed tagged errors (with root cause). +Add regression tests for malformed/partial protocol messages. +PR 9: Native ProviderService orchestration +Rebuild provider logic in Effect using CodexService + CheckpointStore dependencies. +Move event fanout, checkpoint capture/revert orchestration, thread-log routing to Effect state/services. +Remove throw-based flow entirely from provider path. +PR 10: Cleanup + deprecation removal +Remove legacy class implementations/adapters once parity is proven. +Finalize layer composition and startup graph docs. +Add architecture notes for service boundaries and error model. \ No newline at end of file diff --git a/apps/server/src/checkpointing/Errors.ts b/apps/server/src/checkpointing/Errors.ts index 584ce90b7084bd420f74f0ccab8d1e226264bf44..2e9bbed0997a907a27d06ac29ca5314457467102 100644 --- a/apps/server/src/checkpointing/Errors.ts +++ b/apps/server/src/checkpointing/Errors.ts @@ -1,4 +1,5 @@ import { Schema } from "effect"; +import type { CheckpointRepositoryError as CheckpointMetadataRepositoryError } from "../persistence/Errors.ts"; /** * CheckpointValidationError - Invalid checkpoint API input. @@ -67,22 +68,6 @@ export class CheckpointRepositoryError extends Schema.TaggedErrorClass()( - "CheckpointCatalogPersistenceError", - { - operation: Schema.String, - detail: Schema.String, - cause: Schema.optional(Schema.Defect), - }, -) { - override get message(): string { - return `Checkpoint catalog persistence error in ${this.operation}: ${this.detail}`; - } -} - /** * CheckpointServiceValidationError - Invalid application-level checkpoint request. */ @@ -131,18 +116,13 @@ export class CheckpointInvariantError extends Schema.TaggedErrorClass + Effect.acquireRelease( + Effect.promise(() => mkdtemp(path.join(os.tmpdir(), prefix))), + (directory) => Effect.promise(() => rm(directory, { recursive: true, force: true })), + ); + +const runGit = (cwd: string, args: readonly string[]) => + runGitProcess({ + operation: "CheckpointService.test.git", + cwd, + args, + }).pipe( + Effect.map((result) => result.stdout.trim()), + Effect.provide(NodeServices.layer), + ); + +function createRepo() { + return Effect.gen(function* () { + const cwd = yield* makeTemporaryDirectoryScoped("t3-checkpoint-service-repo-"); + yield* runGit(cwd, ["init", "--initial-branch=main"]); + yield* runGit(cwd, ["config", "user.email", "test@example.com"]); + yield* runGit(cwd, ["config", "user.name", "Test User"]); + yield* Effect.promise(() => writeFile(path.join(cwd, "README.md"), "v1\n")); + yield* runGit(cwd, ["add", "."]); + yield* runGit(cwd, ["commit", "-m", "Initial"]); + return cwd; + }); +} + +class InMemoryProviderAdapter implements ProviderAdapterShape { + readonly provider = "codex" as const; + + private active = true; + private snapshot: ProviderThreadSnapshot; + private readonly session: ProviderSession; + + constructor( + readonly sessionId: string, + readonly cwd: string, + turns: ReadonlyArray, + ) { + const now = new Date().toISOString(); + this.session = { + sessionId, + provider: "codex", + status: "ready", + threadId: "thread-1", + cwd, + createdAt: now, + updatedAt: now, + }; + this.snapshot = { + threadId: this.session.threadId ?? "thread-1", + turns, + }; + } + + setTurns(turns: ReadonlyArray): void { + this.snapshot = { + threadId: this.snapshot.threadId, + turns, + }; + } + + getTurns(): ReadonlyArray { + return this.snapshot.turns; + } + + readonly startSession = ( + _input: ProviderSessionStartInput, + ): Effect.Effect => + Effect.fail( + new ProviderAdapterValidationError({ + provider: this.provider, + operation: "startSession", + issue: "not implemented in test adapter", + }), + ); + + readonly sendTurn = ( + _input: ProviderSendTurnInput, + ): Effect.Effect => + Effect.fail( + new ProviderAdapterValidationError({ + provider: this.provider, + operation: "sendTurn", + issue: "not implemented in test adapter", + }), + ); + + readonly interruptTurn = ( + _sessionId: string, + _turnId?: string, + ): Effect.Effect => Effect.void; + + readonly respondToRequest = ( + _sessionId: string, + _requestId: string, + _decision: ProviderApprovalDecision, + ): Effect.Effect => Effect.void; + + readonly stopSession = (sessionId: string): Effect.Effect => + Effect.sync(() => { + if (sessionId === this.sessionId) { + this.active = false; + } + }); + + readonly listSessions = (): Effect.Effect> => + Effect.sync(() => (this.active ? [this.session] : [])); + + readonly hasSession = (sessionId: string): Effect.Effect => + Effect.succeed(this.active && sessionId === this.sessionId); + + readonly readThread = ( + sessionId: string, + ): Effect.Effect => + this.hasSession(sessionId).pipe( + Effect.flatMap((hasSession) => + hasSession + ? Effect.succeed(this.snapshot) + : Effect.fail( + new ProviderAdapterSessionNotFoundError({ + provider: this.provider, + sessionId, + }), + ), + ), + ); + + readonly rollbackThread = ( + sessionId: string, + numTurns: number, + ): Effect.Effect => + this.readThread(sessionId).pipe( + Effect.flatMap((snapshot) => { + if (!Number.isInteger(numTurns) || numTurns < 0 || numTurns > snapshot.turns.length) { + return Effect.fail( + new ProviderAdapterValidationError({ + provider: this.provider, + operation: "rollbackThread", + issue: "invalid rollback input", + }), + ); + } + const next = { + threadId: snapshot.threadId, + turns: snapshot.turns.slice(0, snapshot.turns.length - numTurns), + } satisfies ProviderThreadSnapshot; + this.snapshot = next; + return Effect.succeed(next); + }), + ); + + readonly stopAll = (): Effect.Effect => + Effect.sync(() => { + this.active = false; + }); + + readonly subscribeToEvents = ( + _callback: (event: ProviderEvent) => void, + ): Effect.Effect<() => void, ProviderAdapterError> => Effect.succeed(() => undefined); +} + +function makeFixture(turns: ReadonlyArray) { + return Effect.gen(function* () { + const cwd = yield* createRepo(); + const stateDir = yield* makeTemporaryDirectoryScoped("t3-checkpoint-service-state-"); + const dbPath = path.join(stateDir, "orchestration.sqlite"); + + const sessionId = "sess-1"; + const adapter = new InMemoryProviderAdapter(sessionId, cwd, turns); + + const registry: typeof ProviderAdapterRegistry.Service = { + getByProvider: (provider: ProviderKind) => + provider === "codex" + ? Effect.succeed(adapter) + : Effect.fail(new ProviderUnsupportedError({ provider })), + listProviders: () => Effect.succeed(["codex"]), + }; + + const dependencies = Layer.mergeAll( + Layer.provide(CheckpointStoreLive, NodeServices.layer), + makeSqliteCheckpointRepositoryLive(dbPath), + Layer.succeed(ProviderAdapterRegistry, registry), + ProviderSessionDirectoryLive, + ); + + const layer = CheckpointServiceLive.pipe(Layer.provideMerge(dependencies)); + + return { + layer, + adapter, + sessionId, + cwd, + }; + }); +} + +it.effect( + "CheckpointServiceLive captures checkpoints, diffs refs, and reverts workspace + turns", + () => + Effect.gen(function* () { + const firstTurn: ProviderThreadTurnSnapshot = { + id: "turn-1", + items: [{ type: "userMessage", content: [{ type: "text", text: "first" }] }], + }; + const secondTurn: ProviderThreadTurnSnapshot = { + id: "turn-2", + items: [{ type: "agentMessage", text: "second" }], + }; + + const fixture = yield* makeFixture([]); + + yield* Effect.gen(function* () { + const service = yield* CheckpointService; + const directory = yield* ProviderSessionDirectory; + + yield* directory.upsert({ + sessionId: fixture.sessionId, + provider: "codex", + threadId: "thread-1", + }); + + yield* service.initializeForSession({ + providerSessionId: fixture.sessionId, + cwd: fixture.cwd, + }); + + yield* Effect.promise(() => writeFile(path.join(fixture.cwd, "README.md"), "v2\n")); + fixture.adapter.setTurns([firstTurn]); + + yield* service.captureCurrentTurn({ + providerSessionId: fixture.sessionId, + }); + + const listed = yield* service.listCheckpoints({ sessionId: fixture.sessionId }); + assert.equal(listed.checkpoints.length, 2); + assert.equal(listed.checkpoints[1]?.id, "turn-1"); + assert.equal(listed.checkpoints[1]?.isCurrent, true); + + const diff = yield* service.getCheckpointDiff({ + sessionId: fixture.sessionId, + fromTurnCount: 0, + toTurnCount: 1, + }); + assert.equal(diff.threadId, "thread-1"); + assert.equal(diff.diff.includes("diff --git a/README.md b/README.md"), true); + assert.equal(diff.diff.includes("-v1"), true); + assert.equal(diff.diff.includes("+v2"), true); + + yield* Effect.promise(() => writeFile(path.join(fixture.cwd, "README.md"), "v3\n")); + fixture.adapter.setTurns([firstTurn, secondTurn]); + + yield* service.captureCurrentTurn({ + providerSessionId: fixture.sessionId, + }); + + const beforeRevert = yield* Effect.promise(() => + readFile(path.join(fixture.cwd, "README.md"), "utf8"), + ); + assert.equal(beforeRevert, "v3\n"); + + const reverted = yield* service.revertToCheckpoint({ + sessionId: fixture.sessionId, + turnCount: 1, + }); + + const afterRevert = yield* Effect.promise(() => + readFile(path.join(fixture.cwd, "README.md"), "utf8"), + ); + assert.equal(afterRevert, "v2\n"); + assert.equal(reverted.turnCount, 1); + assert.equal(reverted.rolledBackTurns, 1); + assert.equal(reverted.checkpoints.length, 2); + assert.equal(reverted.checkpoints[1]?.id, "turn-1"); + assert.equal(reverted.checkpoints[1]?.isCurrent, true); + assert.equal(fixture.adapter.getTurns().length, 1); + }).pipe(Effect.provide(fixture.layer)); + }), +); + +it.effect( + "CheckpointServiceLive fails with CheckpointSessionNotFoundError for unknown sessions", + () => + Effect.gen(function* () { + const fixture = yield* makeFixture([]); + + yield* Effect.gen(function* () { + const service = yield* CheckpointService; + const result = yield* service + .listCheckpoints({ + sessionId: "missing", + }) + .pipe(Effect.result); + + assert.equal(result._tag, "Failure"); + if (result._tag === "Failure") { + assert.equal(result.failure._tag, "CheckpointSessionNotFoundError"); + } + }).pipe(Effect.provide(fixture.layer)); + }), +); + +it.effect( + "CheckpointServiceLive fails with CheckpointUnavailableError when a checkpoint ref is missing", + () => + Effect.gen(function* () { + const firstTurn: ProviderThreadTurnSnapshot = { + id: "turn-1", + items: [{ type: "userMessage", content: [{ type: "text", text: "first" }] }], + }; + + const fixture = yield* makeFixture([]); + + yield* Effect.gen(function* () { + const service = yield* CheckpointService; + const store = yield* CheckpointStore; + const checkpointRepository = yield* CheckpointRepository; + const directory = yield* ProviderSessionDirectory; + + yield* directory.upsert({ + sessionId: fixture.sessionId, + provider: "codex", + threadId: "thread-1", + }); + + yield* service.initializeForSession({ + providerSessionId: fixture.sessionId, + cwd: fixture.cwd, + }); + + yield* Effect.promise(() => writeFile(path.join(fixture.cwd, "README.md"), "v2\n")); + fixture.adapter.setTurns([firstTurn]); + yield* service.captureCurrentTurn({ + providerSessionId: fixture.sessionId, + }); + + const checkpoint = yield* checkpointRepository.getCheckpoint({ + providerSessionId: fixture.sessionId, + turnCount: 1, + }); + if (checkpoint._tag !== "Some") { + assert.fail("Expected turn-1 checkpoint row to exist."); + } + + yield* store.deleteCheckpointRefs({ + cwd: fixture.cwd, + checkpointRefs: [checkpoint.value.checkpointRef], + }); + + const result = yield* service + .revertToCheckpoint({ + sessionId: fixture.sessionId, + turnCount: 1, + }) + .pipe(Effect.result); + + assert.equal(result._tag, "Failure"); + if (result._tag === "Failure") { + assert.equal(result.failure._tag, "CheckpointUnavailableError"); + } + }).pipe(Effect.provide(fixture.layer)); + }), +); diff --git a/apps/server/src/checkpointing/Layers/CheckpointService.ts b/apps/server/src/checkpointing/Layers/CheckpointService.ts new file mode 100644 index 0000000000000000000000000000000000000000..0677d7615cf03d7c1e4ad8056a68e33a6885da19 --- /dev/null +++ b/apps/server/src/checkpointing/Layers/CheckpointService.ts @@ -0,0 +1,1008 @@ +/** + * CheckpointServiceLive - Checkpoint workflow orchestration layer. + * + * Coordinates provider thread snapshots, filesystem checkpoint refs + * (`CheckpointStore`), and checkpoint metadata persistence + * (`CheckpointRepository`) for initialize/capture/list/diff/revert workflows. + * + * This layer owns checkpoint workflow rules and invariants. It does not execute + * provider protocol details directly (adapter concern), and it does not perform + * raw SQL or Git operations directly (catalog/store concerns). + * + * @module CheckpointServiceLive + */ +import type { + ProviderCheckpoint, + ProviderGetCheckpointDiffInput, + ProviderListCheckpointsInput, + ProviderRevertToCheckpointInput, +} from "@t3tools/contracts"; +import { + providerGetCheckpointDiffInputSchema, + providerListCheckpointsInputSchema, + providerRevertToCheckpointInputSchema, +} from "@t3tools/contracts"; +import { Effect, Layer, Option, Ref, Semaphore } from "effect"; + +import { + CheckpointInvariantError, + CheckpointRepositoryError, + CheckpointServiceValidationError, + CheckpointSessionNotFoundError, + CheckpointUnavailableError, + type CheckpointServiceError, +} from "../Errors.ts"; +import { + CheckpointService, + type CaptureCurrentTurnInput, + type CheckpointServiceShape, + type InitializeCheckpointSessionInput, +} from "../Services/CheckpointService.ts"; +import { CheckpointStore } from "../Services/CheckpointStore.ts"; +import { ProviderAdapterRegistry } from "../../provider/Services/ProviderAdapterRegistry.ts"; +import { ProviderSessionDirectory } from "../../provider/Services/ProviderSessionDirectory.ts"; +import { CheckpointRepository } from "../../persistence/Services/Checkpoints.ts"; +import type { + ProviderAdapterShape, + ProviderThreadSnapshot, + ProviderThreadTurnSnapshot, +} from "../../provider/Services/ProviderAdapter.ts"; +import type { ProviderAdapterError } from "../../provider/Errors.ts"; + +interface CheckpointCwdResolution { + readonly candidateCwd: string; + readonly checkpointCwd: Option.Option; +} + +const CHECKPOINT_REFS_PREFIX = "refs/t3/checkpoints"; + +function checkpointRefThreadSegment(threadId: string): string { + return Buffer.from(threadId, "utf8").toString("base64url"); +} + +function checkpointRefForThreadTurn(threadId: string, turnCount: number): string { + return `${CHECKPOINT_REFS_PREFIX}/${checkpointRefThreadSegment(threadId)}/turn/${turnCount}`; +} + +function asObject(value: unknown): Record | undefined { + if (!value || typeof value !== "object") { + return undefined; + } + return value as Record; +} + +function asString(value: unknown): string | undefined { + return typeof value === "string" ? value : undefined; +} + +function asArray(value: unknown): unknown[] { + return Array.isArray(value) ? value : []; +} + +function trimToPreview(value: string): string { + const normalized = value.replace(/\s+/g, " ").trim(); + if (normalized.length <= 120) { + return normalized; + } + return `${normalized.slice(0, 117)}...`; +} + +function summarizeUserMessageContent(content: unknown[]): string | undefined { + const segments: string[] = []; + + for (const part of content) { + const record = asObject(part); + const type = asString(record?.type); + if (!type) { + continue; + } + + if (type === "text") { + const text = asString(record?.text); + if (text && text.trim().length > 0) { + segments.push(text.trim()); + } + continue; + } + + if (type === "image") { + segments.push("[Image attachment]"); + continue; + } + + if (type === "localImage") { + segments.push("[Local image attachment]"); + } + } + + if (segments.length === 0) { + return undefined; + } + return trimToPreview(segments.join(" ")); +} + +function summarizeTurn(turn: ProviderThreadTurnSnapshot): { + readonly messageCountDelta: number; + readonly preview?: string; +} { + let messageCountDelta = 0; + let preview: string | undefined; + + for (const item of turn.items) { + const record = asObject(item); + const type = asString(record?.type); + if (!type) { + continue; + } + + if (type === "userMessage") { + messageCountDelta += 1; + if (!preview) { + const content = asArray(record?.content); + preview = summarizeUserMessageContent(content); + } + continue; + } + + if (type === "agentMessage") { + messageCountDelta += 1; + if (!preview) { + const text = asString(record?.text); + if (text && text.trim().length > 0) { + preview = trimToPreview(text); + } + } + } + } + + return { + messageCountDelta, + ...(preview ? { preview } : {}), + }; +} + +function buildCheckpoints(turns: ReadonlyArray): ProviderCheckpoint[] { + const checkpoints: ProviderCheckpoint[] = []; + let messageCount = 0; + const isEmpty = turns.length === 0; + + checkpoints.push({ + id: "root", + turnCount: 0, + messageCount: 0, + label: "Start of conversation", + isCurrent: isEmpty, + }); + + for (let index = 0; index < turns.length; index += 1) { + const turn = turns[index]; + if (!turn) { + continue; + } + + const turnSummary = summarizeTurn(turn); + messageCount += turnSummary.messageCountDelta; + checkpoints.push({ + id: turn.id, + turnCount: index + 1, + messageCount, + label: `Turn ${index + 1}`, + ...(turnSummary.preview ? { preview: turnSummary.preview } : {}), + isCurrent: index === turns.length - 1, + }); + } + + return checkpoints; +} + +function rootCheckpoint(): ProviderCheckpoint { + return { + id: "root", + turnCount: 0, + messageCount: 0, + label: "Start of conversation", + isCurrent: true, + }; +} + +function validationError( + operation: string, + issue: string, + cause?: unknown, +): CheckpointServiceValidationError { + return new CheckpointServiceValidationError({ + operation, + issue, + ...(cause !== undefined ? { cause } : {}), + }); +} + +function invariantError( + operation: string, + detail: string, + cause?: unknown, +): CheckpointInvariantError { + return new CheckpointInvariantError({ + operation, + detail, + ...(cause !== undefined ? { cause } : {}), + }); +} + +function normalizeNonEmpty( + value: string, + operation: string, + field: string, +): Effect.Effect { + const normalized = value.trim(); + if (normalized.length === 0) { + return Effect.fail(validationError(operation, `${field} must be a non-empty string.`)); + } + return Effect.succeed(normalized); +} + +function decodeInitializeInput( + input: InitializeCheckpointSessionInput, +): Effect.Effect { + return Effect.gen(function* () { + const operation = "CheckpointService.initializeForSession"; + const providerSessionId = yield* normalizeNonEmpty( + input.providerSessionId, + operation, + "providerSessionId", + ); + const cwd = yield* normalizeNonEmpty(input.cwd, operation, "cwd"); + return { + providerSessionId, + cwd, + } satisfies InitializeCheckpointSessionInput; + }); +} + +function decodeCaptureInput( + input: CaptureCurrentTurnInput, +): Effect.Effect { + return Effect.gen(function* () { + const operation = "CheckpointService.captureCurrentTurn"; + const providerSessionId = yield* normalizeNonEmpty( + input.providerSessionId, + operation, + "providerSessionId", + ); + const turnId = input.turnId?.trim(); + const status = input.status?.trim(); + + return { + providerSessionId, + ...(turnId ? { turnId } : {}), + ...(status ? { status } : {}), + } satisfies CaptureCurrentTurnInput; + }); +} + +function decodeListInput( + input: ProviderListCheckpointsInput, +): Effect.Effect { + const parsed = providerListCheckpointsInputSchema.safeParse(input); + if (!parsed.success) { + return Effect.fail( + validationError( + "CheckpointService.listCheckpoints", + parsed.error.message, + parsed.error.cause, + ), + ); + } + return Effect.succeed(parsed.data); +} + +function decodeDiffInput( + input: ProviderGetCheckpointDiffInput, +): Effect.Effect { + const parsed = providerGetCheckpointDiffInputSchema.safeParse(input); + if (!parsed.success) { + return Effect.fail( + validationError( + "CheckpointService.getCheckpointDiff", + parsed.error.message, + parsed.error.cause, + ), + ); + } + return Effect.succeed(parsed.data); +} + +function decodeRevertInput( + input: ProviderRevertToCheckpointInput, +): Effect.Effect { + const parsed = providerRevertToCheckpointInputSchema.safeParse(input); + if (!parsed.success) { + return Effect.fail( + validationError( + "CheckpointService.revertToCheckpoint", + parsed.error.message, + parsed.error.cause, + ), + ); + } + return Effect.succeed(parsed.data); +} + +const makeCheckpointService = Effect.gen(function* () { + const store = yield* CheckpointStore; + const checkpointRepository = yield* CheckpointRepository; + const registry = yield* ProviderAdapterRegistry; + const directory = yield* ProviderSessionDirectory; + + const sessionCheckpointCwdsRef = yield* Ref.make>(new Map()); + const sessionSemaphoresRef = yield* Ref.make>( + new Map(), + ); + + const getSessionSemaphore = (sessionId: string) => + Ref.modify(sessionSemaphoresRef, (current) => { + const existing = current.get(sessionId); + if (existing) { + return [existing, current] as const; + } + + const next = new Map(current); + const semaphore = Semaphore.makeUnsafe(1); + next.set(sessionId, semaphore); + return [semaphore, next] as const; + }); + + const withSessionLock = ( + sessionId: string, + effect: Effect.Effect, + ) => + getSessionSemaphore(sessionId).pipe( + Effect.flatMap((semaphore) => semaphore.withPermits(1)(effect)), + ); + + const setTrackedCheckpointCwd = (sessionId: string, cwd: string) => + Ref.update(sessionCheckpointCwdsRef, (current) => { + const next = new Map(current); + next.set(sessionId, cwd); + return next; + }); + + const clearTrackedCheckpointCwd = (sessionId: string) => + Ref.update(sessionCheckpointCwdsRef, (current) => { + if (!current.has(sessionId)) { + return current; + } + const next = new Map(current); + next.delete(sessionId); + return next; + }); + + const getTrackedCheckpointCwd = (sessionId: string) => + Ref.get(sessionCheckpointCwdsRef).pipe( + Effect.map((current) => Option.fromNullishOr(current.get(sessionId))), + ); + + const resolveAdapterForSession = ( + sessionId: string, + ): Effect.Effect, CheckpointServiceError> => + directory.getProvider(sessionId).pipe( + Effect.mapError( + (cause) => + new CheckpointSessionNotFoundError({ + sessionId, + cause, + }), + ), + Effect.flatMap((provider) => + registry + .getByProvider(provider) + .pipe( + Effect.mapError((cause) => + invariantError( + "CheckpointService.resolveAdapterForSession", + `Provider '${provider}' is not registered for session ${sessionId}.`, + cause, + ), + ), + ), + ), + ); + + const ensureSessionIsActive = ( + sessionId: string, + adapter: ProviderAdapterShape, + ): Effect.Effect => + adapter.hasSession(sessionId).pipe( + Effect.flatMap((hasSession) => + hasSession + ? Effect.void + : Effect.fail( + new CheckpointSessionNotFoundError({ + sessionId, + }), + ), + ), + ); + + const readThreadSnapshot = ( + sessionId: string, + adapter: ProviderAdapterShape, + operation: string, + ): Effect.Effect => + adapter + .readThread(sessionId) + .pipe( + Effect.mapError((cause) => + invariantError( + operation, + `Failed to read thread snapshot for session ${sessionId}.`, + cause, + ), + ), + ); + + const rollbackThreadSnapshot = ( + sessionId: string, + adapter: ProviderAdapterShape, + numTurns: number, + operation: string, + ): Effect.Effect => + adapter + .rollbackThread(sessionId, numTurns) + .pipe( + Effect.mapError((cause) => + invariantError(operation, `Failed to rollback thread for session ${sessionId}.`, cause), + ), + ); + + const persistCheckpointMetadata = (input: { + readonly providerSessionId: string; + readonly threadId: string; + readonly checkpoint: ProviderCheckpoint; + }): Effect.Effect => + checkpointRepository.upsertCheckpoint({ + providerSessionId: input.providerSessionId, + threadId: input.threadId, + checkpointId: input.checkpoint.id, + checkpointRef: checkpointRefForThreadTurn(input.threadId, input.checkpoint.turnCount), + turnCount: input.checkpoint.turnCount, + messageCount: input.checkpoint.messageCount, + label: input.checkpoint.label, + ...(input.checkpoint.preview !== undefined ? { preview: input.checkpoint.preview } : {}), + createdAt: new Date().toISOString(), + }); + + const captureAndPersistCheckpoint = (input: { + readonly providerSessionId: string; + readonly cwd: string; + readonly threadId: string; + readonly checkpoint: ProviderCheckpoint; + }): Effect.Effect => { + const checkpointRef = checkpointRefForThreadTurn(input.threadId, input.checkpoint.turnCount); + return Effect.gen(function* () { + yield* store.captureCheckpoint({ + cwd: input.cwd, + checkpointRef, + }); + + yield* persistCheckpointMetadata({ + providerSessionId: input.providerSessionId, + threadId: input.threadId, + checkpoint: input.checkpoint, + }).pipe( + Effect.catch((error) => + store + .deleteCheckpointRefs({ + cwd: input.cwd, + checkpointRefs: [checkpointRef], + }) + .pipe( + Effect.catch(() => Effect.void), + Effect.flatMap(() => Effect.fail(error)), + ), + ), + ); + }); + }; + + const latestCheckpointFromSnapshot = ( + snapshot: ProviderThreadSnapshot, + operation: string, + ): Effect.Effect => { + if (snapshot.turns.length === 0) { + return Effect.fail( + invariantError(operation, "Cannot derive latest checkpoint from an empty thread snapshot."), + ); + } + + const checkpoints = buildCheckpoints(snapshot.turns); + const latest = checkpoints[checkpoints.length - 1]; + if (!latest || latest.turnCount !== snapshot.turns.length) { + return Effect.fail( + invariantError( + operation, + `Unable to derive checkpoint metadata for turn ${snapshot.turns.length}.`, + ), + ); + } + + return Effect.succeed(latest); + }; + + const ensureRootCheckpointMetadata = ( + providerSessionId: string, + operation: string, + ): Effect.Effect => + checkpointRepository.getCheckpoint({ providerSessionId, turnCount: 0 }).pipe( + Effect.flatMap((rootEntry) => + Option.isSome(rootEntry) + ? Effect.void + : Effect.fail( + invariantError( + operation, + `Missing root checkpoint metadata for session ${providerSessionId}.`, + ), + ), + ), + ); + + const getCheckpointRefOrFail = ( + providerSessionId: string, + threadId: string, + turnCount: number, + operation: string, + ): Effect.Effect => + checkpointRepository.getCheckpoint({ providerSessionId, turnCount }).pipe( + Effect.mapError((error): CheckpointServiceError => error), + Effect.flatMap((checkpoint): Effect.Effect => { + if (Option.isNone(checkpoint)) { + return Effect.fail( + new CheckpointUnavailableError({ + threadId, + turnCount, + detail: `Filesystem checkpoint is unavailable for turn ${turnCount} in thread ${threadId}.`, + }), + ); + } + + const checkpointRef = checkpoint.value.checkpointRef.trim(); + if (checkpointRef.length === 0) { + return Effect.fail( + invariantError( + operation, + `Checkpoint metadata for turn ${turnCount} is missing checkpointRef.`, + ), + ); + } + + return Effect.succeed(checkpointRef); + }), + ); + + const resolveCandidateCwd = ( + sessionId: string, + adapter: ProviderAdapterShape, + ): Effect.Effect => + adapter.listSessions().pipe( + Effect.map((sessions) => sessions.find((session) => session.sessionId === sessionId)?.cwd), + Effect.map((cwd) => { + const candidate = cwd?.trim() || process.cwd(); + if (candidate.length > 0) { + return candidate; + } + return process.cwd(); + }), + ); + + const resolveCheckpointCwd = ( + sessionId: string, + adapter: ProviderAdapterShape, + ): Effect.Effect => + Effect.gen(function* () { + const tracked = yield* getTrackedCheckpointCwd(sessionId); + if (Option.isSome(tracked)) { + return { + candidateCwd: tracked.value, + checkpointCwd: tracked, + } satisfies CheckpointCwdResolution; + } + + const candidateCwd = yield* resolveCandidateCwd(sessionId, adapter); + const supportsGit = yield* store.isGitRepository(candidateCwd); + if (!supportsGit) { + yield* clearTrackedCheckpointCwd(sessionId); + return { + candidateCwd, + checkpointCwd: Option.none(), + } satisfies CheckpointCwdResolution; + } + + yield* setTrackedCheckpointCwd(sessionId, candidateCwd); + + return { + candidateCwd, + checkpointCwd: Option.some(candidateCwd), + } satisfies CheckpointCwdResolution; + }); + + const initializeForSession: CheckpointServiceShape["initializeForSession"] = (rawInput) => + decodeInitializeInput(rawInput).pipe( + Effect.flatMap((input) => + withSessionLock( + input.providerSessionId, + Effect.gen(function* () { + const adapter = yield* resolveAdapterForSession(input.providerSessionId); + yield* ensureSessionIsActive(input.providerSessionId, adapter); + const snapshot = yield* readThreadSnapshot( + input.providerSessionId, + adapter, + "CheckpointService.initializeForSession:readThread", + ); + if (snapshot.turns.length !== 0) { + return yield* Effect.fail( + invariantError( + "CheckpointService.initializeForSession", + `Expected empty thread snapshot during initialization, received ${snapshot.turns.length} turn(s).`, + ), + ); + } + + const supportsGit = yield* store.isGitRepository(input.cwd); + if (!supportsGit) { + yield* clearTrackedCheckpointCwd(input.providerSessionId); + return; + } + + const existingRoot = yield* checkpointRepository.getCheckpoint({ + providerSessionId: input.providerSessionId, + turnCount: 0, + }); + if (Option.isSome(existingRoot)) { + const existingRef = existingRoot.value.checkpointRef.trim(); + if (existingRef.length === 0) { + return yield* Effect.fail( + invariantError( + "CheckpointService.initializeForSession", + "Existing root checkpoint metadata is missing checkpointRef.", + ), + ); + } + const exists = yield* store.hasCheckpointRef({ + cwd: input.cwd, + checkpointRef: existingRef, + }); + if (!exists) { + return yield* Effect.fail( + invariantError( + "CheckpointService.initializeForSession", + "Root checkpoint metadata exists but filesystem ref is missing.", + ), + ); + } + yield* setTrackedCheckpointCwd(input.providerSessionId, input.cwd); + return; + } + + yield* captureAndPersistCheckpoint({ + providerSessionId: input.providerSessionId, + cwd: input.cwd, + threadId: snapshot.threadId, + checkpoint: rootCheckpoint(), + }); + yield* setTrackedCheckpointCwd(input.providerSessionId, input.cwd); + }), + ), + ), + ); + + const captureCurrentTurn: CheckpointServiceShape["captureCurrentTurn"] = (rawInput) => + decodeCaptureInput(rawInput).pipe( + Effect.flatMap((input) => + withSessionLock( + input.providerSessionId, + Effect.gen(function* () { + const adapter = yield* resolveAdapterForSession(input.providerSessionId); + yield* ensureSessionIsActive(input.providerSessionId, adapter); + const snapshot = yield* readThreadSnapshot( + input.providerSessionId, + adapter, + "CheckpointService.captureCurrentTurn:readThread", + ); + if (snapshot.turns.length === 0) { + return; + } + + const resolution = yield* resolveCheckpointCwd(input.providerSessionId, adapter); + if (Option.isNone(resolution.checkpointCwd)) { + return; + } + + const checkpoint = yield* latestCheckpointFromSnapshot( + snapshot, + "CheckpointService.captureCurrentTurn", + ); + + yield* ensureRootCheckpointMetadata( + input.providerSessionId, + "CheckpointService.captureCurrentTurn", + ); + + const existing = yield* checkpointRepository.getCheckpoint({ + providerSessionId: input.providerSessionId, + turnCount: checkpoint.turnCount, + }); + if (Option.isSome(existing)) { + const checkpointRef = existing.value.checkpointRef.trim(); + if (checkpointRef.length === 0) { + return yield* Effect.fail( + invariantError( + "CheckpointService.captureCurrentTurn", + `Checkpoint metadata for turn ${checkpoint.turnCount} is missing checkpointRef.`, + ), + ); + } + const exists = yield* store.hasCheckpointRef({ + cwd: resolution.checkpointCwd.value, + checkpointRef, + }); + if (!exists) { + return yield* Effect.fail( + invariantError( + "CheckpointService.captureCurrentTurn", + `Checkpoint metadata exists for turn ${checkpoint.turnCount} but filesystem ref is missing.`, + ), + ); + } + return; + } + + yield* captureAndPersistCheckpoint({ + providerSessionId: input.providerSessionId, + cwd: resolution.checkpointCwd.value, + threadId: snapshot.threadId, + checkpoint, + }); + }), + ), + ), + ); + + const listCheckpoints: CheckpointServiceShape["listCheckpoints"] = (rawInput) => + decodeListInput(rawInput).pipe( + Effect.flatMap((input) => + withSessionLock( + input.sessionId, + Effect.gen(function* () { + const adapter = yield* resolveAdapterForSession(input.sessionId); + yield* ensureSessionIsActive(input.sessionId, adapter); + const snapshot = yield* readThreadSnapshot( + input.sessionId, + adapter, + "CheckpointService.listCheckpoints:readThread", + ); + + const resolution = yield* resolveCheckpointCwd(input.sessionId, adapter); + if (Option.isNone(resolution.checkpointCwd)) { + return yield* Effect.fail( + new CheckpointRepositoryError({ + cwd: resolution.candidateCwd, + detail: "Filesystem checkpoints are unavailable for this session.", + }), + ); + } + + const checkpoints = yield* checkpointRepository.listCheckpoints({ + providerSessionId: input.sessionId, + }); + if (checkpoints.length === 0) { + return yield* Effect.fail( + invariantError( + "CheckpointService.listCheckpoints", + `Missing checkpoint metadata for active session ${input.sessionId}.`, + ), + ); + } + + return { + threadId: snapshot.threadId, + checkpoints: Array.from(checkpoints), + }; + }), + ), + ), + ); + + const getCheckpointDiff: CheckpointServiceShape["getCheckpointDiff"] = (rawInput) => + decodeDiffInput(rawInput).pipe( + Effect.flatMap((input) => + withSessionLock( + input.sessionId, + Effect.gen(function* () { + const adapter = yield* resolveAdapterForSession(input.sessionId); + yield* ensureSessionIsActive(input.sessionId, adapter); + const snapshot = yield* readThreadSnapshot( + input.sessionId, + adapter, + "CheckpointService.getCheckpointDiff:readThread", + ); + + if (input.toTurnCount > snapshot.turns.length) { + return yield* Effect.fail( + validationError( + "CheckpointService.getCheckpointDiff", + `Checkpoint turn count ${input.toTurnCount} exceeds current turn count ${snapshot.turns.length}.`, + ), + ); + } + + const resolution = yield* resolveCheckpointCwd(input.sessionId, adapter); + if (Option.isNone(resolution.checkpointCwd)) { + return yield* Effect.fail( + new CheckpointRepositoryError({ + cwd: resolution.candidateCwd, + detail: "Filesystem checkpoints are unavailable for this session.", + }), + ); + } + + const fromCheckpointRef = yield* getCheckpointRefOrFail( + input.sessionId, + snapshot.threadId, + input.fromTurnCount, + "CheckpointService.getCheckpointDiff:fromCheckpoint", + ); + const toCheckpointRef = yield* getCheckpointRefOrFail( + input.sessionId, + snapshot.threadId, + input.toTurnCount, + "CheckpointService.getCheckpointDiff:toCheckpoint", + ); + + const diff = yield* store.diffCheckpoints({ + cwd: resolution.checkpointCwd.value, + fromCheckpointRef, + toCheckpointRef, + fallbackFromToHead: input.fromTurnCount === 0, + }); + + return { + threadId: snapshot.threadId, + fromTurnCount: input.fromTurnCount, + toTurnCount: input.toTurnCount, + diff, + }; + }), + ), + ), + ); + + const revertToCheckpoint: CheckpointServiceShape["revertToCheckpoint"] = (rawInput) => + decodeRevertInput(rawInput).pipe( + Effect.flatMap((input) => + withSessionLock( + input.sessionId, + Effect.gen(function* () { + const adapter = yield* resolveAdapterForSession(input.sessionId); + yield* ensureSessionIsActive(input.sessionId, adapter); + const beforeSnapshot = yield* readThreadSnapshot( + input.sessionId, + adapter, + "CheckpointService.revertToCheckpoint:readThread", + ); + const currentTurnCount = beforeSnapshot.turns.length; + + if (input.turnCount > currentTurnCount) { + return yield* Effect.fail( + validationError( + "CheckpointService.revertToCheckpoint", + `Checkpoint turn count ${input.turnCount} exceeds current turn count ${currentTurnCount}.`, + ), + ); + } + + const resolution = yield* resolveCheckpointCwd(input.sessionId, adapter); + if (Option.isNone(resolution.checkpointCwd)) { + return yield* Effect.fail( + new CheckpointRepositoryError({ + cwd: resolution.candidateCwd, + detail: "Filesystem checkpoints are unavailable for this session.", + }), + ); + } + + const checkpointCwd = resolution.checkpointCwd.value; + const checkpointRef = yield* getCheckpointRefOrFail( + input.sessionId, + beforeSnapshot.threadId, + input.turnCount, + "CheckpointService.revertToCheckpoint:checkpointLookup", + ); + + const restored = yield* store.restoreCheckpoint({ + cwd: checkpointCwd, + checkpointRef, + fallbackToHead: input.turnCount === 0, + }); + if (!restored) { + return yield* Effect.fail( + new CheckpointUnavailableError({ + threadId: beforeSnapshot.threadId, + turnCount: input.turnCount, + detail: `Filesystem checkpoint is unavailable for turn ${input.turnCount} in thread ${beforeSnapshot.threadId}.`, + }), + ); + } + + const requestedRollbackTurns = currentTurnCount - input.turnCount; + const afterSnapshot = + requestedRollbackTurns > 0 + ? yield* rollbackThreadSnapshot( + input.sessionId, + adapter, + requestedRollbackTurns, + "CheckpointService.revertToCheckpoint:rollbackThread", + ) + : beforeSnapshot; + + const staleTurnCounts: number[] = []; + for (let turn = input.turnCount + 1; turn <= currentTurnCount; turn += 1) { + staleTurnCounts.push(turn); + } + + const staleCheckpointRefs = yield* Effect.forEach( + staleTurnCounts, + (turnCount) => + checkpointRepository.getCheckpoint({ + providerSessionId: input.sessionId, + turnCount, + }), + { discard: false }, + ).pipe( + Effect.map((entries) => + entries + .flatMap((entry) => + Option.isSome(entry) ? [entry.value.checkpointRef.trim()] : [], + ) + .filter((ref) => ref.length > 0), + ), + ); + + yield* store.deleteCheckpointRefs({ + cwd: checkpointCwd, + checkpointRefs: staleCheckpointRefs, + }); + yield* checkpointRepository.deleteAfterTurn({ + providerSessionId: input.sessionId, + maxTurnCount: input.turnCount, + }); + const checkpoints = yield* checkpointRepository.listCheckpoints({ + providerSessionId: input.sessionId, + }); + const currentCheckpoint = + checkpoints.find((checkpoint) => checkpoint.isCurrent) ?? + checkpoints[checkpoints.length - 1] ?? + checkpoints[0]; + const rolledBackTurns = Math.max(0, currentTurnCount - afterSnapshot.turns.length); + + return { + threadId: afterSnapshot.threadId, + turnCount: currentCheckpoint?.turnCount ?? 0, + messageCount: currentCheckpoint?.messageCount ?? 0, + rolledBackTurns, + checkpoints: Array.from(checkpoints), + }; + }), + ), + ), + ); + + return { + initializeForSession, + captureCurrentTurn, + listCheckpoints, + getCheckpointDiff, + revertToCheckpoint, + } satisfies CheckpointServiceShape; +}); + +export const CheckpointServiceLive = Layer.effect(CheckpointService, makeCheckpointService); diff --git a/apps/server/src/checkpointing/Layers/CheckpointStore.ts b/apps/server/src/checkpointing/Layers/CheckpointStore.ts new file mode 100644 index 0000000000000000000000000000000000000000..31b15cc6f02057bf64b3581ec79632f90e469c16 --- /dev/null +++ b/apps/server/src/checkpointing/Layers/CheckpointStore.ts @@ -0,0 +1,418 @@ +/** + * CheckpointStoreLive - Filesystem checkpoint store adapter layer. + * + * Implements hidden Git-ref checkpoint capture/restore directly with + * Effect-native child process execution (`effect/unstable/process`). + * + * This layer owns filesystem/Git interactions only; it does not persist + * checkpoint metadata and does not coordinate provider rollback semantics. + * + * @module CheckpointStoreLive + */ +import { mkdtemp, rm } from "node:fs/promises"; +import { randomUUID } from "node:crypto"; +import os from "node:os"; +import path from "node:path"; + +import { Effect, Layer } from "effect"; +import { ChildProcessSpawner } from "effect/unstable/process"; + +import { + CheckpointGitCommandError, + CheckpointRepositoryError, + CheckpointUnavailableError, + type CheckpointStoreError, +} from "../Errors.ts"; +import { GitCommandError } from "../../git/Errors.ts"; +import { runGit, RunGitInput } from "../../git/Process.ts"; +import { CheckpointStore, type CheckpointStoreShape } from "../Services/CheckpointStore.ts"; + +function causeDetail(cause: unknown, fallback: string): string { + if (cause instanceof GitCommandError) { + return cause.detail; + } + if (cause instanceof Error && cause.message.length > 0) { + return cause.message; + } + return fallback; +} + +function mapFilesystemFailure( + operation: string, + cwd: string, + cause: unknown, +): CheckpointStoreError { + const detail = causeDetail(cause, `${operation} failed.`); + + const unavailable = + /^Filesystem checkpoint is unavailable for turn (\d+) in thread (.+)\.?$/i.exec(detail); + if (unavailable) { + const [, rawTurnCount, threadId] = unavailable; + const turnCount = Number.parseInt(rawTurnCount ?? "", 10); + if (threadId && Number.isInteger(turnCount) && turnCount >= 0) { + return new CheckpointUnavailableError({ + threadId, + turnCount, + detail, + cause, + }); + } + } + + const lowered = detail.toLowerCase(); + if (lowered.includes("not a git repository")) { + return new CheckpointRepositoryError({ + cwd, + detail, + cause, + }); + } + + return new CheckpointGitCommandError({ + operation, + command: cause instanceof GitCommandError ? cause.command : "git", + cwd, + detail, + cause, + }); +} + +function resolveHeadCommit( + cwd: string, +): Effect.Effect { + return runGit({ + operation: "CheckpointStore.resolveHeadCommit", + cwd, + args: ["rev-parse", "--verify", "--quiet", "HEAD^{commit}"], + allowNonZeroExit: true, + }).pipe( + Effect.map((result) => { + if (result.code !== 0) { + return null; + } + const commit = result.stdout.trim(); + return commit.length > 0 ? commit : null; + }), + ); +} + +function hasHeadCommit( + cwd: string, +): Effect.Effect { + return runGit({ + operation: "CheckpointStore.hasHeadCommit", + cwd, + args: ["rev-parse", "--verify", "HEAD"], + allowNonZeroExit: true, + }).pipe(Effect.map((result) => result.code === 0)); +} + +function resolveCheckpointCommit( + cwd: string, + checkpointRef: string, +): Effect.Effect { + return runGit({ + operation: "CheckpointStore.resolveCheckpointCommit", + cwd, + args: ["rev-parse", "--verify", "--quiet", `${checkpointRef}^{commit}`], + allowNonZeroExit: true, + }).pipe( + Effect.map((result) => { + if (result.code !== 0) { + return null; + } + const commit = result.stdout.trim(); + return commit.length > 0 ? commit : null; + }), + ); +} + +function removeDirectory(pathname: string): Effect.Effect { + return Effect.tryPromise({ + try: () => rm(pathname, { recursive: true, force: true }), + catch: (cause) => cause, + }).pipe(Effect.orDie); +} + +const makeCheckpointStore = Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const runGitWithSpawner = (input: RunGitInput) => + runGit(input).pipe(Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner)); + const resolveHeadCommitWithSpawner = (cwd: string) => + resolveHeadCommit(cwd).pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + ); + const hasHeadCommitWithSpawner = (cwd: string) => + hasHeadCommit(cwd).pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + ); + const resolveCheckpointCommitWithSpawner = (cwd: string, checkpointRef: string) => + resolveCheckpointCommit(cwd, checkpointRef).pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + ); + + const isGitRepository: CheckpointStoreShape["isGitRepository"] = (cwd) => + Effect.gen(function* () { + const normalizedCwd = cwd.trim(); + + return yield* runGitWithSpawner({ + operation: "CheckpointStore.isGitRepository", + cwd: normalizedCwd, + args: ["rev-parse", "--is-inside-work-tree"], + allowNonZeroExit: true, + }).pipe( + Effect.map((result) => result.code === 0 && result.stdout.trim() === "true"), + Effect.catch(() => Effect.succeed(false)), + ); + }); + + const captureCheckpoint: CheckpointStoreShape["captureCheckpoint"] = (input) => + Effect.gen(function* () { + const operation = "CheckpointStore.captureCheckpoint"; + const normalizedCwd = input.cwd.trim(); + const checkpointRef = input.checkpointRef.trim(); + + yield* Effect.acquireUseRelease( + Effect.tryPromise({ + try: () => mkdtemp(path.join(os.tmpdir(), "t3-fs-checkpoint-")), + catch: (cause) => cause, + }), + (tempDir) => + Effect.gen(function* () { + const tempIndexPath = path.join(tempDir, `index-${randomUUID()}`); + const commitEnv: NodeJS.ProcessEnv = { + ...process.env, + GIT_INDEX_FILE: tempIndexPath, + GIT_AUTHOR_NAME: "T3 Code", + GIT_AUTHOR_EMAIL: "codex@users.noreply.github.com", + GIT_COMMITTER_NAME: "T3 Code", + GIT_COMMITTER_EMAIL: "codex@users.noreply.github.com", + }; + + const headExists = yield* hasHeadCommitWithSpawner(normalizedCwd); + if (headExists) { + yield* runGitWithSpawner({ + operation, + cwd: normalizedCwd, + args: ["read-tree", "HEAD"], + env: commitEnv, + }); + } + + yield* runGitWithSpawner({ + operation, + cwd: normalizedCwd, + args: ["add", "-A", "--", "."], + env: commitEnv, + }); + + const writeTreeResult = yield* runGitWithSpawner({ + operation, + cwd: normalizedCwd, + args: ["write-tree"], + env: commitEnv, + }); + const treeOid = writeTreeResult.stdout.trim(); + if (treeOid.length === 0) { + return yield* Effect.fail( + new CheckpointGitCommandError({ + operation, + command: "git write-tree", + cwd: normalizedCwd, + detail: "git write-tree returned an empty tree oid.", + }), + ); + } + + const message = `t3 checkpoint ref=${checkpointRef}`; + const commitTreeResult = yield* runGitWithSpawner({ + operation, + cwd: normalizedCwd, + args: ["commit-tree", treeOid, "-m", message], + env: commitEnv, + }); + const commitOid = commitTreeResult.stdout.trim(); + if (commitOid.length === 0) { + return yield* Effect.fail( + new CheckpointGitCommandError({ + operation, + command: "git commit-tree", + cwd: normalizedCwd, + detail: "git commit-tree returned an empty commit oid.", + }), + ); + } + + yield* runGitWithSpawner({ + operation, + cwd: normalizedCwd, + args: ["update-ref", checkpointRef, commitOid], + }); + }), + removeDirectory, + ).pipe(Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause))); + }); + + const hasCheckpointRef: CheckpointStoreShape["hasCheckpointRef"] = (input) => + Effect.gen(function* () { + const operation = "CheckpointStore.hasCheckpointRef"; + const normalizedCwd = input.cwd.trim(); + const checkpointRef = input.checkpointRef.trim(); + + return yield* resolveCheckpointCommitWithSpawner(normalizedCwd, checkpointRef).pipe( + Effect.map((commit) => commit !== null), + Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)), + ); + }); + + const ensureRootCheckpoint: CheckpointStoreShape["ensureRootCheckpoint"] = (input) => + Effect.gen(function* () { + const operation = "CheckpointStore.ensureRootCheckpoint"; + const normalizedCwd = input.cwd.trim(); + const checkpointRef = input.checkpointRef.trim(); + + const existing = yield* resolveCheckpointCommitWithSpawner(normalizedCwd, checkpointRef).pipe( + Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)), + ); + if (existing) { + return true; + } + + const headCommit = yield* resolveHeadCommitWithSpawner(normalizedCwd).pipe( + Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)), + ); + if (!headCommit) { + return false; + } + + yield* runGitWithSpawner({ + operation, + cwd: normalizedCwd, + args: ["update-ref", checkpointRef, headCommit], + }).pipe(Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause))); + return true; + }); + + const restoreCheckpoint: CheckpointStoreShape["restoreCheckpoint"] = (input) => + Effect.gen(function* () { + const operation = "CheckpointStore.restoreCheckpoint"; + const normalizedCwd = input.cwd.trim(); + const checkpointRef = input.checkpointRef.trim(); + + let commitOid = yield* resolveCheckpointCommitWithSpawner(normalizedCwd, checkpointRef).pipe( + Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)), + ); + + if (!commitOid && input.fallbackToHead === true) { + commitOid = yield* resolveHeadCommitWithSpawner(normalizedCwd).pipe( + Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)), + ); + } + + if (!commitOid) { + return false; + } + + yield* runGitWithSpawner({ + operation, + cwd: normalizedCwd, + args: ["restore", "--source", commitOid, "--worktree", "--staged", "--", "."], + }).pipe(Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause))); + yield* runGitWithSpawner({ + operation, + cwd: normalizedCwd, + args: ["clean", "-fd", "--", "."], + }).pipe(Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause))); + + const headExists = yield* hasHeadCommitWithSpawner(normalizedCwd).pipe( + Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)), + ); + if (headExists) { + yield* runGitWithSpawner({ + operation, + cwd: normalizedCwd, + args: ["reset", "--quiet", "--", "."], + }).pipe(Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause))); + } + + return true; + }); + + const diffCheckpoints: CheckpointStoreShape["diffCheckpoints"] = (input) => + Effect.gen(function* () { + const operation = "CheckpointStore.diffCheckpoints"; + const normalizedCwd = input.cwd.trim(); + const fromCheckpointRef = input.fromCheckpointRef.trim(); + const toCheckpointRef = input.toCheckpointRef.trim(); + + let fromCommitOid = yield* resolveCheckpointCommitWithSpawner( + normalizedCwd, + fromCheckpointRef, + ).pipe(Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause))); + const toCommitOid = yield* resolveCheckpointCommitWithSpawner( + normalizedCwd, + toCheckpointRef, + ).pipe(Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause))); + + if (!fromCommitOid && input.fallbackFromToHead === true) { + const headCommit = yield* resolveHeadCommitWithSpawner(normalizedCwd).pipe( + Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)), + ); + if (headCommit) { + fromCommitOid = headCommit; + } + } + + if (!fromCommitOid || !toCommitOid) { + return yield* Effect.fail( + new CheckpointGitCommandError({ + operation, + command: "git diff", + cwd: normalizedCwd, + detail: "Checkpoint ref is unavailable for diff operation.", + }), + ); + } + + const result = yield* runGitWithSpawner({ + operation, + cwd: normalizedCwd, + args: ["diff", "--patch", "--minimal", "--no-color", fromCommitOid, toCommitOid], + }).pipe(Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause))); + + return result.stdout; + }); + + const deleteCheckpointRefs: CheckpointStoreShape["deleteCheckpointRefs"] = (input) => + Effect.gen(function* () { + const operation = "CheckpointStore.deleteCheckpointRefs"; + const normalizedCwd = input.cwd.trim(); + const checkpointRefs = input.checkpointRefs.map((ref) => ref.trim()); + + yield* Effect.forEach( + checkpointRefs, + (checkpointRef) => + runGitWithSpawner({ + operation, + cwd: normalizedCwd, + args: ["update-ref", "-d", checkpointRef], + allowNonZeroExit: true, + }).pipe( + Effect.mapError((cause) => mapFilesystemFailure(operation, normalizedCwd, cause)), + ), + { discard: true }, + ); + }); + + return { + isGitRepository, + captureCheckpoint, + hasCheckpointRef, + ensureRootCheckpoint, + restoreCheckpoint, + diffCheckpoints, + deleteCheckpointRefs, + } satisfies CheckpointStoreShape; +}); + +export const CheckpointStoreLive = Layer.effect(CheckpointStore, makeCheckpointStore); diff --git a/apps/server/src/checkpointing/Services/CheckpointCatalog.ts b/apps/server/src/checkpointing/Services/CheckpointCatalog.ts deleted file mode 100644 index b9e00e08a90dc6742c0a9b138b277d99d3f4f3de..0000000000000000000000000000000000000000 --- a/apps/server/src/checkpointing/Services/CheckpointCatalog.ts +++ /dev/null @@ -1,83 +0,0 @@ -/** - * CheckpointCatalog - Metadata repository for checkpoints. - * - * Stores user-facing checkpoint metadata used by list/revert APIs. It does not - * touch the filesystem or hidden Git refs; `CheckpointStore` owns that state. - * It also does not drive provider conversation rollback. - * - * @module CheckpointCatalog - */ -import type { ProviderCheckpoint } from "@t3tools/contracts"; -import { Option, ServiceMap } from "effect"; -import type { Effect } from "effect"; - -import type { CheckpointCatalogError } from "../Errors.ts"; - -export interface UpsertCheckpointInput { - readonly providerSessionId: string; - readonly threadId: string; - readonly turnCount: number; - readonly messageCount: number; - readonly label: string; - readonly preview?: string; - readonly createdAt: string; -} - -export interface ListSessionCheckpointsInput { - readonly providerSessionId: string; -} - -export interface GetCheckpointInput { - readonly providerSessionId: string; - readonly turnCount: number; -} - -export interface DeleteAfterTurnInput { - readonly providerSessionId: string; - readonly maxTurnCount: number; -} - -export interface DeleteAllForSessionInput { - readonly providerSessionId: string; -} - -export interface CheckpointCatalogShape { - /** - * Insert or update one checkpoint metadata row. - */ - readonly upsertCheckpoint: (input: UpsertCheckpointInput) => Effect.Effect; - - /** - * List checkpoints for a provider session. - */ - readonly listCheckpoints: ( - input: ListSessionCheckpointsInput, - ) => Effect.Effect, CheckpointCatalogError>; - - /** - * Read one checkpoint by turn count. - */ - readonly getCheckpoint: ( - input: GetCheckpointInput, - ) => Effect.Effect, CheckpointCatalogError>; - - /** - * Delete checkpoint metadata newer than the provided turn count. - */ - readonly deleteAfterTurn: (input: DeleteAfterTurnInput) => Effect.Effect; - - /** - * Delete all checkpoint metadata for a provider session. - */ - readonly deleteAllForSession: ( - input: DeleteAllForSessionInput, - ) => Effect.Effect; -} - -/** - * CheckpointCatalog - Service tag for checkpoint metadata persistence. - */ -export class CheckpointCatalog extends ServiceMap.Service< - CheckpointCatalog, - CheckpointCatalogShape ->()("checkpointing/CheckpointCatalog") {} diff --git a/apps/server/src/checkpointing/Services/CheckpointService.ts b/apps/server/src/checkpointing/Services/CheckpointService.ts index 4622b1d18263960c6490d7cd06830b07236d0c6c..6a5260eb6c9e7993110da84b982893262cb4cc0a 100644 --- a/apps/server/src/checkpointing/Services/CheckpointService.ts +++ b/apps/server/src/checkpointing/Services/CheckpointService.ts @@ -1,7 +1,7 @@ /** * CheckpointService - Application orchestrator for checkpoint workflows. * - * Coordinates `CheckpointStore` (Git refs), `CheckpointCatalog` (metadata), and + * Coordinates `CheckpointStore` (Git refs), `CheckpointRepository` (metadata), and * provider timeline operations resolved through `ProviderSessionDirectory` and * `ProviderAdapterRegistry` (read/rollback/hasSession on adapters). * @@ -45,7 +45,9 @@ export interface CheckpointServiceShape { /** * Capture the current thread turn into checkpoint refs/metadata. */ - readonly captureCurrentTurn: (input: CaptureCurrentTurnInput) => Effect.Effect; + readonly captureCurrentTurn: ( + input: CaptureCurrentTurnInput, + ) => Effect.Effect; /** * List checkpoints for a provider session. diff --git a/apps/server/src/checkpointing/Services/CheckpointStore.ts b/apps/server/src/checkpointing/Services/CheckpointStore.ts index 81fa60c305f6f3928c11506f80919e446479ce33..7dece4884468d4842e06a3e81ea573458064931c 100644 --- a/apps/server/src/checkpointing/Services/CheckpointStore.ts +++ b/apps/server/src/checkpointing/Services/CheckpointStore.ts @@ -17,32 +17,30 @@ import type { CheckpointStoreError } from "../Errors.ts"; export interface CaptureCheckpointInput { readonly cwd: string; - readonly threadId: string; - readonly turnCount: number; + readonly checkpointRef: string; } export interface EnsureRootCheckpointInput { readonly cwd: string; - readonly threadId: string; + readonly checkpointRef: string; } export interface RestoreCheckpointInput { readonly cwd: string; - readonly threadId: string; - readonly turnCount: number; + readonly checkpointRef: string; + readonly fallbackToHead?: boolean; } export interface DiffCheckpointsInput { readonly cwd: string; - readonly threadId: string; - readonly fromTurnCount: number; - readonly toTurnCount: number; + readonly fromCheckpointRef: string; + readonly toCheckpointRef: string; + readonly fallbackFromToHead?: boolean; } -export interface PruneAfterTurnInput { +export interface DeleteCheckpointRefsInput { readonly cwd: string; - readonly threadId: string; - readonly maxTurnCount: number; + readonly checkpointRefs: ReadonlyArray; } export interface CheckpointStoreShape { @@ -52,17 +50,21 @@ export interface CheckpointStoreShape { readonly isGitRepository: (cwd: string) => Effect.Effect; /** - * Capture a checkpoint for a thread at a turn count. + * Capture a checkpoint commit and store it at the provided checkpoint ref. */ - readonly captureCheckpoint: (input: CaptureCheckpointInput) => Effect.Effect; + readonly captureCheckpoint: ( + input: CaptureCheckpointInput, + ) => Effect.Effect; /** * Check whether a checkpoint ref exists. */ - readonly hasCheckpoint: (input: RestoreCheckpointInput) => Effect.Effect; + readonly hasCheckpointRef: ( + input: Omit, + ) => Effect.Effect; /** - * Ensure a turn-0 checkpoint exists for a thread. + * Ensure a root checkpoint ref exists. */ readonly ensureRootCheckpoint: ( input: EnsureRootCheckpointInput, @@ -76,20 +78,23 @@ export interface CheckpointStoreShape { ) => Effect.Effect; /** - * Compute patch diff between two checkpoint turn counts. + * Compute patch diff between two checkpoint refs. */ - readonly diffCheckpoints: (input: DiffCheckpointsInput) => Effect.Effect; + readonly diffCheckpoints: ( + input: DiffCheckpointsInput, + ) => Effect.Effect; /** - * Delete checkpoints newer than the provided turn count. + * Delete the provided checkpoint refs. */ - readonly pruneAfterTurn: (input: PruneAfterTurnInput) => Effect.Effect; + readonly deleteCheckpointRefs: ( + input: DeleteCheckpointRefsInput, + ) => Effect.Effect; } /** * CheckpointStore - Service tag for checkpoint persistence and restore operations. */ -export class CheckpointStore extends ServiceMap.Service< - CheckpointStore, - CheckpointStoreShape ->()("checkpointing/CheckpointStore") {} +export class CheckpointStore extends ServiceMap.Service()( + "checkpointing/CheckpointStore", +) {} diff --git a/apps/server/src/git/Errors.ts b/apps/server/src/git/Errors.ts new file mode 100644 index 0000000000000000000000000000000000000000..9e2ee14f72826d8c5fa915aa37d30900e634cf5e --- /dev/null +++ b/apps/server/src/git/Errors.ts @@ -0,0 +1,16 @@ +import { Schema } from "effect"; + +/** + * GitCommandError - Git command execution failed. + */ +export class GitCommandError extends Schema.TaggedErrorClass()("GitCommandError", { + operation: Schema.String, + command: Schema.String, + cwd: Schema.String, + detail: Schema.String, + cause: Schema.optional(Schema.Defect), +}) { + override get message(): string { + return `Git command failed in ${this.operation}: ${this.command} (${this.cwd}) - ${this.detail}`; + } +} diff --git a/apps/server/src/git/process.test.ts b/apps/server/src/git/process.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..58242a4e7a561c68abe8b3b12aeba4f5b6e9cb43 --- /dev/null +++ b/apps/server/src/git/process.test.ts @@ -0,0 +1,51 @@ +import { NodeServices } from "@effect/platform-node"; +import { it, assert } from "@effect/vitest"; +import { Effect } from "effect"; + +import { GitCommandError } from "./Errors.ts"; +import { runGit } from "./Process.ts"; + +it.effect("runGit executes successful git commands", () => + Effect.gen(function* () { + const result = yield* runGit({ + operation: "GitProcess.test.version", + cwd: process.cwd(), + args: ["--version"], + }); + + assert.equal(result.code, 0); + assert.ok(result.stdout.toLowerCase().includes("git version")); + }).pipe(Effect.provide(NodeServices.layer)), +); + +it.effect("runGit can return non-zero exit codes when allowed", () => + Effect.gen(function* () { + const result = yield* runGit({ + operation: "GitProcess.test.allowNonZero", + cwd: process.cwd(), + args: ["rev-parse", "--verify", "__definitely_missing_ref__"], + allowNonZeroExit: true, + }); + + assert.notEqual(result.code, 0); + }).pipe(Effect.provide(NodeServices.layer)), +); + +it.effect("runGit fails with GitCommandError when non-zero exits are not allowed", () => + Effect.gen(function* () { + const result = yield* Effect.result( + runGit({ + operation: "GitProcess.test.failOnNonZero", + cwd: process.cwd(), + args: ["rev-parse", "--verify", "__definitely_missing_ref__"], + }), + ); + + assert.equal(result._tag, "Failure"); + if (result._tag === "Failure") { + assert.ok(result.failure instanceof GitCommandError); + assert.equal(result.failure.operation, "GitProcess.test.failOnNonZero"); + assert.equal(result.failure.command, "git rev-parse --verify __definitely_missing_ref__"); + } + }).pipe(Effect.provide(NodeServices.layer)), +); diff --git a/apps/server/src/git/process.ts b/apps/server/src/git/process.ts new file mode 100644 index 0000000000000000000000000000000000000000..d5ed04de081c951522d578eb9172756ad0a1d0d6 --- /dev/null +++ b/apps/server/src/git/process.ts @@ -0,0 +1,174 @@ +/** + * Git process helpers - Effect-native git execution with typed errors. + * + * Centralizes child-process git invocation for server modules. This module + * only executes git commands and reports structured failures. + * + * @module GitProcess + */ +import { Effect, Option, Stream } from "effect"; +import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; +import { GitCommandError } from "./Errors.ts"; + +const DEFAULT_TIMEOUT_MS = 30_000; +const DEFAULT_MAX_OUTPUT_BYTES = 1_000_000; + +export interface RunGitInput { + readonly operation: string; + readonly cwd: string; + readonly args: ReadonlyArray; + readonly env?: NodeJS.ProcessEnv; + readonly allowNonZeroExit?: boolean; + readonly timeoutMs?: number; + readonly maxOutputBytes?: number; +} + +export interface GitRunResult { + readonly code: number; + readonly stdout: string; + readonly stderr: string; +} + +function quoteGitCommand(args: ReadonlyArray): string { + return `git ${args.join(" ")}`; +} + +function causeMessage(cause: unknown, fallback: string): string { + if (cause instanceof Error && cause.message.length > 0) { + return cause.message; + } + return fallback; +} + +function gitError( + input: Pick, + detail: string, + cause?: unknown, +): GitCommandError { + return new GitCommandError({ + operation: input.operation, + command: quoteGitCommand(input.args), + cwd: input.cwd, + detail, + ...(cause !== undefined ? { cause } : {}), + }); +} + +function collectOutput( + input: Pick, + stream: Stream.Stream, + maxOutputBytes: number, +): Effect.Effect { + return Effect.gen(function* () { + const decoder = new TextDecoder(); + let bytes = 0; + let text = ""; + + yield* Stream.runForEach(stream, (chunk) => + Effect.gen(function* () { + bytes += chunk.byteLength; + if (bytes > maxOutputBytes) { + return yield* Effect.fail( + gitError( + input, + `${quoteGitCommand(input.args)} output exceeded ${maxOutputBytes} bytes and was truncated.`, + ), + ); + } + text += decoder.decode(chunk, { stream: true }); + }), + ); + + text += decoder.decode(); + return text; + }).pipe( + Effect.mapError((cause) => + gitError( + input, + causeMessage(cause, `${quoteGitCommand(input.args)} output stream failed.`), + cause, + ), + ), + ); +} + +export function runGit( + input: RunGitInput, +): Effect.Effect { + const normalizedCwd = input.cwd.trim(); + const commandInput = { + ...input, + cwd: normalizedCwd, + args: [...input.args], + } as const; + const timeoutMs = input.timeoutMs ?? DEFAULT_TIMEOUT_MS; + const maxOutputBytes = input.maxOutputBytes ?? DEFAULT_MAX_OUTPUT_BYTES; + + const commandEffect = Effect.gen(function* () { + const child = yield* ChildProcess.spawn( + ChildProcess.make("git", commandInput.args, { + cwd: commandInput.cwd, + ...(input.env ? { env: input.env } : {}), + }), + ).pipe( + Effect.mapError((cause) => + gitError( + commandInput, + causeMessage(cause, `Failed to spawn ${quoteGitCommand(commandInput.args)}.`), + cause, + ), + ), + ); + + const [stdout, stderr, exitCode] = yield* Effect.all( + [ + collectOutput(commandInput, child.stdout, maxOutputBytes), + collectOutput(commandInput, child.stderr, maxOutputBytes), + child.exitCode.pipe( + Effect.map((value) => Number(value)), + Effect.mapError((cause) => + gitError( + commandInput, + causeMessage( + cause, + `${quoteGitCommand(commandInput.args)} failed to report exit code.`, + ), + cause, + ), + ), + ), + ], + { concurrency: "unbounded" }, + ); + + if (!input.allowNonZeroExit && exitCode !== 0) { + const trimmedStderr = stderr.trim(); + return yield* Effect.fail( + gitError( + commandInput, + trimmedStderr.length > 0 + ? `${quoteGitCommand(commandInput.args)} failed: ${trimmedStderr}` + : `${quoteGitCommand(commandInput.args)} failed with code ${exitCode}.`, + ), + ); + } + + return { + code: exitCode, + stdout, + stderr, + } satisfies GitRunResult; + }); + + return commandEffect.pipe( + Effect.scoped, + Effect.timeoutOption(timeoutMs), + Effect.flatMap((result) => + Option.match(result, { + onNone: () => + Effect.fail(gitError(commandInput, `${quoteGitCommand(commandInput.args)} timed out.`)), + onSome: Effect.succeed, + }), + ), + ); +} diff --git a/apps/server/src/persistence/Errors.ts b/apps/server/src/persistence/Errors.ts index d6363f6abeb7f6b64eca519a70c862ce8a809f81..75f23a89e10f2eb6573ab3c70c6bb9cfbb4e2b00 100644 --- a/apps/server/src/persistence/Errors.ts +++ b/apps/server/src/persistence/Errors.ts @@ -1,5 +1,9 @@ import { Schema, SchemaIssue } from "effect"; +// =============================== +// Core Persistence Errors +// =============================== + export class PersistenceSqlError extends Schema.TaggedErrorClass()( "PersistenceSqlError", { @@ -39,6 +43,46 @@ export class PersistenceSerializationError extends Schema.TaggedErrorClass + new PersistenceSqlError({ + operation, + detail: `Failed to execute ${operation}`, + cause, + }); +} + +export function toPersistenceDecodeError(operation: string) { + return (error: Schema.SchemaError): PersistenceDecodeError => + new PersistenceDecodeError({ + operation, + issue: SchemaIssue.makeFormatterDefault()(error.issue), + cause: error, + }); +} + +export function toPersistenceDecodeCauseError(operation: string) { + return (cause: unknown): PersistenceDecodeError => + new PersistenceDecodeError({ + operation, + issue: `Failed to execute ${operation}`, + cause, + }); +} + +export function toPersistenceSerializationError(operation: string) { + return (cause: unknown): PersistenceSerializationError => + new PersistenceSerializationError({ + operation, + detail: `Failed to execute ${operation}`, + cause, + }); +} + +// =============================== +// Project Repository Errors +// =============================== + export class ProjectValidationError extends Schema.TaggedErrorClass()( "ProjectValidationError", { @@ -76,6 +120,36 @@ export class ProjectNotFoundError extends Schema.TaggedErrorClass()( + "CheckpointRepositoryValidationError", + { + operation: Schema.String, + issue: Schema.String, + cause: Schema.optional(Schema.Defect), + }, +) { + override get message(): string { + return `Checkpoint repository validation failed in ${this.operation}: ${this.issue}`; + } +} + +export class CheckpointRepositoryPersistenceError extends Schema.TaggedErrorClass()( + "CheckpointRepositoryPersistenceError", + { + operation: Schema.String, + detail: Schema.String, + cause: Schema.optional(Schema.Defect), + }, +) { + override get message(): string { + return `Checkpoint repository persistence error in ${this.operation}: ${this.detail}`; + } +} + export type ProjectRepositoryError = | PersistenceSqlError | PersistenceDecodeError @@ -89,38 +163,6 @@ export type OrchestrationEventRepositoryError = | PersistenceDecodeError | PersistenceSerializationError; -export function toPersistenceSqlError(operation: string) { - return (cause: unknown): PersistenceSqlError => - new PersistenceSqlError({ - operation, - detail: `Failed to execute ${operation}`, - cause, - }); -} - -export function toPersistenceDecodeError(operation: string) { - return (error: Schema.SchemaError): PersistenceDecodeError => - new PersistenceDecodeError({ - operation, - issue: SchemaIssue.makeFormatterDefault()(error.issue), - cause: error, - }); -} - -export function toPersistenceDecodeCauseError(operation: string) { - return (cause: unknown): PersistenceDecodeError => - new PersistenceDecodeError({ - operation, - issue: `Failed to execute ${operation}`, - cause, - }); -} - -export function toPersistenceSerializationError(operation: string) { - return (cause: unknown): PersistenceSerializationError => - new PersistenceSerializationError({ - operation, - detail: `Failed to execute ${operation}`, - cause, - }); -} +export type CheckpointRepositoryError = + | CheckpointRepositoryValidationError + | CheckpointRepositoryPersistenceError; diff --git a/apps/server/src/persistence/Layers/Checkpoints.test.ts b/apps/server/src/persistence/Layers/Checkpoints.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..d13b5c14f8b4780c8e72e8fb216e5b315127d00f --- /dev/null +++ b/apps/server/src/persistence/Layers/Checkpoints.test.ts @@ -0,0 +1,95 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; + +import { it, assert } from "@effect/vitest"; +import { Effect, Layer, Option } from "effect"; + +import { CheckpointRepository } from "../Services/Checkpoints.ts"; +import { CheckpointRepositoryLive } from "./Checkpoints.ts"; +import { makeSqlitePersistenceLive } from "./Sqlite.ts"; + +const makeTemporaryDirectoryScoped = (prefix: string) => + Effect.acquireRelease( + Effect.promise(() => mkdtemp(path.join(os.tmpdir(), prefix))), + (directory) => Effect.promise(() => rm(directory, { recursive: true, force: true })), + ); + +function makeCheckpointRepositoryLayer(dbPath: string) { + return CheckpointRepositoryLive.pipe(Layer.provide(makeSqlitePersistenceLive(dbPath))); +} + +it.effect("CheckpointRepositoryLive upserts, lists, reads, and prunes checkpoint metadata", () => + Effect.gen(function* () { + const stateDir = yield* makeTemporaryDirectoryScoped("t3code-checkpoint-repository-"); + const dbPath = path.join(stateDir, "orchestration.sqlite"); + + yield* Effect.gen(function* () { + const repository = yield* CheckpointRepository; + + yield* repository.upsertCheckpoint({ + providerSessionId: "sess-1", + threadId: "thread-1", + checkpointId: "root", + checkpointRef: "refs/t3/checkpoints/thread-1/turn/0", + turnCount: 0, + messageCount: 0, + label: "Start of conversation", + createdAt: new Date().toISOString(), + }); + + yield* repository.upsertCheckpoint({ + providerSessionId: "sess-1", + threadId: "thread-1", + checkpointId: "turn-1", + checkpointRef: "refs/t3/checkpoints/thread-1/turn/1", + turnCount: 1, + messageCount: 2, + label: "Turn 1", + preview: "hello", + createdAt: new Date().toISOString(), + }); + + yield* repository.upsertCheckpoint({ + providerSessionId: "sess-1", + threadId: "thread-1", + checkpointId: "turn-2", + checkpointRef: "refs/t3/checkpoints/thread-1/turn/2", + turnCount: 2, + messageCount: 4, + label: "Turn 2", + preview: "done", + createdAt: new Date().toISOString(), + }); + + const checkpoints = yield* repository.listCheckpoints({ providerSessionId: "sess-1" }); + assert.equal(checkpoints.length, 3); + assert.equal(checkpoints[0]?.id, "root"); + assert.equal(checkpoints[0]?.isCurrent, false); + assert.equal(checkpoints[2]?.id, "turn-2"); + assert.equal(checkpoints[2]?.isCurrent, true); + + const one = yield* repository.getCheckpoint({ providerSessionId: "sess-1", turnCount: 1 }); + assert.equal(Option.isSome(one), true); + if (Option.isSome(one)) { + assert.equal(one.value.id, "turn-1"); + assert.equal(one.value.isCurrent, false); + assert.equal(one.value.checkpointRef, "refs/t3/checkpoints/thread-1/turn/1"); + } + + yield* repository.deleteAfterTurn({ + providerSessionId: "sess-1", + maxTurnCount: 1, + }); + + const pruned = yield* repository.listCheckpoints({ providerSessionId: "sess-1" }); + assert.equal(pruned.length, 2); + assert.equal(pruned[1]?.turnCount, 1); + assert.equal(pruned[1]?.isCurrent, true); + + yield* repository.deleteAllForSession({ providerSessionId: "sess-1" }); + const empty = yield* repository.listCheckpoints({ providerSessionId: "sess-1" }); + assert.equal(empty.length, 0); + }).pipe(Effect.provide(makeCheckpointRepositoryLayer(dbPath))); + }), +); diff --git a/apps/server/src/persistence/Layers/Checkpoints.ts b/apps/server/src/persistence/Layers/Checkpoints.ts new file mode 100644 index 0000000000000000000000000000000000000000..b9efe5b9b0b3f46b685b0800f589b6a37d0ae317 --- /dev/null +++ b/apps/server/src/persistence/Layers/Checkpoints.ts @@ -0,0 +1,363 @@ +/** + * CheckpointRepositoryLive - SQLite-backed checkpoint metadata repository layer. + * + * Persists and queries checkpoint metadata (`turnCount`, labels, previews, + * message counts) and the associated hidden Git ref for each turn. + * + * This layer does not read/write Git refs and does not perform provider-thread + * rollback. Those responsibilities remain in `CheckpointStore` and + * `CheckpointService`. + * + * @module CheckpointRepositoryLive + */ +import type { ProviderCheckpoint } from "@t3tools/contracts"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as SqlSchema from "effect/unstable/sql/SqlSchema"; +import { Effect, Layer, Option, Schema } from "effect"; + +import { + CheckpointRepositoryPersistenceError, + CheckpointRepositoryValidationError, + type CheckpointRepositoryError, +} from "../Errors.ts"; +import { makeSqlitePersistenceLive } from "./Sqlite.ts"; +import { + CheckpointRepository, + type CheckpointRepositoryEntry, + type CheckpointRepositoryShape, +} from "../Services/Checkpoints.ts"; + +const NonEmptyString = Schema.NonEmptyString; + +const CheckpointRowSchema = Schema.Struct({ + providerSessionId: Schema.String, + threadId: Schema.String, + checkpointId: Schema.String, + checkpointRef: Schema.String, + turnCount: Schema.Number, + messageCount: Schema.Number, + label: Schema.String, + preview: Schema.NullOr(Schema.String), + createdAt: Schema.String, +}); + +const SessionRequestSchema = Schema.Struct({ + providerSessionId: NonEmptyString, +}); + +const UpsertRequestSchema = Schema.Struct({ + providerSessionId: NonEmptyString, + threadId: NonEmptyString, + checkpointId: NonEmptyString, + checkpointRef: NonEmptyString, + turnCount: Schema.Int, + messageCount: Schema.Int, + label: NonEmptyString, + preview: Schema.NullOr(Schema.String), + createdAt: NonEmptyString, +}); + +const GetCheckpointRequestSchema = Schema.Struct({ + providerSessionId: NonEmptyString, + turnCount: Schema.Int, +}); + +const DeleteAfterTurnRequestSchema = Schema.Struct({ + providerSessionId: NonEmptyString, + maxTurnCount: Schema.Int, +}); + +function errorMessage(cause: unknown, fallback: string): string { + if (cause instanceof Error && cause.message.length > 0) { + return cause.message; + } + return fallback; +} + +function toValidationError(operation: string, cause: unknown): CheckpointRepositoryValidationError { + return new CheckpointRepositoryValidationError({ + operation, + issue: errorMessage(cause, "Invalid checkpoint repository input."), + cause, + }); +} + +function decodeInput( + schema: Schema.Schema, + input: unknown, + operation: string, +): Effect.Effect { + return Schema.decodeUnknownEffect(schema)(input).pipe( + Effect.mapError((cause) => toValidationError(operation, cause)), + ) as Effect.Effect; +} + +function ensureNonNegativeInt( + operation: string, + field: string, + value: number, +): Effect.Effect { + if (value < 0) { + return Effect.fail( + new CheckpointRepositoryValidationError({ + operation, + issue: `${field} must be an integer >= 0.`, + }), + ); + } + return Effect.succeed(value); +} + +function toPersistenceError( + operation: string, + cause: unknown, +): CheckpointRepositoryPersistenceError { + return new CheckpointRepositoryPersistenceError({ + operation, + detail: `Failed to execute ${operation}.`, + cause, + }); +} + +function toCheckpointEntry( + row: Schema.Schema.Type, + isCurrent: boolean, +): CheckpointRepositoryEntry { + return { + id: row.checkpointId, + threadId: row.threadId, + checkpointRef: row.checkpointRef, + turnCount: row.turnCount, + messageCount: row.messageCount, + label: row.label, + ...(row.preview !== null ? { preview: row.preview } : {}), + createdAt: row.createdAt, + isCurrent, + }; +} + +function toProviderCheckpoint(entry: CheckpointRepositoryEntry): ProviderCheckpoint { + return { + id: entry.id, + turnCount: entry.turnCount, + messageCount: entry.messageCount, + label: entry.label, + ...(entry.preview !== undefined ? { preview: entry.preview } : {}), + isCurrent: entry.isCurrent, + }; +} + +const makeCheckpointRepository = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + const upsertCheckpointRow = SqlSchema.findOne({ + Request: UpsertRequestSchema, + Result: CheckpointRowSchema, + execute: (request) => + sql` + INSERT INTO provider_checkpoints ( + provider_session_id, + thread_id, + checkpoint_id, + checkpoint_ref, + turn_count, + message_count, + label, + preview, + created_at + ) + VALUES ( + ${request.providerSessionId}, + ${request.threadId}, + ${request.checkpointId}, + ${request.checkpointRef}, + ${request.turnCount}, + ${request.messageCount}, + ${request.label}, + ${request.preview}, + ${request.createdAt} + ) + ON CONFLICT (provider_session_id, turn_count) + DO UPDATE SET + thread_id = excluded.thread_id, + checkpoint_id = excluded.checkpoint_id, + checkpoint_ref = excluded.checkpoint_ref, + message_count = excluded.message_count, + label = excluded.label, + preview = excluded.preview, + created_at = excluded.created_at + RETURNING + provider_session_id AS "providerSessionId", + thread_id AS "threadId", + checkpoint_id AS "checkpointId", + checkpoint_ref AS "checkpointRef", + turn_count AS "turnCount", + message_count AS "messageCount", + label, + preview, + created_at AS "createdAt" + `, + }); + + const listCheckpointRows = SqlSchema.findAll({ + Request: SessionRequestSchema, + Result: CheckpointRowSchema, + execute: ({ providerSessionId }) => + sql` + SELECT + provider_session_id AS "providerSessionId", + thread_id AS "threadId", + checkpoint_id AS "checkpointId", + checkpoint_ref AS "checkpointRef", + turn_count AS "turnCount", + message_count AS "messageCount", + label, + preview, + created_at AS "createdAt" + FROM provider_checkpoints + WHERE provider_session_id = ${providerSessionId} + ORDER BY turn_count ASC + `, + }); + + const listForSession = ( + providerSessionId: string, + ): Effect.Effect, CheckpointRepositoryError> => + listCheckpointRows({ providerSessionId }).pipe( + Effect.mapError((cause) => + toPersistenceError("CheckpointRepository.listCheckpoints:query", cause), + ), + Effect.map((rows) => { + const currentTurnCount = rows[rows.length - 1]?.turnCount; + return rows.map((row) => toCheckpointEntry(row, row.turnCount === currentTurnCount)); + }), + ); + + const upsertCheckpoint: CheckpointRepositoryShape["upsertCheckpoint"] = (input) => + Effect.gen(function* () { + const operation = "CheckpointRepository.upsertCheckpoint"; + const parsed = yield* decodeInput( + UpsertRequestSchema, + { + providerSessionId: input.providerSessionId.trim(), + threadId: input.threadId.trim(), + checkpointId: input.checkpointId.trim(), + checkpointRef: input.checkpointRef.trim(), + turnCount: input.turnCount, + messageCount: input.messageCount, + label: input.label.trim(), + preview: input.preview?.trim() || null, + createdAt: input.createdAt.trim(), + }, + operation, + ); + yield* ensureNonNegativeInt(operation, "turnCount", parsed.turnCount); + yield* ensureNonNegativeInt(operation, "messageCount", parsed.messageCount); + + yield* upsertCheckpointRow(parsed).pipe( + Effect.mapError((cause) => + toPersistenceError("CheckpointRepository.upsertCheckpoint:query", cause), + ), + Effect.asVoid, + ); + }); + + const listCheckpoints: CheckpointRepositoryShape["listCheckpoints"] = (input) => + Effect.gen(function* () { + const parsed = yield* decodeInput( + SessionRequestSchema, + { + providerSessionId: input.providerSessionId.trim(), + }, + "CheckpointRepository.listCheckpoints", + ); + + const entries = yield* listForSession(parsed.providerSessionId); + return entries.map(toProviderCheckpoint); + }); + + const getCheckpoint: CheckpointRepositoryShape["getCheckpoint"] = (input) => + Effect.gen(function* () { + const parsed = yield* decodeInput( + GetCheckpointRequestSchema, + { + providerSessionId: input.providerSessionId.trim(), + turnCount: input.turnCount, + }, + "CheckpointRepository.getCheckpoint", + ); + yield* ensureNonNegativeInt( + "CheckpointRepository.getCheckpoint", + "turnCount", + parsed.turnCount, + ); + + const entries = yield* listForSession(parsed.providerSessionId); + const checkpoint = entries.find((entry) => entry.turnCount === parsed.turnCount); + return Option.fromNullishOr(checkpoint); + }); + + const deleteAfterTurn: CheckpointRepositoryShape["deleteAfterTurn"] = (input) => + Effect.gen(function* () { + const parsed = yield* decodeInput( + DeleteAfterTurnRequestSchema, + { + providerSessionId: input.providerSessionId.trim(), + maxTurnCount: input.maxTurnCount, + }, + "CheckpointRepository.deleteAfterTurn", + ); + yield* ensureNonNegativeInt( + "CheckpointRepository.deleteAfterTurn", + "maxTurnCount", + parsed.maxTurnCount, + ); + + yield* sql` + DELETE FROM provider_checkpoints + WHERE provider_session_id = ${parsed.providerSessionId} + AND turn_count > ${parsed.maxTurnCount} + `.pipe( + Effect.mapError((cause) => + toPersistenceError("CheckpointRepository.deleteAfterTurn:query", cause), + ), + ); + }); + + const deleteAllForSession: CheckpointRepositoryShape["deleteAllForSession"] = (input) => + Effect.gen(function* () { + const parsed = yield* decodeInput( + SessionRequestSchema, + { + providerSessionId: input.providerSessionId.trim(), + }, + "CheckpointRepository.deleteAllForSession", + ); + + yield* sql` + DELETE FROM provider_checkpoints + WHERE provider_session_id = ${parsed.providerSessionId} + `.pipe( + Effect.mapError((cause) => + toPersistenceError("CheckpointRepository.deleteAllForSession:query", cause), + ), + ); + }); + + return { + upsertCheckpoint, + listCheckpoints, + getCheckpoint, + deleteAfterTurn, + deleteAllForSession, + } satisfies CheckpointRepositoryShape; +}); + +export const CheckpointRepositoryLive = Layer.effect( + CheckpointRepository, + makeCheckpointRepository, +); + +export function makeSqliteCheckpointRepositoryLive(dbPath: string) { + return CheckpointRepositoryLive.pipe(Layer.provide(makeSqlitePersistenceLive(dbPath))); +} diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index d289361f913252c4d849690f265e39847f8443cf..46dba5ae2deb1324a971e8ab2ba65cfce7364b97 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -14,6 +14,7 @@ import * as Layer from "effect/Layer"; // Import all migrations statically import Migration0001 from "./Migrations/001_OrchestrationEvents.ts"; import Migration0002 from "./Migrations/002_Projects.ts"; +import Migration0003 from "./Migrations/003_Checkpoints.ts"; /** * Migration loader with all migrations defined inline. @@ -28,6 +29,7 @@ import Migration0002 from "./Migrations/002_Projects.ts"; const loader = Migrator.fromRecord({ "1_OrchestrationEvents": Migration0001, "2_Projects": Migration0002, + "3_CheckpointCatalog": Migration0003, }); /** diff --git a/apps/server/src/persistence/Migrations/003_Checkpoints.ts b/apps/server/src/persistence/Migrations/003_Checkpoints.ts new file mode 100644 index 0000000000000000000000000000000000000000..c7dc9ea3c7eaf4a7587404693c78fc9736f3d346 --- /dev/null +++ b/apps/server/src/persistence/Migrations/003_Checkpoints.ts @@ -0,0 +1,26 @@ +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as Effect from "effect/Effect"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE TABLE IF NOT EXISTS provider_checkpoints ( + provider_session_id TEXT NOT NULL, + thread_id TEXT NOT NULL, + checkpoint_id TEXT NOT NULL, + checkpoint_ref TEXT NOT NULL, + turn_count INTEGER NOT NULL, + message_count INTEGER NOT NULL, + label TEXT NOT NULL, + preview TEXT, + created_at TEXT NOT NULL, + PRIMARY KEY (provider_session_id, turn_count) + ) + `; + + yield* sql` + CREATE INDEX IF NOT EXISTS idx_provider_checkpoints_session_turn + ON provider_checkpoints(provider_session_id, turn_count ASC) + `; +}); diff --git a/apps/server/src/persistence/Services/Checkpoints.ts b/apps/server/src/persistence/Services/Checkpoints.ts new file mode 100644 index 0000000000000000000000000000000000000000..954d1c8b74c04ba12415196bac14cce1f2ad3c92 --- /dev/null +++ b/apps/server/src/persistence/Services/Checkpoints.ts @@ -0,0 +1,95 @@ +/** + * CheckpointRepository - Repository interface for checkpoint metadata persistence. + * + * Stores and queries checkpoint metadata and checkpoint Git refs keyed by + * provider session + turn count. It does not read/write Git refs itself and it + * does not coordinate provider rollback workflows. + * + * @module CheckpointRepository + */ +import type { ProviderCheckpoint } from "@t3tools/contracts"; +import { Option, ServiceMap } from "effect"; +import type { Effect } from "effect"; + +import type { CheckpointRepositoryError } from "../Errors.ts"; + +export interface UpsertCheckpointInput { + readonly providerSessionId: string; + readonly threadId: string; + readonly checkpointId: string; + readonly checkpointRef: string; + readonly turnCount: number; + readonly messageCount: number; + readonly label: string; + readonly preview?: string; + readonly createdAt: string; +} + +export interface CheckpointRepositoryEntry extends ProviderCheckpoint { + readonly threadId: string; + readonly checkpointRef: string; + readonly createdAt: string; +} + +export interface ListSessionCheckpointsInput { + readonly providerSessionId: string; +} + +export interface GetCheckpointInput { + readonly providerSessionId: string; + readonly turnCount: number; +} + +export interface DeleteAfterTurnInput { + readonly providerSessionId: string; + readonly maxTurnCount: number; +} + +export interface DeleteAllForSessionInput { + readonly providerSessionId: string; +} + +export interface CheckpointRepositoryShape { + /** + * Insert or update one checkpoint metadata row. + */ + readonly upsertCheckpoint: ( + input: UpsertCheckpointInput, + ) => Effect.Effect; + + /** + * List user-facing checkpoints for a provider session. + */ + readonly listCheckpoints: ( + input: ListSessionCheckpointsInput, + ) => Effect.Effect, CheckpointRepositoryError>; + + /** + * Read one checkpoint record (including checkpoint ref) by turn count. + */ + readonly getCheckpoint: ( + input: GetCheckpointInput, + ) => Effect.Effect, CheckpointRepositoryError>; + + /** + * Delete checkpoint metadata newer than the provided turn count. + */ + readonly deleteAfterTurn: ( + input: DeleteAfterTurnInput, + ) => Effect.Effect; + + /** + * Delete all checkpoint metadata for a provider session. + */ + readonly deleteAllForSession: ( + input: DeleteAllForSessionInput, + ) => Effect.Effect; +} + +/** + * CheckpointRepository - Service tag for checkpoint metadata persistence. + */ +export class CheckpointRepository extends ServiceMap.Service< + CheckpointRepository, + CheckpointRepositoryShape +>()("persistence/CheckpointRepository") {} diff --git a/apps/server/src/provider/contracts.test.ts b/apps/server/src/provider/contracts.test.ts index 73756e09cedc9ae74022c4a2c706ed0e847a35b0..7f216b1c08c9c89b9d2dcbfac0368ab506761ff5 100644 --- a/apps/server/src/provider/contracts.test.ts +++ b/apps/server/src/provider/contracts.test.ts @@ -11,14 +11,18 @@ import { type ProviderServiceError, } from "./Errors.ts"; import { - CheckpointValidationError, - type CheckpointCatalogError, + CheckpointGitCommandError, + CheckpointServiceValidationError, type CheckpointServiceError, type CheckpointStoreError, } from "../checkpointing/Errors.ts"; -import { CheckpointCatalog } from "../checkpointing/Services/CheckpointCatalog.ts"; +import { + CheckpointRepositoryValidationError, + type CheckpointRepositoryError, +} from "../persistence/Errors.ts"; import { CheckpointService } from "../checkpointing/Services/CheckpointService.ts"; import { CheckpointStore } from "../checkpointing/Services/CheckpointStore.ts"; +import { CheckpointRepository } from "../persistence/Services/Checkpoints.ts"; import { CodexAdapter } from "./Services/CodexAdapter.ts"; import { ProviderAdapterRegistry } from "./Services/ProviderAdapterRegistry.ts"; import { ProviderService } from "./Services/ProviderService.ts"; @@ -107,20 +111,22 @@ describe("provider/checkpoint service contracts", () => { const checkpointStoreLive: typeof CheckpointStore.Service = { isGitRepository: () => Effect.succeed(false), captureCheckpoint: () => Effect.void, - hasCheckpoint: () => Effect.succeed(false), + hasCheckpointRef: () => Effect.succeed(false), ensureRootCheckpoint: () => Effect.succeed(false), restoreCheckpoint: () => Effect.succeed(false), diffCheckpoints: () => Effect.fail( - new CheckpointValidationError({ + new CheckpointGitCommandError({ operation: "diffCheckpoints", - issue: "not implemented", + command: "git diff", + cwd: "/tmp", + detail: "not implemented", }), ), - pruneAfterTurn: () => Effect.void, + deleteCheckpointRefs: () => Effect.void, }; - const checkpointCatalogLive: typeof CheckpointCatalog.Service = { + const checkpointRepositoryLive: typeof CheckpointRepository.Service = { upsertCheckpoint: () => Effect.void, listCheckpoints: () => Effect.succeed([]), getCheckpoint: () => Effect.succeed(Option.none()), @@ -133,18 +139,21 @@ describe("provider/checkpoint service contracts", () => { captureCurrentTurn: () => Effect.void, listCheckpoints: () => Effect.fail( - new CheckpointValidationError({ operation: "listCheckpoints", issue: "not implemented" }), + new CheckpointServiceValidationError({ + operation: "listCheckpoints", + issue: "not implemented", + }), ), getCheckpointDiff: () => Effect.fail( - new CheckpointValidationError({ + new CheckpointServiceValidationError({ operation: "getCheckpointDiff", issue: "not implemented", }), ), revertToCheckpoint: () => Effect.fail( - new CheckpointValidationError({ + new CheckpointServiceValidationError({ operation: "revertToCheckpoint", issue: "not implemented", }), @@ -174,7 +183,7 @@ describe("provider/checkpoint service contracts", () => { Layer.succeed(ProviderService, providerLive), Layer.succeed(CodexAdapter, codexLive), Layer.succeed(CheckpointStore, checkpointStoreLive), - Layer.succeed(CheckpointCatalog, checkpointCatalogLive), + Layer.succeed(CheckpointRepository, checkpointRepositoryLive), Layer.succeed(CheckpointService, checkpointServiceLive), Layer.succeed(ProviderAdapterRegistry, providerAdapterRegistryLive), Layer.succeed(ProviderSessionDirectory, providerSessionDirectoryLive), @@ -185,7 +194,7 @@ describe("provider/checkpoint service contracts", () => { provider, codex, checkpointStore, - checkpointCatalog, + checkpointRepository, checkpointService, providerAdapterRegistry, providerSessionDirectory, @@ -193,7 +202,7 @@ describe("provider/checkpoint service contracts", () => { runtime.runPromise(Effect.service(ProviderService)), runtime.runPromise(Effect.service(CodexAdapter)), runtime.runPromise(Effect.service(CheckpointStore)), - runtime.runPromise(Effect.service(CheckpointCatalog)), + runtime.runPromise(Effect.service(CheckpointRepository)), runtime.runPromise(Effect.service(CheckpointService)), runtime.runPromise(Effect.service(ProviderAdapterRegistry)), runtime.runPromise(Effect.service(ProviderSessionDirectory)), @@ -202,7 +211,7 @@ describe("provider/checkpoint service contracts", () => { expect(provider).toBe(providerLive); expect(codex).toBe(codexLive); expect(checkpointStore).toBe(checkpointStoreLive); - expect(checkpointCatalog).toBe(checkpointCatalogLive); + expect(checkpointRepository).toBe(checkpointRepositoryLive); expect(checkpointService).toBe(checkpointServiceLive); expect(providerAdapterRegistry).toBe(providerAdapterRegistryLive); expect(providerSessionDirectory).toBe(providerSessionDirectoryLive); @@ -225,12 +234,12 @@ describe("provider/checkpoint errors", () => { expect(fsError.cause).toBe(fsCause); }); - it("accepts checkpoint catalog error channel type", () => { - const error: CheckpointCatalogError = new CheckpointValidationError({ + it("accepts checkpoint repository error channel type", () => { + const error: CheckpointRepositoryError = new CheckpointRepositoryValidationError({ operation: "upsertCheckpoint", issue: "turnCount must be >= 0", }); - expect(error._tag).toBe("CheckpointValidationError"); + expect(error._tag).toBe("CheckpointRepositoryValidationError"); }); });