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 { 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,    streamEvents: 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(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);  });   it("captures pre-turn and completion checkpoints for claude runtime events", async () => {    const harness = await createHarness({      seedFilesystemCheckpoints: false,      providerName: "claudeAgent",

Lines 1–500 of 1048. Load more