From 414b2aa9f9bcfe10943d54313867e1b9a19072f4 Mon Sep 17 00:00:00 2001 From: Adam Firestone Date: Wed, 12 Aug 2026 17:52:34 -0500 Subject: [PATCH 1/3] fix(server): replay the full event backlog when rebuilding projections Migration 045 wipes all projection tables and resets the replay cursor to 0, but bootstrap read the backlog with readFromSequence's default 1,000-event limit. On a 27k-event store the rebuild silently stopped after the oldest 1,000 events, live events then stamped the cursor to the head, and nearly every thread and project vanished from the UI with no error. Pass Number.MAX_SAFE_INTEGER so bootstrap replays everything past the cursor, matching readAll's behavior. Fable 5 via Claude Code --- apps/server/src/orchestration/Layers/ProjectionPipeline.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index b28f7c9e46d1..8d72fc915d7c 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -1858,7 +1858,11 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti // Match live event-major ordering so a revert sees turn/session evidence // before later projectors apply that same event's trims. - yield* Stream.runForEach(eventStore.readFromSequence(firstSequence), (event) => + // The full backlog, not the default 1k page: a projection rebuild (migration + // 045) resets the cursor to 0 and needs every event replayed. + yield* Stream.runForEach( + eventStore.readFromSequence(firstSequence, Number.MAX_SAFE_INTEGER), + (event) => Effect.forEach( projectors, (projector) => From 2008290113a46ea764aec27fdef710758e1afa51 Mon Sep 17 00:00:00 2001 From: Adam Firestone Date: Wed, 12 Aug 2026 19:28:52 -0500 Subject: [PATCH 2/3] fix(server): retain checkpointless turns across replayed reverts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The turns projector's thread.reverted handler kept only turns with checkpoint evidence at or below the revert baseline. Turns recorded while checkpointing was unavailable have no checkpoint_turn_count, so the first revert on the thread deleted them — and since thread detail paginates by turns, their retained messages became unreachable, indistinguishable from data loss. Use the same collectRevertedTurnIds denylist as the message, activity, and plan projectors so a revert only ever removes the turns it actually reverted. Fable 5 via Claude Code --- .../Layers/ProjectionPipeline.test.ts | 194 ++++++++++++++++++ .../Layers/ProjectionPipeline.ts | 14 +- 2 files changed, 204 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index ad929a6f55d1..543d469458ed 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -2911,6 +2911,200 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { }), ); + it.effect("retains checkpointless turns and their messages across a revert", () => + Effect.gen(function* () { + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread-revert-turns"); + const appendAndProject = (event: Parameters[0]) => + eventStore + .append(event) + .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent))); + const sessionSet = ( + suffix: string, + occurredAt: string, + status: "running" | "ready", + activeTurnId: string | null, + ) => + appendAndProject({ + type: "thread.session-set", + eventId: EventId.make(`evt-rt-${suffix}`), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt, + commandId: CommandId.make(`cmd-rt-${suffix}`), + causationEventId: null, + correlationId: CorrelationId.make(`cmd-rt-${suffix}`), + metadata: {}, + payload: { + threadId, + session: { + threadId, + status, + providerName: "claude", + runtimeMode: "full-access", + activeTurnId: activeTurnId === null ? null : TurnId.make(activeTurnId), + lastError: null, + updatedAt: occurredAt, + }, + }, + }); + const messageSent = (suffix: string, occurredAt: string, messageId: string, turnId: string) => + appendAndProject({ + type: "thread.message-sent", + eventId: EventId.make(`evt-rt-${suffix}`), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt, + commandId: CommandId.make(`cmd-rt-${suffix}`), + causationEventId: null, + correlationId: CorrelationId.make(`cmd-rt-${suffix}`), + metadata: {}, + payload: { + threadId, + messageId: MessageId.make(messageId), + role: "assistant" as const, + text: messageId, + turnId: TurnId.make(turnId), + streaming: false, + createdAt: occurredAt, + updatedAt: occurredAt, + }, + }); + const diffCompleted = ( + suffix: string, + occurredAt: string, + turnId: string, + checkpointTurnCount: number, + assistantMessageId: string, + ) => + appendAndProject({ + type: "thread.turn-diff-completed", + eventId: EventId.make(`evt-rt-${suffix}`), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt, + commandId: CommandId.make(`cmd-rt-${suffix}`), + causationEventId: null, + correlationId: CorrelationId.make(`cmd-rt-${suffix}`), + metadata: {}, + payload: { + threadId, + turnId: TurnId.make(turnId), + checkpointTurnCount, + checkpointRef: CheckpointRef.make( + `refs/t3/checkpoints/thread-revert-turns/turn/${checkpointTurnCount}`, + ), + status: "ready" as const, + files: [], + assistantMessageId: MessageId.make(assistantMessageId), + completedAt: occurredAt, + }, + }); + + yield* appendAndProject({ + type: "project.created", + eventId: EventId.make("evt-rt-1"), + aggregateKind: "project", + aggregateId: ProjectId.make("project-revert-turns"), + occurredAt: "2026-02-26T13:00:00.000Z", + commandId: CommandId.make("cmd-rt-1"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-rt-1"), + metadata: {}, + payload: { + projectId: ProjectId.make("project-revert-turns"), + title: "Project Revert Turns", + workspaceRoot: "/tmp/project-revert-turns", + defaultModelSelection: null, + scripts: [], + createdAt: "2026-02-26T13:00:00.000Z", + updatedAt: "2026-02-26T13:00:00.000Z", + }, + }); + yield* appendAndProject({ + type: "thread.created", + eventId: EventId.make("evt-rt-2"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-02-26T13:00:01.000Z", + commandId: CommandId.make("cmd-rt-2"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-rt-2"), + metadata: {}, + payload: { + threadId, + projectId: ProjectId.make("project-revert-turns"), + title: "Thread Revert Turns", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + runtimeMode: "full-access", + branch: null, + worktreePath: null, + createdAt: "2026-02-26T13:00:01.000Z", + updatedAt: "2026-02-26T13:00:01.000Z", + }, + }); + + // A turn recorded while checkpointing was unavailable: it settles + // without ever getting a checkpoint_turn_count. + yield* sessionSet("3", "2026-02-26T13:00:02.000Z", "running", "turn-early"); + yield* messageSent("4", "2026-02-26T13:00:03.000Z", "assistant-early", "turn-early"); + yield* sessionSet("5", "2026-02-26T13:00:04.000Z", "ready", null); + + // A checkpointed turn at the revert baseline, then one past it. + yield* sessionSet("6", "2026-02-26T13:00:05.000Z", "running", "turn-mid"); + yield* diffCompleted("7", "2026-02-26T13:00:06.000Z", "turn-mid", 1, "assistant-mid"); + yield* messageSent("8", "2026-02-26T13:00:06.500Z", "assistant-mid", "turn-mid"); + yield* sessionSet("9", "2026-02-26T13:00:07.000Z", "ready", null); + yield* sessionSet("10", "2026-02-26T13:00:08.000Z", "running", "turn-late"); + yield* diffCompleted("11", "2026-02-26T13:00:09.000Z", "turn-late", 2, "assistant-late"); + yield* messageSent("12", "2026-02-26T13:00:09.500Z", "assistant-late", "turn-late"); + yield* sessionSet("13", "2026-02-26T13:00:10.000Z", "ready", null); + + yield* appendAndProject({ + type: "thread.reverted", + eventId: EventId.make("evt-rt-14"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-02-26T13:00:11.000Z", + commandId: CommandId.make("cmd-rt-14"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-rt-14"), + metadata: {}, + payload: { + threadId, + turnCount: 1, + }, + }); + + // The post-baseline checkpointed turn is reverted; the checkpointless + // turn survives so its message stays reachable to turn-anchored + // pagination. + const turnRows = yield* sql<{ readonly turnId: string | null }>` + SELECT turn_id AS "turnId" + FROM projection_turns + WHERE thread_id = ${threadId} + ORDER BY turn_id ASC + `; + assert.deepEqual(turnRows, [{ turnId: "turn-early" }, { turnId: "turn-mid" }]); + + const messageRowsAfterRevert = yield* sql<{ readonly messageId: string }>` + SELECT message_id AS "messageId" + FROM projection_thread_messages + WHERE thread_id = ${threadId} + ORDER BY message_id ASC + `; + assert.deepEqual(messageRowsAfterRevert, [ + { messageId: "assistant-early" }, + { messageId: "assistant-mid" }, + ]); + }), + ); + it.effect("excludes a completed retraction message from SQLite projection", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 8d72fc915d7c..54057048d9a1 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -1466,11 +1466,17 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti const existingTurns = yield* projectionTurnRepository.listByThreadId({ threadId: event.payload.threadId, }); + // Same denylist as the message/activity/plan projectors: turns + // without checkpoint evidence (e.g. recorded while checkpointing + // was unavailable) must survive a revert, or their retained + // messages become unreachable to turn-anchored pagination. + const revertedTurnIds = yield* getRevertedTurnIds({ + threadId: event.payload.threadId, + baselineTurnCount: event.payload.turnCount, + retractionTurnId: event.payload.retraction?.turnId ?? null, + }); const keptTurns = existingTurns.filter( - (turn) => - turn.turnId !== null && - turn.checkpointTurnCount !== null && - turn.checkpointTurnCount <= event.payload.turnCount, + (turn) => turn.turnId !== null && !revertedTurnIds.has(turn.turnId), ); yield* projectionTurnRepository.deleteByThreadId({ threadId: event.payload.threadId, From a211e61b011b011ce768aa9a3e53c85cf1fed232 Mon Sep 17 00:00:00 2001 From: Adam Firestone Date: Wed, 12 Aug 2026 19:28:52 -0500 Subject: [PATCH 3/3] feat(server,web): rebuild projections with retained turns Migration 046 re-runs 045's wipe-and-replay so the event history rebuilds through the fixed turns projector, restoring turn rows the first rebuild dropped. Bump the thread snapshot cache to v7 so warm browser caches cold-load the rebuilt projections instead of resuming past them. Fable 5 via Claude Code --- apps/server/src/persistence/Migrations.ts | 2 ++ .../Migrations/045_RebuildProjectionsFromEvents.test.ts | 4 +++- .../046_RebuildProjectionsWithRetainedTurns.ts | 9 +++++++++ apps/web/src/connection/storage.ts | 6 ++++-- 4 files changed, 18 insertions(+), 3 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/046_RebuildProjectionsWithRetainedTurns.ts diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index a14e914baf97..e57c82fdd5fe 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -58,6 +58,7 @@ import Migration0042 from "./Migrations/042_ProjectionTurnDispatchOwnership.ts"; import Migration0043 from "./Migrations/043_ProjectionManagedWorktrees.ts"; import Migration0044 from "./Migrations/044_CleanupCompletedRetractionMessages.ts"; import Migration0045 from "./Migrations/045_RebuildProjectionsFromEvents.ts"; +import Migration0046 from "./Migrations/046_RebuildProjectionsWithRetainedTurns.ts"; /** * Migration loader with all migrations defined inline. @@ -115,6 +116,7 @@ export const migrationEntries = [ [43, "ProjectionManagedWorktrees", Migration0043], [44, "CleanupCompletedRetractionMessages", Migration0044], [45, "RebuildProjectionsFromEvents", Migration0045], + [46, "RebuildProjectionsWithRetainedTurns", Migration0046], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/045_RebuildProjectionsFromEvents.test.ts b/apps/server/src/persistence/Migrations/045_RebuildProjectionsFromEvents.test.ts index 7eb840090aa3..67c32e8c9a63 100644 --- a/apps/server/src/persistence/Migrations/045_RebuildProjectionsFromEvents.test.ts +++ b/apps/server/src/persistence/Migrations/045_RebuildProjectionsFromEvents.test.ts @@ -54,7 +54,9 @@ layer("045_RebuildProjectionsFromEvents", (it) => { yield* sql`DROP TABLE projection_thread_proposed_plans`; yield* rebuildProjectionsFromEvents; - assert.deepEqual(migrationManifest.at(-1), [45, "RebuildProjectionsFromEvents"]); + // The manifest must end on a full rebuild so boot replays every event + // through the current projectors after the wipe. + assert.deepEqual(migrationManifest.at(-1), [46, "RebuildProjectionsWithRetainedTurns"]); }), ); }); diff --git a/apps/server/src/persistence/Migrations/046_RebuildProjectionsWithRetainedTurns.ts b/apps/server/src/persistence/Migrations/046_RebuildProjectionsWithRetainedTurns.ts new file mode 100644 index 000000000000..b0d2c101582e --- /dev/null +++ b/apps/server/src/persistence/Migrations/046_RebuildProjectionsWithRetainedTurns.ts @@ -0,0 +1,9 @@ +import Migration0045 from "./045_RebuildProjectionsFromEvents.ts"; + +// Re-run migration 045's wipe-and-replay. The first rebuild ran through a +// turns projector whose thread.reverted handler kept only checkpointed turns, +// deleting checkpointless turns and stranding their messages outside +// turn-anchored pagination. The projector now shares the revert denylist with +// the other projectors, so replaying the event history again rebuilds those +// turn rows. +export default Migration0045; diff --git a/apps/web/src/connection/storage.ts b/apps/web/src/connection/storage.ts index 766d56f14dfe..474e4fc5c2ef 100644 --- a/apps/web/src/connection/storage.ts +++ b/apps/web/src/connection/storage.ts @@ -63,8 +63,10 @@ const StoredShellSnapshotJson = Schema.fromJsonString(StoredShellSnapshot); // a warm cache resuming via `afterSequence` would render the ghost messages // forever. Any server-side row surgery needs a bump here to reach clients. // v6 pairs with server migration 045's full projection rebuild. +// v7 pairs with server migration 046, which rebuilds again now that the +// turns projector retains checkpointless turns across replayed reverts. const StoredThreadSnapshot = Schema.Struct({ - schemaVersion: Schema.Literal(6), + schemaVersion: Schema.Literal(7), environmentId: EnvironmentId, threadId: ThreadId, snapshot: OrchestrationThreadDetailSnapshot, @@ -572,7 +574,7 @@ export const connectionStorageLayer = Layer.effectContext( saveThread: (environmentId, snapshot) => Effect.gen(function* () { const encoded = yield* encodeStoredThreadSnapshot({ - schemaVersion: 6, + schemaVersion: 7, environmentId, threadId: snapshot.thread.id, snapshot,