import {  CheckpointRef,  CommandId,  DEFAULT_PROVIDER_INTERACTION_MODE,  MessageId,  ProjectId,  ThreadId,  TurnId,  type OrchestrationEvent,} from "@t3tools/contracts";import { Effect, Layer, ManagedRuntime, Metric, Option, Queue, Stream } from "effect";import { describe, expect, it } from "vitest"; 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 { 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";import * as NodeServices from "@effect/platform-node/NodeServices"; const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value);const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value);const asCheckpointRef = (value: string): CheckpointRef => CheckpointRef.makeUnsafe(value); async function createOrchestrationSystem() {  const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), {    prefix: "t3-orchestration-engine-test-",  });  const orchestrationLayer = OrchestrationEngineLive.pipe(    Layer.provide(OrchestrationProjectionSnapshotQueryLive),    Layer.provide(OrchestrationProjectionPipelineLive),    Layer.provide(OrchestrationEventStoreLive),    Layer.provide(OrchestrationCommandReceiptRepositoryLive),    Layer.provide(SqlitePersistenceMemory),    Layer.provideMerge(ServerConfigLayer),    Layer.provideMerge(NodeServices.layer),  );  const runtime = ManagedRuntime.make(orchestrationLayer);  const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));  return {    engine,    run: <A, E>(effect: Effect.Effect<A, E>) => runtime.runPromise(effect),    dispose: () => runtime.dispose(),  };} function now() {  return new Date().toISOString();} 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 the in-memory read model from persisted projections", async () => {    const failOnHistoricalReplayStore: OrchestrationEventStoreShape = {      append: () =>        Effect.fail(          new PersistenceSqlError({            operation: "test.append",            detail: "append should not be called during bootstrap",          }),        ),      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: {            provider: "codex" as const,            model: "gpt-5-codex",          },          scripts: [],          createdAt: "2026-03-03T00:00:00.000Z",          updatedAt: "2026-03-03T00:00:01.000Z",          deletedAt: null,        },      ],      threads: [        {          id: ThreadId.makeUnsafe("thread-bootstrap"),          projectId: asProjectId("project-bootstrap"),          title: "Bootstrap Thread",          modelSelection: {            provider: "codex" as const,            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 layer = OrchestrationEngineLive.pipe(      Layer.provide(        Layer.succeed(ProjectionSnapshotQuery, {          getSnapshot: () => Effect.succeed(projectionSnapshot),          getCounts: () => Effect.succeed({ projectCount: 1, threadCount: 1 }),          getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),          getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),          getThreadCheckpointContext: () => Effect.succeed(Option.none()),        }),      ),      Layer.provide(        Layer.succeed(OrchestrationProjectionPipeline, {          bootstrap: Effect.void,          projectEvent: () => Effect.void,        } satisfies OrchestrationProjectionPipelineShape),      ),      Layer.provide(Layer.succeed(OrchestrationEventStore, failOnHistoricalReplayStore)),      Layer.provide(OrchestrationCommandReceiptRepositoryLive),      Layer.provide(SqlitePersistenceMemory),    );     const runtime = ManagedRuntime.make(layer);     const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));    const readModel = await runtime.runPromise(engine.getReadModel());     expect(readModel.snapshotSequence).toBe(7);    expect(readModel.projects).toHaveLength(1);    expect(readModel.projects[0]?.title).toBe("Bootstrap Project");    expect(readModel.threads).toHaveLength(1);    expect(readModel.threads[0]?.title).toBe("Bootstrap Thread");     await runtime.dispose();  });   it("returns deterministic read models for repeated reads", async () => {    const createdAt = now();    const system = await createOrchestrationSystem();    const { engine } = system;     await system.run(      engine.dispatch({        type: "project.create",        commandId: CommandId.makeUnsafe("cmd-project-1-create"),        projectId: asProjectId("project-1"),        title: "Project 1",        workspaceRoot: "/tmp/project-1",        defaultModelSelection: {          provider: "codex",          model: "gpt-5-codex",        },        createdAt,      }),    );    await system.run(      engine.dispatch({        type: "thread.create",        commandId: CommandId.makeUnsafe("cmd-thread-1-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: null,        createdAt,      }),    );    await system.run(      engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.makeUnsafe("cmd-turn-start-1"),        threadId: ThreadId.makeUnsafe("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.run(engine.getReadModel());    const readModelB = await system.run(engine.getReadModel());    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.makeUnsafe("cmd-project-archive-create"),        projectId: asProjectId("project-archive"),        title: "Project Archive",        workspaceRoot: "/tmp/project-archive",        defaultModelSelection: {          provider: "codex",          model: "gpt-5-codex",        },        createdAt,      }),    );    await system.run(      engine.dispatch({        type: "thread.create",        commandId: CommandId.makeUnsafe("cmd-thread-archive-create"),        threadId: ThreadId.makeUnsafe("thread-archive"),        projectId: asProjectId("project-archive"),        title: "Archive me",        modelSelection: {          provider: "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.makeUnsafe("cmd-thread-archive"),        threadId: ThreadId.makeUnsafe("thread-archive"),      }),    );    expect(      (await system.run(engine.getReadModel())).threads.find(        (thread) => thread.id === "thread-archive",      )?.archivedAt,    ).not.toBeNull();     await system.run(      engine.dispatch({        type: "thread.unarchive",        commandId: CommandId.makeUnsafe("cmd-thread-unarchive"),        threadId: ThreadId.makeUnsafe("thread-archive"),      }),    );    expect(      (await system.run(engine.getReadModel())).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.makeUnsafe("cmd-project-replay-create"),        projectId: asProjectId("project-replay"),        title: "Replay Project",        workspaceRoot: "/tmp/project-replay",        defaultModelSelection: {          provider: "codex",          model: "gpt-5-codex",        },        createdAt,      }),    );    await system.run(      engine.dispatch({        type: "thread.create",        commandId: CommandId.makeUnsafe("cmd-thread-replay-create"),        threadId: ThreadId.makeUnsafe("thread-replay"),        projectId: asProjectId("project-replay"),        title: "replay",        modelSelection: {          provider: "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.makeUnsafe("cmd-thread-replay-delete"),        threadId: ThreadId.makeUnsafe("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.makeUnsafe("cmd-project-stream-create"),        projectId: asProjectId("project-stream"),        title: "Stream Project",        workspaceRoot: "/tmp/project-stream",        defaultModelSelection: {          provider: "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.makeUnsafe("cmd-stream-thread-create"),          threadId: ThreadId.makeUnsafe("thread-stream"),          projectId: asProjectId("project-stream"),          title: "domain-stream",          modelSelection: {            provider: "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.makeUnsafe("cmd-stream-thread-update"),          threadId: ThreadId.makeUnsafe("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.makeUnsafe("cmd-project-ack-create"),        projectId: asProjectId("project-ack"),        title: "Ack Project",        workspaceRoot: "/tmp/project-ack",        defaultModelSelection: {          provider: "codex",          model: "gpt-5-codex",        },        createdAt,      }),    );     await system.run(      engine.dispatch({        type: "thread.create",        commandId: CommandId.makeUnsafe("cmd-thread-ack-create"),        threadId: ThreadId.makeUnsafe("thread-ack"),        projectId: asProjectId("project-ack"),        title: "Ack Thread",        modelSelection: {          provider: "codex",          model: "gpt-5-codex",        },        interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,        runtimeMode: "full-access",        branch: null,        worktreePath: null,        createdAt,      }),    );     const snapshots = await system.run(Metric.snapshot);    expect(      hasMetricSnapshot(snapshots, "t3_orchestration_command_ack_duration", {        commandType: "thread.create",        aggregateKind: "thread",        ackEventType: "thread.created",      }),    ).toBe(true);     await system.dispose();  });   it("records failed command dispatches as metric failures", async () => {    const system = await createOrchestrationSystem();    const { engine } = system;    const createdAt = now();     await expect(      system.run(        engine.dispatch({          type: "thread.create",          commandId: CommandId.makeUnsafe("cmd-thread-missing-project"),          threadId: ThreadId.makeUnsafe("thread-missing-project"),          projectId: asProjectId("project-missing"),          title: "Missing Project Thread",          modelSelection: {            provider: "codex",            model: "gpt-5-codex",          },          interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,          runtimeMode: "full-access",          branch: null,          worktreePath: null,          createdAt,        }),      ),    ).rejects.toThrow("does not exist");     const snapshots = await system.run(Metric.snapshot);    expect(

Lines 1–500 of 1012. Load more