diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 543d469458ed..ea3f11aa83e0 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -10,11 +10,12 @@ import { ProviderInstanceId, } from "@t3tools/contracts"; import * as NodeServices from "@effect/platform-node/NodeServices"; -import { assert, it } from "@effect/vitest"; +import { assert, it, vi } 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 Stream from "effect/Stream"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts"; @@ -1609,6 +1610,102 @@ it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-bootstrap-attac }, ); +it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-bootstrap-incomplete-")))( + "OrchestrationProjectionPipeline attachment bootstrap reconciliation", + (it) => { + it.effect("preserves attachment files when replay stops before the event log head", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const { attachmentsDir } = yield* ServerConfig; + const now = "2026-01-03T00:00:00.000Z"; + const threadId = ThreadId.make("bootstrap-incomplete-thread"); + const attachmentId = "bootstrap-incomplete-thread-00000000-0000-4000-8000-000000000001"; + const attachmentPath = path.join(attachmentsDir, `${attachmentId}.png`); + + yield* eventStore.append({ + type: "thread.created", + eventId: EventId.make("evt-bootstrap-incomplete-1"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: now, + commandId: CommandId.make("cmd-bootstrap-incomplete-1"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-bootstrap-incomplete-1"), + metadata: {}, + payload: { + threadId, + projectId: ProjectId.make("project-bootstrap-incomplete"), + title: "Incomplete replay", + 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-bootstrap-incomplete-2"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: now, + commandId: CommandId.make("cmd-bootstrap-incomplete-2"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-bootstrap-incomplete-2"), + metadata: {}, + payload: { + threadId, + messageId: MessageId.make("message-bootstrap-incomplete"), + role: "user", + text: "not replayed", + attachments: [ + { + type: "image", + id: attachmentId, + name: "preserve.png", + mimeType: "image/png", + sizeBytes: 4, + }, + ], + turnId: null, + streaming: false, + createdAt: now, + updatedAt: now, + }, + }); + yield* fileSystem.makeDirectory(attachmentsDir, { recursive: true }); + yield* fileSystem.writeFileString(attachmentPath, "keep"); + + const readFromSequence = eventStore.readFromSequence.bind(eventStore); + const replaySpy = vi + .spyOn(eventStore, "readFromSequence") + .mockImplementation((sequenceExclusive, limit) => + readFromSequence(sequenceExclusive, limit).pipe(Stream.take(1)), + ); + yield* projectionPipeline.bootstrap.pipe( + Effect.ensuring(Effect.sync(() => replaySpy.mockRestore())), + ); + + const projectedMessages = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS "count" + FROM projection_thread_messages + WHERE message_id = 'message-bootstrap-incomplete' + `; + assert.equal(projectedMessages[0]?.count ?? 0, 0); + assert.isTrue(yield* exists(attachmentPath)); + }), + ); + }, +); + it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { it.effect("resumes from projector last_applied_sequence without replaying older events", () => Effect.gen(function* () { diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 54057048d9a1..13a380d1734d 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -1725,6 +1725,40 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti ]; const reconcileAttachmentFiles = Effect.fn("reconcileAttachmentFiles")(function* () { + const eventHeadRows = yield* sql<{ readonly eventHead: number }>` + SELECT COALESCE(MAX(sequence), 0) AS "eventHead" + FROM orchestration_events + `; + const eventHead = eventHeadRows[0]?.eventHead ?? 0; + const projectorStateRows = yield* sql<{ + readonly projector: string; + readonly lastAppliedSequence: number; + }>` + SELECT + projector, + last_applied_sequence AS "lastAppliedSequence" + FROM projection_state + `; + const lastAppliedSequenceByProjector = new Map( + projectorStateRows.map((row) => [row.projector, row.lastAppliedSequence]), + ); + const incompleteProjectors = projectors.filter( + (projector) => lastAppliedSequenceByProjector.get(projector.name) !== eventHead, + ); + if (incompleteProjectors.length > 0) { + yield* Effect.logWarning( + "skipping attachment reconciliation until every projector reaches the event log head", + { + eventHead, + incompleteProjectors: incompleteProjectors.map((projector) => ({ + projector: projector.name, + lastAppliedSequence: lastAppliedSequenceByProjector.get(projector.name) ?? null, + })), + }, + ); + return; + } + const entries = yield* fileSystem .readDirectory(serverConfig.attachmentsDir, { recursive: false }) .pipe(Effect.orElseSucceed(() => [] as Array)); @@ -1869,14 +1903,14 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti yield* Stream.runForEach( eventStore.readFromSequence(firstSequence, Number.MAX_SAFE_INTEGER), (event) => - Effect.forEach( - projectors, - (projector) => - event.sequence > (lastAppliedSequenceByProjector.get(projector.name) ?? 0) - ? runProjectorForEvent(projector, event, "bootstrap") - : Effect.void, - { concurrency: 1 }, - ), + Effect.forEach( + projectors, + (projector) => + event.sequence > (lastAppliedSequenceByProjector.get(projector.name) ?? 0) + ? runProjectorForEvent(projector, event, "bootstrap") + : Effect.void, + { concurrency: 1 }, + ), ); });