import {  CheckpointRef,  CommandId,  DEFAULT_PROVIDER_INTERACTION_MODE,  MessageId,  ProjectId,  ThreadId,  TurnId,  type OrchestrationEvent,} from "@t3tools/contracts";import { Effect, Layer, ManagedRuntime, 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 { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";import {  OrchestrationProjectionPipeline,  type OrchestrationProjectionPipelineShape,} from "../Services/ProjectionPipeline.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(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();} describe("OrchestrationEngine", () => {  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("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("stores completed checkpoint summaries even when no files changed", async () => {    const system = await createOrchestrationSystem();    const { engine } = system;    const createdAt = now();     await system.run(      engine.dispatch({        type: "project.create",        commandId: CommandId.makeUnsafe("cmd-project-turn-diff-create"),        projectId: asProjectId("project-turn-diff"),        title: "Turn Diff Project",        workspaceRoot: "/tmp/project-turn-diff",        defaultModelSelection: {          provider: "codex",          model: "gpt-5-codex",        },        createdAt,      }),    );    await system.run(      engine.dispatch({        type: "thread.create",        commandId: CommandId.makeUnsafe("cmd-thread-turn-diff-create"),        threadId: ThreadId.makeUnsafe("thread-turn-diff"),        projectId: asProjectId("project-turn-diff"),        title: "Turn diff 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.diff.complete",        commandId: CommandId.makeUnsafe("cmd-turn-diff-complete"),        threadId: ThreadId.makeUnsafe("thread-turn-diff"),        turnId: asTurnId("turn-1"),        completedAt: createdAt,        checkpointRef: asCheckpointRef("refs/t3/checkpoints/thread-turn-diff/turn/1"),        status: "ready",        files: [],        checkpointTurnCount: 1,        createdAt,      }),    );     const thread = (await system.run(engine.getReadModel())).threads.find(      (entry) => entry.id === "thread-turn-diff",    );    expect(thread?.checkpoints).toEqual([      {        turnId: asTurnId("turn-1"),        checkpointTurnCount: 1,        checkpointRef: asCheckpointRef("refs/t3/checkpoints/thread-turn-diff/turn/1"),        status: "ready",        files: [],        assistantMessageId: null,        completedAt: createdAt,      },    ]);    await system.dispose();  });   it("keeps processing queued commands after a storage failure", async () => {    type StoredEvent =      ReturnType<OrchestrationEventStoreShape["append"]> extends Effect.Effect<infer A, any, any>        ? A        : never;    const events: StoredEvent[] = [];    let nextSequence = 1;    let shouldFailFirstAppend = true;     const flakyStore: OrchestrationEventStoreShape = {      append(event) {        if (shouldFailFirstAppend && event.commandId === CommandId.makeUnsafe("cmd-flaky-1")) {          shouldFailFirstAppend = false;          return Effect.fail(            new PersistenceSqlError({              operation: "test.append",              detail: "append failed",            }),          );        }        const savedEvent = {          ...event,          sequence: nextSequence,        } as StoredEvent;        nextSequence += 1;        events.push(savedEvent);        return Effect.succeed(savedEvent);      },      readFromSequence(sequenceExclusive) {        return Stream.fromIterable(events.filter((event) => event.sequence > sequenceExclusive));      },      readAll() {        return Stream.fromIterable(events);      },    };     const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), {      prefix: "t3-orchestration-engine-test-",    });     const runtime = ManagedRuntime.make(      OrchestrationEngineLive.pipe(        Layer.provide(OrchestrationProjectionPipelineLive),        Layer.provide(Layer.succeed(OrchestrationEventStore, flakyStore)),        Layer.provide(OrchestrationCommandReceiptRepositoryLive),        Layer.provide(SqlitePersistenceMemory),        Layer.provideMerge(ServerConfigLayer),        Layer.provideMerge(NodeServices.layer),      ),    );    const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));    const createdAt = now();     await runtime.runPromise(      engine.dispatch({        type: "project.create",        commandId: CommandId.makeUnsafe("cmd-project-flaky-create"),        projectId: asProjectId("project-flaky"),        title: "Flaky Project",        workspaceRoot: "/tmp/project-flaky",        defaultModelSelection: {          provider: "codex",          model: "gpt-5-codex",        },        createdAt,      }),    );     await expect(      runtime.runPromise(        engine.dispatch({          type: "thread.create",          commandId: CommandId.makeUnsafe("cmd-flaky-1"),          threadId: ThreadId.makeUnsafe("thread-flaky-fail"),          projectId: asProjectId("project-flaky"),          title: "flaky-fail",          modelSelection: {            provider: "codex",            model: "gpt-5-codex",          },          interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,          runtimeMode: "approval-required",          branch: null,          worktreePath: null,          createdAt,        }),      ),    ).rejects.toThrow("append failed");     const result = await runtime.runPromise(      engine.dispatch({        type: "thread.create",        commandId: CommandId.makeUnsafe("cmd-flaky-2"),        threadId: ThreadId.makeUnsafe("thread-flaky-ok"),        projectId: asProjectId("project-flaky"),        title: "flaky-ok",        modelSelection: {          provider: "codex",          model: "gpt-5-codex",        },        interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,        runtimeMode: "approval-required",        branch: null,        worktreePath: null,        createdAt,      }),    );     expect(result.sequence).toBe(2);    expect((await runtime.runPromise(engine.getReadModel())).snapshotSequence).toBe(2);    await runtime.dispose();  });   it("rolls back all events for a multi-event command when projection fails mid-dispatch", async () => {    let shouldFailRequestedProjection = true;    const flakyProjectionPipeline: OrchestrationProjectionPipelineShape = {      bootstrap: Effect.void,      projectEvent: (event) => {        if (          shouldFailRequestedProjection &&          event.commandId === CommandId.makeUnsafe("cmd-turn-start-atomic") &&          event.type === "thread.turn-start-requested"        ) {          shouldFailRequestedProjection = false;          return Effect.fail(            new PersistenceSqlError({              operation: "test.projection",              detail: "projection failed",            }),          );        }        return Effect.void;      },    };     const runtime = ManagedRuntime.make(      OrchestrationEngineLive.pipe(        Layer.provide(Layer.succeed(OrchestrationProjectionPipeline, flakyProjectionPipeline)),        Layer.provide(OrchestrationEventStoreLive),        Layer.provide(OrchestrationCommandReceiptRepositoryLive),        Layer.provide(SqlitePersistenceMemory),      ),    );    const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));    const createdAt = now();     await runtime.runPromise(      engine.dispatch({        type: "project.create",        commandId: CommandId.makeUnsafe("cmd-project-atomic-create"),        projectId: asProjectId("project-atomic"),        title: "Atomic Project",        workspaceRoot: "/tmp/project-atomic",        defaultModelSelection: {          provider: "codex",          model: "gpt-5-codex",        },        createdAt,      }),    );    await runtime.runPromise(      engine.dispatch({        type: "thread.create",        commandId: CommandId.makeUnsafe("cmd-thread-atomic-create"),        threadId: ThreadId.makeUnsafe("thread-atomic"),        projectId: asProjectId("project-atomic"),        title: "atomic",        modelSelection: {          provider: "codex",          model: "gpt-5-codex",        },        interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,        runtimeMode: "approval-required",        branch: null,        worktreePath: null,        createdAt,      }),    );     const turnStartCommand = {      type: "thread.turn.start" as const,      commandId: CommandId.makeUnsafe("cmd-turn-start-atomic"),      threadId: ThreadId.makeUnsafe("thread-atomic"),      message: {        messageId: asMessageId("msg-atomic-1"),        role: "user" as const,        text: "hello",        attachments: [],      },      interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,

Lines 1–500 of 740. Load more