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 b28f7c9e46d1..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, @@ -1858,7 +1864,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) => 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,