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 { it as effectIt } from "@effect/vitest";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 { TestClock } from "effect/testing";import { describe, expect, it } from "vite-plus/test"; import { PersistenceSqlError } from "../../persistence/Errors.ts";import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts";import * as OrchestrationCommandReceipts from "../../persistence/Services/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 * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts";import * as ThreadPlanProgress from "../ThreadPlanProgress.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); function makeOrchestrationLayer() { const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), { prefix: "t3-orchestration-engine-test-", }); return Layer.mergeAll( OrchestrationEngineLive.pipe( Layer.provide(OrchestrationProjectionSnapshotQueryLive), Layer.provide(OrchestrationProjectionPipelineLive), ), OrchestrationProjectionSnapshotQueryLive, ).pipe( Layer.provideMerge(ThreadBackgroundLiveness.layer), Layer.provide(ThreadPlanProgress.layer), Layer.provide(OrchestrationEventStoreLive), Layer.provideMerge(OrchestrationCommandReceiptRepositoryLive), Layer.provide(RepositoryIdentityResolver.layer), Layer.provide(SqlitePersistenceMemory), Layer.provideMerge(ServerConfigLayer), Layer.provideMerge(NodeServices.layer), );} async function createOrchestrationSystem() { const runtime = ManagedRuntime.make(makeOrchestrationLayer()); 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", }), ), hasEventAfter: () => Effect.succeed(false), }; 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, settledOverride: null, settledAt: 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 }), getEventReplayStats: () => Effect.die("unused"), 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()), searchThreads: () => Effect.succeed({ matches: [] }), }), ), Layer.provide( Layer.succeed(OrchestrationProjectionPipeline, { bootstrap: Effect.void, projectEvent: () => Effect.void, projectEventDeferred: () => Effect.succeed(Effect.void), } satisfies OrchestrationProjectionPipelineShape), ), Layer.provide(Layer.succeed(OrchestrationEventStore, eventStore)), Layer.provide(ThreadBackgroundLiveness.layer), Layer.provide(OrchestrationCommandReceiptRepositoryLive), Layer.provide(SqlitePersistenceMemory), Layer.provideMerge(NodeServices.layer), ); const runtime = ManagedRuntime.make(layer); const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); expect(await runtime.runPromise(engine.latestSequence)).toBe(7); 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(await runtime.runPromise(engine.latestSequence)).toBe(8); expect(fullSnapshotReadCount).toBe(0); await runtime.dispose(); }); effectIt.effect("preserves the blocked-settle error and persists its rejected receipt", () => Effect.gen(function* () { const engine = yield* OrchestrationEngineService; const receipts = yield* OrchestrationCommandReceipts.OrchestrationCommandReceiptRepository; const projectId = ProjectId.make("project-blocked-settle"); const threadId = ThreadId.make("thread-blocked-settle"); const commandId = CommandId.make("cmd-blocked-settle"); const createdAt = now(); yield* engine.dispatch({ type: "project.create", commandId: CommandId.make("cmd-blocked-settle-project-create"), projectId, title: "Project", workspaceRoot: "/tmp/project-blocked-settle", createdAt, }); yield* engine.dispatch({ type: "thread.create", commandId: CommandId.make("cmd-blocked-settle-thread-create"), threadId, projectId, title: "Thread", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "full-access", branch: null, worktreePath: null, createdAt, }); yield* engine.dispatch({ type: "thread.session.set", commandId: CommandId.make("cmd-blocked-settle-session-set"), threadId, createdAt, session: { threadId, status: "running", providerName: "codex", runtimeMode: "full-access", activeTurnId: null, lastError: null, updatedAt: createdAt, }, }); const sequence = yield* engine.latestSequence; const error = yield* engine .dispatch({ type: "thread.settle", commandId, threadId }) .pipe(Effect.flip); const message = "This thread still needs attention. Resolve or interrupt it first, then try again."; expect(error).toMatchObject({ _tag: "OrchestrationThreadSettleBlockedError", threadId, message, }); expect(Option.getOrNull(yield* receipts.getByCommandId({ commandId }))).toMatchObject({ commandId, aggregateKind: "thread", aggregateId: threadId, status: "rejected", error: message, resultSequence: sequence, }); expect(yield* engine.latestSequence).toBe(sequence); }).pipe(Effect.provide(makeOrchestrationLayer())), ); effectIt.effect( "rejects persisted changes and live background work without blocking unrelated threads", () => Effect.gen(function* () { yield* TestClock.setTime(Date.parse(now())); const engine = yield* OrchestrationEngineService; const snapshots = yield* ProjectionSnapshotQuery; const backgroundLiveness = yield* ThreadBackgroundLiveness.ThreadBackgroundLivenessService; const projectId = ProjectId.make("project-auto-settle-guard"); const guardedThreadId = ThreadId.make("thread-auto-settle-guarded"); const unrelatedThreadId = ThreadId.make("thread-auto-settle-unrelated"); const liveThreadId = ThreadId.make("thread-auto-settle-live"); yield* engine.dispatch({ type: "project.create", commandId: CommandId.make("cmd-auto-settle-guard-project"), projectId, title: "Project", workspaceRoot: "/tmp/project-auto-settle-guard", createdAt: now(), }); for (const threadId of [guardedThreadId, unrelatedThreadId, liveThreadId]) { yield* engine.dispatch({ type: "thread.create", commandId: CommandId.make(`cmd-create-${threadId}`), threadId, projectId, title: "Thread", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "full-access", branch: null, worktreePath: null, createdAt: now(), }); } const beforeUpdate = yield* snapshots.getSnapshot(); const snapshotSequence = beforeUpdate.snapshotSequence; const originalUpdatedAt = beforeUpdate.threads.find( (thread) => thread.id === guardedThreadId, )?.updatedAt; yield* engine.dispatch({ type: "thread.meta.update", commandId: CommandId.make("cmd-auto-settle-guard-meta"), threadId: guardedThreadId, branch: "new-branch", }); const afterUpdate = yield* snapshots.getSnapshot(); expect(afterUpdate.threads.find((thread) => thread.id === guardedThreadId)?.updatedAt).toBe( originalUpdatedAt, ); // Automatic settlement stamps the last activity, never the sweep time. const lastActivityAt = "2025-12-20T00:00:00.000Z"; const staleError = yield* engine .dispatch({ type: "thread.auto-settle", commandId: CommandId.make("cmd-auto-settle-stale-snapshot"), threadId: guardedThreadId, snapshotSequence, settledAt: lastActivityAt, }) .pipe(Effect.flip); expect(staleError._tag).toBe("OrchestrationCommandInvariantError"); const livenessSnapshotSequence = yield* engine.latestSequence; for (const [taskType, expectedLiveness] of [ ["subagent", "working"], ["local_bash", "monitoring"], ] as const) { backgroundLiveness.recordTaskLiveness({ threadId: liveThreadId, taskId: `task-${expectedLiveness}`, taskType, status: undefined, kind: "started", }); expect(backgroundLiveness.getThreadBackgroundLiveness(liveThreadId)).toBe( expectedLiveness, ); expect(yield* engine.latestSequence).toBe(livenessSnapshotSequence); const livenessError = yield* engine .dispatch({ type: "thread.auto-settle", commandId: CommandId.make(`cmd-auto-settle-${expectedLiveness}`), threadId: liveThreadId, snapshotSequence: livenessSnapshotSequence, settledAt: lastActivityAt, }) .pipe(Effect.flip); expect(livenessError._tag).toBe("OrchestrationCommandInvariantError"); expect(yield* engine.latestSequence).toBe(livenessSnapshotSequence); backgroundLiveness.clearThreadLiveness(liveThreadId); } yield* engine.dispatch({ type: "thread.auto-settle", commandId: CommandId.make("cmd-auto-settle-after-liveness-cleared"), threadId: liveThreadId, snapshotSequence: livenessSnapshotSequence, settledAt: lastActivityAt, }); const freshSnapshotSequence = yield* engine.latestSequence; yield* engine.dispatch({ type: "thread.meta.update", commandId: CommandId.make("cmd-auto-settle-unrelated-meta"), threadId: unrelatedThreadId, title: "Unrelated update", }); yield* engine.dispatch({ type: "thread.auto-settle", commandId: CommandId.make("cmd-auto-settle-after-unrelated-update"), threadId: guardedThreadId, snapshotSequence: freshSnapshotSequence, settledAt: lastActivityAt, }); const settled = yield* snapshots.getSnapshot(); for (const threadId of [guardedThreadId, liveThreadId]) { const thread = settled.threads.find((candidate) => candidate.id === threadId); expect(thread?.settledOverride).toBe("settled"); expect(thread?.settledAt).toBe(lastActivityAt); expect(thread?.updatedAt).toBe(now()); } }).pipe(Effect.provide(makeOrchestrationLayer())), ); 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"),Lines 1–500 of 1652. Load more
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 { it as effectIt } from "@effect/vitest"; 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 { TestClock } from "effect/testing"; import { describe, expect, it } from "vite-plus/test"; import { PersistenceSqlError } from "../../persistence/Errors.ts"; import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts"; import * as OrchestrationCommandReceipts from "../../persistence/Services/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 * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts"; import * as ThreadPlanProgress from "../ThreadPlanProgress.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); function makeOrchestrationLayer() { const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), { prefix: "t3-orchestration-engine-test-", }); return Layer.mergeAll( OrchestrationEngineLive.pipe( Layer.provide(OrchestrationProjectionSnapshotQueryLive), Layer.provide(OrchestrationProjectionPipelineLive), ), OrchestrationProjectionSnapshotQueryLive, ).pipe( Layer.provideMerge(ThreadBackgroundLiveness.layer), Layer.provide(ThreadPlanProgress.layer), Layer.provide(OrchestrationEventStoreLive), Layer.provideMerge(OrchestrationCommandReceiptRepositoryLive), Layer.provide(RepositoryIdentityResolver.layer), Layer.provide(SqlitePersistenceMemory), Layer.provideMerge(ServerConfigLayer), Layer.provideMerge(NodeServices.layer), ); } async function createOrchestrationSystem() { const runtime = ManagedRuntime.make(makeOrchestrationLayer()); 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", }), ), hasEventAfter: () => Effect.succeed(false), }; 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, settledOverride: null, settledAt: 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 }), getEventReplayStats: () => Effect.die("unused"), 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()), searchThreads: () => Effect.succeed({ matches: [] }), }), ), Layer.provide( Layer.succeed(OrchestrationProjectionPipeline, { bootstrap: Effect.void, projectEvent: () => Effect.void, projectEventDeferred: () => Effect.succeed(Effect.void), } satisfies OrchestrationProjectionPipelineShape), ), Layer.provide(Layer.succeed(OrchestrationEventStore, eventStore)), Layer.provide(ThreadBackgroundLiveness.layer), Layer.provide(OrchestrationCommandReceiptRepositoryLive), Layer.provide(SqlitePersistenceMemory), Layer.provideMerge(NodeServices.layer), ); const runtime = ManagedRuntime.make(layer); const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); expect(await runtime.runPromise(engine.latestSequence)).toBe(7); 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(await runtime.runPromise(engine.latestSequence)).toBe(8); expect(fullSnapshotReadCount).toBe(0); await runtime.dispose(); }); effectIt.effect("preserves the blocked-settle error and persists its rejected receipt", () => Effect.gen(function* () { const engine = yield* OrchestrationEngineService; const receipts = yield* OrchestrationCommandReceipts.OrchestrationCommandReceiptRepository; const projectId = ProjectId.make("project-blocked-settle"); const threadId = ThreadId.make("thread-blocked-settle"); const commandId = CommandId.make("cmd-blocked-settle"); const createdAt = now(); yield* engine.dispatch({ type: "project.create", commandId: CommandId.make("cmd-blocked-settle-project-create"), projectId, title: "Project", workspaceRoot: "/tmp/project-blocked-settle", createdAt, }); yield* engine.dispatch({ type: "thread.create", commandId: CommandId.make("cmd-blocked-settle-thread-create"), threadId, projectId, title: "Thread", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "full-access", branch: null, worktreePath: null, createdAt, }); yield* engine.dispatch({ type: "thread.session.set", commandId: CommandId.make("cmd-blocked-settle-session-set"), threadId, createdAt, session: { threadId, status: "running", providerName: "codex", runtimeMode: "full-access", activeTurnId: null, lastError: null, updatedAt: createdAt, }, }); const sequence = yield* engine.latestSequence; const error = yield* engine .dispatch({ type: "thread.settle", commandId, threadId }) .pipe(Effect.flip); const message = "This thread still needs attention. Resolve or interrupt it first, then try again."; expect(error).toMatchObject({ _tag: "OrchestrationThreadSettleBlockedError", threadId, message, }); expect(Option.getOrNull(yield* receipts.getByCommandId({ commandId }))).toMatchObject({ commandId, aggregateKind: "thread", aggregateId: threadId, status: "rejected", error: message, resultSequence: sequence, }); expect(yield* engine.latestSequence).toBe(sequence); }).pipe(Effect.provide(makeOrchestrationLayer())), ); effectIt.effect( "rejects persisted changes and live background work without blocking unrelated threads", () => Effect.gen(function* () { yield* TestClock.setTime(Date.parse(now())); const engine = yield* OrchestrationEngineService; const snapshots = yield* ProjectionSnapshotQuery; const backgroundLiveness = yield* ThreadBackgroundLiveness.ThreadBackgroundLivenessService; const projectId = ProjectId.make("project-auto-settle-guard"); const guardedThreadId = ThreadId.make("thread-auto-settle-guarded"); const unrelatedThreadId = ThreadId.make("thread-auto-settle-unrelated"); const liveThreadId = ThreadId.make("thread-auto-settle-live"); yield* engine.dispatch({ type: "project.create", commandId: CommandId.make("cmd-auto-settle-guard-project"), projectId, title: "Project", workspaceRoot: "/tmp/project-auto-settle-guard", createdAt: now(), }); for (const threadId of [guardedThreadId, unrelatedThreadId, liveThreadId]) { yield* engine.dispatch({ type: "thread.create", commandId: CommandId.make(`cmd-create-${threadId}`), threadId, projectId, title: "Thread", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "full-access", branch: null, worktreePath: null, createdAt: now(), }); } const beforeUpdate = yield* snapshots.getSnapshot(); const snapshotSequence = beforeUpdate.snapshotSequence; const originalUpdatedAt = beforeUpdate.threads.find( (thread) => thread.id === guardedThreadId, )?.updatedAt; yield* engine.dispatch({ type: "thread.meta.update", commandId: CommandId.make("cmd-auto-settle-guard-meta"), threadId: guardedThreadId, branch: "new-branch", }); const afterUpdate = yield* snapshots.getSnapshot(); expect(afterUpdate.threads.find((thread) => thread.id === guardedThreadId)?.updatedAt).toBe( originalUpdatedAt, ); // Automatic settlement stamps the last activity, never the sweep time. const lastActivityAt = "2025-12-20T00:00:00.000Z"; const staleError = yield* engine .dispatch({ type: "thread.auto-settle", commandId: CommandId.make("cmd-auto-settle-stale-snapshot"), threadId: guardedThreadId, snapshotSequence, settledAt: lastActivityAt, }) .pipe(Effect.flip); expect(staleError._tag).toBe("OrchestrationCommandInvariantError"); const livenessSnapshotSequence = yield* engine.latestSequence; for (const [taskType, expectedLiveness] of [ ["subagent", "working"], ["local_bash", "monitoring"], ] as const) { backgroundLiveness.recordTaskLiveness({ threadId: liveThreadId, taskId: `task-${expectedLiveness}`, taskType, status: undefined, kind: "started", }); expect(backgroundLiveness.getThreadBackgroundLiveness(liveThreadId)).toBe( expectedLiveness, ); expect(yield* engine.latestSequence).toBe(livenessSnapshotSequence); const livenessError = yield* engine .dispatch({ type: "thread.auto-settle", commandId: CommandId.make(`cmd-auto-settle-${expectedLiveness}`), threadId: liveThreadId, snapshotSequence: livenessSnapshotSequence, settledAt: lastActivityAt, }) .pipe(Effect.flip); expect(livenessError._tag).toBe("OrchestrationCommandInvariantError"); expect(yield* engine.latestSequence).toBe(livenessSnapshotSequence); backgroundLiveness.clearThreadLiveness(liveThreadId); } yield* engine.dispatch({ type: "thread.auto-settle", commandId: CommandId.make("cmd-auto-settle-after-liveness-cleared"), threadId: liveThreadId, snapshotSequence: livenessSnapshotSequence, settledAt: lastActivityAt, }); const freshSnapshotSequence = yield* engine.latestSequence; yield* engine.dispatch({ type: "thread.meta.update", commandId: CommandId.make("cmd-auto-settle-unrelated-meta"), threadId: unrelatedThreadId, title: "Unrelated update", }); yield* engine.dispatch({ type: "thread.auto-settle", commandId: CommandId.make("cmd-auto-settle-after-unrelated-update"), threadId: guardedThreadId, snapshotSequence: freshSnapshotSequence, settledAt: lastActivityAt, }); const settled = yield* snapshots.getSnapshot(); for (const threadId of [guardedThreadId, liveThreadId]) { const thread = settled.threads.find((candidate) => candidate.id === threadId); expect(thread?.settledOverride).toBe("settled"); expect(thread?.settledAt).toBe(lastActivityAt); expect(thread?.updatedAt).toBe(now()); } }).pipe(Effect.provide(makeOrchestrationLayer())), ); 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"),