import {  CheckpointRef,  CommandId,  DEFAULT_PROVIDER_INTERACTION_MODE,  MessageId,  ProjectId,  ThreadId,  TurnId,  type OrchestrationEvent,  ProviderInstanceId,} from "@t3tools/contracts";import * as NodeServices from "@effect/platform-node/NodeServices";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as ManagedRuntime from "effect/ManagedRuntime";import * as Metric from "effect/Metric";import * as Option from "effect/Option";import * as Queue from "effect/Queue";import * as Stream from "effect/Stream";import { describe, expect, it } from "vite-plus/test"; import { PersistenceSqlError } from "../../persistence/Errors.ts";import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts";import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";import {  OrchestrationEventStore,  type OrchestrationEventStoreShape,} from "../../persistence/Services/OrchestrationEventStore.ts";import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityResolver.ts";import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";import {  OrchestrationProjectionPipeline,  type OrchestrationProjectionPipelineShape,} from "../Services/ProjectionPipeline.ts";import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";import { ServerConfig } from "../../config.ts"; const asProjectId = (value: string): ProjectId => ProjectId.make(value);const asMessageId = (value: string): MessageId => MessageId.make(value);const asTurnId = (value: string): TurnId => TurnId.make(value);const asCheckpointRef = (value: string): CheckpointRef => CheckpointRef.make(value); async function createOrchestrationSystem() {  const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), {    prefix: "t3-orchestration-engine-test-",  });  const orchestrationLayer = Layer.mergeAll(    OrchestrationEngineLive.pipe(      Layer.provide(OrchestrationProjectionSnapshotQueryLive),      Layer.provide(OrchestrationProjectionPipelineLive),    ),    OrchestrationProjectionSnapshotQueryLive,  ).pipe(    Layer.provide(OrchestrationEventStoreLive),    Layer.provide(OrchestrationCommandReceiptRepositoryLive),    Layer.provide(RepositoryIdentityResolver.layer),    Layer.provide(SqlitePersistenceMemory),    Layer.provideMerge(ServerConfigLayer),    Layer.provideMerge(NodeServices.layer),  );  const runtime = ManagedRuntime.make(orchestrationLayer);  const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));  const snapshotQuery = await runtime.runPromise(Effect.service(ProjectionSnapshotQuery));  return {    engine,    readModel: () => runtime.runPromise(snapshotQuery.getSnapshot()),    run: <A, E>(effect: Effect.Effect<A, E>) => runtime.runPromise(effect),    dispose: () => runtime.dispose(),  };} function now() {  return "2026-01-01T00:00:00.000Z";} const hasMetricSnapshot = (  snapshots: ReadonlyArray<Metric.Metric.Snapshot>,  id: string,  attributes: Readonly<Record<string, string>>,) =>  snapshots.some(    (snapshot) =>      snapshot.id === id &&      Object.entries(attributes).every(([key, value]) => snapshot.attributes?.[key] === value),  ); describe("OrchestrationEngine", () => {  it("bootstraps command handling from persisted projections without reading the full snapshot", async () => {    let nextSequence = 8;    const eventStore: OrchestrationEventStoreShape = {      append: (event) =>        Effect.sync(() => {          const savedEvent = {            ...event,            sequence: nextSequence,          } as OrchestrationEvent;          nextSequence += 1;          return savedEvent;        }),      readFromSequence: () => Stream.empty,      readAll: () =>        Stream.fail(          new PersistenceSqlError({            operation: "test.readAll",            detail: "historical replay should not be used during bootstrap",          }),        ),    };     const projectionSnapshot = {      snapshotSequence: 7,      updatedAt: "2026-03-03T00:00:04.000Z",      projects: [        {          id: asProjectId("project-bootstrap"),          title: "Bootstrap Project",          workspaceRoot: "/tmp/project-bootstrap",          defaultModelSelection: {            instanceId: ProviderInstanceId.make("codex"),            model: "gpt-5-codex",          },          scripts: [],          createdAt: "2026-03-03T00:00:00.000Z",          updatedAt: "2026-03-03T00:00:01.000Z",          deletedAt: null,        },      ],      threads: [        {          id: ThreadId.make("thread-bootstrap"),          projectId: asProjectId("project-bootstrap"),          title: "Bootstrap Thread",          modelSelection: {            instanceId: ProviderInstanceId.make("codex"),            model: "gpt-5-codex",          },          interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,          runtimeMode: "full-access" as const,          branch: null,          worktreePath: null,          latestTurn: null,          createdAt: "2026-03-03T00:00:02.000Z",          updatedAt: "2026-03-03T00:00:03.000Z",          archivedAt: null,          deletedAt: null,          messages: [],          proposedPlans: [],          activities: [],          checkpoints: [],          session: null,        },      ],    };    const commandReadModel = {      ...projectionSnapshot,      threads: projectionSnapshot.threads.map((thread) => ({        ...thread,        messages: [],        proposedPlans: [],        activities: [],        checkpoints: [],      })),    };    let fullSnapshotReadCount = 0;     const layer = OrchestrationEngineLive.pipe(      Layer.provide(        Layer.succeed(ProjectionSnapshotQuery, {          getCommandReadModel: () => Effect.succeed(commandReadModel),          getSnapshot: () =>            Effect.sync(() => {              fullSnapshotReadCount += 1;              return projectionSnapshot;            }),          getShellSnapshot: () =>            Effect.succeed({              snapshotSequence: projectionSnapshot.snapshotSequence,              projects: [],              threads: [],              updatedAt: projectionSnapshot.updatedAt,            }),          getArchivedShellSnapshot: () =>            Effect.succeed({              snapshotSequence: projectionSnapshot.snapshotSequence,              projects: [],              threads: [],              updatedAt: projectionSnapshot.updatedAt,            }),          getSnapshotSequence: () =>            Effect.succeed({ snapshotSequence: projectionSnapshot.snapshotSequence }),          getCounts: () => Effect.succeed({ projectCount: 1, threadCount: 1 }),          getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),          getProjectShellById: () => Effect.succeed(Option.none()),          getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),          getThreadCheckpointContext: () => Effect.succeed(Option.none()),          getFullThreadDiffContext: () => Effect.succeed(Option.none()),          getThreadShellById: () => Effect.succeed(Option.none()),          getThreadDetailById: () => Effect.succeed(Option.none()),        }),      ),      Layer.provide(        Layer.succeed(OrchestrationProjectionPipeline, {          bootstrap: Effect.void,          projectEvent: () => Effect.void,        } satisfies OrchestrationProjectionPipelineShape),      ),      Layer.provide(Layer.succeed(OrchestrationEventStore, eventStore)),      Layer.provide(OrchestrationCommandReceiptRepositoryLive),      Layer.provide(SqlitePersistenceMemory),      Layer.provideMerge(NodeServices.layer),    );     const runtime = ManagedRuntime.make(layer);     const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));    const result = await runtime.runPromise(      engine.dispatch({        type: "thread.meta.update",        commandId: CommandId.make("cmd-bootstrap-thread-update"),        threadId: ThreadId.make("thread-bootstrap"),        title: "Updated Bootstrap Thread",      }),    );     expect(result.sequence).toBe(8);    expect(fullSnapshotReadCount).toBe(0);     await runtime.dispose();  });   it("persists deterministic read models for repeated snapshot reads", async () => {    const createdAt = now();    const system = await createOrchestrationSystem();    const { engine } = system;     await system.run(      engine.dispatch({        type: "project.create",        commandId: CommandId.make("cmd-project-1-create"),        projectId: asProjectId("project-1"),        title: "Project 1",        workspaceRoot: "/tmp/project-1",        defaultModelSelection: {          instanceId: ProviderInstanceId.make("codex"),          model: "gpt-5-codex",        },        createdAt,      }),    );    await system.run(      engine.dispatch({        type: "thread.create",        commandId: CommandId.make("cmd-thread-1-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: null,        worktreePath: null,        createdAt,      }),    );    await system.run(      engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.make("cmd-turn-start-1"),        threadId: ThreadId.make("thread-1"),        message: {          messageId: asMessageId("msg-1"),          role: "user",          text: "hello",          attachments: [],        },        interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,        runtimeMode: "approval-required",        createdAt,      }),    );     const readModelA = await system.readModel();    const readModelB = await system.readModel();    expect(readModelB).toEqual(readModelA);    await system.dispose();  });   it("archives and unarchives threads through orchestration commands", async () => {    const system = await createOrchestrationSystem();    const { engine } = system;    const createdAt = now();     await system.run(      engine.dispatch({        type: "project.create",        commandId: CommandId.make("cmd-project-archive-create"),        projectId: asProjectId("project-archive"),        title: "Project Archive",        workspaceRoot: "/tmp/project-archive",        defaultModelSelection: {          instanceId: ProviderInstanceId.make("codex"),          model: "gpt-5-codex",        },        createdAt,      }),    );    await system.run(      engine.dispatch({        type: "thread.create",        commandId: CommandId.make("cmd-thread-archive-create"),        threadId: ThreadId.make("thread-archive"),        projectId: asProjectId("project-archive"),        title: "Archive me",        modelSelection: {          instanceId: ProviderInstanceId.make("codex"),          model: "gpt-5-codex",        },        interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,        runtimeMode: "full-access",        branch: null,        worktreePath: null,        createdAt,      }),    );     await system.run(      engine.dispatch({        type: "thread.archive",        commandId: CommandId.make("cmd-thread-archive"),        threadId: ThreadId.make("thread-archive"),      }),    );    expect(      (await system.readModel()).threads.find((thread) => thread.id === "thread-archive")        ?.archivedAt,    ).not.toBeNull();     await system.run(      engine.dispatch({        type: "thread.unarchive",        commandId: CommandId.make("cmd-thread-unarchive"),        threadId: ThreadId.make("thread-archive"),      }),    );    expect(      (await system.readModel()).threads.find((thread) => thread.id === "thread-archive")        ?.archivedAt,    ).toBeNull();     await system.dispose();  });   it("replays append-only events from sequence", async () => {    const system = await createOrchestrationSystem();    const { engine } = system;    const createdAt = now();     await system.run(      engine.dispatch({        type: "project.create",        commandId: CommandId.make("cmd-project-replay-create"),        projectId: asProjectId("project-replay"),        title: "Replay Project",        workspaceRoot: "/tmp/project-replay",        defaultModelSelection: {          instanceId: ProviderInstanceId.make("codex"),          model: "gpt-5-codex",        },        createdAt,      }),    );    await system.run(      engine.dispatch({        type: "thread.create",        commandId: CommandId.make("cmd-thread-replay-create"),        threadId: ThreadId.make("thread-replay"),        projectId: asProjectId("project-replay"),        title: "replay",        modelSelection: {          instanceId: ProviderInstanceId.make("codex"),          model: "gpt-5-codex",        },        interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,        runtimeMode: "approval-required",        branch: null,        worktreePath: null,        createdAt,      }),    );    await system.run(      engine.dispatch({        type: "thread.delete",        commandId: CommandId.make("cmd-thread-replay-delete"),        threadId: ThreadId.make("thread-replay"),      }),    );     const events = await system.run(      Stream.runCollect(engine.readEvents(0)).pipe(        Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk)),      ),    );    expect(events.map((event) => event.type)).toEqual([      "project.created",      "thread.created",      "thread.deleted",    ]);    await system.dispose();  });   it("streams persisted domain events in order", async () => {    const system = await createOrchestrationSystem();    const { engine } = system;    const createdAt = now();     await system.run(      engine.dispatch({        type: "project.create",        commandId: CommandId.make("cmd-project-stream-create"),        projectId: asProjectId("project-stream"),        title: "Stream Project",        workspaceRoot: "/tmp/project-stream",        defaultModelSelection: {          instanceId: ProviderInstanceId.make("codex"),          model: "gpt-5-codex",        },        createdAt,      }),    );     const eventTypes: string[] = [];    await system.run(      Effect.gen(function* () {        const eventQueue = yield* Queue.unbounded<OrchestrationEvent>();        yield* Effect.forkScoped(          Stream.take(engine.streamDomainEvents, 2).pipe(            Stream.runForEach((event) => Queue.offer(eventQueue, event).pipe(Effect.asVoid)),          ),        );        yield* Effect.sleep("10 millis");        yield* engine.dispatch({          type: "thread.create",          commandId: CommandId.make("cmd-stream-thread-create"),          threadId: ThreadId.make("thread-stream"),          projectId: asProjectId("project-stream"),          title: "domain-stream",          modelSelection: {            instanceId: ProviderInstanceId.make("codex"),            model: "gpt-5-codex",          },          interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,          runtimeMode: "approval-required",          branch: null,          worktreePath: null,          createdAt,        });        yield* engine.dispatch({          type: "thread.meta.update",          commandId: CommandId.make("cmd-stream-thread-update"),          threadId: ThreadId.make("thread-stream"),          title: "domain-stream-updated",        });        eventTypes.push((yield* Queue.take(eventQueue)).type);        eventTypes.push((yield* Queue.take(eventQueue)).type);      }).pipe(Effect.scoped),    );     expect(eventTypes).toEqual(["thread.created", "thread.meta-updated"]);    await system.dispose();  });   it("records command ack duration using the first committed event type", async () => {    const system = await createOrchestrationSystem();    const { engine } = system;    const createdAt = now();     await system.run(      engine.dispatch({        type: "project.create",        commandId: CommandId.make("cmd-project-ack-create"),        projectId: asProjectId("project-ack"),        title: "Ack Project",        workspaceRoot: "/tmp/project-ack",        defaultModelSelection: {          instanceId: ProviderInstanceId.make("codex"),          model: "gpt-5-codex",        },        createdAt,      }),    );     await system.run(      engine.dispatch({

Lines 1–500 of 1083. Load more