import fs from "node:fs";import os from "node:os";import path from "node:path";import { execFileSync } from "node:child_process"; import type { ProviderKind, ProviderRuntimeEvent, ProviderSession } from "@t3tools/contracts";import { CommandId, DEFAULT_PROVIDER_INTERACTION_MODE, EventId, MessageId, ProjectId, ThreadId, TurnId,} from "@t3tools/contracts";import * as NodeServices from "@effect/platform-node/NodeServices";import { Effect, Exit, Layer, ManagedRuntime, PubSub, Scope, Stream } from "effect";import { afterEach, describe, expect, it, vi } from "vitest"; import { CheckpointStoreLive } from "../../checkpointing/Layers/CheckpointStore.ts";import { CheckpointStore } from "../../checkpointing/Services/CheckpointStore.ts";import { GitCoreLive } from "../../git/Layers/GitCore.ts";import { CheckpointReactorLive } from "./CheckpointReactor.ts";import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";import { RuntimeReceiptBusLive } from "./RuntimeReceiptBus.ts";import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts";import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";import { OrchestrationEngineService, type OrchestrationEngineShape,} from "../Services/OrchestrationEngine.ts";import { CheckpointReactor } from "../Services/CheckpointReactor.ts";import { ProviderService, type ProviderServiceShape,} from "../../provider/Services/ProviderService.ts";import { checkpointRefForThreadTurn } from "../../checkpointing/Utils.ts";import { ServerConfig } from "../../config.ts";import { WorkspaceEntriesLive } from "../../workspace/Layers/WorkspaceEntries.ts";import { WorkspacePathsLive } from "../../workspace/Layers/WorkspacePaths.ts"; const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); type LegacyProviderRuntimeEvent = { readonly type: string; readonly eventId: EventId; readonly provider: ProviderKind; readonly createdAt: string; readonly threadId: ThreadId; readonly turnId?: string | undefined; readonly itemId?: string | undefined; readonly requestId?: string | undefined; readonly payload?: unknown | undefined; readonly [key: string]: unknown;}; function createProviderServiceHarness( cwd: string, hasSession = true, sessionCwd = cwd, providerName: ProviderSession["provider"] = "codex",) { const now = new Date().toISOString(); const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>()); const rollbackConversation = vi.fn( (_input: { readonly threadId: ThreadId; readonly numTurns: number }) => Effect.void, ); const unsupported = <A>() => Effect.die(new Error("Unsupported provider call in test")) as Effect.Effect<A, never>; const listSessions = () => hasSession ? Effect.succeed([ { provider: providerName, status: "ready", runtimeMode: "full-access", threadId: ThreadId.makeUnsafe("thread-1"), cwd: sessionCwd, createdAt: now, updatedAt: now, }, ] satisfies ReadonlyArray<ProviderSession>) : Effect.succeed([] as ReadonlyArray<ProviderSession>); const service: ProviderServiceShape = { startSession: () => unsupported(), sendTurn: () => unsupported(), interruptTurn: () => unsupported(), respondToRequest: () => unsupported(), respondToUserInput: () => unsupported(), stopSession: () => unsupported(), listSessions, getCapabilities: () => Effect.succeed({ sessionModelSwitch: "in-session" }), rollbackConversation, get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); }, }; const emit = (event: LegacyProviderRuntimeEvent): void => { Effect.runSync(PubSub.publish(runtimeEventPubSub, event as unknown as ProviderRuntimeEvent)); }; return { service, rollbackConversation, emit, };} async function waitForThread( engine: OrchestrationEngineShape, predicate: (thread: { latestTurn: { turnId: string } | null; checkpoints: ReadonlyArray<{ checkpointTurnCount: number }>; activities: ReadonlyArray<{ kind: string }>; }) => boolean, timeoutMs = 15_000,) { const deadline = Date.now() + timeoutMs; const poll = async (): Promise<{ latestTurn: { turnId: string } | null; checkpoints: ReadonlyArray<{ checkpointTurnCount: number }>; activities: ReadonlyArray<{ kind: string }>; }> => { const readModel = await Effect.runPromise(engine.getReadModel()); const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); if (thread && predicate(thread)) { return thread; } if (Date.now() >= deadline) { throw new Error("Timed out waiting for thread state."); } await new Promise((resolve) => setTimeout(resolve, 10)); return poll(); }; return poll();} async function waitForEvent( engine: OrchestrationEngineShape, predicate: (event: { type: string }) => boolean, timeoutMs = 15_000,) { const deadline = Date.now() + timeoutMs; const poll = async () => { const events = await Effect.runPromise( Stream.runCollect(engine.readEvents(0)).pipe(Effect.map((chunk) => Array.from(chunk))), ); if (events.some(predicate)) { return events; } if (Date.now() >= deadline) { throw new Error("Timed out waiting for orchestration event."); } await new Promise((resolve) => setTimeout(resolve, 10)); return poll(); }; return poll();} function runGit(cwd: string, args: ReadonlyArray<string>) { return execFileSync("git", args, { cwd, stdio: ["ignore", "pipe", "pipe"], encoding: "utf8", });} function createGitRepository() { const cwd = fs.mkdtempSync(path.join(os.tmpdir(), "t3-checkpoint-handler-")); runGit(cwd, ["init", "--initial-branch=main"]); runGit(cwd, ["config", "user.email", "test@example.com"]); runGit(cwd, ["config", "user.name", "Test User"]); fs.writeFileSync(path.join(cwd, "README.md"), "v1\n", "utf8"); runGit(cwd, ["add", "."]); runGit(cwd, ["commit", "-m", "Initial"]); return cwd;} function gitRefExists(cwd: string, ref: string): boolean { try { runGit(cwd, ["show-ref", "--verify", "--quiet", ref]); return true; } catch { return false; }} function gitShowFileAtRef(cwd: string, ref: string, filePath: string): string { return runGit(cwd, ["show", `${ref}:${filePath}`]);} async function waitForGitRefExists(cwd: string, ref: string, timeoutMs = 15_000) { const deadline = Date.now() + timeoutMs; const poll = async (): Promise<void> => { if (gitRefExists(cwd, ref)) { return; } if (Date.now() >= deadline) { throw new Error(`Timed out waiting for git ref '${ref}'.`); } await new Promise((resolve) => setTimeout(resolve, 10)); return poll(); }; return poll();} describe("CheckpointReactor", () => { let runtime: ManagedRuntime.ManagedRuntime< OrchestrationEngineService | CheckpointReactor | CheckpointStore, unknown > | null = null; let scope: Scope.Closeable | null = null; const tempDirs: string[] = []; afterEach(async () => { if (scope) { await Effect.runPromise(Scope.close(scope, Exit.void)); } scope = null; if (runtime) { await runtime.dispose(); } runtime = null; while (tempDirs.length > 0) { const dir = tempDirs.pop(); if (dir) { fs.rmSync(dir, { recursive: true, force: true }); } } }); async function createHarness(options?: { readonly hasSession?: boolean; readonly seedFilesystemCheckpoints?: boolean; readonly projectWorkspaceRoot?: string; readonly threadWorktreePath?: string | null; readonly providerSessionCwd?: string; readonly providerName?: ProviderKind; }) { const cwd = createGitRepository(); tempDirs.push(cwd); const provider = createProviderServiceHarness( cwd, options?.hasSession ?? true, options?.providerSessionCwd ?? cwd, options?.providerName ?? "codex", ); const orchestrationLayer = OrchestrationEngineLive.pipe( Layer.provide(OrchestrationProjectionSnapshotQueryLive), Layer.provide(OrchestrationProjectionPipelineLive), Layer.provide(OrchestrationEventStoreLive), Layer.provide(OrchestrationCommandReceiptRepositoryLive), Layer.provide(SqlitePersistenceMemory), ); const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), { prefix: "t3-checkpoint-reactor-test-", }); const layer = CheckpointReactorLive.pipe( Layer.provideMerge(orchestrationLayer), Layer.provideMerge(RuntimeReceiptBusLive), Layer.provideMerge(Layer.succeed(ProviderService, provider.service)), Layer.provideMerge(CheckpointStoreLive), Layer.provideMerge(WorkspaceEntriesLive.pipe(Layer.provide(WorkspacePathsLive))), Layer.provideMerge(WorkspacePathsLive), Layer.provideMerge(GitCoreLive), Layer.provideMerge(ServerConfigLayer), Layer.provideMerge(NodeServices.layer), ); runtime = ManagedRuntime.make(layer); const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); const reactor = await runtime.runPromise(Effect.service(CheckpointReactor)); const checkpointStore = await runtime.runPromise(Effect.service(CheckpointStore)); scope = await Effect.runPromise(Scope.make("sequential")); await Effect.runPromise(reactor.start().pipe(Scope.provide(scope))); const drain = () => Effect.runPromise(reactor.drain); const createdAt = new Date().toISOString(); await Effect.runPromise( engine.dispatch({ type: "project.create", commandId: CommandId.makeUnsafe("cmd-project-create"), projectId: asProjectId("project-1"), title: "Test Project", workspaceRoot: options?.projectWorkspaceRoot ?? cwd, defaultModelSelection: { provider: "codex", model: "gpt-5-codex", }, createdAt, }), ); await Effect.runPromise( engine.dispatch({ type: "thread.create", commandId: CommandId.makeUnsafe("cmd-thread-create"), threadId: ThreadId.makeUnsafe("thread-1"), projectId: asProjectId("project-1"), title: "Thread", modelSelection: { provider: "codex", model: "gpt-5-codex", }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required", branch: null, worktreePath: options?.threadWorktreePath ?? cwd, createdAt, }), ); if (options?.seedFilesystemCheckpoints ?? true) { await runtime.runPromise( checkpointStore.captureCheckpoint({ cwd, checkpointRef: checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0), }), ); fs.writeFileSync(path.join(cwd, "README.md"), "v2\n", "utf8"); await runtime.runPromise( checkpointStore.captureCheckpoint({ cwd, checkpointRef: checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 1), }), ); fs.writeFileSync(path.join(cwd, "README.md"), "v3\n", "utf8"); await runtime.runPromise( checkpointStore.captureCheckpoint({ cwd, checkpointRef: checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 2), }), ); } return { engine, provider, cwd, drain, }; } it("captures pre-turn baseline on turn.started and post-turn checkpoint on turn.completed", async () => { const harness = await createHarness({ seedFilesystemCheckpoints: false }); const createdAt = new Date().toISOString(); await Effect.runPromise( harness.engine.dispatch({ type: "thread.session.set", commandId: CommandId.makeUnsafe("cmd-session-set-capture"), threadId: ThreadId.makeUnsafe("thread-1"), session: { threadId: ThreadId.makeUnsafe("thread-1"), status: "ready", providerName: "codex", runtimeMode: "approval-required", activeTurnId: null, lastError: null, updatedAt: createdAt, }, createdAt, }), ); harness.provider.emit({ type: "turn.started", eventId: EventId.makeUnsafe("evt-turn-started-1"), provider: "codex", createdAt: new Date().toISOString(), threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-1"), }); await waitForGitRefExists( harness.cwd, checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0), ); fs.writeFileSync(path.join(harness.cwd, "README.md"), "v2\n", "utf8"); harness.provider.emit({ type: "turn.completed", eventId: EventId.makeUnsafe("evt-turn-completed-1"), provider: "codex", createdAt: new Date().toISOString(), threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-1"), payload: { state: "completed" }, }); await waitForEvent(harness.engine, (event) => event.type === "thread.turn-diff-completed"); const thread = await waitForThread( harness.engine, (entry) => entry.latestTurn?.turnId === "turn-1" && entry.checkpoints.length === 1, ); expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1); expect( gitRefExists(harness.cwd, checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0)), ).toBe(true); expect( gitRefExists(harness.cwd, checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 1)), ).toBe(true); expect( gitShowFileAtRef( harness.cwd, checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0), "README.md", ), ).toBe("v1\n"); expect( gitShowFileAtRef( harness.cwd, checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 1), "README.md", ), ).toBe("v2\n"); }); it("ignores auxiliary thread turn completion while primary turn is active", async () => { const harness = await createHarness({ seedFilesystemCheckpoints: false }); const createdAt = new Date().toISOString(); await Effect.runPromise( harness.engine.dispatch({ type: "thread.session.set", commandId: CommandId.makeUnsafe("cmd-session-set-primary-running"), threadId: ThreadId.makeUnsafe("thread-1"), session: { threadId: ThreadId.makeUnsafe("thread-1"), status: "running", providerName: "codex", runtimeMode: "approval-required", activeTurnId: asTurnId("turn-main"), lastError: null, updatedAt: createdAt, }, createdAt, }), ); harness.provider.emit({ type: "turn.started", eventId: EventId.makeUnsafe("evt-turn-started-main"), provider: "codex", createdAt: new Date().toISOString(), threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-main"), }); await waitForGitRefExists( harness.cwd, checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0), ); fs.writeFileSync(path.join(harness.cwd, "README.md"), "v2\n", "utf8"); harness.provider.emit({ type: "turn.completed", eventId: EventId.makeUnsafe("evt-turn-completed-aux"), provider: "codex", createdAt: new Date().toISOString(), threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-aux"), payload: { state: "completed" }, }); await harness.drain(); const midReadModel = await Effect.runPromise(harness.engine.getReadModel()); const midThread = midReadModel.threads.find( (entry) => entry.id === ThreadId.makeUnsafe("thread-1"), ); expect(midThread?.checkpoints).toHaveLength(0); harness.provider.emit({ type: "turn.completed", eventId: EventId.makeUnsafe("evt-turn-completed-main"), provider: "codex", createdAt: new Date().toISOString(), threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-main"), payload: { state: "completed" }, }); const thread = await waitForThread( harness.engine, (entry) => entry.latestTurn?.turnId === "turn-main" && entry.checkpoints.length === 1, ); expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1); }); Lines 1–500 of 1052. Load more
import fs from "node:fs"; import os from "node:os"; import path from "node:path"; import { execFileSync } from "node:child_process"; import type { ProviderKind, ProviderRuntimeEvent, ProviderSession } from "@t3tools/contracts"; import { CommandId, DEFAULT_PROVIDER_INTERACTION_MODE, EventId, MessageId, ProjectId, ThreadId, TurnId, } from "@t3tools/contracts"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { Effect, Exit, Layer, ManagedRuntime, PubSub, Scope, Stream } from "effect"; import { afterEach, describe, expect, it, vi } from "vitest"; import { CheckpointStoreLive } from "../../checkpointing/Layers/CheckpointStore.ts"; import { CheckpointStore } from "../../checkpointing/Services/CheckpointStore.ts"; import { GitCoreLive } from "../../git/Layers/GitCore.ts"; import { CheckpointReactorLive } from "./CheckpointReactor.ts"; import { OrchestrationEngineLive } from "./OrchestrationEngine.ts"; import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts"; import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts"; import { RuntimeReceiptBusLive } from "./RuntimeReceiptBus.ts"; import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts"; import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts"; import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; import { OrchestrationEngineService, type OrchestrationEngineShape, } from "../Services/OrchestrationEngine.ts"; import { CheckpointReactor } from "../Services/CheckpointReactor.ts"; import { ProviderService, type ProviderServiceShape, } from "../../provider/Services/ProviderService.ts"; import { checkpointRefForThreadTurn } from "../../checkpointing/Utils.ts"; import { ServerConfig } from "../../config.ts"; import { WorkspaceEntriesLive } from "../../workspace/Layers/WorkspaceEntries.ts"; import { WorkspacePathsLive } from "../../workspace/Layers/WorkspacePaths.ts"; const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); type LegacyProviderRuntimeEvent = { readonly type: string; readonly eventId: EventId; readonly provider: ProviderKind; readonly createdAt: string; readonly threadId: ThreadId; readonly turnId?: string | undefined; readonly itemId?: string | undefined; readonly requestId?: string | undefined; readonly payload?: unknown | undefined; readonly [key: string]: unknown; }; function createProviderServiceHarness( cwd: string, hasSession = true, sessionCwd = cwd, providerName: ProviderSession["provider"] = "codex", ) { const now = new Date().toISOString(); const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>()); const rollbackConversation = vi.fn( (_input: { readonly threadId: ThreadId; readonly numTurns: number }) => Effect.void, ); const unsupported = <A>() => Effect.die(new Error("Unsupported provider call in test")) as Effect.Effect<A, never>; const listSessions = () => hasSession ? Effect.succeed([ { provider: providerName, status: "ready", runtimeMode: "full-access", threadId: ThreadId.makeUnsafe("thread-1"), cwd: sessionCwd, createdAt: now, updatedAt: now, }, ] satisfies ReadonlyArray<ProviderSession>) : Effect.succeed([] as ReadonlyArray<ProviderSession>); const service: ProviderServiceShape = { startSession: () => unsupported(), sendTurn: () => unsupported(), interruptTurn: () => unsupported(), respondToRequest: () => unsupported(), respondToUserInput: () => unsupported(), stopSession: () => unsupported(), listSessions, getCapabilities: () => Effect.succeed({ sessionModelSwitch: "in-session" }), rollbackConversation, get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); }, }; const emit = (event: LegacyProviderRuntimeEvent): void => { Effect.runSync(PubSub.publish(runtimeEventPubSub, event as unknown as ProviderRuntimeEvent)); }; return { service, rollbackConversation, emit, }; } async function waitForThread( engine: OrchestrationEngineShape, predicate: (thread: { latestTurn: { turnId: string } | null; checkpoints: ReadonlyArray<{ checkpointTurnCount: number }>; activities: ReadonlyArray<{ kind: string }>; }) => boolean, timeoutMs = 15_000, ) { const deadline = Date.now() + timeoutMs; const poll = async (): Promise<{ latestTurn: { turnId: string } | null; checkpoints: ReadonlyArray<{ checkpointTurnCount: number }>; activities: ReadonlyArray<{ kind: string }>; }> => { const readModel = await Effect.runPromise(engine.getReadModel()); const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); if (thread && predicate(thread)) { return thread; } if (Date.now() >= deadline) { throw new Error("Timed out waiting for thread state."); } await new Promise((resolve) => setTimeout(resolve, 10)); return poll(); }; return poll(); } async function waitForEvent( engine: OrchestrationEngineShape, predicate: (event: { type: string }) => boolean, timeoutMs = 15_000, ) { const deadline = Date.now() + timeoutMs; const poll = async () => { const events = await Effect.runPromise( Stream.runCollect(engine.readEvents(0)).pipe(Effect.map((chunk) => Array.from(chunk))), ); if (events.some(predicate)) { return events; } if (Date.now() >= deadline) { throw new Error("Timed out waiting for orchestration event."); } await new Promise((resolve) => setTimeout(resolve, 10)); return poll(); }; return poll(); } function runGit(cwd: string, args: ReadonlyArray<string>) { return execFileSync("git", args, { cwd, stdio: ["ignore", "pipe", "pipe"], encoding: "utf8", }); } function createGitRepository() { const cwd = fs.mkdtempSync(path.join(os.tmpdir(), "t3-checkpoint-handler-")); runGit(cwd, ["init", "--initial-branch=main"]); runGit(cwd, ["config", "user.email", "test@example.com"]); runGit(cwd, ["config", "user.name", "Test User"]); fs.writeFileSync(path.join(cwd, "README.md"), "v1\n", "utf8"); runGit(cwd, ["add", "."]); runGit(cwd, ["commit", "-m", "Initial"]); return cwd; } function gitRefExists(cwd: string, ref: string): boolean { try { runGit(cwd, ["show-ref", "--verify", "--quiet", ref]); return true; } catch { return false; } } function gitShowFileAtRef(cwd: string, ref: string, filePath: string): string { return runGit(cwd, ["show", `${ref}:${filePath}`]); } async function waitForGitRefExists(cwd: string, ref: string, timeoutMs = 15_000) { const deadline = Date.now() + timeoutMs; const poll = async (): Promise<void> => { if (gitRefExists(cwd, ref)) { return; } if (Date.now() >= deadline) { throw new Error(`Timed out waiting for git ref '${ref}'.`); } await new Promise((resolve) => setTimeout(resolve, 10)); return poll(); }; return poll(); } describe("CheckpointReactor", () => { let runtime: ManagedRuntime.ManagedRuntime< OrchestrationEngineService | CheckpointReactor | CheckpointStore, unknown > | null = null; let scope: Scope.Closeable | null = null; const tempDirs: string[] = []; afterEach(async () => { if (scope) { await Effect.runPromise(Scope.close(scope, Exit.void)); } scope = null; if (runtime) { await runtime.dispose(); } runtime = null; while (tempDirs.length > 0) { const dir = tempDirs.pop(); if (dir) { fs.rmSync(dir, { recursive: true, force: true }); } } }); async function createHarness(options?: { readonly hasSession?: boolean; readonly seedFilesystemCheckpoints?: boolean; readonly projectWorkspaceRoot?: string; readonly threadWorktreePath?: string | null; readonly providerSessionCwd?: string; readonly providerName?: ProviderKind; }) { const cwd = createGitRepository(); tempDirs.push(cwd); const provider = createProviderServiceHarness( cwd, options?.hasSession ?? true, options?.providerSessionCwd ?? cwd, options?.providerName ?? "codex", ); const orchestrationLayer = OrchestrationEngineLive.pipe( Layer.provide(OrchestrationProjectionSnapshotQueryLive), Layer.provide(OrchestrationProjectionPipelineLive), Layer.provide(OrchestrationEventStoreLive), Layer.provide(OrchestrationCommandReceiptRepositoryLive), Layer.provide(SqlitePersistenceMemory), ); const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), { prefix: "t3-checkpoint-reactor-test-", }); const layer = CheckpointReactorLive.pipe( Layer.provideMerge(orchestrationLayer), Layer.provideMerge(RuntimeReceiptBusLive), Layer.provideMerge(Layer.succeed(ProviderService, provider.service)), Layer.provideMerge(CheckpointStoreLive), Layer.provideMerge(WorkspaceEntriesLive.pipe(Layer.provide(WorkspacePathsLive))), Layer.provideMerge(WorkspacePathsLive), Layer.provideMerge(GitCoreLive), Layer.provideMerge(ServerConfigLayer), Layer.provideMerge(NodeServices.layer), ); runtime = ManagedRuntime.make(layer); const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); const reactor = await runtime.runPromise(Effect.service(CheckpointReactor)); const checkpointStore = await runtime.runPromise(Effect.service(CheckpointStore)); scope = await Effect.runPromise(Scope.make("sequential")); await Effect.runPromise(reactor.start().pipe(Scope.provide(scope))); const drain = () => Effect.runPromise(reactor.drain); const createdAt = new Date().toISOString(); await Effect.runPromise( engine.dispatch({ type: "project.create", commandId: CommandId.makeUnsafe("cmd-project-create"), projectId: asProjectId("project-1"), title: "Test Project", workspaceRoot: options?.projectWorkspaceRoot ?? cwd, defaultModelSelection: { provider: "codex", model: "gpt-5-codex", }, createdAt, }), ); await Effect.runPromise( engine.dispatch({ type: "thread.create", commandId: CommandId.makeUnsafe("cmd-thread-create"), threadId: ThreadId.makeUnsafe("thread-1"), projectId: asProjectId("project-1"), title: "Thread", modelSelection: { provider: "codex", model: "gpt-5-codex", }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required", branch: null, worktreePath: options?.threadWorktreePath ?? cwd, createdAt, }), ); if (options?.seedFilesystemCheckpoints ?? true) { await runtime.runPromise( checkpointStore.captureCheckpoint({ cwd, checkpointRef: checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0), }), ); fs.writeFileSync(path.join(cwd, "README.md"), "v2\n", "utf8"); await runtime.runPromise( checkpointStore.captureCheckpoint({ cwd, checkpointRef: checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 1), }), ); fs.writeFileSync(path.join(cwd, "README.md"), "v3\n", "utf8"); await runtime.runPromise( checkpointStore.captureCheckpoint({ cwd, checkpointRef: checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 2), }), ); } return { engine, provider, cwd, drain, }; } it("captures pre-turn baseline on turn.started and post-turn checkpoint on turn.completed", async () => { const harness = await createHarness({ seedFilesystemCheckpoints: false }); const createdAt = new Date().toISOString(); await Effect.runPromise( harness.engine.dispatch({ type: "thread.session.set", commandId: CommandId.makeUnsafe("cmd-session-set-capture"), threadId: ThreadId.makeUnsafe("thread-1"), session: { threadId: ThreadId.makeUnsafe("thread-1"), status: "ready", providerName: "codex", runtimeMode: "approval-required", activeTurnId: null, lastError: null, updatedAt: createdAt, }, createdAt, }), ); harness.provider.emit({ type: "turn.started", eventId: EventId.makeUnsafe("evt-turn-started-1"), provider: "codex", createdAt: new Date().toISOString(), threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-1"), }); await waitForGitRefExists( harness.cwd, checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0), ); fs.writeFileSync(path.join(harness.cwd, "README.md"), "v2\n", "utf8"); harness.provider.emit({ type: "turn.completed", eventId: EventId.makeUnsafe("evt-turn-completed-1"), provider: "codex", createdAt: new Date().toISOString(), threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-1"), payload: { state: "completed" }, }); await waitForEvent(harness.engine, (event) => event.type === "thread.turn-diff-completed"); const thread = await waitForThread( harness.engine, (entry) => entry.latestTurn?.turnId === "turn-1" && entry.checkpoints.length === 1, ); expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1); expect( gitRefExists(harness.cwd, checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0)), ).toBe(true); expect( gitRefExists(harness.cwd, checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 1)), ).toBe(true); expect( gitShowFileAtRef( harness.cwd, checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0), "README.md", ), ).toBe("v1\n"); expect( gitShowFileAtRef( harness.cwd, checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 1), "README.md", ), ).toBe("v2\n"); }); it("ignores auxiliary thread turn completion while primary turn is active", async () => { const harness = await createHarness({ seedFilesystemCheckpoints: false }); const createdAt = new Date().toISOString(); await Effect.runPromise( harness.engine.dispatch({ type: "thread.session.set", commandId: CommandId.makeUnsafe("cmd-session-set-primary-running"), threadId: ThreadId.makeUnsafe("thread-1"), session: { threadId: ThreadId.makeUnsafe("thread-1"), status: "running", providerName: "codex", runtimeMode: "approval-required", activeTurnId: asTurnId("turn-main"), lastError: null, updatedAt: createdAt, }, createdAt, }), ); harness.provider.emit({ type: "turn.started", eventId: EventId.makeUnsafe("evt-turn-started-main"), provider: "codex", createdAt: new Date().toISOString(), threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-main"), }); await waitForGitRefExists( harness.cwd, checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0), ); fs.writeFileSync(path.join(harness.cwd, "README.md"), "v2\n", "utf8"); harness.provider.emit({ type: "turn.completed", eventId: EventId.makeUnsafe("evt-turn-completed-aux"), provider: "codex", createdAt: new Date().toISOString(), threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-aux"), payload: { state: "completed" }, }); await harness.drain(); const midReadModel = await Effect.runPromise(harness.engine.getReadModel()); const midThread = midReadModel.threads.find( (entry) => entry.id === ThreadId.makeUnsafe("thread-1"), ); expect(midThread?.checkpoints).toHaveLength(0); harness.provider.emit({ type: "turn.completed", eventId: EventId.makeUnsafe("evt-turn-completed-main"), provider: "codex", createdAt: new Date().toISOString(), threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-main"), payload: { state: "completed" }, }); const thread = await waitForThread( harness.engine, (entry) => entry.latestTurn?.turnId === "turn-main" && entry.checkpoints.length === 1, ); expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1); });