import { ApprovalRequestId, CheckpointRef, CommandId, CorrelationId, DEFAULT_PROVIDER_INTERACTION_MODE, EventId, MessageId, ProjectId, ThreadId, type ThreadPullRequestSnapshot, ThreadLinkedPullRequest, TurnId, ProviderInstanceId,} from "@t3tools/contracts";import * as Option from "effect/Option";import * as NodeServices from "@effect/platform-node/NodeServices";import { assert, it } from "@effect/vitest";import * as Effect from "effect/Effect";import * as FileSystem from "effect/FileSystem";import * as Layer from "effect/Layer";import * as Path from "effect/Path";import * as Schema from "effect/Schema";import * as SqlClient from "effect/unstable/sql/SqlClient"; import { makeSqlStatementCounter } from "../../../integration/SqlStatementCounter.integration.ts";import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts";import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";import { makeSqlitePersistenceLive, SqlitePersistenceMemory,} from "../../persistence/Layers/Sqlite.ts";import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";import { ProjectionStateRepository } from "../../persistence/Services/ProjectionState.ts";import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityResolver.ts";import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";import { ORCHESTRATION_PROJECTOR_NAMES, OrchestrationProjectionPipelineLive,} from "./ProjectionPipeline.ts";import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts";import * as ThreadPlanProgress from "../ThreadPlanProgress.ts";import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";import { OrchestrationProjectionPipeline } from "../Services/ProjectionPipeline.ts";import { ServerConfig } from "../../config.ts"; const makeProjectionPipelinePrefixedTestLayer = (prefix: string) => OrchestrationProjectionPipelineLive.pipe( Layer.provideMerge(OrchestrationEventStoreLive), Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix })), Layer.provideMerge(SqlitePersistenceMemory), Layer.provideMerge(NodeServices.layer), ); const exists = (filePath: string) => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; const fileInfo = yield* Effect.result(fileSystem.stat(filePath)); return fileInfo._tag === "Success"; }); const BaseTestLayer = makeProjectionPipelinePrefixedTestLayer("t3-projection-pipeline-test-");const encodeThreadLinkedPullRequest = Schema.encodeSync( Schema.fromJsonString(ThreadLinkedPullRequest),); it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-cursor-batch-")))( "OrchestrationProjectionPipeline cursor batches", (it) => { it.effect("writes a project and all projector cursors in two statements", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; const eventStore = yield* OrchestrationEventStore; const projectionState = yield* ProjectionStateRepository; const counter = makeSqlStatementCounter(); const createdAt = "2026-01-01T00:00:00.000Z"; const event = yield* eventStore.append({ type: "project.created", eventId: EventId.make("evt-cursor-batch-project"), aggregateKind: "project", aggregateId: ProjectId.make("project-cursor-batch"), occurredAt: createdAt, commandId: CommandId.make("cmd-cursor-batch-project"), causationEventId: null, correlationId: null, metadata: {}, payload: { projectId: ProjectId.make("project-cursor-batch"), title: "Cursor batch project", workspaceRoot: "/tmp/project-cursor-batch", defaultModelSelection: null, scripts: [], createdAt, updatedAt: createdAt, }, }); yield* projectionPipeline.projectEvent(event).pipe(Effect.withTracer(counter.tracer)); assert.strictEqual(counter.count(), 2); assert.deepEqual( yield* projectionState.listAll(), Object.values(ORCHESTRATION_PROJECTOR_NAMES) .sort() .map((projector) => ({ projector, lastAppliedSequence: event.sequence, updatedAt: createdAt, })), ); }), ); },); it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-import-shell-")))( "imported thread shell projection", (it) => { it.effect("does not mark imported user messages as queued work in thread shells", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; const eventStore = yield* OrchestrationEventStore; const sql = yield* SqlClient.SqlClient; const createdAt = "2026-08-24T10:00:00.000Z"; const threadId = ThreadId.make("import:codex:shell-session"); yield* eventStore.append({ type: "thread.created", eventId: EventId.make("evt-import-shell-thread"), aggregateKind: "thread", aggregateId: threadId, occurredAt: createdAt, commandId: CommandId.make("cmd-import-shell-thread"), causationEventId: null, correlationId: CommandId.make("cmd-import-shell-thread"), metadata: {}, payload: { threadId, projectId: ProjectId.make("project-import-shell"), title: "Imported thread", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" }, runtimeMode: "full-access", interactionMode: "default", branch: null, worktreePath: null, createdAt, updatedAt: createdAt, }, }); yield* eventStore.append({ type: "thread.message-sent", eventId: EventId.make("evt-import-shell-message"), aggregateKind: "thread", aggregateId: threadId, occurredAt: createdAt, commandId: CommandId.make("cmd-import-shell-message"), causationEventId: null, correlationId: CommandId.make("cmd-import-shell-message"), metadata: { historyImport: true }, payload: { threadId, messageId: MessageId.make("import:codex:shell-session:0"), role: "user", text: "Imported user prompt", turnId: null, streaming: false, createdAt, updatedAt: createdAt, }, }); yield* projectionPipeline.bootstrap; const readLatestUserMessageAt = sql<{ readonly latestUserMessageAt: string | null }>` SELECT latest_user_message_at AS "latestUserMessageAt" FROM projection_threads WHERE thread_id = ${threadId} `; assert.deepEqual(yield* readLatestUserMessageAt, [{ latestUserMessageAt: null }]); const sessionEvent = yield* eventStore.append({ type: "thread.session-set", eventId: EventId.make("evt-import-shell-session"), aggregateKind: "thread", aggregateId: threadId, occurredAt: createdAt, commandId: CommandId.make("cmd-import-shell-session"), causationEventId: null, correlationId: CommandId.make("cmd-import-shell-session"), metadata: {}, payload: { threadId, session: { threadId, status: "ready", providerName: "codex", providerInstanceId: ProviderInstanceId.make("codex"), runtimeMode: "full-access", activeTurnId: null, lastError: null, updatedAt: createdAt, }, }, }); yield* projectionPipeline.projectEvent(sessionEvent); assert.deepEqual(yield* readLatestUserMessageAt, [{ latestUserMessageAt: null }]); }), ); },); it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-branch-pr-projection-")))( "branch pull request projection", (it) => { it.effect("persists branch pull request updates without changing manual links", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; const eventStore = yield* OrchestrationEventStore; const sql = yield* SqlClient.SqlClient; const now = "2026-01-01T00:00:00.000Z"; const threadId = ThreadId.make("thread-pull-request"); const projectId = ProjectId.make("project-pull-request"); const eventFields = { aggregateKind: "thread" as const, aggregateId: threadId, occurredAt: now, commandId: null, causationEventId: null, correlationId: null, metadata: {}, }; const created = yield* eventStore.append({ ...eventFields, type: "thread.created", eventId: EventId.make("evt-pull-request-created"), payload: { threadId, projectId, title: "Pull request thread", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" }, runtimeMode: "full-access", interactionMode: "default", branch: "feature", worktreePath: null, createdAt: now, updatedAt: now, }, }); yield* projectionPipeline.projectEvent(created); const linkedPullRequest = { projectId, repository: "pingdotgg/t3code", number: 42, url: "https://github.com/pingdotgg/t3code/pull/42", }; const branchPullRequest = { ...linkedPullRequest, number: 43, url: "https://github.com/pingdotgg/t3code/pull/43", }; const updates = [ { payload: { linkedPullRequest, branchPullRequest }, expected: branchPullRequest }, { payload: { title: "Renamed thread" }, expected: branchPullRequest }, { payload: { branchPullRequest: null }, expected: null }, ]; for (const [index, update] of updates.entries()) { const event = yield* eventStore.append({ ...eventFields, type: "thread.meta-updated", eventId: EventId.make(`evt-pull-request-update-${index}`), payload: { threadId, updatedAt: now, ...update.payload }, }); yield* projectionPipeline.projectEvent(event); const rows = yield* sql<{ readonly linkedPullRequest: string | null; readonly branchPullRequest: string | null; }>` SELECT linked_pull_request_json AS "linkedPullRequest", branch_pull_request_json AS "branchPullRequest" FROM projection_threads WHERE thread_id = ${threadId} `; assert.deepEqual(rows, [ { linkedPullRequest: encodeThreadLinkedPullRequest(linkedPullRequest), branchPullRequest: update.expected === null ? null : encodeThreadLinkedPullRequest(update.expected), }, ]); } }), ); },); it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { it.effect("bootstraps all projection states and writes projection rows", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; const eventStore = yield* OrchestrationEventStore; const sql = yield* SqlClient.SqlClient; const now = "2026-01-01T00:00:00.000Z"; yield* eventStore.append({ type: "project.created", eventId: EventId.make("evt-1"), aggregateKind: "project", aggregateId: ProjectId.make("project-1"), occurredAt: now, commandId: CommandId.make("cmd-1"), causationEventId: null, correlationId: CommandId.make("cmd-1"), metadata: {}, payload: { projectId: ProjectId.make("project-1"), title: "Project 1", workspaceRoot: "/tmp/project-1", defaultModelSelection: null, scripts: [], createdAt: now, updatedAt: now, }, }); yield* eventStore.append({ type: "thread.created", eventId: EventId.make("evt-2"), aggregateKind: "thread", aggregateId: ThreadId.make("thread-1"), occurredAt: now, commandId: CommandId.make("cmd-2"), causationEventId: null, correlationId: CommandId.make("cmd-2"), metadata: {}, payload: { threadId: ThreadId.make("thread-1"), projectId: ProjectId.make("project-1"), title: "Thread 1", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", }, runtimeMode: "full-access", branch: null, worktreePath: null, createdAt: now, updatedAt: now, }, }); yield* eventStore.append({ type: "thread.message-sent", eventId: EventId.make("evt-3"), aggregateKind: "thread", aggregateId: ThreadId.make("thread-1"), occurredAt: now, commandId: CommandId.make("cmd-3"), causationEventId: null, correlationId: CommandId.make("cmd-3"), metadata: {}, payload: { threadId: ThreadId.make("thread-1"), messageId: MessageId.make("message-1"), role: "assistant", text: "hello", turnId: null, streaming: false, createdAt: now, updatedAt: now, }, }); yield* projectionPipeline.bootstrap; const projectRows = yield* sql<{ readonly projectId: string; readonly title: string; readonly scriptsJson: string; }>` SELECT project_id AS "projectId", title, scripts_json AS "scriptsJson" FROM projection_projects `; assert.deepEqual(projectRows, [ { projectId: "project-1", title: "Project 1", scriptsJson: "[]" }, ]); const messageRows = yield* sql<{ readonly messageId: string; readonly text: string; }>` SELECT message_id AS "messageId", text FROM projection_thread_messages `; assert.deepEqual(messageRows, [{ messageId: "message-1", text: "hello" }]); const stateRows = yield* sql<{ readonly projector: string; readonly lastAppliedSequence: number; }>` SELECT projector, last_applied_sequence AS "lastAppliedSequence" FROM projection_state ORDER BY projector ASC `; assert.equal(stateRows.length, Object.keys(ORCHESTRATION_PROJECTOR_NAMES).length + 1); for (const row of stateRows) { assert.equal(row.lastAppliedSequence, 3); } yield* sql`CREATE TABLE thread_shell_updates (count INTEGER NOT NULL)`; yield* sql`INSERT INTO thread_shell_updates (count) VALUES (0)`; yield* sql` CREATE TRIGGER count_thread_shell_updates AFTER UPDATE ON projection_threads WHEN NEW.thread_id = 'thread-1' BEGIN UPDATE thread_shell_updates SET count = count + 1; END; `; yield* eventStore.append({ type: "thread.message-sent", eventId: EventId.make("evt-assistant-update"), aggregateKind: "thread", aggregateId: ThreadId.make("thread-1"), occurredAt: "2026-01-01T00:00:00.100Z", commandId: CommandId.make("cmd-assistant-update"), causationEventId: null, correlationId: CommandId.make("cmd-assistant-update"), metadata: {}, payload: { threadId: ThreadId.make("thread-1"), messageId: MessageId.make("message-2"), role: "assistant", text: "more work", turnId: null, streaming: false, createdAt: "2026-01-01T00:00:00.100Z", updatedAt: "2026-01-01T00:00:00.100Z", }, }); yield* projectionPipeline.bootstrap; let threadShellUpdates = yield* sql<{ readonly count: number }>` SELECT count FROM thread_shell_updates `; assert.deepEqual(threadShellUpdates, [{ count: 1 }]); yield* sql`UPDATE thread_shell_updates SET count = 0`; yield* eventStore.append({ type: "thread.activity-appended", eventId: EventId.make("evt-routine-activity"), aggregateKind: "thread", aggregateId: ThreadId.make("thread-1"), occurredAt: "2026-01-01T00:00:00.200Z", commandId: CommandId.make("cmd-routine-activity"), causationEventId: null, correlationId: CommandId.make("cmd-routine-activity"), metadata: {}, payload: { threadId: ThreadId.make("thread-1"), activity: { id: EventId.make("activity-routine"), tone: "tool", kind: "tool.updated", summary: "Tool made progress", payload: {}, turnId: null, createdAt: "2026-01-01T00:00:00.200Z", }, }, }); yield* projectionPipeline.bootstrap; threadShellUpdates = yield* sql<{ readonly count: number }>` SELECT count FROM thread_shell_updates `; assert.deepEqual(threadShellUpdates, [{ count: 1 }]); yield* sql`DROP TRIGGER count_thread_shell_updates`; yield* sql`DROP TABLE thread_shell_updates`; // Replayed order events must survive later lifecycle upserts, whose // complete SQL row writes otherwise risk dropping the placement. const orderUpdatedAt = "2026-01-01T00:00:00.200Z"; const orderEvents = [ { type: "thread.meta-updated", payload: { activeOrderKey: "gm" } }, { type: "thread.pinned", payload: { pinnedAt: now, pinOrderKey: "m" } }, { type: "thread.snoozed", payload: { snoozedAt: now, snoozedUntil: "2026-01-02T00:00:00.000Z" },Lines 1–500 of 4610. Load more
import { ApprovalRequestId, CheckpointRef, CommandId, CorrelationId, DEFAULT_PROVIDER_INTERACTION_MODE, EventId, MessageId, ProjectId, ThreadId, type ThreadPullRequestSnapshot, ThreadLinkedPullRequest, TurnId, ProviderInstanceId, } from "@t3tools/contracts"; import * as Option from "effect/Option"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import { makeSqlStatementCounter } from "../../../integration/SqlStatementCounter.integration.ts"; import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts"; import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts"; import { makeSqlitePersistenceLive, SqlitePersistenceMemory, } from "../../persistence/Layers/Sqlite.ts"; import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts"; import { ProjectionStateRepository } from "../../persistence/Services/ProjectionState.ts"; import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityResolver.ts"; import { OrchestrationEngineLive } from "./OrchestrationEngine.ts"; import { ORCHESTRATION_PROJECTOR_NAMES, OrchestrationProjectionPipelineLive, } from "./ProjectionPipeline.ts"; import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts"; import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts"; import * as ThreadPlanProgress from "../ThreadPlanProgress.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; import { OrchestrationProjectionPipeline } from "../Services/ProjectionPipeline.ts"; import { ServerConfig } from "../../config.ts"; const makeProjectionPipelinePrefixedTestLayer = (prefix: string) => OrchestrationProjectionPipelineLive.pipe( Layer.provideMerge(OrchestrationEventStoreLive), Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix })), Layer.provideMerge(SqlitePersistenceMemory), Layer.provideMerge(NodeServices.layer), ); const exists = (filePath: string) => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; const fileInfo = yield* Effect.result(fileSystem.stat(filePath)); return fileInfo._tag === "Success"; }); const BaseTestLayer = makeProjectionPipelinePrefixedTestLayer("t3-projection-pipeline-test-"); const encodeThreadLinkedPullRequest = Schema.encodeSync( Schema.fromJsonString(ThreadLinkedPullRequest), ); it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-cursor-batch-")))( "OrchestrationProjectionPipeline cursor batches", (it) => { it.effect("writes a project and all projector cursors in two statements", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; const eventStore = yield* OrchestrationEventStore; const projectionState = yield* ProjectionStateRepository; const counter = makeSqlStatementCounter(); const createdAt = "2026-01-01T00:00:00.000Z"; const event = yield* eventStore.append({ type: "project.created", eventId: EventId.make("evt-cursor-batch-project"), aggregateKind: "project", aggregateId: ProjectId.make("project-cursor-batch"), occurredAt: createdAt, commandId: CommandId.make("cmd-cursor-batch-project"), causationEventId: null, correlationId: null, metadata: {}, payload: { projectId: ProjectId.make("project-cursor-batch"), title: "Cursor batch project", workspaceRoot: "/tmp/project-cursor-batch", defaultModelSelection: null, scripts: [], createdAt, updatedAt: createdAt, }, }); yield* projectionPipeline.projectEvent(event).pipe(Effect.withTracer(counter.tracer)); assert.strictEqual(counter.count(), 2); assert.deepEqual( yield* projectionState.listAll(), Object.values(ORCHESTRATION_PROJECTOR_NAMES) .sort() .map((projector) => ({ projector, lastAppliedSequence: event.sequence, updatedAt: createdAt, })), ); }), ); }, ); it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-import-shell-")))( "imported thread shell projection", (it) => { it.effect("does not mark imported user messages as queued work in thread shells", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; const eventStore = yield* OrchestrationEventStore; const sql = yield* SqlClient.SqlClient; const createdAt = "2026-08-24T10:00:00.000Z"; const threadId = ThreadId.make("import:codex:shell-session"); yield* eventStore.append({ type: "thread.created", eventId: EventId.make("evt-import-shell-thread"), aggregateKind: "thread", aggregateId: threadId, occurredAt: createdAt, commandId: CommandId.make("cmd-import-shell-thread"), causationEventId: null, correlationId: CommandId.make("cmd-import-shell-thread"), metadata: {}, payload: { threadId, projectId: ProjectId.make("project-import-shell"), title: "Imported thread", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" }, runtimeMode: "full-access", interactionMode: "default", branch: null, worktreePath: null, createdAt, updatedAt: createdAt, }, }); yield* eventStore.append({ type: "thread.message-sent", eventId: EventId.make("evt-import-shell-message"), aggregateKind: "thread", aggregateId: threadId, occurredAt: createdAt, commandId: CommandId.make("cmd-import-shell-message"), causationEventId: null, correlationId: CommandId.make("cmd-import-shell-message"), metadata: { historyImport: true }, payload: { threadId, messageId: MessageId.make("import:codex:shell-session:0"), role: "user", text: "Imported user prompt", turnId: null, streaming: false, createdAt, updatedAt: createdAt, }, }); yield* projectionPipeline.bootstrap; const readLatestUserMessageAt = sql<{ readonly latestUserMessageAt: string | null }>` SELECT latest_user_message_at AS "latestUserMessageAt" FROM projection_threads WHERE thread_id = ${threadId} `; assert.deepEqual(yield* readLatestUserMessageAt, [{ latestUserMessageAt: null }]); const sessionEvent = yield* eventStore.append({ type: "thread.session-set", eventId: EventId.make("evt-import-shell-session"), aggregateKind: "thread", aggregateId: threadId, occurredAt: createdAt, commandId: CommandId.make("cmd-import-shell-session"), causationEventId: null, correlationId: CommandId.make("cmd-import-shell-session"), metadata: {}, payload: { threadId, session: { threadId, status: "ready", providerName: "codex", providerInstanceId: ProviderInstanceId.make("codex"), runtimeMode: "full-access", activeTurnId: null, lastError: null, updatedAt: createdAt, }, }, }); yield* projectionPipeline.projectEvent(sessionEvent); assert.deepEqual(yield* readLatestUserMessageAt, [{ latestUserMessageAt: null }]); }), ); }, ); it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-branch-pr-projection-")))( "branch pull request projection", (it) => { it.effect("persists branch pull request updates without changing manual links", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; const eventStore = yield* OrchestrationEventStore; const sql = yield* SqlClient.SqlClient; const now = "2026-01-01T00:00:00.000Z"; const threadId = ThreadId.make("thread-pull-request"); const projectId = ProjectId.make("project-pull-request"); const eventFields = { aggregateKind: "thread" as const, aggregateId: threadId, occurredAt: now, commandId: null, causationEventId: null, correlationId: null, metadata: {}, }; const created = yield* eventStore.append({ ...eventFields, type: "thread.created", eventId: EventId.make("evt-pull-request-created"), payload: { threadId, projectId, title: "Pull request thread", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" }, runtimeMode: "full-access", interactionMode: "default", branch: "feature", worktreePath: null, createdAt: now, updatedAt: now, }, }); yield* projectionPipeline.projectEvent(created); const linkedPullRequest = { projectId, repository: "pingdotgg/t3code", number: 42, url: "https://github.com/pingdotgg/t3code/pull/42", }; const branchPullRequest = { ...linkedPullRequest, number: 43, url: "https://github.com/pingdotgg/t3code/pull/43", }; const updates = [ { payload: { linkedPullRequest, branchPullRequest }, expected: branchPullRequest }, { payload: { title: "Renamed thread" }, expected: branchPullRequest }, { payload: { branchPullRequest: null }, expected: null }, ]; for (const [index, update] of updates.entries()) { const event = yield* eventStore.append({ ...eventFields, type: "thread.meta-updated", eventId: EventId.make(`evt-pull-request-update-${index}`), payload: { threadId, updatedAt: now, ...update.payload }, }); yield* projectionPipeline.projectEvent(event); const rows = yield* sql<{ readonly linkedPullRequest: string | null; readonly branchPullRequest: string | null; }>` SELECT linked_pull_request_json AS "linkedPullRequest", branch_pull_request_json AS "branchPullRequest" FROM projection_threads WHERE thread_id = ${threadId} `; assert.deepEqual(rows, [ { linkedPullRequest: encodeThreadLinkedPullRequest(linkedPullRequest), branchPullRequest: update.expected === null ? null : encodeThreadLinkedPullRequest(update.expected), }, ]); } }), ); }, ); it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { it.effect("bootstraps all projection states and writes projection rows", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; const eventStore = yield* OrchestrationEventStore; const sql = yield* SqlClient.SqlClient; const now = "2026-01-01T00:00:00.000Z"; yield* eventStore.append({ type: "project.created", eventId: EventId.make("evt-1"), aggregateKind: "project", aggregateId: ProjectId.make("project-1"), occurredAt: now, commandId: CommandId.make("cmd-1"), causationEventId: null, correlationId: CommandId.make("cmd-1"), metadata: {}, payload: { projectId: ProjectId.make("project-1"), title: "Project 1", workspaceRoot: "/tmp/project-1", defaultModelSelection: null, scripts: [], createdAt: now, updatedAt: now, }, }); yield* eventStore.append({ type: "thread.created", eventId: EventId.make("evt-2"), aggregateKind: "thread", aggregateId: ThreadId.make("thread-1"), occurredAt: now, commandId: CommandId.make("cmd-2"), causationEventId: null, correlationId: CommandId.make("cmd-2"), metadata: {}, payload: { threadId: ThreadId.make("thread-1"), projectId: ProjectId.make("project-1"), title: "Thread 1", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", }, runtimeMode: "full-access", branch: null, worktreePath: null, createdAt: now, updatedAt: now, }, }); yield* eventStore.append({ type: "thread.message-sent", eventId: EventId.make("evt-3"), aggregateKind: "thread", aggregateId: ThreadId.make("thread-1"), occurredAt: now, commandId: CommandId.make("cmd-3"), causationEventId: null, correlationId: CommandId.make("cmd-3"), metadata: {}, payload: { threadId: ThreadId.make("thread-1"), messageId: MessageId.make("message-1"), role: "assistant", text: "hello", turnId: null, streaming: false, createdAt: now, updatedAt: now, }, }); yield* projectionPipeline.bootstrap; const projectRows = yield* sql<{ readonly projectId: string; readonly title: string; readonly scriptsJson: string; }>` SELECT project_id AS "projectId", title, scripts_json AS "scriptsJson" FROM projection_projects `; assert.deepEqual(projectRows, [ { projectId: "project-1", title: "Project 1", scriptsJson: "[]" }, ]); const messageRows = yield* sql<{ readonly messageId: string; readonly text: string; }>` SELECT message_id AS "messageId", text FROM projection_thread_messages `; assert.deepEqual(messageRows, [{ messageId: "message-1", text: "hello" }]); const stateRows = yield* sql<{ readonly projector: string; readonly lastAppliedSequence: number; }>` SELECT projector, last_applied_sequence AS "lastAppliedSequence" FROM projection_state ORDER BY projector ASC `; assert.equal(stateRows.length, Object.keys(ORCHESTRATION_PROJECTOR_NAMES).length + 1); for (const row of stateRows) { assert.equal(row.lastAppliedSequence, 3); } yield* sql`CREATE TABLE thread_shell_updates (count INTEGER NOT NULL)`; yield* sql`INSERT INTO thread_shell_updates (count) VALUES (0)`; yield* sql` CREATE TRIGGER count_thread_shell_updates AFTER UPDATE ON projection_threads WHEN NEW.thread_id = 'thread-1' BEGIN UPDATE thread_shell_updates SET count = count + 1; END; `; yield* eventStore.append({ type: "thread.message-sent", eventId: EventId.make("evt-assistant-update"), aggregateKind: "thread", aggregateId: ThreadId.make("thread-1"), occurredAt: "2026-01-01T00:00:00.100Z", commandId: CommandId.make("cmd-assistant-update"), causationEventId: null, correlationId: CommandId.make("cmd-assistant-update"), metadata: {}, payload: { threadId: ThreadId.make("thread-1"), messageId: MessageId.make("message-2"), role: "assistant", text: "more work", turnId: null, streaming: false, createdAt: "2026-01-01T00:00:00.100Z", updatedAt: "2026-01-01T00:00:00.100Z", }, }); yield* projectionPipeline.bootstrap; let threadShellUpdates = yield* sql<{ readonly count: number }>` SELECT count FROM thread_shell_updates `; assert.deepEqual(threadShellUpdates, [{ count: 1 }]); yield* sql`UPDATE thread_shell_updates SET count = 0`; yield* eventStore.append({ type: "thread.activity-appended", eventId: EventId.make("evt-routine-activity"), aggregateKind: "thread", aggregateId: ThreadId.make("thread-1"), occurredAt: "2026-01-01T00:00:00.200Z", commandId: CommandId.make("cmd-routine-activity"), causationEventId: null, correlationId: CommandId.make("cmd-routine-activity"), metadata: {}, payload: { threadId: ThreadId.make("thread-1"), activity: { id: EventId.make("activity-routine"), tone: "tool", kind: "tool.updated", summary: "Tool made progress", payload: {}, turnId: null, createdAt: "2026-01-01T00:00:00.200Z", }, }, }); yield* projectionPipeline.bootstrap; threadShellUpdates = yield* sql<{ readonly count: number }>` SELECT count FROM thread_shell_updates `; assert.deepEqual(threadShellUpdates, [{ count: 1 }]); yield* sql`DROP TRIGGER count_thread_shell_updates`; yield* sql`DROP TABLE thread_shell_updates`; // Replayed order events must survive later lifecycle upserts, whose // complete SQL row writes otherwise risk dropping the placement. const orderUpdatedAt = "2026-01-01T00:00:00.200Z"; const orderEvents = [ { type: "thread.meta-updated", payload: { activeOrderKey: "gm" } }, { type: "thread.pinned", payload: { pinnedAt: now, pinOrderKey: "m" } }, { type: "thread.snoozed", payload: { snoozedAt: now, snoozedUntil: "2026-01-02T00:00:00.000Z" },