authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-01 18:25:55-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-01 18:25:55-07:00
log716069f40f000887cb01685099a16e19d9194307
treec7381f01cf7e62fa6a53a819e9fb02e8972d9b9f
parent0e77fbd3d0d79eec5247e75583a04590a1785133
signature Signed by PGP key B5690EEEBB952194

fix(server): keep attachments until the command commits (#7941)

13 files changed, 368 insertions(+), 38 deletions(-)

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts+7-4
......@@ -222,6 +222,7 @@ describe("OrchestrationEngine", () => {
222222 Layer.succeed(OrchestrationProjectionPipeline, {
223223 bootstrap: Effect.void,
224224 projectEvent: () => Effect.void,
225 projectEventDeferred: () => Effect.succeed(Effect.void),
225226 } satisfies OrchestrationProjectionPipelineShape),
226227 ),
227228 Layer.provide(Layer.succeed(OrchestrationEventStore, eventStore)),
......@@ -1116,7 +1117,8 @@ describe("OrchestrationEngine", () => {
11161117 let shouldFailRequestedProjection = true;
11171118 const flakyProjectionPipeline: OrchestrationProjectionPipelineShape = {
11181119 bootstrap: Effect.void,
1119 projectEvent: (event) => {
1120 projectEvent: () => Effect.void,
1121 projectEventDeferred: (event) => {
11201122 if (
11211123 shouldFailRequestedProjection &&
11221124 event.commandId === CommandId.make("cmd-turn-start-atomic") &&
......@@ -1130,7 +1132,7 @@ describe("OrchestrationEngine", () => {
11301132 }),
11311133 );
11321134 }
1133 return Effect.void;
1135 return Effect.succeed(Effect.void);
11341136 },
11351137 };
11361138
......@@ -1263,7 +1265,8 @@ describe("OrchestrationEngine", () => {
12631265 let shouldFailProjection = true;
12641266 const flakyProjectionPipeline: OrchestrationProjectionPipelineShape = {
12651267 bootstrap: Effect.void,
1266 projectEvent: (event) => {
1268 projectEvent: () => Effect.void,
1269 projectEventDeferred: (event) => {
12671270 if (
12681271 shouldFailProjection &&
12691272 event.commandId === CommandId.make("cmd-thread-archive-sync-fail")
......@@ -1276,7 +1279,7 @@ describe("OrchestrationEngine", () => {
12761279 }),
12771280 );
12781281 }
1279 return Effect.void;
1282 return Effect.succeed(Effect.void);
12801283 },
12811284 };
12821285
apps/server/src/orchestration/Layers/OrchestrationEngine.ts+8-1
......@@ -224,12 +224,14 @@ const makeOrchestrationEngine = Effect.gen(function* () {
224224 .withTransaction(
225225 Effect.gen(function* () {
226226 const committedEvents: OrchestrationEvent[] = [];
227 const attachmentCleanups: Effect.Effect<void>[] = [];
227228 let nextCommandReadModel = commandReadModel;
228229
229230 for (const nextEvent of eventBases) {
230231 const savedEvent = yield* eventStore.append(nextEvent);
231232 nextCommandReadModel = yield* projectEvent(nextCommandReadModel, savedEvent);
232 yield* projectionPipeline.projectEvent(savedEvent);
233 const cleanup = yield* projectionPipeline.projectEventDeferred(savedEvent);
234 attachmentCleanups.push(cleanup);
233235 committedEvents.push(savedEvent);
234236 }
235237
......@@ -253,6 +255,7 @@ const makeOrchestrationEngine = Effect.gen(function* () {
253255
254256 return {
255257 committedEvents,
258 attachmentCleanups,
256259 lastSequence: lastSavedEvent.sequence,
257260 nextCommandReadModel,
258261 } as const;
......@@ -267,6 +270,9 @@ const makeOrchestrationEngine = Effect.gen(function* () {
267270 );
268271
269272 commandReadModel = committedCommand.nextCommandReadModel;
273 for (const cleanup of committedCommand.attachmentCleanups) {
274 yield* cleanup;
275 }
270276 for (const [index, event] of committedCommand.committedEvents.entries()) {
271277 yield* PubSub.publish(eventPubSub, event);
272278 if (index === 0) {
......@@ -381,6 +387,7 @@ const makeOrchestrationEngine = Effect.gen(function* () {
381387 return {
382388 readEvents,
383389 dispatch,
390 subscribeDomainEvents: PubSub.subscribe(eventPubSub).pipe(Effect.map(Stream.fromSubscription)),
384391 // Each access creates a fresh PubSub subscription so that multiple
385392 // consumers (wsServer, ProviderRuntimeIngestion, CheckpointReactor, etc.)
386393 // each independently receive all domain events.
apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+177-2
......@@ -2,6 +2,7 @@ import {
22 CheckpointRef,
33 CommandId,
44 CorrelationId,
5 DEFAULT_PROVIDER_INTERACTION_MODE,
56 EventId,
67 MessageId,
78 ProjectId,
......@@ -874,12 +875,13 @@ it.layer(
874875it.layer(
875876 Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-attachments-overwrite-")),
876877)("OrchestrationProjectionPipeline", (it) => {
877 it.effect("removes unreferenced attachment files when a thread is reverted", () =>
878 it.effect("prunes reverted attachments only after every projector commits", () =>
878879 Effect.gen(function* () {
879880 const fileSystem = yield* FileSystem.FileSystem;
880881 const path = yield* Path.Path;
881882 const projectionPipeline = yield* OrchestrationProjectionPipeline;
882883 const eventStore = yield* OrchestrationEventStore;
884 const sql = yield* SqlClient.SqlClient;
883885 const { attachmentsDir } = yield* ServerConfig;
884886 const now = "2026-01-01T00:00:00.000Z";
885887 const threadId = ThreadId.make("Thread Revert.Files");
......@@ -1067,7 +1069,7 @@ it.layer(
10671069 assert.isTrue(yield* exists(removePath));
10681070 assert.isTrue(yield* exists(otherThreadPath));
10691071
1070 yield* appendAndProject({
1072 const revertedEvent = yield* eventStore.append({
10711073 type: "thread.reverted",
10721074 eventId: EventId.make("evt-revert-files-7"),
10731075 aggregateKind: "thread",
......@@ -1083,9 +1085,75 @@ it.layer(
10831085 },
10841086 });
10851087
1088 yield* sql`
1089 CREATE TRIGGER fail_revert_projection
1090 BEFORE UPDATE ON projection_state
1091 WHEN NEW.projector = 'projection.threads'
1092 BEGIN
1093 SELECT RAISE(FAIL, 'forced later projector failure');
1094 END
1095 `;
1096 const projectionError = yield* projectionPipeline
1097 .projectEvent(revertedEvent)
1098 .pipe(Effect.flip);
1099 assert.equal(projectionError._tag, "PersistenceSqlError");
1100 assert.isTrue(yield* exists(removePath));
1101 const rolledBackMessages = yield* sql<{ readonly messageId: string }>`
1102 SELECT message_id AS "messageId" FROM projection_thread_messages
1103 WHERE message_id = 'message-remove'
1104 `;
1105 assert.deepEqual(rolledBackMessages, [{ messageId: "message-remove" }]);
1106 yield* sql`DROP TRIGGER fail_revert_projection`;
1107
1108 const laterAttachmentId = "thread-revert-files-00000000-0000-4000-8000-000000000005";
1109 const laterPath = path.join(attachmentsDir, `${laterAttachmentId}.png`);
1110 yield* fileSystem.writeFileString(laterPath, "added after revert");
1111 const cleanup = yield* sql.withTransaction(
1112 Effect.gen(function* () {
1113 const cleanup = yield* projectionPipeline.projectEventDeferred(revertedEvent);
1114 yield* appendAndProject({
1115 type: "thread.message-sent",
1116 eventId: EventId.make("evt-revert-files-later"),
1117 aggregateKind: "thread",
1118 aggregateId: threadId,
1119 occurredAt: now,
1120 commandId: CommandId.make("cmd-revert-files-later"),
1121 causationEventId: null,
1122 correlationId: CorrelationId.make("cmd-revert-files-later"),
1123 metadata: {},
1124 payload: {
1125 threadId,
1126 messageId: MessageId.make("message-later"),
1127 role: "user",
1128 text: "Later attachment",
1129 attachments: [
1130 {
1131 type: "image",
1132 id: laterAttachmentId,
1133 name: "later.png",
1134 mimeType: "image/png",
1135 sizeBytes: 5,
1136 },
1137 ],
1138 turnId: null,
1139 streaming: false,
1140 createdAt: now,
1141 updatedAt: now,
1142 },
1143 });
1144 assert.isTrue(yield* exists(removePath));
1145 // Return the cleanup effect so the caller runs it after the outer transaction commits.
1146 // @effect-diagnostics-next-line returnEffectInGen:off
1147 return cleanup;
1148 }),
1149 );
1150 assert.isTrue(yield* exists(removePath));
1151 yield* cleanup;
1152
10861153 assert.isTrue(yield* exists(keepPath));
10871154 assert.isTrue(yield* exists(keepFilePath));
10881155 assert.isFalse(yield* exists(removePath));
1156 assert.isTrue(yield* exists(laterPath));
10891157 assert.isTrue(yield* exists(otherThreadPath));
10901158 }),
10911159 );
......@@ -3522,4 +3590,111 @@ engineLayer("OrchestrationProjectionPipeline via engine dispatch", (it) => {
35223590 assert.isNull(detail.session);
35233591 }),
35243592 );
3593
3594 it.effect("cleans attachments only after the command receipt commits", () =>
3595 Effect.gen(function* () {
3596 const engine = yield* OrchestrationEngineService;
3597 const sql = yield* SqlClient.SqlClient;
3598 const fileSystem = yield* FileSystem.FileSystem;
3599 const path = yield* Path.Path;
3600 const { attachmentsDir } = yield* ServerConfig;
3601 const createdAt = "2026-01-01T00:00:00.000Z";
3602 const projectId = ProjectId.make("project-outer-rollback");
3603 const threadId = ThreadId.make("thread-outer-rollback");
3604 const cleanupFailureThreadId = ThreadId.make("thread-cleanup-failure");
3605 const commandId = CommandId.make("cmd-outer-rollback-delete");
3606 const attachmentPath = path.join(
3607 attachmentsDir,
3608 "thread-outer-rollback-00000000-0000-4000-8000-000000000001.png",
3609 );
3610 const blockedAttachmentPath = path.join(
3611 attachmentsDir,
3612 "thread-cleanup-failure-00000000-0000-4000-8000-000000000001.png",
3613 );
3614
3615 yield* engine.dispatch({
3616 type: "project.create",
3617 commandId: CommandId.make("cmd-outer-rollback-project"),
3618 projectId,
3619 title: "Outer rollback project",
3620 workspaceRoot: "/tmp/project-outer-rollback",
3621 createdAt,
3622 });
3623 for (const id of [threadId, cleanupFailureThreadId]) {
3624 yield* engine.dispatch({
3625 type: "thread.create",
3626 commandId: CommandId.make(`cmd-create-${id}`),
3627 threadId: id,
3628 projectId,
3629 title: "Attachment cleanup thread",
3630 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
3631 modelSelection: {
3632 instanceId: ProviderInstanceId.make("codex"),
3633 model: "gpt-5-codex",
3634 },
3635 runtimeMode: "full-access",
3636 branch: null,
3637 worktreePath: null,
3638 createdAt,
3639 });
3640 }
3641
3642 yield* fileSystem.makeDirectory(attachmentsDir, { recursive: true });
3643 yield* fileSystem.writeFileString(attachmentPath, "keep this attachment");
3644 yield* sql`
3645 CREATE TRIGGER fail_attachment_command_receipt
3646 BEFORE INSERT ON orchestration_command_receipts
3647 WHEN NEW.command_id = 'cmd-outer-rollback-delete' AND NEW.status = 'accepted'
3648 BEGIN
3649 SELECT RAISE(FAIL, 'forced receipt failure');
3650 END
3651 `;
3652 const deleteCommand = { type: "thread.delete", commandId, threadId } as const;
3653 const dispatchError = yield* engine.dispatch(deleteCommand).pipe(Effect.flip);
3654 assert.equal(dispatchError._tag, "PersistenceSqlError");
3655 assert.equal(yield* fileSystem.readFileString(attachmentPath), "keep this attachment");
3656 const rolledBackThreads = yield* sql<{ readonly deletedAt: string | null }>`
3657 SELECT deleted_at AS "deletedAt" FROM projection_threads WHERE thread_id = ${threadId}
3658 `;
3659 assert.deepEqual(rolledBackThreads, [{ deletedAt: null }]);
3660 const rolledBackEvents = yield* sql`
3661 SELECT sequence FROM orchestration_events WHERE command_id = ${commandId}
3662 `;
3663 assert.deepEqual(rolledBackEvents, []);
3664 const rolledBackReceipts = yield* sql`
3665 SELECT status FROM orchestration_command_receipts WHERE command_id = ${commandId}
3666 `;
3667 assert.deepEqual(rolledBackReceipts, []);
3668 yield* sql`DROP TRIGGER fail_attachment_command_receipt`;
3669
3670 const result = yield* engine.dispatch(deleteCommand);
3671 assert.isFalse(yield* exists(attachmentPath));
3672 const committedReceipts = yield* sql<{
3673 readonly status: string;
3674 readonly resultSequence: number;
3675 }>`
3676 SELECT status, result_sequence AS "resultSequence"
3677 FROM orchestration_command_receipts WHERE command_id = ${commandId}
3678 `;
3679 assert.deepEqual(committedReceipts, [
3680 { status: "accepted", resultSequence: result.sequence },
3681 ]);
3682
3683 // Removing a nonempty directory as a file fails after the command commits.
3684 yield* fileSystem.makeDirectory(blockedAttachmentPath);
3685 yield* fileSystem.writeFileString(path.join(blockedAttachmentPath, "keep.txt"), "keep");
3686 const cleanupFailureCommandId = CommandId.make("cmd-cleanup-failure-delete");
3687 yield* engine.dispatch({
3688 type: "thread.delete",
3689 commandId: cleanupFailureCommandId,
3690 threadId: cleanupFailureThreadId,
3691 });
3692 assert.isTrue(yield* exists(blockedAttachmentPath));
3693 const cleanupFailureReceipts = yield* sql<{ readonly status: string }>`
3694 SELECT status FROM orchestration_command_receipts
3695 WHERE command_id = ${cleanupFailureCommandId}
3696 `;
3697 assert.deepEqual(cleanupFailureReceipts, [{ status: "accepted" }]);
3698 }),
3699 );
35253700});
apps/server/src/orchestration/Layers/ProjectionPipeline.ts+92-27
......@@ -1765,6 +1765,70 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
17651765 },
17661766 ];
17671767
1768 const applyAttachmentSideEffects = Effect.fn("applyAttachmentSideEffects")(
1769 function* (event: OrchestrationEvent, sideEffects: AttachmentSideEffects) {
1770 if (
1771 sideEffects.deletedThreadIds.size === 0 &&
1772 sideEffects.prunedThreadRelativePaths.size === 0
1773 ) {
1774 return;
1775 }
1776
1777 const deletedThreadIds = new Set<string>();
1778 for (const threadId of sideEffects.deletedThreadIds) {
1779 const recreatedLater = yield* eventStore.hasEventAfter({
1780 aggregateKind: "thread",
1781 aggregateId: ThreadId.make(threadId),
1782 type: "thread.created",
1783 sequenceExclusive: event.sequence,
1784 });
1785 if (!recreatedLater) {
1786 deletedThreadIds.add(threadId);
1787 }
1788 }
1789
1790 // Later events in the same transaction can add attachment references.
1791 const prunedThreadRelativePaths = new Map<string, Set<string>>();
1792 for (const threadId of sideEffects.prunedThreadRelativePaths.keys()) {
1793 const messages = yield* projectionThreadMessageRepository.listByThreadId({
1794 threadId: ThreadId.make(threadId),
1795 });
1796 prunedThreadRelativePaths.set(
1797 threadId,
1798 collectThreadAttachmentRelativePaths(threadId, messages),
1799 );
1800 }
1801
1802 yield* runAttachmentSideEffects({ deletedThreadIds, prunedThreadRelativePaths });
1803 },
1804 Effect.provideService(FileSystem.FileSystem, fileSystem),
1805 Effect.provideService(Path.Path, path),
1806 Effect.provideService(ServerConfig, serverConfig),
1807 (effect, event) =>
1808 effect.pipe(
1809 Effect.catch((cause) =>
1810 Effect.logWarning("failed to apply projected attachment side-effects", {
1811 sequence: event.sequence,
1812 eventType: event.type,
1813 cause,
1814 }),
1815 ),
1816 ),
1817 );
1818
1819 const applyProjectorForEvent = Effect.fn("applyProjectorForEvent")(function* (
1820 projector: ProjectorDefinition,
1821 event: OrchestrationEvent,
1822 attachmentSideEffects: AttachmentSideEffects,
1823 ) {
1824 yield* projector.apply(event, attachmentSideEffects);
1825 yield* projectionStateRepository.upsert({
1826 projector: projector.name,
1827 lastAppliedSequence: event.sequence,
1828 updatedAt: event.occurredAt,
1829 });
1830 });
1831
17681832 const runProjectorForEvent = Effect.fn("runProjectorForEvent")(function* (
17691833 projector: ProjectorDefinition,
17701834 event: OrchestrationEvent,
......@@ -1774,28 +1838,8 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
17741838 prunedThreadRelativePaths: new Map<string, Set<string>>(),
17751839 };
17761840
1777 yield* sql.withTransaction(
1778 projector.apply(event, attachmentSideEffects).pipe(
1779 Effect.flatMap(() =>
1780 projectionStateRepository.upsert({
1781 projector: projector.name,
1782 lastAppliedSequence: event.sequence,
1783 updatedAt: event.occurredAt,
1784 }),
1785 ),
1786 ),
1787 );
1788
1789 yield* runAttachmentSideEffects(attachmentSideEffects).pipe(
1790 Effect.catch((cause) =>
1791 Effect.logWarning("failed to apply projected attachment side-effects", {
1792 projector: projector.name,
1793 sequence: event.sequence,
1794 eventType: event.type,
1795 cause,
1796 }),
1797 ),
1798 );
1841 yield* sql.withTransaction(applyProjectorForEvent(projector, event, attachmentSideEffects));
1842 yield* applyAttachmentSideEffects(event, attachmentSideEffects);
17991843 });
18001844
18011845 const bootstrapProjector = (projector: ProjectorDefinition) =>
......@@ -1815,19 +1859,39 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
18151859 ),
18161860 );
18171861
1818 const projectEvent: OrchestrationProjectionPipelineShape["projectEvent"] = (event) =>
1819 Effect.forEach(projectors, (projector) => runProjectorForEvent(projector, event), {
1820 concurrency: 1,
1821 }).pipe(
1862 const projectEventDeferred: OrchestrationProjectionPipelineShape["projectEventDeferred"] =
1863 Effect.fn("projectEventDeferred")(
1864 function* (event) {
1865 const attachmentSideEffects: AttachmentSideEffects = {
1866 deletedThreadIds: new Set<string>(),
1867 prunedThreadRelativePaths: new Map<string, Set<string>>(),
1868 };
1869 yield* sql.withTransaction(
1870 Effect.forEach(
1871 projectors,
1872 (projector) => applyProjectorForEvent(projector, event, attachmentSideEffects),
1873 { concurrency: 1, discard: true },
1874 ),
1875 );
1876 // Return the cleanup effect so the caller runs it after the outer transaction commits.
1877 // @effect-diagnostics-next-line returnEffectInGen:off
1878 return applyAttachmentSideEffects(event, attachmentSideEffects);
1879 },
18221880 Effect.provideService(FileSystem.FileSystem, fileSystem),
18231881 Effect.provideService(Path.Path, path),
18241882 Effect.provideService(ServerConfig, serverConfig),
1825 Effect.asVoid,
18261883 Effect.catchTag("SqlError", (sqlError) =>
18271884 Effect.fail(toPersistenceSqlError("ProjectionPipeline.projectEvent:query")(sqlError)),
18281885 ),
18291886 );
18301887
1888 const projectEvent: OrchestrationProjectionPipelineShape["projectEvent"] = Effect.fn(
1889 "projectEvent",
1890 )(function* (event) {
1891 const cleanup = yield* projectEventDeferred(event);
1892 yield* cleanup;
1893 });
1894
18311895 const bootstrap: OrchestrationProjectionPipelineShape["bootstrap"] = Effect.forEach(
18321896 projectors,
18331897 bootstrapProjector,
......@@ -1850,6 +1914,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
18501914 return {
18511915 bootstrap,
18521916 projectEvent,
1917 projectEventDeferred,
18531918 } satisfies OrchestrationProjectionPipelineShape;
18541919 },
18551920);
apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts+50-1
......@@ -61,6 +61,7 @@ import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"
6161import * as NodeServices from "@effect/platform-node/NodeServices";
6262import * as Clock from "effect/Clock";
6363import { ServerSettingsService } from "../../serverSettings.ts";
64import { ServerActivation } from "../../serverActivation.ts";
6465import { VcsStatusBroadcaster } from "../../vcs/VcsStatusBroadcaster.ts";
6566import * as GitWorkflowService from "../../git/GitWorkflowService.ts";
6667
......@@ -150,6 +151,7 @@ describe("ProviderCommandReactor", () => {
150151 readonly requiresNewThreadForModelChange?: boolean;
151152 readonly titleRegenerationCompletionDispatchFailures?: number;
152153 readonly titleRegenerationBeforeStart?: "one" | "two";
154 readonly serverActivation?: Effect.Effect<void>;
153155 readonly interruptTurnEffect?: () => Effect.Effect<void, ProviderAdapterRequestError>;
154156 readonly stopSessionEffect?: () => Effect.Effect<void, ProviderAdapterRequestError>;
155157 readonly startSessionEffect?: (
......@@ -395,6 +397,7 @@ describe("ProviderCommandReactor", () => {
395397 get streamDomainEvents() {
396398 return engine.streamDomainEvents;
397399 },
400 subscribeDomainEvents: engine.subscribeDomainEvents,
398401 latestSequence: engine.latestSequence,
399402 } satisfies OrchestrationEngineService["Service"];
400403 }),
......@@ -500,7 +503,14 @@ describe("ProviderCommandReactor", () => {
500503 }
501504
502505 scope = await Effect.runPromise(Scope.make("sequential"));
503 await Effect.runPromise(reactor.start().pipe(Scope.provide(scope)));
506 await Effect.runPromise(
507 reactor
508 .start()
509 .pipe(
510 Scope.provide(scope),
511 Effect.provideService(ServerActivation, input?.serverActivation),
512 ),
513 );
504514 const drain = () => Effect.runPromise(reactor.drain);
505515
506516 return {
......@@ -568,6 +578,45 @@ describe("ProviderCommandReactor", () => {
568578 expect(thread?.session?.runtimeMode).toBe("approval-required");
569579 });
570580
581 effectIt.effect("retains a turn dispatched immediately after start until activation", () =>
582 Effect.gen(function* () {
583 const activation = yield* Deferred.make<void>();
584 const started = yield* Deferred.make<ProviderSession>();
585 const harness = yield* Effect.promise(() =>
586 createHarness({
587 serverActivation: Deferred.await(activation),
588 startSessionEffect: (session) =>
589 Deferred.succeed(started, session).pipe(Effect.as(session)),
590 }),
591 );
592
593 yield* harness.engine.dispatch({
594 type: "thread.turn.start",
595 commandId: CommandId.make("cmd-turn-start-before-activation"),
596 threadId: ThreadId.make("thread-1"),
597 message: {
598 messageId: MessageId.make("message-before-activation"),
599 role: "user",
600 text: "Start after activation",
601 attachments: [],
602 },
603 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
604 runtimeMode: "approval-required",
605 createdAt: "2026-01-01T00:00:00.000Z",
606 });
607 expect(yield* Deferred.isDone(started)).toBe(false);
608
609 yield* Deferred.succeed(activation, undefined);
610 const session = yield* Deferred.await(started);
611 yield* Effect.promise(() => harness.drain());
612 expect(session.threadId).toBe(ThreadId.make("thread-1"));
613 expect(harness.sendTurn.mock.calls[0]?.[0]).toMatchObject({
614 threadId: ThreadId.make("thread-1"),
615 input: "Start after activation",
616 });
617 }),
618 );
619
571620 effectIt.effect("projects starting before a slow provider session finishes", () =>
572621 Effect.gen(function* () {
573622 const releaseStart = yield* Deferred.make<void>();
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts+3-1
......@@ -1554,7 +1554,9 @@ const make = Effect.gen(function* () {
15541554 }
15551555 });
15561556
1557 yield* forkParked(Stream.runForEach(orchestrationEngine.streamDomainEvents, processEvent));
1557 // Subscribe before returning, even while event handling waits for server activation.
1558 const domainEvents = yield* orchestrationEngine.subscribeDomainEvents;
1559 yield* forkParked(Stream.runForEach(domainEvents, processEvent));
15581560
15591561 // The domain event stream is hot, so work pending before this reactor
15601562 // starts cannot be resumed. Correlated completions only clear the request
apps/server/src/orchestration/Services/OrchestrationEngine.ts+11
......@@ -17,6 +17,7 @@ import type {
1717} from "@t3tools/contracts";
1818import * as Context from "effect/Context";
1919import type * as Effect from "effect/Effect";
20import type * as Scope from "effect/Scope";
2021import type * as Stream from "effect/Stream";
2122
2223import type { OrchestrationDispatchError } from "../Errors.ts";
......@@ -64,6 +65,16 @@ export interface OrchestrationEngineShape {
6465 */
6566 readonly streamDomainEvents: Stream.Stream<OrchestrationEvent>;
6667
68 /**
69 * Acquire a domain-event subscription before starting a consumer.
70 * The subscription is ready when this effect returns and closes with the scope.
71 */
72 readonly subscribeDomainEvents: Effect.Effect<
73 Stream.Stream<OrchestrationEvent>,
74 never,
75 Scope.Scope
76 >;
77
6778 /**
6879 * The latest sequence reflected in the engine's authoritative command read
6980 * model (0 if none). Used to gauge how far behind a resuming client is before
apps/server/src/orchestration/Services/ProjectionPipeline.ts+10-1
......@@ -26,11 +26,20 @@ export interface OrchestrationProjectionPipelineShape {
2626 /**
2727 * Project a single orchestration event into projection repositories.
2828 *
29 * Projectors are executed sequentially to preserve deterministic ordering.
29 * Projectors run sequentially in one transaction. Attachment cleanup runs
30 * after that transaction commits.
3031 */
3132 readonly projectEvent: (
3233 event: OrchestrationEvent,
3334 ) => Effect.Effect<void, ProjectionRepositoryError>;
35
36 /**
37 * Project an event inside a caller's transaction and return its attachment
38 * cleanup. Run the returned effect only after the outer transaction commits.
39 */
40 readonly projectEventDeferred: (
41 event: OrchestrationEvent,
42 ) => Effect.Effect<Effect.Effect<void>, ProjectionRepositoryError>;
3443}
3544
3645/**
apps/server/src/orchestration/Services/ProviderCommandReactor.ts+1
......@@ -19,6 +19,7 @@ export interface ProviderCommandReactorShape {
1919 *
2020 * The returned effect must be run in a scope so all worker fibers can be
2121 * finalized on shutdown.
22 * It subscribes before returning. Event handling waits for server activation.
2223 *
2324 * Filters orchestration domain events to provider-intent types before
2425 * processing.
apps/server/src/relay/AgentAwarenessRelay.test.ts+2
......@@ -474,6 +474,7 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {
474474 readEvents: () => Stream.empty,
475475 dispatch: () => Effect.succeed({ sequence: 1 }),
476476 streamDomainEvents: Stream.fromQueue(events),
477 subscribeDomainEvents: Effect.succeed(Stream.fromQueue(events)),
477478 latestSequence: Effect.succeed(0),
478479 } satisfies OrchestrationEngineShape;
479480
......@@ -666,6 +667,7 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {
666667 readEvents: () => Stream.empty,
667668 dispatch: () => Effect.succeed({ sequence: 1 }),
668669 streamDomainEvents: Stream.fromQueue(events),
670 subscribeDomainEvents: Effect.succeed(Stream.fromQueue(events)),
669671 latestSequence: Effect.succeed(0),
670672 } satisfies OrchestrationEngineShape),
671673 Layer.succeed(ProjectionSnapshotQuery, {
apps/server/src/serverRuntimeStartup.reconcile.test.ts+2
......@@ -84,6 +84,7 @@ const runReconciliation = (input: {
8484 readEvents: () => Stream.empty,
8585 dispatch: input.dispatch,
8686 streamDomainEvents: Stream.empty,
87 subscribeDomainEvents: Effect.succeed(Stream.empty),
8788 latestSequence: Effect.succeed(0),
8889 }),
8990 Effect.provide(NodeServices.layer),
......@@ -289,6 +290,7 @@ it.effect("does not fail startup when the live provider session inventory cannot
289290 readEvents: () => Stream.empty,
290291 dispatch: () => Effect.die("unused"),
291292 streamDomainEvents: Stream.empty,
293 subscribeDomainEvents: Effect.succeed(Stream.empty),
292294 latestSequence: Effect.succeed(0),
293295 }),
294296 Effect.provide(NodeServices.layer),
apps/server/src/serverRuntimeStartup.test.ts+3
......@@ -181,6 +181,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa
181181 Effect.as({ sequence: 1 }),
182182 ),
183183 streamDomainEvents: Stream.empty,
184 subscribeDomainEvents: Effect.succeed(Stream.empty),
184185 latestSequence: Effect.succeed(0),
185186 } satisfies OrchestrationEngine.OrchestrationEngineService["Service"]),
186187 Effect.provide(NodeServices.layer),
......@@ -227,6 +228,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets creates a project and thread when
227228 Effect.as({ sequence: 1 }),
228229 ),
229230 streamDomainEvents: Stream.empty,
231 subscribeDomainEvents: Effect.succeed(Stream.empty),
230232 latestSequence: Effect.succeed(0),
231233 } satisfies OrchestrationEngine.OrchestrationEngineService["Service"]),
232234 Effect.provide(NodeServices.layer),
......@@ -279,6 +281,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa
279281 Effect.as({ sequence: 1 }),
280282 ),
281283 streamDomainEvents: Stream.empty,
284 subscribeDomainEvents: Effect.succeed(Stream.empty),
282285 latestSequence: Effect.succeed(0),
283286 } satisfies OrchestrationEngine.OrchestrationEngineService["Service"]),
284287 Effect.provideService(Crypto.Crypto, {
docs/internals/overview.md+2-1
......@@ -73,7 +73,8 @@ processing is totally ordered. For each envelope `processEnvelope`:
73733. inside one SQL transaction, appends events to the event store, applies them to the in-memory read
7474 model via [`projector.ts`][projector], projects them into persisted tables, and writes the
7575 accepted receipt;
764. after commit, swaps in the new read model and publishes committed events to subscribers.
764. after commit, swaps in the new read model, cleans up attachments, and publishes committed events
77 to subscribers. Attachment cleanup failures are logged and do not reject committed commands.
7778
7879Because persistence and projection share a transaction, the read model cannot durably disagree with
7980the event log. On dispatch failure the engine rereads persisted events past the starting sequence and