// @effect-diagnostics nodeBuiltinImport:offimport * as NodeFS from "node:fs";import * as NodeOS from "node:os";import * as NodePath from "node:path";import * as NodeChildProcess from "node:child_process"; import {  ProviderDriverKind,  ProviderRuntimeEvent,  ProviderSession,  ProviderInstanceId,} from "@t3tools/contracts";import {  CommandId,  CheckpointRef,  DEFAULT_PROVIDER_INTERACTION_MODE,  EventId,  MessageId,  ProjectId,  ThreadId,  TurnId,} from "@t3tools/contracts";import * as NodeServices from "@effect/platform-node/NodeServices";import * as Clock from "effect/Clock";import * as Deferred from "effect/Deferred";import * as Effect from "effect/Effect";import * as Exit from "effect/Exit";import * as Layer from "effect/Layer";import * as ManagedRuntime from "effect/ManagedRuntime";import * as PubSub from "effect/PubSub";import * as Queue from "effect/Queue";import * as Scope from "effect/Scope";import * as Stream from "effect/Stream";import { it as effectIt } from "@effect/vitest";import { afterEach, describe, expect, it, vi } from "vite-plus/test"; import * as CheckpointStore from "../../checkpointing/CheckpointStore.ts";import * as VcsDriverRegistry from "../../vcs/VcsDriverRegistry.ts";import * as VcsProcess from "../../vcs/VcsProcess.ts";import { VcsStatusBroadcaster } from "../../vcs/VcsStatusBroadcaster.ts";import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityResolver.ts";import { CheckpointReactorLive } from "./CheckpointReactor.ts";import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts";import * as ThreadPlanProgress from "../ThreadPlanProgress.ts";import { RuntimeReceiptBusTest } from "./RuntimeReceiptBus.ts";import * as RuntimeReceiptBus from "../Services/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 { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";import {  ProviderService,  type ProviderServiceShape,} from "../../provider/Services/ProviderService.ts";import { checkpointRefForThreadTurn } from "../../checkpointing/Utils.ts";import { ProviderValidationError } from "../../provider/Errors.ts";import { ServerConfig } from "../../config.ts";import * as WorkspaceEntries from "../../workspace/WorkspaceEntries.ts";import * as WorkspacePaths from "../../workspace/WorkspacePaths.ts";import { PullRequestService } from "../../pullRequest/PullRequestService.ts"; const asProjectId = (value: string): ProjectId => ProjectId.make(value);const asTurnId = (value: string): TurnId => TurnId.make(value); type LegacyProviderRuntimeEvent = {  readonly type: string;  readonly eventId: EventId;  readonly provider: ProviderDriverKind;  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"] = ProviderDriverKind.make("codex"),) {  const now = "2026-01-01T00:00:00.000Z";  const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());  const rollbackConversation = vi.fn(    (_input: { readonly threadId: ThreadId; readonly numTurns: number }) => Effect.void,  );  const assertConversationRollbackSupported = vi.fn<    ProviderServiceShape["assertConversationRollbackSupported"]  >(() => 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.make("thread-1"),            cwd: sessionCwd,            createdAt: now,            updatedAt: now,          },        ] satisfies ReadonlyArray<ProviderSession>)      : Effect.succeed([] as ReadonlyArray<ProviderSession>);  const service: ProviderServiceShape = {    startSession: () => unsupported(),    sendTurn: () => unsupported(),    compactThread: () => unsupported(),    interruptTurn: () => unsupported(),    respondToRequest: () => unsupported(),    respondToUserInput: () => unsupported(),    stopSession: () => unsupported(),    listSessions,    getCapabilities: () => Effect.succeed({ sessionModelSwitch: "in-session" }),    assertConversationRollbackSupported,    getInstanceInfo: (instanceId) =>      Effect.succeed({        instanceId,        driverKind: ProviderDriverKind.make(providerName),        displayName: undefined,        enabled: true,        continuationIdentity: {          driverKind: ProviderDriverKind.make(providerName),          continuationKey: `${providerName}:instance:${instanceId}`,        },      }),    rollbackConversation,    uploadFeedback: () => unsupported(),    get streamEvents() {      return Stream.fromPubSub(runtimeEventPubSub);    },  };   const emit = (event: LegacyProviderRuntimeEvent): void => {    Effect.runSync(PubSub.publish(runtimeEventPubSub, event as unknown as ProviderRuntimeEvent));  };   return {    service,    assertConversationRollbackSupported,    rollbackConversation,    emit,  };} async function waitForThread(  readModel: () => Promise<{    readonly threads: ReadonlyArray<{      readonly id: ThreadId;      readonly latestTurn: { readonly turnId: string } | null;      readonly checkpoints: ReadonlyArray<{ readonly checkpointTurnCount: number }>;      readonly activities: ReadonlyArray<{ readonly kind: string }>;    }>;  }>,  predicate: (thread: {    latestTurn: { turnId: string } | null;    checkpoints: ReadonlyArray<{ checkpointTurnCount: number }>;    activities: ReadonlyArray<{ kind: string }>;  }) => boolean,  timeoutMs = 15_000,) {  const deadline = (await Effect.runPromise(Clock.currentTimeMillis)) + timeoutMs;  const poll = async (): Promise<{    latestTurn: { turnId: string } | null;    checkpoints: ReadonlyArray<{ checkpointTurnCount: number }>;    activities: ReadonlyArray<{ kind: string }>;  }> => {    const snapshot = await readModel();    const thread = snapshot.threads.find((entry) => entry.id === ThreadId.make("thread-1"));    if (thread && predicate(thread)) {      return thread;    }    if ((await Effect.runPromise(Clock.currentTimeMillis)) >= deadline) {      throw new Error("Timed out waiting for thread state.");    }    await Effect.runPromise(Effect.sleep("10 millis"));    return poll();  };  return poll();} async function waitForEvent(  engine: OrchestrationEngineShape,  predicate: (event: { type: string }) => boolean,  timeoutMs = 15_000,) {  const deadline = (await Effect.runPromise(Clock.currentTimeMillis)) + 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 ((await Effect.runPromise(Clock.currentTimeMillis)) >= deadline) {      throw new Error("Timed out waiting for orchestration event.");    }    await Effect.runPromise(Effect.sleep("10 millis"));    return poll();  };  return poll();} function runGit(cwd: string, args: ReadonlyArray<string>) {  return NodeChildProcess.execFileSync("git", args, {    cwd,    stdio: ["ignore", "pipe", "pipe"],    encoding: "utf8",  });} function createGitRepository() {  const cwd = NodeFS.mkdtempSync(NodePath.join(NodeOS.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"]);  NodeFS.writeFileSync(NodePath.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 = (await Effect.runPromise(Clock.currentTimeMillis)) + timeoutMs;  const poll = async (): Promise<void> => {    if (gitRefExists(cwd, ref)) {      return;    }    if ((await Effect.runPromise(Clock.currentTimeMillis)) >= deadline) {      throw new Error(`Timed out waiting for git ref '${ref}'.`);    }    await Effect.runPromise(Effect.sleep("10 millis"));    return poll();  };  return poll();} describe("CheckpointReactor", () => {  let runtime: ManagedRuntime.ManagedRuntime<    | OrchestrationEngineService    | CheckpointReactor    | CheckpointStore.CheckpointStore    | ProjectionSnapshotQuery    | RuntimeReceiptBus.RuntimeReceiptBus,    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) {        NodeFS.rmSync(dir, { recursive: true, force: true });      }    }  });   async function createHarness(options?: {    readonly hasSession?: boolean;    readonly seedFilesystemCheckpoints?: boolean;    readonly initializeGit?: boolean;    readonly projectWorkspaceRoot?: string;    readonly threadWorktreePath?: string | null;    readonly threadBranch?: string | null;    readonly secondThreadSharingWorktree?: boolean;    readonly localStatusRefName?: string | null;    readonly providerSessionCwd?: string;    readonly providerName?: ProviderDriverKind;    readonly gitStatusRefreshCalls?: Array<string>;    readonly pullRequestRefreshCalls?: Array<string>;    readonly pullRequestRefresh?: Effect.Effect<void>;  }) {    const cwd = createGitRepository();    if (options?.initializeGit === false) {      NodeFS.rmSync(NodePath.join(cwd, ".git"), { recursive: true });    }    tempDirs.push(cwd);    const provider = createProviderServiceHarness(      cwd,      options?.hasSession ?? true,      options?.providerSessionCwd ?? cwd,      options?.providerName ?? ProviderDriverKind.make("codex"),    );    const orchestrationLayer = OrchestrationEngineLive.pipe(      Layer.provide(OrchestrationProjectionSnapshotQueryLive),      Layer.provide(ThreadBackgroundLiveness.layer),      Layer.provide(ThreadPlanProgress.layer),      Layer.provide(OrchestrationProjectionPipelineLive),      Layer.provide(OrchestrationEventStoreLive),      Layer.provide(OrchestrationCommandReceiptRepositoryLive),      Layer.provide(RepositoryIdentityResolver.layer),      Layer.provide(SqlitePersistenceMemory),    );    const projectionSnapshotLayer = OrchestrationProjectionSnapshotQueryLive.pipe(      Layer.provide(ThreadBackgroundLiveness.layer),      Layer.provide(ThreadPlanProgress.layer),      Layer.provide(RepositoryIdentityResolver.layer),      Layer.provide(SqlitePersistenceMemory),    );     const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), {      prefix: "t3-checkpoint-reactor-test-",    });    const pullRequestRefreshes: number[] = [];    const refreshAfterTurn = Effect.sync(() => void pullRequestRefreshes.push(1));    const vcsStatusBroadcasterLayer = Layer.succeed(VcsStatusBroadcaster, {      getStatus: () => Effect.die("getStatus should not be called in this test"),      refreshLocalStatus: (cwd: string) =>        Effect.sync(() => {          options?.gitStatusRefreshCalls?.push(cwd);        }).pipe(          Effect.as({            isRepo: true,            hasPrimaryRemote: false,            isDefaultRef:              options?.localStatusRefName === undefined || options.localStatusRefName === "main",            refName:              options?.localStatusRefName !== undefined ? options.localStatusRefName : "main",            hasWorkingTreeChanges: false,            workingTree: { files: [], insertions: 0, deletions: 0 },          }),        ),      refreshStatus: () => Effect.die("refreshStatus should not be called in this test"),      refreshPullRequestStatus: (cwd: string) =>        Effect.sync(() => {          options?.pullRequestRefreshCalls?.push(cwd);        }).pipe(Effect.andThen(options?.pullRequestRefresh ?? Effect.void), Effect.as(null)),      streamStatus: () => Stream.empty,    });     const layer = CheckpointReactorLive.pipe(      Layer.provideMerge(orchestrationLayer),      Layer.provideMerge(projectionSnapshotLayer),      Layer.provideMerge(RuntimeReceiptBusTest),      Layer.provideMerge(Layer.succeed(ProviderService, provider.service)),      Layer.provideMerge(Layer.mock(PullRequestService)({ refreshAfterTurn })),      Layer.provideMerge(vcsStatusBroadcasterLayer),      Layer.provideMerge(CheckpointStore.layer.pipe(Layer.provide(VcsDriverRegistry.layer))),      Layer.provideMerge(        WorkspaceEntries.layer.pipe(          Layer.provide(WorkspacePaths.layer),          Layer.provideMerge(VcsDriverRegistry.layer),        ),      ),      Layer.provideMerge(WorkspacePaths.layer),      Layer.provideMerge(VcsProcess.layer),      Layer.provideMerge(ServerConfigLayer),      Layer.provideMerge(NodeServices.layer),    );     runtime = ManagedRuntime.make(layer);    const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));    const snapshotQuery = await runtime.runPromise(Effect.service(ProjectionSnapshotQuery));    const reactor = await runtime.runPromise(Effect.service(CheckpointReactor));    const checkpointStore = await runtime.runPromise(      Effect.service(CheckpointStore.CheckpointStore),    );    const receiptBus = await runtime.runPromise(      Effect.service(RuntimeReceiptBus.RuntimeReceiptBus),    );    const testScope = await Effect.runPromise(Scope.make("sequential"));    scope = testScope;    const receipts = await Effect.runPromise(      Effect.gen(function* () {        const receipts = yield* Queue.unbounded<RuntimeReceiptBus.OrchestrationRuntimeReceipt>();        yield* Stream.runForEach(receiptBus.streamEventsForTest, (receipt) =>          Queue.offer(receipts, receipt),        ).pipe(Effect.forkIn(testScope, { startImmediately: true }));        yield* reactor.start().pipe(Scope.provide(testScope));        return receipts;      }),    );    const drain = () => Effect.runPromise(reactor.drain);     const createdAt = "2026-01-01T00:00:00.000Z";    await Effect.runPromise(      engine.dispatch({        type: "project.create",        commandId: CommandId.make("cmd-project-create"),        projectId: asProjectId("project-1"),        title: "Test Project",        workspaceRoot: options?.projectWorkspaceRoot ?? cwd,        defaultModelSelection: {          instanceId: ProviderInstanceId.make("codex"),          model: "gpt-5-codex",        },        createdAt,      }),    );    await Effect.runPromise(      engine        .dispatch({          type: "thread.create",          commandId: CommandId.make("cmd-thread-create"),          threadId: ThreadId.make("thread-1"),          projectId: asProjectId("project-1"),          title: "Thread",          modelSelection: {            instanceId: ProviderInstanceId.make("codex"),            model: "gpt-5-codex",          },          interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,          runtimeMode: "approval-required",          branch: options?.threadBranch ?? null,          worktreePath: options?.threadWorktreePath ?? cwd,          createdAt,        })        .pipe(          options?.secondThreadSharingWorktree            ? Effect.andThen(                engine.dispatch({                  type: "thread.create",                  commandId: CommandId.make("cmd-thread-create-2"),                  threadId: ThreadId.make("thread-2"),                  projectId: asProjectId("project-1"),                  title: "Thread 2",                  modelSelection: {                    instanceId: ProviderInstanceId.make("codex"),                    model: "gpt-5-codex",                  },                  interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,                  runtimeMode: "approval-required",                  branch: null,                  worktreePath: options?.threadWorktreePath ?? cwd,                  createdAt,                }),              )            : Effect.asVoid,        ),    );     if (options?.seedFilesystemCheckpoints ?? true) {      await runtime.runPromise(        checkpointStore.captureCheckpoint({          cwd,          checkpointRef: checkpointRefForThreadTurn(ThreadId.make("thread-1"), 0),        }),      );      NodeFS.writeFileSync(NodePath.join(cwd, "README.md"), "v2\n", "utf8");      await runtime.runPromise(        checkpointStore.captureCheckpoint({          cwd,          checkpointRef: checkpointRefForThreadTurn(ThreadId.make("thread-1"), 1),        }),      );      NodeFS.writeFileSync(NodePath.join(cwd, "README.md"), "v3\n", "utf8");      await runtime.runPromise(        checkpointStore.captureCheckpoint({          cwd,          checkpointRef: checkpointRefForThreadTurn(ThreadId.make("thread-1"), 2),        }),      );    }     return {      engine,      readModel: () => Effect.runPromise(snapshotQuery.getSnapshot()),      provider,      cwd,      drain,      nextReceipt: Queue.take(receipts),      pullRequestRefreshes,    };  }   effectIt.effect("captures baseline and large turn summaries before completion receipts", () =>

Lines 1–500 of 1942. Load more