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: (effect: Effect.Effect) => runtime.runPromise(effect), dispose: () => runtime.dispose(), }; } function now() { return "2026-01-01T00:00:00.000Z"; } const hasMetricSnapshot = ( snapshots: ReadonlyArray, id: string, attributes: Readonly>, ) => 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()), getThreadDetailSnapshot: () => 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(); 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({ type: "thread.create", commandId: CommandId.make("cmd-thread-ack-create"), threadId: ThreadId.make("thread-ack"), projectId: asProjectId("project-ack"), title: "Ack Thread", modelSelection: { instanceId: ProviderInstanceId.make("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.make("cmd-thread-missing-project"), threadId: ThreadId.make("thread-missing-project"), projectId: asProjectId("project-missing"), title: "Missing Project Thread", modelSelection: { instanceId: ProviderInstanceId.make("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( hasMetricSnapshot(snapshots, "t3_orchestration_commands_total", { commandType: "thread.create", aggregateKind: "thread", outcome: "failure", }), ).toBe(true); 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.make("cmd-project-turn-diff-create"), projectId: asProjectId("project-turn-diff"), title: "Turn Diff Project", workspaceRoot: "/tmp/project-turn-diff", defaultModelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", }, createdAt, }), ); await system.run( engine.dispatch({ type: "thread.create", commandId: CommandId.make("cmd-thread-turn-diff-create"), threadId: ThreadId.make("thread-turn-diff"), projectId: asProjectId("project-turn-diff"), title: "Turn diff 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.diff.complete", commandId: CommandId.make("cmd-turn-diff-complete"), threadId: ThreadId.make("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.readModel()).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 extends Effect.Effect ? A : never; const events: StoredEvent[] = []; let nextSequence = 1; let shouldFailFirstAppend = true; const flakyStore: OrchestrationEventStoreShape = { append(event) { if (shouldFailFirstAppend && event.commandId === CommandId.make("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(OrchestrationProjectionSnapshotQueryLive), Layer.provide(OrchestrationProjectionPipelineLive), Layer.provide(Layer.succeed(OrchestrationEventStore, flakyStore)), Layer.provide(OrchestrationCommandReceiptRepositoryLive), Layer.provide(RepositoryIdentityResolver.layer), 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.make("cmd-project-flaky-create"), projectId: asProjectId("project-flaky"), title: "Flaky Project", workspaceRoot: "/tmp/project-flaky", defaultModelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", }, createdAt, }), ); await expect( runtime.runPromise( engine.dispatch({ type: "thread.create", commandId: CommandId.make("cmd-flaky-1"), threadId: ThreadId.make("thread-flaky-fail"), projectId: asProjectId("project-flaky"), title: "flaky-fail", modelSelection: { instanceId: ProviderInstanceId.make("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.make("cmd-flaky-2"), threadId: ThreadId.make("thread-flaky-ok"), projectId: asProjectId("project-flaky"), title: "flaky-ok", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required", branch: null, worktreePath: null, createdAt, }), ); expect(result.sequence).toBe(2); const eventsAfterRetry = await runtime.runPromise( Stream.runCollect(engine.readEvents(0)).pipe( Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk)), ), ); expect(eventsAfterRetry.map((event) => event.type)).toEqual([ "project.created", "thread.created", ]); 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.make("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(OrchestrationProjectionSnapshotQueryLive), Layer.provide(Layer.succeed(OrchestrationProjectionPipeline, flakyProjectionPipeline)), Layer.provide(OrchestrationEventStoreLive), Layer.provide(OrchestrationCommandReceiptRepositoryLive), Layer.provide(RepositoryIdentityResolver.layer), Layer.provide(SqlitePersistenceMemory), Layer.provide(NodeServices.layer), ), ); const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); const createdAt = now(); await runtime.runPromise( engine.dispatch({ type: "project.create", commandId: CommandId.make("cmd-project-atomic-create"), projectId: asProjectId("project-atomic"), title: "Atomic Project", workspaceRoot: "/tmp/project-atomic", defaultModelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", }, createdAt, }), ); await runtime.runPromise( engine.dispatch({ type: "thread.create", commandId: CommandId.make("cmd-thread-atomic-create"), threadId: ThreadId.make("thread-atomic"), projectId: asProjectId("project-atomic"), title: "atomic", modelSelection: { instanceId: ProviderInstanceId.make("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.make("cmd-turn-start-atomic"), threadId: ThreadId.make("thread-atomic"), message: { messageId: asMessageId("msg-atomic-1"), role: "user" as const, text: "hello", attachments: [], }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required" as const, createdAt, }; await expect(runtime.runPromise(engine.dispatch(turnStartCommand))).rejects.toThrow( "projection failed", ); const eventsAfterFailure = await runtime.runPromise( Stream.runCollect(engine.readEvents(0)).pipe( Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk)), ), ); expect(eventsAfterFailure.map((event) => event.type)).toEqual([ "project.created", "thread.created", ]); const retryResult = await runtime.runPromise(engine.dispatch(turnStartCommand)); expect(retryResult.sequence).toBe(4); const eventsAfterRetry = await runtime.runPromise( Stream.runCollect(engine.readEvents(0)).pipe( Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk)), ), ); expect(eventsAfterRetry.map((event) => event.type)).toEqual([ "project.created", "thread.created", "thread.message-sent", "thread.turn-start-requested", ]); expect( eventsAfterRetry.filter((event) => event.commandId === turnStartCommand.commandId), ).toHaveLength(2); await runtime.dispose(); }); it("reconciles command state when append persists but projection fails", async () => { type StoredEvent = ReturnType extends Effect.Effect ? A : never; const events: StoredEvent[] = []; let nextSequence = 1; const nonTransactionalStore: OrchestrationEventStoreShape = { append(event) { 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); }, }; let shouldFailProjection = true; const flakyProjectionPipeline: OrchestrationProjectionPipelineShape = { bootstrap: Effect.void, projectEvent: (event) => { if ( shouldFailProjection && event.commandId === CommandId.make("cmd-thread-archive-sync-fail") ) { shouldFailProjection = false; return Effect.fail( new PersistenceSqlError({ operation: "test.projection", detail: "projection failed", }), ); } return Effect.void; }, }; const runtime = ManagedRuntime.make( OrchestrationEngineLive.pipe( Layer.provide(OrchestrationProjectionSnapshotQueryLive), Layer.provide(Layer.succeed(OrchestrationProjectionPipeline, flakyProjectionPipeline)), Layer.provide(Layer.succeed(OrchestrationEventStore, nonTransactionalStore)), Layer.provide(OrchestrationCommandReceiptRepositoryLive), Layer.provide(RepositoryIdentityResolver.layer), Layer.provide(SqlitePersistenceMemory), Layer.provide(NodeServices.layer), ), ); const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); const createdAt = now(); await runtime.runPromise( engine.dispatch({ type: "project.create", commandId: CommandId.make("cmd-project-sync-create"), projectId: asProjectId("project-sync"), title: "Sync Project", workspaceRoot: "/tmp/project-sync", defaultModelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", }, createdAt, }), ); await runtime.runPromise( engine.dispatch({ type: "thread.create", commandId: CommandId.make("cmd-thread-sync-create"), threadId: ThreadId.make("thread-sync"), projectId: asProjectId("project-sync"), title: "sync-before", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required", branch: null, worktreePath: null, createdAt, }), ); await expect( runtime.runPromise( engine.dispatch({ type: "thread.archive", commandId: CommandId.make("cmd-thread-archive-sync-fail"), threadId: ThreadId.make("thread-sync"), }), ), ).rejects.toThrow("projection failed"); await expect( runtime.runPromise( engine.dispatch({ type: "thread.archive", commandId: CommandId.make("cmd-thread-archive-sync-retry"), threadId: ThreadId.make("thread-sync"), }), ), ).rejects.toThrow("already archived"); await runtime.dispose(); }); it("fails command dispatch when command invariants are violated", async () => { const system = await createOrchestrationSystem(); const { engine } = system; await expect( system.run( engine.dispatch({ type: "thread.turn.start", commandId: CommandId.make("cmd-invariant-missing-thread"), threadId: ThreadId.make("thread-missing"), message: { messageId: asMessageId("msg-missing"), role: "user", text: "hello", attachments: [], }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required", createdAt: now(), }), ), ).rejects.toThrow("Thread 'thread-missing' does not exist"); await system.dispose(); }); it("rejects duplicate thread creation", async () => { const system = await createOrchestrationSystem(); const { engine } = system; const createdAt = now(); await system.run( engine.dispatch({ type: "project.create", commandId: CommandId.make("cmd-project-duplicate-create"), projectId: asProjectId("project-duplicate"), title: "Duplicate Project", workspaceRoot: "/tmp/project-duplicate", defaultModelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", }, createdAt, }), ); await system.run( engine.dispatch({ type: "thread.create", commandId: CommandId.make("cmd-thread-duplicate-1"), threadId: ThreadId.make("thread-duplicate"), projectId: asProjectId("project-duplicate"), title: "duplicate", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required", branch: null, worktreePath: null, createdAt, }), ); await expect( system.run( engine.dispatch({ type: "thread.create", commandId: CommandId.make("cmd-thread-duplicate-2"), threadId: ThreadId.make("thread-duplicate"), projectId: asProjectId("project-duplicate"), title: "duplicate", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required", branch: null, worktreePath: null, createdAt, }), ), ).rejects.toThrow("already exists"); await system.dispose(); }); });