// @effect-diagnostics nodeBuiltinImport:offimport * as NodeFSP from "node:fs/promises";import * as NodeOS from "node:os";import * as NodePath from "node:path"; import { ApprovalRequestId, EventId, CheckpointRef, CommandId, DEFAULT_PROVIDER_INTERACTION_MODE, MessageId, ProjectId, ThreadId, TurnId, type OrchestrationCommand, 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, vi } 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 { makeSqlitePersistenceLive, 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( databasePath?: string, repositoryIdentityResolver?: RepositoryIdentityResolver.RepositoryIdentityResolver["Service"],) { const persistence = databasePath ? makeSqlitePersistenceLive(databasePath) : SqlitePersistenceMemory; 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.succeed( RepositoryIdentityResolver.RepositoryIdentityResolver, repositoryIdentityResolver, ) : RepositoryIdentityResolver.layer, ), Layer.provide(persistence), Layer.provideMerge(ServerConfigLayer), Layer.provideMerge(NodeServices.layer), );} async function createOrchestrationSystem( databasePath?: string, repositoryIdentityResolver?: RepositoryIdentityResolver.RepositoryIdentityResolver["Service"],) { const runtime = ManagedRuntime.make( makeOrchestrationLayer(databasePath, repositoryIdentityResolver), ); const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); const snapshotQuery = await runtime.runPromise(Effect.service(ProjectionSnapshotQuery)); return { engine, readModel: () => runtime.runPromise(snapshotQuery.getSnapshot()), readThread: (threadId: ThreadId) => runtime.runPromise(snapshotQuery.getThreadDetailById(threadId)), 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.each(["running", "stopped"] as const)( "sends async answers with a %s session and rejects old duplicate replies", async (status) => { const directory = await NodeFSP.mkdtemp( NodePath.join(NodeOS.tmpdir(), "t3-async-questions-"), ); const databasePath = NodePath.join(directory, "state.sqlite"); let system = await createOrchestrationSystem(databasePath); const threadId = ThreadId.make("async-thread"); const projectId = ProjectId.make("async-project"); const requestId = ApprovalRequestId.make("codex-async:question-1"); try { await system.run( system.engine.dispatch({ type: "project.create", commandId: CommandId.make("async-project"), projectId, title: "Async questions", workspaceRoot: "/tmp/async-questions", createdAt: now(), }), ); await system.run( system.engine.dispatch({ type: "thread.create", commandId: CommandId.make("async-thread"), threadId, projectId, title: "Async questions", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, runtimeMode: "full-access", interactionMode: "default", branch: null, worktreePath: null, createdAt: now(), }), ); await system.run( system.engine.dispatch({ type: "thread.session.set", commandId: CommandId.make("async-session"), threadId, createdAt: now(), session: { threadId, status, providerName: "codex", runtimeMode: "full-access", activeTurnId: status === "running" ? TurnId.make("turn-1") : null, lastError: null, updatedAt: now(), }, }), ); await system.run( system.engine.dispatch({ type: "thread.activity.append", commandId: CommandId.make("async-question"), threadId, createdAt: now(), activity: { id: EventId.make("async-question"), kind: "user-input.requested", summary: "User input requested", tone: "info", turnId: TurnId.make("turn-1"), createdAt: now(), payload: { requestId, responseMode: "message", questions: [ { id: "0", header: "Question", question: "Which package manager?", options: [{ label: "pnpm", description: "" }], }, { id: "1", header: "Question", question: "What should it be named?", options: [], }, ], }, }, }), ); const appendWork = async (prefix: string, createdAt: string) => { for (let index = 0; index < 501; index += 1) { await system.run( system.engine.dispatch({ type: "thread.activity.append", commandId: CommandId.make(`${prefix}-${index}`), threadId, createdAt, activity: { id: EventId.make(`${prefix}-${index}`), kind: "tool.completed", summary: "Work continued", payload: {}, tone: "info", turnId: TurnId.make("turn-1"), createdAt, }, }), ); } }; await appendWork("work", "2026-01-01T00:00:01.000Z"); const before = await system.readModel(); expect( before.threads[0]?.activities.some((activity) => activity.id === "async-question"), ).toBe(true); if (status === "stopped") { await system.dispose(); system = await createOrchestrationSystem(databasePath); } const response = { type: "thread.user-input.respond" as const, commandId: CommandId.make("async-response"), threadId, requestId, answers: { "0": "pnpm", "1": "Example" }, attachmentsByQuestionId: { "1": [ { type: "file" as const, id: "thread-1-00000000-0000-4000-8000-0000000000aa-txt", name: "spec.txt", mimeType: "text/plain", sizeBytes: 4, }, ], }, createdAt: "2026-01-01T00:00:02.000Z", }; await expect( system.run( system.engine.dispatch({ ...response, commandId: CommandId.make("incomplete-answer"), answers: { "0": "pnpm" }, }), ), ).rejects.toThrow("Answer each question before sending."); await system.run(system.engine.dispatch(response)); const after = await system.readModel(); const userMessages = after.threads[0]?.messages.filter( (message) => message.role === "user", ); expect(userMessages).toHaveLength(1); expect(userMessages?.[0]?.attachments).toEqual(response.attachmentsByQuestionId["1"]); expect(userMessages?.[0]?.text).toBe( "Which package manager?\npnpm\n\nWhat should it be named?\nExample\nAttached file: spec.txt (thread-1-00000000-0000-4000-8000-0000000000aa-txt)", ); expect( after.threads[0]?.activities.find((activity) => activity.kind === "user-input.resolved") ?.payload, ).toMatchObject({ requestId, responseMode: "message", answers: response.answers }); const events = await system.run(Stream.runCollect(system.engine.readEvents(0))); expect( Array.from(events) .filter((event) => event.commandId === response.commandId) .map((event) => event.type), ).toEqual([ "thread.activity-appended", "thread.message-sent", "thread.turn-start-requested", ]); await expect( system.run( system.engine.dispatch({ ...response, commandId: CommandId.make("second-client-reply"), }), ), ).rejects.toThrow("This question has already been answered."); await appendWork("later-work", "2026-01-01T00:00:03.000Z"); const afterEviction = Option.getOrThrow(await system.readThread(threadId)); expect( afterEviction.activities.some((activity) => activity.kind === "user-input.resolved"), ).toBe(false); if (status === "stopped") { await system.dispose(); system = await createOrchestrationSystem(databasePath); } await expect( system.run( system.engine.dispatch({ ...response, commandId: CommandId.make("reply-after-eviction"), }), ), ).rejects.toThrow("This question has already been answered."); } finally { await system.dispose(); await NodeFSP.rm(directory, { recursive: true, force: true }); } }, ); 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), readAggregateRange: () => Stream.die("unused aggregate replay"), getAggregateReplayStats: () => Effect.die("unused aggregate replay stats"), }; 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, pullRequests: [], 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, { getUserInputActivity: () => Effect.die("unused"), 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()), getImportedAgentSessionSources: () => Effect.die("unused"), getThreadCheckpointContext: () => Effect.succeed(Option.none()), getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadRuntimeContext: () => Effect.die("unused"), getTurnStartMessage: () => Effect.die("unused"), 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");Lines 1–500 of 2126. Load more
// @effect-diagnostics nodeBuiltinImport:off import * as NodeFSP from "node:fs/promises"; import * as NodeOS from "node:os"; import * as NodePath from "node:path"; import { ApprovalRequestId, EventId, CheckpointRef, CommandId, DEFAULT_PROVIDER_INTERACTION_MODE, MessageId, ProjectId, ThreadId, TurnId, type OrchestrationCommand, 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, vi } 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 { makeSqlitePersistenceLive, 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( databasePath?: string, repositoryIdentityResolver?: RepositoryIdentityResolver.RepositoryIdentityResolver["Service"], ) { const persistence = databasePath ? makeSqlitePersistenceLive(databasePath) : SqlitePersistenceMemory; 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.succeed( RepositoryIdentityResolver.RepositoryIdentityResolver, repositoryIdentityResolver, ) : RepositoryIdentityResolver.layer, ), Layer.provide(persistence), Layer.provideMerge(ServerConfigLayer), Layer.provideMerge(NodeServices.layer), ); } async function createOrchestrationSystem( databasePath?: string, repositoryIdentityResolver?: RepositoryIdentityResolver.RepositoryIdentityResolver["Service"], ) { const runtime = ManagedRuntime.make( makeOrchestrationLayer(databasePath, repositoryIdentityResolver), ); const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); const snapshotQuery = await runtime.runPromise(Effect.service(ProjectionSnapshotQuery)); return { engine, readModel: () => runtime.runPromise(snapshotQuery.getSnapshot()), readThread: (threadId: ThreadId) => runtime.runPromise(snapshotQuery.getThreadDetailById(threadId)), 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.each(["running", "stopped"] as const)( "sends async answers with a %s session and rejects old duplicate replies", async (status) => { const directory = await NodeFSP.mkdtemp( NodePath.join(NodeOS.tmpdir(), "t3-async-questions-"), ); const databasePath = NodePath.join(directory, "state.sqlite"); let system = await createOrchestrationSystem(databasePath); const threadId = ThreadId.make("async-thread"); const projectId = ProjectId.make("async-project"); const requestId = ApprovalRequestId.make("codex-async:question-1"); try { await system.run( system.engine.dispatch({ type: "project.create", commandId: CommandId.make("async-project"), projectId, title: "Async questions", workspaceRoot: "/tmp/async-questions", createdAt: now(), }), ); await system.run( system.engine.dispatch({ type: "thread.create", commandId: CommandId.make("async-thread"), threadId, projectId, title: "Async questions", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, runtimeMode: "full-access", interactionMode: "default", branch: null, worktreePath: null, createdAt: now(), }), ); await system.run( system.engine.dispatch({ type: "thread.session.set", commandId: CommandId.make("async-session"), threadId, createdAt: now(), session: { threadId, status, providerName: "codex", runtimeMode: "full-access", activeTurnId: status === "running" ? TurnId.make("turn-1") : null, lastError: null, updatedAt: now(), }, }), ); await system.run( system.engine.dispatch({ type: "thread.activity.append", commandId: CommandId.make("async-question"), threadId, createdAt: now(), activity: { id: EventId.make("async-question"), kind: "user-input.requested", summary: "User input requested", tone: "info", turnId: TurnId.make("turn-1"), createdAt: now(), payload: { requestId, responseMode: "message", questions: [ { id: "0", header: "Question", question: "Which package manager?", options: [{ label: "pnpm", description: "" }], }, { id: "1", header: "Question", question: "What should it be named?", options: [], }, ], }, }, }), ); const appendWork = async (prefix: string, createdAt: string) => { for (let index = 0; index < 501; index += 1) { await system.run( system.engine.dispatch({ type: "thread.activity.append", commandId: CommandId.make(`${prefix}-${index}`), threadId, createdAt, activity: { id: EventId.make(`${prefix}-${index}`), kind: "tool.completed", summary: "Work continued", payload: {}, tone: "info", turnId: TurnId.make("turn-1"), createdAt, }, }), ); } }; await appendWork("work", "2026-01-01T00:00:01.000Z"); const before = await system.readModel(); expect( before.threads[0]?.activities.some((activity) => activity.id === "async-question"), ).toBe(true); if (status === "stopped") { await system.dispose(); system = await createOrchestrationSystem(databasePath); } const response = { type: "thread.user-input.respond" as const, commandId: CommandId.make("async-response"), threadId, requestId, answers: { "0": "pnpm", "1": "Example" }, attachmentsByQuestionId: { "1": [ { type: "file" as const, id: "thread-1-00000000-0000-4000-8000-0000000000aa-txt", name: "spec.txt", mimeType: "text/plain", sizeBytes: 4, }, ], }, createdAt: "2026-01-01T00:00:02.000Z", }; await expect( system.run( system.engine.dispatch({ ...response, commandId: CommandId.make("incomplete-answer"), answers: { "0": "pnpm" }, }), ), ).rejects.toThrow("Answer each question before sending."); await system.run(system.engine.dispatch(response)); const after = await system.readModel(); const userMessages = after.threads[0]?.messages.filter( (message) => message.role === "user", ); expect(userMessages).toHaveLength(1); expect(userMessages?.[0]?.attachments).toEqual(response.attachmentsByQuestionId["1"]); expect(userMessages?.[0]?.text).toBe( "Which package manager?\npnpm\n\nWhat should it be named?\nExample\nAttached file: spec.txt (thread-1-00000000-0000-4000-8000-0000000000aa-txt)", ); expect( after.threads[0]?.activities.find((activity) => activity.kind === "user-input.resolved") ?.payload, ).toMatchObject({ requestId, responseMode: "message", answers: response.answers }); const events = await system.run(Stream.runCollect(system.engine.readEvents(0))); expect( Array.from(events) .filter((event) => event.commandId === response.commandId) .map((event) => event.type), ).toEqual([ "thread.activity-appended", "thread.message-sent", "thread.turn-start-requested", ]); await expect( system.run( system.engine.dispatch({ ...response, commandId: CommandId.make("second-client-reply"), }), ), ).rejects.toThrow("This question has already been answered."); await appendWork("later-work", "2026-01-01T00:00:03.000Z"); const afterEviction = Option.getOrThrow(await system.readThread(threadId)); expect( afterEviction.activities.some((activity) => activity.kind === "user-input.resolved"), ).toBe(false); if (status === "stopped") { await system.dispose(); system = await createOrchestrationSystem(databasePath); } await expect( system.run( system.engine.dispatch({ ...response, commandId: CommandId.make("reply-after-eviction"), }), ), ).rejects.toThrow("This question has already been answered."); } finally { await system.dispose(); await NodeFSP.rm(directory, { recursive: true, force: true }); } }, ); 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), readAggregateRange: () => Stream.die("unused aggregate replay"), getAggregateReplayStats: () => Effect.die("unused aggregate replay stats"), }; 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, pullRequests: [], 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, { getUserInputActivity: () => Effect.die("unused"), 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()), getImportedAgentSessionSources: () => Effect.die("unused"), getThreadCheckpointContext: () => Effect.succeed(Option.none()), getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadRuntimeContext: () => Effect.die("unused"), getTurnStartMessage: () => Effect.die("unused"), 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");