authorgravatar for 91240327+shivamhwp@users.noreply.github.comshivam <91240327+shivamhwp@users.noreply.github.com> 2026-08-29 02:46:37+05:30
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-08-29 02:46:37+05:30
loga40aef4ccbc10b7bcd7bc817151db8d5a9b1c257
treec652816f82524c36207701ce54e7f2c87363abb5
parent48c176b3cfda7a714c10f18a12d398528ff4c2f9
signature Signed by PGP key B5690EEEBB952194

fix(server): a draft can retry its first send after a failed bootstrap (#8226)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

16 files changed, 690 insertions(+), 25 deletions(-)

apps/server/integration/OrchestrationEngineHarness.integration.ts+1-1
......@@ -373,7 +373,7 @@ export const makeOrchestrationIntegrationHarness = (
373373 Layer.provideMerge(
374374 Layer.succeed(ThreadDeletionReactor, {
375375 start: () => Effect.void,
376 drain: Effect.void,
376 drainThrough: () => Effect.void,
377377 }),
378378 ),
379379 Layer.provideMerge(
apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts+3
......@@ -113,6 +113,7 @@ describe("OrchestrationEngine", () => {
113113 detail: "historical replay should not be used during bootstrap",
114114 }),
115115 ),
116 hasEventAfter: () => Effect.succeed(false),
116117 };
117118
118119 const projectionSnapshot = {
......@@ -812,6 +813,7 @@ describe("OrchestrationEngine", () => {
812813 readAll() {
813814 return Stream.fromIterable(events);
814815 },
816 hasEventAfter: () => Effect.succeed(false),
815817 };
816818
817819 const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), {
......@@ -1048,6 +1050,7 @@ describe("OrchestrationEngine", () => {
10481050 readAll() {
10491051 return Stream.fromIterable(events);
10501052 },
1053 hasEventAfter: () => Effect.succeed(false),
10511054 };
10521055
10531056 let shouldFailProjection = true;
apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts+1-1
......@@ -61,7 +61,7 @@ describe("OrchestrationReactor", () => {
6161 started.push("thread-deletion-reactor");
6262 return Effect.void;
6363 },
64 drain: Effect.void,
64 drainThrough: () => Effect.void,
6565 }),
6666 ),
6767 Layer.provideMerge(
apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+251-1
......@@ -9,6 +9,7 @@ import {
99 TurnId,
1010 ProviderInstanceId,
1111} from "@t3tools/contracts";
12import * as Option from "effect/Option";
1213import * as NodeServices from "@effect/platform-node/NodeServices";
1314import { assert, it } from "@effect/vitest";
1415import * as Effect from "effect/Effect";
......@@ -31,6 +32,7 @@ import {
3132 OrchestrationProjectionPipelineLive,
3233} from "./ProjectionPipeline.ts";
3334import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";
35import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";
3436import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts";
3537import * as ThreadPlanProgress from "../ThreadPlanProgress.ts";
3638import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
......@@ -1277,6 +1279,111 @@ it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-atta
12771279 },
12781280);
12791281
1282it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-attachments-replay-")))(
1283 "OrchestrationProjectionPipeline",
1284 (it) => {
1285 it.effect("replaying a superseded thread.deleted keeps the re-created thread's files", () =>
1286 Effect.gen(function* () {
1287 const fileSystem = yield* FileSystem.FileSystem;
1288 const path = yield* Path.Path;
1289 const projectionPipeline = yield* OrchestrationProjectionPipeline;
1290 const eventStore = yield* OrchestrationEventStore;
1291 const { attachmentsDir } = yield* ServerConfig;
1292 const now = "2026-01-01T00:00:00.000Z";
1293 const projectId = ProjectId.make("project-replay");
1294 const retriedThreadId = ThreadId.make("thread-replay-retried");
1295 const goneThreadId = ThreadId.make("thread-replay-gone");
1296 const retriedAttachmentPath = path.join(
1297 attachmentsDir,
1298 "thread-replay-retried-00000000-0000-4000-8000-000000000001.png",
1299 );
1300 const goneAttachmentPath = path.join(
1301 attachmentsDir,
1302 "thread-replay-gone-00000000-0000-4000-8000-000000000002.png",
1303 );
1304 const threadCreated = (threadId: ThreadId, suffix: string) =>
1305 eventStore.append({
1306 type: "thread.created",
1307 eventId: EventId.make(`evt-replay-create-${suffix}`),
1308 aggregateKind: "thread",
1309 aggregateId: threadId,
1310 occurredAt: now,
1311 commandId: CommandId.make(`cmd-replay-create-${suffix}`),
1312 causationEventId: null,
1313 correlationId: CorrelationId.make(`cmd-replay-create-${suffix}`),
1314 metadata: {},
1315 payload: {
1316 threadId,
1317 projectId,
1318 title: `Thread ${suffix}`,
1319 modelSelection: {
1320 instanceId: ProviderInstanceId.make("codex"),
1321 model: "gpt-5-codex",
1322 },
1323 runtimeMode: "full-access",
1324 branch: null,
1325 worktreePath: null,
1326 createdAt: now,
1327 updatedAt: now,
1328 },
1329 });
1330 const threadDeleted = (threadId: ThreadId, suffix: string) =>
1331 eventStore.append({
1332 type: "thread.deleted",
1333 eventId: EventId.make(`evt-replay-delete-${suffix}`),
1334 aggregateKind: "thread",
1335 aggregateId: threadId,
1336 occurredAt: now,
1337 commandId: CommandId.make(`cmd-replay-delete-${suffix}`),
1338 causationEventId: null,
1339 correlationId: CorrelationId.make(`cmd-replay-delete-${suffix}`),
1340 metadata: {},
1341 payload: { threadId, deletedAt: now },
1342 });
1343
1344 yield* eventStore.append({
1345 type: "project.created",
1346 eventId: EventId.make("evt-replay-project"),
1347 aggregateKind: "project",
1348 aggregateId: projectId,
1349 occurredAt: now,
1350 commandId: CommandId.make("cmd-replay-project"),
1351 causationEventId: null,
1352 correlationId: CorrelationId.make("cmd-replay-project"),
1353 metadata: {},
1354 payload: {
1355 projectId,
1356 title: "Replay",
1357 workspaceRoot: "/tmp/project-replay",
1358 defaultModelSelection: null,
1359 scripts: [],
1360 createdAt: now,
1361 updatedAt: now,
1362 },
1363 });
1364 // A failed first send: create, roll back, then the draft retries the id.
1365 yield* threadCreated(retriedThreadId, "retried-1");
1366 yield* threadDeleted(retriedThreadId, "retried");
1367 yield* threadCreated(retriedThreadId, "retried-2");
1368 // A thread that was deleted for good.
1369 yield* threadCreated(goneThreadId, "gone");
1370 yield* threadDeleted(goneThreadId, "gone");
1371
1372 // Files on disk are not event-sourced: by the time anything replays,
1373 // the retried thread's attachments already belong to its second life.
1374 yield* fileSystem.makeDirectory(attachmentsDir, { recursive: true });
1375 yield* fileSystem.writeFileString(retriedAttachmentPath, "second incarnation");
1376 yield* fileSystem.writeFileString(goneAttachmentPath, "gone");
1377
1378 yield* projectionPipeline.bootstrap;
1379
1380 assert.isTrue(yield* exists(retriedAttachmentPath));
1381 assert.isFalse(yield* exists(goneAttachmentPath));
1382 }),
1383 );
1384 },
1385);
1386
12801387it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
12811388 it.effect("replays a bootstrap backlog larger than the event store default limit", () =>
12821389 Effect.gen(function* () {
......@@ -2850,7 +2957,7 @@ it.effect("restores pending turn-start metadata across projection pipeline resta
28502957
28512958const engineLayer = it.layer(
28522959 OrchestrationEngineLive.pipe(
2853 Layer.provide(OrchestrationProjectionSnapshotQueryLive),
2960 Layer.provideMerge(OrchestrationProjectionSnapshotQueryLive),
28542961 Layer.provide(ThreadBackgroundLiveness.layer),
28552962 Layer.provide(ThreadPlanProgress.layer),
28562963 Layer.provide(OrchestrationProjectionPipelineLive),
......@@ -2967,4 +3074,147 @@ engineLayer("OrchestrationProjectionPipeline via engine dispatch", (it) => {
29673074 ]);
29683075 }),
29693076 );
3077
3078 it.effect("re-creating a deleted thread id starts from an empty projection", () =>
3079 Effect.gen(function* () {
3080 const engine = yield* OrchestrationEngineService;
3081 const snapshotQuery = yield* ProjectionSnapshotQuery;
3082 const sql = yield* SqlClient.SqlClient;
3083 const createdAt = "2026-01-01T00:00:00.000Z";
3084 const projectId = ProjectId.make("project-retry");
3085 const threadId = ThreadId.make("thread-retry");
3086 const modelSelection = {
3087 instanceId: ProviderInstanceId.make("codex"),
3088 model: "gpt-5-codex",
3089 };
3090 const createThread = (commandId: string, title: string) =>
3091 engine.dispatch({
3092 type: "thread.create",
3093 commandId: CommandId.make(commandId),
3094 threadId,
3095 projectId,
3096 title,
3097 modelSelection,
3098 runtimeMode: "full-access",
3099 interactionMode: "default",
3100 branch: null,
3101 worktreePath: null,
3102 createdAt,
3103 });
3104 const countRowsForThread = (table: string) =>
3105 sql<{ readonly count: number }>`
3106 SELECT COUNT(*) AS count FROM ${sql(table)} WHERE thread_id = ${threadId}
3107 `.pipe(Effect.map((rows) => rows[0]?.count ?? 0));
3108 const perThreadTables = [
3109 "projection_thread_messages",
3110 "projection_thread_activities",
3111 "projection_thread_sessions",
3112 "projection_turns",
3113 "projection_thread_proposed_plans",
3114 "projection_pending_approvals",
3115 ];
3116
3117 yield* engine.dispatch({
3118 type: "project.create",
3119 commandId: CommandId.make("cmd-retry-project"),
3120 projectId,
3121 title: "Retry Project",
3122 workspaceRoot: "/tmp/project-retry",
3123 defaultModelSelection: modelSelection,
3124 createdAt,
3125 });
3126
3127 // First attempt: the thread gets a turn, a message, an activity, and a
3128 // running session before its bootstrap fails and the server rolls back.
3129 yield* createThread("cmd-retry-create-1", "First attempt");
3130 yield* engine.dispatch({
3131 type: "thread.turn.start",
3132 commandId: CommandId.make("cmd-retry-turn-1"),
3133 threadId,
3134 message: {
3135 messageId: MessageId.make("message-retry-1"),
3136 role: "user",
3137 text: "first attempt",
3138 attachments: [],
3139 },
3140 runtimeMode: "full-access",
3141 interactionMode: "default",
3142 createdAt,
3143 });
3144 yield* engine.dispatch({
3145 type: "thread.activity.append",
3146 commandId: CommandId.make("cmd-retry-activity-1"),
3147 threadId,
3148 activity: {
3149 id: EventId.make("activity-retry-1"),
3150 tone: "info",
3151 kind: "approval.requested",
3152 summary: "approval requested",
3153 payload: { requestId: "request-retry-1" },
3154 turnId: null,
3155 createdAt,
3156 },
3157 createdAt,
3158 });
3159 yield* engine.dispatch({
3160 type: "thread.proposed-plan.upsert",
3161 commandId: CommandId.make("cmd-retry-plan-1"),
3162 threadId,
3163 proposedPlan: {
3164 id: "plan-retry-1",
3165 turnId: null,
3166 planMarkdown: "# Plan",
3167 implementedAt: null,
3168 implementationThreadId: null,
3169 createdAt,
3170 updatedAt: createdAt,
3171 },
3172 createdAt,
3173 });
3174 yield* engine.dispatch({
3175 type: "thread.session.set",
3176 commandId: CommandId.make("cmd-retry-session-1"),
3177 threadId,
3178 session: {
3179 threadId,
3180 status: "running",
3181 providerName: "codex",
3182 runtimeMode: "full-access",
3183 activeTurnId: TurnId.make("turn-retry-1"),
3184 lastError: null,
3185 updatedAt: createdAt,
3186 },
3187 createdAt,
3188 });
3189 for (const table of perThreadTables) {
3190 assert.isAbove(yield* countRowsForThread(table), 0, `${table} should be populated`);
3191 }
3192 const populatedShell = Option.getOrThrow(yield* snapshotQuery.getThreadShellById(threadId));
3193 assert.isTrue(populatedShell.hasPendingApprovals);
3194 assert.isTrue(populatedShell.hasActionableProposedPlan);
3195
3196 yield* engine.dispatch({
3197 type: "thread.delete",
3198 commandId: CommandId.make("cmd-retry-delete"),
3199 threadId,
3200 });
3201 assert.isTrue(Option.isNone(yield* snapshotQuery.getThreadShellById(threadId)));
3202
3203 // Retry from the same draft reuses the thread id.
3204 yield* createThread("cmd-retry-create-2", "Second attempt");
3205
3206 const shell = Option.getOrThrow(yield* snapshotQuery.getThreadShellById(threadId));
3207 assert.strictEqual(shell.title, "Second attempt");
3208 assert.isFalse(shell.hasPendingApprovals);
3209 assert.isFalse(shell.hasActionableProposedPlan);
3210 for (const table of perThreadTables) {
3211 assert.strictEqual(yield* countRowsForThread(table), 0, `${table} should be empty`);
3212 }
3213 const detail = Option.getOrThrow(yield* snapshotQuery.getThreadDetailById(threadId));
3214 assert.deepEqual(detail.messages, []);
3215 assert.deepEqual(detail.activities, []);
3216 assert.isNull(detail.latestTurn);
3217 assert.isNull(detail.session);
3218 }),
3219 );
29703220});
apps/server/src/orchestration/Layers/ProjectionPipeline.ts+51-1
......@@ -871,7 +871,18 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
871871 }
872872
873873 case "thread.deleted": {
874 attachmentSideEffects.deletedThreadIds.add(event.payload.threadId);
874 // A draft retry can re-create this id later in the log. During
875 // replay the attachment files on disk already belong to that later
876 // incarnation, so only an unsuperseded deletion removes them.
877 const recreatedLater = yield* eventStore.hasEventAfter({
878 aggregateKind: "thread",
879 aggregateId: event.payload.threadId,
880 type: "thread.created",
881 sequenceExclusive: event.sequence,
882 });
883 if (!recreatedLater) {
884 attachmentSideEffects.deletedThreadIds.add(event.payload.threadId);
885 }
875886 const existingRow = yield* projectionThreadRepository.getById({
876887 threadId: event.payload.threadId,
877888 });
......@@ -987,6 +998,15 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
987998 "applyThreadMessagesProjection",
988999 )(function* (event, attachmentSideEffects) {
9891000 switch (event.type) {
1001 // A draft retry re-creates a soft-deleted thread id. Every projector
1002 // drops its own rows for the old incarnation here so replay from any
1003 // per-projector cursor rebuilds the new thread without stale history.
1004 case "thread.created":
1005 yield* projectionThreadMessageRepository.deleteByThreadId({
1006 threadId: event.payload.threadId,
1007 });
1008 return;
1009
9901010 case "thread.message-sent": {
9911011 const existingMessage = yield* projectionThreadMessageRepository.getByMessageId({
9921012 messageId: event.payload.messageId,
......@@ -1066,6 +1086,12 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
10661086 "applyThreadProposedPlansProjection",
10671087 )(function* (event, _attachmentSideEffects) {
10681088 switch (event.type) {
1089 case "thread.created":
1090 yield* projectionThreadProposedPlanRepository.deleteByThreadId({
1091 threadId: event.payload.threadId,
1092 });
1093 return;
1094
10691095 case "thread.proposed-plan-upserted":
10701096 yield* projectionThreadProposedPlanRepository.upsert({
10711097 planId: event.payload.proposedPlan.id,
......@@ -1117,6 +1143,12 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
11171143 "applyThreadActivitiesProjection",
11181144 )(function* (event, _attachmentSideEffects) {
11191145 switch (event.type) {
1146 case "thread.created":
1147 yield* projectionThreadActivityRepository.deleteByThreadId({
1148 threadId: event.payload.threadId,
1149 });
1150 return;
1151
11201152 case "thread.activity-appended":
11211153 yield* projectionThreadActivityRepository.upsert({
11221154 activityId: event.payload.activity.id,
......@@ -1168,6 +1200,12 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
11681200 const applyThreadSessionsProjection: ProjectorDefinition["apply"] = Effect.fn(
11691201 "applyThreadSessionsProjection",
11701202 )(function* (event, _attachmentSideEffects) {
1203 if (event.type === "thread.created") {
1204 yield* projectionThreadSessionRepository.deleteByThreadId({
1205 threadId: event.payload.threadId,
1206 });
1207 return;
1208 }
11711209 if (event.type !== "thread.session-set") {
11721210 return;
11731211 }
......@@ -1187,6 +1225,12 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
11871225 "applyThreadTurnsProjection",
11881226 )(function* (event, _attachmentSideEffects) {
11891227 switch (event.type) {
1228 case "thread.created":
1229 yield* projectionTurnRepository.deleteByThreadId({
1230 threadId: event.payload.threadId,
1231 });
1232 return;
1233
11901234 case "thread.turn-start-requested": {
11911235 yield* projectionTurnRepository.replacePendingTurnStart({
11921236 threadId: event.payload.threadId,
......@@ -1524,6 +1568,12 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
15241568 "applyPendingApprovalsProjection",
15251569 )(function* (event, _attachmentSideEffects) {
15261570 switch (event.type) {
1571 case "thread.created":
1572 yield* projectionPendingApprovalRepository.deleteByThreadId({
1573 threadId: event.payload.threadId,
1574 });
1575 return;
1576
15271577 case "thread.activity-appended": {
15281578 const requestId =
15291579 extractActivityRequestId(event.payload.activity.payload) ??
apps/server/src/orchestration/Layers/ThreadDeletionReactor.test.ts+103-2
......@@ -1,10 +1,35 @@
1import { ThreadId } from "@t3tools/contracts";
1import {
2 CommandId,
3 CorrelationId,
4 EventId,
5 type OrchestrationEvent,
6 ThreadId,
7} from "@t3tools/contracts";
8import { it as effectIt } from "@effect/vitest";
29import * as Cause from "effect/Cause";
10import * as Deferred from "effect/Deferred";
311import * as Effect from "effect/Effect";
412import * as Exit from "effect/Exit";
13import * as Fiber from "effect/Fiber";
14import * as Layer from "effect/Layer";
15import * as Ref from "effect/Ref";
16import * as Stream from "effect/Stream";
517import { describe, expect, it } from "vite-plus/test";
618
7import { logCleanupCauseUnlessInterrupted } from "./ThreadDeletionReactor.ts";
19import {
20 ProviderService,
21 type ProviderServiceShape,
22} from "../../provider/Services/ProviderService.ts";
23import * as TerminalManager from "../../terminal/Manager.ts";
24import {
25 OrchestrationEngineService,
26 type OrchestrationEngineShape,
27} from "../Services/OrchestrationEngine.ts";
28import { ThreadDeletionReactor } from "../Services/ThreadDeletionReactor.ts";
29import {
30 logCleanupCauseUnlessInterrupted,
31 ThreadDeletionReactorLive,
32} from "./ThreadDeletionReactor.ts";
833
934describe("logCleanupCauseUnlessInterrupted", () => {
1035 const threadId = ThreadId.make("thread-deletion-reactor-test");
......@@ -36,3 +61,79 @@ describe("logCleanupCauseUnlessInterrupted", () => {
3661 }
3762 });
3863});
64
65describe("ThreadDeletionReactor drain", () => {
66 const now = "2026-01-01T00:00:00.000Z";
67 const threadId = ThreadId.make("thread-deletion-reactor-drain");
68 const deletedEvent = (sequence: number): OrchestrationEvent => ({
69 sequence,
70 eventId: EventId.make(`evt-deleted-${sequence}`),
71 aggregateKind: "thread",
72 aggregateId: threadId,
73 type: "thread.deleted",
74 occurredAt: now,
75 commandId: CommandId.make(`cmd-deleted-${sequence}`),
76 causationEventId: null,
77 correlationId: CorrelationId.make(`cmd-deleted-${sequence}`),
78 metadata: {},
79 payload: { threadId, deletedAt: now },
80 });
81
82 effectIt.effect("waits for a published deletion the subscriber has not consumed yet", () =>
83 Effect.gen(function* () {
84 const stops: Array<number> = [];
85 const firstCleanupDone = yield* Deferred.make<void>();
86 // The engine has already committed and published sequence 2, but the
87 // subscriber has not received it yet: the stream releases it on demand.
88 const releaseSecondEvent = yield* Deferred.make<void>();
89 const latestSequence = yield* Ref.make(0);
90 const engine = {
91 latestSequence: Ref.get(latestSequence),
92 streamDomainEvents: Stream.concat(
93 Stream.make(deletedEvent(1)),
94 Stream.fromEffect(Deferred.await(releaseSecondEvent)).pipe(
95 Stream.map(() => deletedEvent(2)),
96 ),
97 ),
98 } as unknown as OrchestrationEngineShape;
99 const providerService = {
100 stopSession: () =>
101 Effect.gen(function* () {
102 stops.push(stops.length + 1);
103 if (stops.length === 1) {
104 yield* Deferred.succeed(firstCleanupDone, undefined);
105 }
106 }),
107 } as unknown as ProviderServiceShape;
108 const terminalManager = {
109 close: () => Effect.void,
110 } as unknown as TerminalManager.TerminalManager["Service"];
111 const layer = ThreadDeletionReactorLive.pipe(
112 Layer.provide(Layer.succeed(ProviderService, providerService)),
113 Layer.provide(Layer.succeed(TerminalManager.TerminalManager, terminalManager)),
114 Layer.provide(Layer.succeed(OrchestrationEngineService, engine)),
115 );
116
117 yield* Effect.scoped(
118 Effect.gen(function* () {
119 const reactor = yield* ThreadDeletionReactor;
120 yield* reactor.start();
121 yield* Deferred.await(firstCleanupDone);
122
123 // Sequence 1 is fully cleaned and the worker queue is idle. Sequence
124 // 2 is committed and published but still in flight to the subscriber.
125 yield* Ref.set(latestSequence, 2);
126 const drained = yield* Effect.forkChild(reactor.drainThrough(2));
127 yield* Effect.yieldNow;
128 yield* Effect.yieldNow;
129 expect(stops).toEqual([1]);
130 expect(drained.pollUnsafe()).toBeUndefined();
131
132 yield* Deferred.succeed(releaseSecondEvent, undefined);
133 yield* Fiber.join(drained);
134 expect(stops).toEqual([1, 2]);
135 }),
136 ).pipe(Effect.provide(layer));
137 }),
138 );
139});
apps/server/src/orchestration/Layers/ThreadDeletionReactor.ts+31-7
......@@ -4,6 +4,7 @@ import * as Cause from "effect/Cause";
44import * as Effect from "effect/Effect";
55import * as Layer from "effect/Layer";
66import * as Stream from "effect/Stream";
7import * as SubscriptionRef from "effect/SubscriptionRef";
78
89import { ProviderService } from "../../provider/Services/ProviderService.ts";
910import * as TerminalManager from "../../terminal/Manager.ts";
......@@ -80,20 +81,43 @@ const make = Effect.gen(function* () {
8081
8182 const worker = yield* makeDrainableWorker(processThreadDeletedSafely);
8283
84 // Highest event sequence the subscriber has handed to the worker. Waiting
85 // through a successful thread.created sequence covers every deletion that
86 // was ahead of that create in the engine queue; the worker drain then covers
87 // the in-flight cleanup.
88 const seenSequence = yield* SubscriptionRef.make(0);
89 const noteSeen = (sequence: number) =>
90 SubscriptionRef.update(seenSequence, (seen) => Math.max(seen, sequence));
91
8392 const start: ThreadDeletionReactorShape["start"] = Effect.fn("start")(function* () {
8493 yield* forkParked(
85 Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) => {
86 if (event.type !== "thread.deleted") {
87 return Effect.void;
88 }
89 return worker.enqueue(event);
90 }),
94 Stream.runForEach(
95 orchestrationEngine.streamDomainEvents.pipe(
96 // Events that landed before the subscription are not replayed, so
97 // start the watermark at the current head instead of zero.
98 Stream.onStart(orchestrationEngine.latestSequence.pipe(Effect.flatMap(noteSeen))),
99 ),
100 (event) =>
101 (event.type === "thread.deleted" ? worker.enqueue(event) : Effect.void).pipe(
102 Effect.andThen(noteSeen(event.sequence)),
103 ),
104 ),
105 );
106 });
107
108 const drainThrough: ThreadDeletionReactorShape["drainThrough"] = Effect.fn(
109 "ThreadDeletionReactor.drainThrough",
110 )(function* (target) {
111 yield* SubscriptionRef.changes(seenSequence).pipe(
112 Stream.filter((seen) => seen >= target),
113 Stream.runHead,
91114 );
115 yield* worker.drain;
92116 });
93117
94118 return {
95119 start,
96 drain: worker.drain,
120 drainThrough,
97121 } satisfies ThreadDeletionReactorShape;
98122});
99123
apps/server/src/orchestration/Services/ThreadDeletionReactor.ts+5-3
......@@ -23,10 +23,12 @@ export interface ThreadDeletionReactorShape {
2323 readonly start: () => Effect.Effect<void, never, Scope.Scope>;
2424
2525 /**
26 * Resolves when the internal processing queue is empty and idle.
27 * Intended for test use to replace timing-sensitive sleeps.
26 * Resolves once every thread.deleted at or before the supplied event
27 * sequence has been handed to the worker and the worker is empty and idle.
28 * A successful thread.create sequence is the fence callers use before the
29 * new incarnation can own runtime resources.
2830 */
29 readonly drain: Effect.Effect<void>;
31 readonly drainThrough: (sequence: number) => Effect.Effect<void>;
3032}
3133
3234/**
apps/server/src/orchestration/commandInvariants.test.ts+30
......@@ -199,4 +199,34 @@ describe("commandInvariants", () => {
199199 ),
200200 ).rejects.toThrow("already exists");
201201 });
202
203 it("lets a draft retry re-create a thread id after its first attempt was deleted", async () => {
204 const threadId = ThreadId.make("thread-1");
205 const firstAttempt = readModel.threads.find((thread) => thread.id === threadId)!;
206 const afterRollback: OrchestrationReadModel = {
207 ...readModel,
208 threads: readModel.threads.map((thread) =>
209 thread.id === threadId ? { ...thread, deletedAt: now, updatedAt: now } : thread,
210 ),
211 };
212 const retry: OrchestrationCommand = {
213 type: "thread.create",
214 commandId: CommandId.make("cmd-retry"),
215 threadId,
216 projectId: firstAttempt.projectId,
217 title: firstAttempt.title,
218 modelSelection: firstAttempt.modelSelection,
219 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
220 runtimeMode: "approval-required",
221 branch: null,
222 worktreePath: null,
223 createdAt: now,
224 };
225
226 await expect(
227 Effect.runPromise(
228 requireThreadAbsent({ readModel: afterRollback, command: retry, threadId }),
229 ),
230 ).resolves.toBeUndefined();
231 });
202232});
apps/server/src/orchestration/commandInvariants.ts+5-1
......@@ -156,7 +156,11 @@ export function requireThreadAbsent(input: {
156156 readonly command: OrchestrationCommand;
157157 readonly threadId: ThreadId;
158158}): Effect.Effect<void, OrchestrationCommandInvariantError> {
159 if (!findThreadById(input.readModel, input.threadId)) {
159 // Thread deletion is a soft delete and a draft keeps its client-minted id
160 // across retries, so only a live row blocks creation. Projectors reset the
161 // thread's rows when the id is created again.
162 const existing = findThreadById(input.readModel, input.threadId);
163 if (existing === undefined || existing.deletedAt !== null) {
160164 return Effect.void;
161165 }
162166 return Effect.fail(
apps/server/src/persistence/Layers/OrchestrationEventStore.ts+35
......@@ -15,6 +15,7 @@ import * as SqlClient from "effect/unstable/sql/SqlClient";
1515import * as SqlSchema from "effect/unstable/sql/SqlSchema";
1616import * as Effect from "effect/Effect";
1717import * as Layer from "effect/Layer";
18import * as Option from "effect/Option";
1819import * as Schema from "effect/Schema";
1920import * as Stream from "effect/Stream";
2021
......@@ -60,6 +61,13 @@ const OrchestrationEventPersistedRowSchema = Schema.Struct({
6061 metadata: EventMetadataFromJsonString,
6162});
6263
64const HasEventAfterRequestSchema = Schema.Struct({
65 aggregateKind: Schema.String,
66 aggregateId: Schema.String,
67 type: Schema.String,
68 sequenceExclusive: NonNegativeInt,
69});
70
6371const ReadFromSequenceRequestSchema = Schema.Struct({
6472 sequenceExclusive: NonNegativeInt,
6573 limit: Schema.Number,
......@@ -260,10 +268,37 @@ const makeEventStore = Effect.gen(function* () {
260268 return readPage(sequenceExclusive, normalizedLimit);
261269 };
262270
271 const findEventAfter = SqlSchema.findOneOption({
272 Request: HasEventAfterRequestSchema,
273 Result: Schema.Struct({ sequence: Schema.Number }),
274 execute: (request) =>
275 sql`
276 SELECT sequence
277 FROM orchestration_events
278 WHERE aggregate_kind = ${request.aggregateKind}
279 AND stream_id = ${request.aggregateId}
280 AND event_type = ${request.type}
281 AND sequence > ${request.sequenceExclusive}
282 LIMIT 1
283 `,
284 });
285
286 const hasEventAfter: OrchestrationEventStoreShape["hasEventAfter"] = (input) =>
287 findEventAfter(input).pipe(
288 Effect.map(Option.isSome),
289 Effect.mapError(
290 toPersistenceSqlOrDecodeError(
291 "OrchestrationEventStore.hasEventAfter:query",
292 "OrchestrationEventStore.hasEventAfter:decodeRow",
293 ),
294 ),
295 );
296
263297 return {
264298 append,
265299 readFromSequence,
266300 readAll: () => readFromSequence(0, Number.MAX_SAFE_INTEGER),
301 hasEventAfter,
267302 } satisfies OrchestrationEventStoreShape;
268303});
269304
apps/server/src/persistence/Layers/ProjectionPendingApprovals.ts+17
......@@ -95,6 +95,15 @@ const makeProjectionPendingApprovalRepository = Effect.gen(function* () {
9595 `,
9696 });
9797
98 const deleteProjectionPendingApprovalRowsByThread = SqlSchema.void({
99 Request: ListProjectionPendingApprovalsInput,
100 execute: ({ threadId }) =>
101 sql`
102 DELETE FROM projection_pending_approvals
103 WHERE thread_id = ${threadId}
104 `,
105 });
106
98107 const upsert: ProjectionPendingApprovalRepositoryShape["upsert"] = (row) =>
99108 upsertProjectionPendingApprovalRow(row).pipe(
100109 Effect.mapError(toPersistenceSqlError("ProjectionPendingApprovalRepository.upsert:query")),
......@@ -123,11 +132,19 @@ const makeProjectionPendingApprovalRepository = Effect.gen(function* () {
123132 ),
124133 );
125134
135 const deleteByThreadId: ProjectionPendingApprovalRepositoryShape["deleteByThreadId"] = (input) =>
136 deleteProjectionPendingApprovalRowsByThread(input).pipe(
137 Effect.mapError(
138 toPersistenceSqlError("ProjectionPendingApprovalRepository.deleteByThreadId:query"),
139 ),
140 );
141
126142 return {
127143 upsert,
128144 listByThreadId,
129145 getByRequestId,
130146 deleteByRequestId,
147 deleteByThreadId,
131148 } satisfies ProjectionPendingApprovalRepositoryShape;
132149});
133150
apps/server/src/persistence/Services/OrchestrationEventStore.ts+13
......@@ -52,6 +52,19 @@ export interface OrchestrationEventStoreShape {
5252 * @returns Stream containing all stored events.
5353 */
5454 readonly readAll: () => Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError>;
55
56 /**
57 * Check whether an aggregate has an event of the given type after a sequence.
58 *
59 * Used during replay to tell whether a later event supersedes the one being
60 * applied, without streaming the rest of the log.
61 */
62 readonly hasEventAfter: (input: {
63 readonly aggregateKind: OrchestrationEvent["aggregateKind"];
64 readonly aggregateId: string;
65 readonly type: OrchestrationEvent["type"];
66 readonly sequenceExclusive: number;
67 }) => Effect.Effect<boolean, OrchestrationEventStoreError>;
5568}
5669
5770/**
apps/server/src/persistence/Services/ProjectionPendingApprovals.ts+7
......@@ -82,6 +82,13 @@ export interface ProjectionPendingApprovalRepositoryShape {
8282 readonly deleteByRequestId: (
8383 input: DeleteProjectionPendingApprovalInput,
8484 ) => Effect.Effect<void, ProjectionRepositoryError>;
85
86 /**
87 * Delete every pending approval row for a thread.
88 */
89 readonly deleteByThreadId: (
90 input: ListProjectionPendingApprovalsInput,
91 ) => Effect.Effect<void, ProjectionRepositoryError>;
8592}
8693
8794/**
apps/server/src/server.test.ts+121-7
......@@ -116,6 +116,7 @@ import * as RemoteOpenTargets from "./environment/RemoteOpenTargets.ts";
116116import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts";
117117import { OrchestrationListenerCallbackError } from "./orchestration/Errors.ts";
118118import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts";
119import { ThreadDeletionReactor } from "./orchestration/Services/ThreadDeletionReactor.ts";
119120import { SqlitePersistenceMemory } from "./persistence/Layers/Sqlite.ts";
120121import { PersistenceSqlError } from "./persistence/Errors.ts";
121122import * as ProviderRegistry from "./provider/Services/ProviderRegistry.ts";
......@@ -411,6 +412,7 @@ const buildAppUnderTest = (options?: {
411412 >;
412413 terminalManager?: Partial<TerminalManager.TerminalManager["Service"]>;
413414 orchestrationEngine?: Partial<OrchestrationEngine.OrchestrationEngineService["Service"]>;
415 threadDeletionReactor?: Partial<ThreadDeletionReactor["Service"]>;
414416 analyticsService?: Partial<AnalyticsService.AnalyticsService["Service"]>;
415417 projectionSnapshotQuery?: Partial<ProjectionSnapshotQuery.ProjectionSnapshotQuery["Service"]>;
416418 checkpointDiffQuery?: Partial<CheckpointDiffQuery.CheckpointDiffQuery["Service"]>;
......@@ -787,13 +789,20 @@ const buildAppUnderTest = (options?: {
787789 ),
788790 ),
789791 Layer.provide(
790 Layer.mock(OrchestrationEngine.OrchestrationEngineService)({
791 readEvents: () => Stream.empty,
792 dispatch: () => Effect.succeed({ sequence: 0 }),
793 streamDomainEvents: Stream.empty,
794 latestSequence: Effect.succeed(0),
795 ...options?.layers?.orchestrationEngine,
796 }),
792 Layer.mergeAll(
793 Layer.mock(OrchestrationEngine.OrchestrationEngineService)({
794 readEvents: () => Stream.empty,
795 dispatch: () => Effect.succeed({ sequence: 0 }),
796 streamDomainEvents: Stream.empty,
797 latestSequence: Effect.succeed(0),
798 ...options?.layers?.orchestrationEngine,
799 }),
800 Layer.mock(ThreadDeletionReactor)({
801 start: () => Effect.void,
802 drainThrough: () => Effect.void,
803 ...options?.layers?.threadDeletionReactor,
804 }),
805 ),
797806 ),
798807 Layer.provide(
799808 Layer.mock(ProjectionSnapshotQuery.ProjectionSnapshotQuery)({
......@@ -8476,6 +8485,111 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
84768485 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
84778486 );
84788487
8488 it.effect("drains deletion cleanup through the re-created thread event", () =>
8489 Effect.gen(function* () {
8490 // A draft retry reuses the thread id its failed bootstrap deleted. The
8491 // deletion reactor stops sessions and closes terminals by that id, so
8492 // both thread.create paths use the created event as a fence, then drain
8493 // cleanup before handing the new incarnation to resource-owning work.
8494 const trace: Array<string> = [];
8495 const drainRequested = yield* Deferred.make<void>();
8496 const cleanupDone = yield* Deferred.make<void>();
8497 yield* buildAppUnderTest({
8498 layers: {
8499 threadDeletionReactor: {
8500 drainThrough: (sequence) =>
8501 Effect.gen(function* () {
8502 trace.push(`drain:${sequence}`);
8503 yield* Deferred.succeed(drainRequested, undefined);
8504 yield* Deferred.await(cleanupDone);
8505 }),
8506 },
8507 orchestrationEngine: {
8508 dispatch: (command) =>
8509 Effect.sync(() => {
8510 trace.push(command.type);
8511 return { sequence: trace.length };
8512 }),
8513 readEvents: () => Stream.empty,
8514 },
8515 },
8516 });
8517
8518 const createdAt = "2026-01-01T00:00:00.000Z";
8519 const threadId = ThreadId.make("thread-retry-after-delete");
8520 const wsUrl = yield* getWsServerUrl("/ws");
8521
8522 yield* Effect.scoped(
8523 withWsRpcClient(wsUrl, (client) =>
8524 Effect.gen(function* () {
8525 const directCreate = yield* Effect.forkChild(
8526 client[ORCHESTRATION_WS_METHODS.dispatchCommand]({
8527 type: "thread.create",
8528 commandId: CommandId.make("cmd-retry-create"),
8529 threadId,
8530 projectId: defaultProjectId,
8531 title: "Retry",
8532 modelSelection: defaultModelSelection,
8533 runtimeMode: "full-access",
8534 interactionMode: "default",
8535 branch: null,
8536 worktreePath: null,
8537 createdAt,
8538 }),
8539 );
8540 yield* Deferred.await(drainRequested);
8541 assert.deepEqual(trace, ["thread.create", "drain:1"]);
8542 yield* Deferred.succeed(cleanupDone, undefined);
8543 yield* Fiber.join(directCreate);
8544 }),
8545 ),
8546 );
8547 assert.deepEqual(trace, ["thread.create", "drain:1"]);
8548
8549 // Cleanup is already released; the bootstrap path must still drain
8550 // between creating the thread and starting its turn.
8551 trace.length = 0;
8552 yield* Effect.scoped(
8553 withWsRpcClient(wsUrl, (client) =>
8554 Effect.gen(function* () {
8555 const bootstrapCreate = yield* Effect.forkChild(
8556 client[ORCHESTRATION_WS_METHODS.dispatchCommand]({
8557 type: "thread.turn.start",
8558 commandId: CommandId.make("cmd-retry-bootstrap"),
8559 threadId,
8560 message: {
8561 messageId: MessageId.make("msg-retry-bootstrap"),
8562 role: "user",
8563 text: "hello",
8564 attachments: [],
8565 },
8566 modelSelection: defaultModelSelection,
8567 runtimeMode: "full-access",
8568 interactionMode: "default",
8569 bootstrap: {
8570 createThread: {
8571 projectId: defaultProjectId,
8572 title: "Retry",
8573 modelSelection: defaultModelSelection,
8574 runtimeMode: "full-access",
8575 interactionMode: "default",
8576 branch: null,
8577 worktreePath: null,
8578 createdAt,
8579 },
8580 runSetupScript: false,
8581 },
8582 createdAt,
8583 }),
8584 );
8585 yield* Fiber.join(bootstrapCreate);
8586 }),
8587 ),
8588 );
8589 assert.deepEqual(trace, ["thread.create", "drain:1", "thread.turn.start"]);
8590 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
8591 );
8592
84798593 it.effect("does not report a deleted bootstrap thread when cleanup fails", () =>
84808594 Effect.gen(function* () {
84818595 const dispatchedCommands: Array<OrchestrationCommand> = [];
apps/server/src/ws.ts+16-1
......@@ -85,6 +85,7 @@ import {
8585} from "./orchestration/Normalizer.ts";
8686import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts";
8787import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts";
88import { ThreadDeletionReactor } from "./orchestration/Services/ThreadDeletionReactor.ts";
8889import {
8990 observeRpcEffect as instrumentRpcEffect,
9091 observeRpcStream as instrumentRpcStream,
......@@ -460,6 +461,7 @@ const makeWsRpcLayer = (
460461 const crypto = yield* Crypto.Crypto;
461462 const projectionSnapshotQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;
462463 const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService;
464 const threadDeletionReactor = yield* ThreadDeletionReactor;
463465 const analytics = yield* AnalyticsService.AnalyticsService;
464466 // Every command dispatched on this connection carries the connecting
465467 // client's origin, including server-generated bootstrap sub-commands:
......@@ -1026,7 +1028,7 @@ const makeWsRpcLayer = (
10261028
10271029 const bootstrapProgram = Effect.gen(function* () {
10281030 if (bootstrap?.createThread) {
1029 yield* dispatchFromClient({
1031 const created = yield* dispatchFromClient({
10301032 type: "thread.create",
10311033 commandId: yield* serverCommandId("bootstrap-thread-create"),
10321034 threadId: command.threadId,
......@@ -1039,6 +1041,11 @@ const makeWsRpcLayer = (
10391041 worktreePath: bootstrap.createThread.worktreePath,
10401042 createdAt: bootstrap.createThread.createdAt,
10411043 });
1044 // The successful create is a fence in the engine command queue:
1045 // every delete for the prior incarnation committed before it.
1046 // Drain through that event before setup or turn start can own
1047 // terminals and provider sessions under the reused thread id.
1048 yield* threadDeletionReactor.drainThrough(created.sequence);
10421049 createdThread = true;
10431050 }
10441051
......@@ -1126,6 +1133,14 @@ const makeWsRpcLayer = (
11261133 normalizedCommand.type === "thread.turn.start" && normalizedCommand.bootstrap
11271134 ? dispatchBootstrapTurnStart(normalizedCommand)
11281135 : dispatchFromClient(normalizedCommand).pipe(
1136 Effect.tap(({ sequence }) =>
1137 // Returning from thread.create is the handoff point at which
1138 // clients may start resources for the new incarnation. Use
1139 // its event sequence as the exact deletion-cleanup fence.
1140 normalizedCommand.type === "thread.create"
1141 ? threadDeletionReactor.drainThrough(sequence)
1142 : Effect.void,
1143 ),
11291144 Effect.mapError((cause) =>
11301145 toDispatchCommandError(cause, "Failed to dispatch orchestration command"),
11311146 ),