import fs from "node:fs";import os from "node:os";import path from "node:path";import { execFileSync } from "node:child_process"; import type { ProviderRuntimeEvent, ProviderSession } from "@t3tools/contracts";import {  CommandId,  EventId,  MessageId,  ProjectId,  ProviderSessionId,  ProviderThreadId,  ThreadId,  TurnId,} from "@t3tools/contracts";import { NodeServices } from "@effect/platform-node";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 { CheckpointReactorLive } from "./CheckpointReactor.ts";import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.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"; const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value);const asProviderThreadId = (value: string): ProviderThreadId => ProviderThreadId.makeUnsafe(value);const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); function createProviderServiceHarness(cwd: string, hasSession = true, sessionCwd = cwd) {  const now = new Date().toISOString();  const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());  const rollbackConversation = vi.fn(    (_input: { readonly sessionId: ProviderSessionId; 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([          {            sessionId: asSessionId("sess-1"),            provider: "codex",            status: "ready",            threadId: asProviderThreadId("provider-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(),    stopSession: () => unsupported(),    listSessions,    rollbackConversation,    stopAll: () => Effect.void,    streamEvents: Stream.fromPubSub(runtimeEventPubSub),  };   const emit = (event: ProviderRuntimeEvent): void => {    Effect.runSync(PubSub.publish(runtimeEventPubSub, event));  };   return {    service,    rollbackConversation,    emit,  };} async function waitForThread(  engine: OrchestrationEngineShape,  predicate: (thread: {    latestTurnId: string | null;    checkpoints: ReadonlyArray<{ checkpointTurnCount: number }>;    activities: ReadonlyArray<{ kind: string }>;  }) => boolean,  timeoutMs = 2000,) {  const deadline = Date.now() + timeoutMs;  const poll = async (): Promise<{    latestTurnId: 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 = 2000,) {  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 = 2000) {  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;  }) {    const cwd = createGitRepository();    tempDirs.push(cwd);    const provider = createProviderServiceHarness(      cwd,      options?.hasSession ?? true,      options?.providerSessionCwd ?? cwd,    );    const orchestrationLayer = OrchestrationEngineLive.pipe(      Layer.provide(OrchestrationProjectionPipelineLive),      Layer.provide(OrchestrationEventStoreLive),      Layer.provide(OrchestrationCommandReceiptRepositoryLive),      Layer.provide(SqlitePersistenceMemory),    );    const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer));    const layer = CheckpointReactorLive.pipe(      Layer.provideMerge(orchestrationLayer),      Layer.provideMerge(Layer.succeed(ProviderService, provider.service)),      Layer.provideMerge(checkpointStoreLayer),    );     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)));    await Effect.runPromise(Effect.sleep("10 millis"));     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,        defaultModel: "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",        model: "gpt-5-codex",        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,    };  }   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",          providerSessionId: asSessionId("sess-1"),          providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-1"),          activeTurnId: null,          lastError: null,          updatedAt: createdAt,        },        createdAt,      }),    );     harness.provider.emit({      type: "turn.started",      eventId: EventId.makeUnsafe("evt-turn-started-1"),      provider: "codex",      sessionId: asSessionId("sess-1"),      createdAt: new Date().toISOString(),      threadId: ProviderThreadId.makeUnsafe("provider-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",      sessionId: asSessionId("sess-1"),      createdAt: new Date().toISOString(),      threadId: ProviderThreadId.makeUnsafe("provider-thread-1"),      turnId: asTurnId("turn-1"),      status: "completed",    });     await waitForEvent(harness.engine, (event) => event.type === "thread.turn-diff-completed");    const thread = await waitForThread(      harness.engine,      (entry) => entry.latestTurnId === "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("appends capture failure activity when turn diff summary cannot be derived", 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-missing-baseline-diff"),        threadId: ThreadId.makeUnsafe("thread-1"),        session: {          threadId: ThreadId.makeUnsafe("thread-1"),          status: "ready",          providerName: "codex",          providerSessionId: asSessionId("sess-1"),          providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-1"),          activeTurnId: null,          lastError: null,          updatedAt: createdAt,        },        createdAt,      }),    );     harness.provider.emit({      type: "turn.completed",      eventId: EventId.makeUnsafe("evt-turn-completed-missing-baseline"),      provider: "codex",      sessionId: asSessionId("sess-1"),      createdAt: new Date().toISOString(),      threadId: ProviderThreadId.makeUnsafe("provider-thread-1"),      turnId: asTurnId("turn-missing-baseline"),      status: "completed",    });     await waitForEvent(harness.engine, (event) => event.type === "thread.turn-diff-completed");    const thread = await waitForThread(      harness.engine,      (entry) =>        entry.checkpoints.length === 1 &&        entry.activities.some((activity) => activity.kind === "checkpoint.capture.failed"),    );     expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1);    expect(      thread.activities.some((activity) => activity.kind === "checkpoint.capture.failed"),    ).toBe(true);  });   it("captures pre-turn baseline from project workspace root when thread worktree is unset", async () => {    const harness = await createHarness({      hasSession: false,      seedFilesystemCheckpoints: false,      threadWorktreePath: null,    });     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.makeUnsafe("cmd-turn-start-for-baseline"),        threadId: ThreadId.makeUnsafe("thread-1"),        message: {          messageId: MessageId.makeUnsafe("message-user-1"),          role: "user",          text: "start turn",          attachments: [],        },        createdAt: new Date().toISOString(),      }),    );     await waitForGitRefExists(      harness.cwd,      checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0),    );    expect(      gitShowFileAtRef(        harness.cwd,        checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0),        "README.md",      ),    ).toBe("v1\n");  });   it("captures turn completion checkpoint from project workspace root when provider session cwd is unavailable", async () => {    const harness = await createHarness({      hasSession: false,      seedFilesystemCheckpoints: false,      threadWorktreePath: null,    });    const createdAt = new Date().toISOString();     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.session.set",        commandId: CommandId.makeUnsafe("cmd-session-set-missing-provider-cwd"),        threadId: ThreadId.makeUnsafe("thread-1"),        session: {          threadId: ThreadId.makeUnsafe("thread-1"),          status: "running",          providerName: "codex",          providerSessionId: asSessionId("sess-missing"),          providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-missing"),          activeTurnId: asTurnId("turn-missing-cwd"),          lastError: null,          updatedAt: createdAt,        },        createdAt,      }),    );     fs.writeFileSync(path.join(harness.cwd, "README.md"), "v2\n", "utf8");    harness.provider.emit({      type: "turn.completed",      eventId: EventId.makeUnsafe("evt-turn-completed-missing-provider-cwd"),      provider: "codex",      sessionId: asSessionId("sess-missing"),      createdAt: new Date().toISOString(),      threadId: ProviderThreadId.makeUnsafe("provider-thread-missing"),      turnId: asTurnId("turn-missing-cwd"),

Lines 1–500 of 819. Load more