fix(server): keep fork history stable across repeated source rewinds

apps/server/src/orchestration-v2/ContextHandoffService.test.ts

@@ -293,6 +293,7 @@ it.effect("inherited history stops at a missing fork ancestor", () =>
     };
     const items = yield* ContextHandoffService.readInheritedHandoffItems(
       {
+        getForkSourceRunIds: () => Effect.succeed([]),
         getThreadRecords: (missing) =>
           Effect.fail(new ProjectionStoreThreadNotFoundError({ threadId: missing })),
       },

apps/server/src/orchestration-v2/ContextHandoffService.ts

@@ -148,25 +148,42 @@ type ForkHistorySource = Pick<OrchestrationV2ThreadProjection, "thread" | "runs"
 /** Visible rows of `source`'s own runs through `throughRun`, plus its v1 imported rows. */
 const readOwnHistoryItems = Effect.fn("orchestrationV2.contextHandoff.readOwnHistoryItems")(
   function* (
-    projectionStore: Pick<ProjectionStoreV2Shape, "getThreadRecords">,
+    projectionStore: Pick<ProjectionStoreV2Shape, "getThreadRecords" | "getForkSourceRunIds">,
     source: ForkHistorySource,
     throughRun: Pick<OrchestrationV2Run, "ordinal">,
+    forkThreadId: ThreadId,
   ) {
+    const inheritedRunIds = new Set(yield* projectionStore.getForkSourceRunIds(forkThreadId));
     const { turnItems } = yield* projectionStore.getThreadRecords(source.thread.id, ["turnItems"], {
       turnItemRunIds: [
-        ...source.runs.filter((run) => run.ordinal <= throughRun.ordinal).map((run) => run.id),
+        ...source.runs
+          .filter((run) => run.ordinal <= throughRun.ordinal && inheritedRunIds.has(run.id))
+          .map((run) => run.id),
         ...(source.thread.historyOrigin === "v1_import" ? [null] : []),
       ],
     });
-    return visibleConversationItems(source, turnItems);
+    return visibleConversationItems(
+      {
+        ...source,
+        runs: source.runs.map((run) =>
+          run.status === "rolled_back" && inheritedRunIds.has(run.id)
+            ? {
+                ...run,
+                status: run.startedAt === null ? ("cancelled" as const) : ("completed" as const),
+              }
+            : run,
+        ),
+      },
+      turnItems,
+    );
   },
 );
 
 /** History `thread` shows from its fork ancestors, oldest first. Side chats inherit no rows. */
 export const readInheritedHandoffItems = Effect.fn(
   "orchestrationV2.contextHandoff.readInheritedHandoffItems",
 )(function* (
-  projectionStore: Pick<ProjectionStoreV2Shape, "getThreadRecords">,
+  projectionStore: Pick<ProjectionStoreV2Shape, "getThreadRecords" | "getForkSourceRunIds">,
   thread: OrchestrationV2AppThread,
 ) {
   const segments: Array<ReadonlyArray<OrchestrationV2TurnItem>> = [];
@@ -184,7 +201,7 @@ export const readInheritedHandoffItems = Effect.fn(
       .pipe(Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) }));
     const forkRun = source?.runs.find((run) => run.id === runId);
     if (source === null || forkRun === undefined) break;
-    segments.unshift(yield* readOwnHistoryItems(projectionStore, source, forkRun));
+    segments.unshift(yield* readOwnHistoryItems(projectionStore, source, forkRun, current.id));
     current = source.thread;
   }
   return segments.flat();
@@ -194,13 +211,14 @@ export const readInheritedHandoffItems = Effect.fn(
 export const readForkHistoryItems = Effect.fn(
   "orchestrationV2.contextHandoff.readForkHistoryItems",
 )(function* (
-  projectionStore: Pick<ProjectionStoreV2Shape, "getThreadRecords">,
+  projectionStore: Pick<ProjectionStoreV2Shape, "getThreadRecords" | "getForkSourceRunIds">,
   source: ForkHistorySource,
   throughRun: Pick<OrchestrationV2Run, "ordinal">,
+  forkThreadId: ThreadId,
 ) {
   return [
     ...(yield* readInheritedHandoffItems(projectionStore, source.thread)),
-    ...(yield* readOwnHistoryItems(projectionStore, source, throughRun)),
+    ...(yield* readOwnHistoryItems(projectionStore, source, throughRun, forkThreadId)),
   ];
 });
 

apps/server/src/orchestration-v2/Orchestrator.ts

@@ -7358,7 +7358,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
     const portableForkItems =
       !requiresPortableFork || sourceProjection === null || sourceRun === null
         ? []
-        : yield* readForkHistoryItems(projectionStore, sourceProjection, sourceRun).pipe(
+        : yield* readForkHistoryItems(
+            projectionStore,
+            sourceProjection,
+            sourceRun,
+            projection.thread.id,
+          ).pipe(
             Effect.mapError(
               (cause) =>
                 new OrchestratorProjectionError({ threadId: sourceProjection.thread.id, cause }),

apps/server/src/orchestration-v2/ProjectionStore.test.ts

@@ -279,6 +279,14 @@ it.effect("routes full projection reads to readers and retains uncommitted write
     const sql = yield* SqlClient.SqlClient;
     const parentId = yield* addRolledBackRecoveryCandidate("reader-routing");
     const parent = yield* store.getThreadProjection(parentId);
+    yield* store.apply({
+      id: EventId.make("event:reader-routing:restored"),
+      type: "run.updated",
+      threadId: parentId,
+      runId: parent.runs[0]!.id,
+      occurredAt: parent.thread.createdAt,
+      payload: { ...parent.runs[0]!, status: "completed" },
+    });
     const threadId = ThreadId.make("thread:reader-routing:fork");
     yield* store.apply({
       id: EventId.make("event:reader-routing:fork"),
@@ -291,6 +299,14 @@ it.effect("routes full projection reads to readers and retains uncommitted write
         forkedFrom: { type: "run", threadId: parentId, runId: parent.runs[0]!.id },
       },
     });
+    yield* store.apply({
+      id: EventId.make("event:reader-routing:rewound-after-fork"),
+      type: "run.updated",
+      threadId: parentId,
+      runId: parent.runs[0]!.id,
+      occurredAt: parent.thread.createdAt,
+      payload: parent.runs[0]!,
+    });
     const messageId = MessageId.make("message:reader-routing");
     yield* store.apply({
       id: EventId.make("event:reader-routing:message"),
@@ -7149,6 +7165,33 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
           deletedAt: null,
         },
       });
+      yield* projectionStore.apply({
+        id: EventId.make("event:projection-imported-fork-shell:source-run"),
+        type: "run.created",
+        threadId: sourceThreadId,
+        runId: sourceRunId,
+        nodeId: rootNodeId,
+        driver,
+        occurredAt: now,
+        payload: {
+          id: sourceRunId,
+          threadId: sourceThreadId,
+          ordinal: 1,
+          providerInstanceId,
+          modelSelection,
+          providerThreadId: null,
+          userMessageId: MessageId.make("message:projection-imported-fork-shell:run"),
+          rootNodeId,
+          activeAttemptId: null,
+          status: "completed",
+          requestedAt: now,
+          startedAt: now,
+          completedAt: now,
+          checkpointId: null,
+          contextHandoffId: null,
+        },
+      });
+
       for (const [threadId, sideChat] of [
         [targetThreadId, false],
         [sideChatThreadId, true],
@@ -7192,32 +7235,6 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
           },
         });
       }
-      yield* projectionStore.apply({
-        id: EventId.make("event:projection-imported-fork-shell:source-run"),
-        type: "run.created",
-        threadId: sourceThreadId,
-        runId: sourceRunId,
-        nodeId: rootNodeId,
-        driver,
-        occurredAt: now,
-        payload: {
-          id: sourceRunId,
-          threadId: sourceThreadId,
-          ordinal: 1,
-          providerInstanceId,
-          modelSelection,
-          providerThreadId: null,
-          userMessageId: MessageId.make("message:projection-imported-fork-shell:run"),
-          rootNodeId,
-          activeAttemptId: null,
-          status: "completed",
-          requestedAt: now,
-          startedAt: now,
-          completedAt: now,
-          checkpointId: null,
-          contextHandoffId: null,
-        },
-      });
 
       const applyAssistantItem = (suffix: string, runId: RunId | null, ordinal: number) =>
         projectionStore.apply({
@@ -7737,43 +7754,6 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
           deletedAt: null,
         },
       });
-      yield* projectionStore.apply({
-        id: EventId.make("event:projection-fork-source-rollback:target-thread"),
-        type: "thread.created",
-        threadId: targetThreadId,
-        occurredAt: now,
-        payload: {
-          createdBy: "user",
-          creationSource: "web",
-          id: targetThreadId,
-          projectId,
-          title: "Projection fork source rollback target",
-          providerInstanceId,
-          modelSelection,
-          runtimeMode: "full-access",
-          interactionMode: "default",
-          branch: null,
-          worktreePath: null,
-          activeProviderThreadId: targetProviderThreadId,
-          lineage: {
-            parentThreadId: sourceThreadId,
-            relationshipToParent: "fork",
-            rootThreadId: sourceThreadId,
-          },
-          forkedFrom: {
-            type: "run",
-            threadId: sourceThreadId,
-            runId: sourceRun2Id,
-          },
-          createdAt: now,
-          updatedAt: now,
-          archivedAt: null,
-          settledOverride: null,
-          settledAt: null,
-          lastVisitedAt: null,
-          deletedAt: null,
-        },
-      });
 
       for (const [ordinal, runId, nodeId, promptText, responseText] of [
         [1, sourceRun1Id, sourceRun1NodeId, "source one", "one"],
@@ -7872,6 +7852,44 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
         });
       }
 
+      yield* projectionStore.apply({
+        id: EventId.make("event:projection-fork-source-rollback:target-thread"),
+        type: "thread.created",
+        threadId: targetThreadId,
+        occurredAt: now,
+        payload: {
+          createdBy: "user",
+          creationSource: "web",
+          id: targetThreadId,
+          projectId,
+          title: "Projection fork source rollback target",
+          providerInstanceId,
+          modelSelection,
+          runtimeMode: "full-access",
+          interactionMode: "default",
+          branch: null,
+          worktreePath: null,
+          activeProviderThreadId: targetProviderThreadId,
+          lineage: {
+            parentThreadId: sourceThreadId,
+            relationshipToParent: "fork",
+            rootThreadId: sourceThreadId,
+          },
+          forkedFrom: {
+            type: "run",
+            threadId: sourceThreadId,
+            runId: sourceRun2Id,
+          },
+          createdAt: now,
+          updatedAt: now,
+          archivedAt: null,
+          settledOverride: null,
+          settledAt: null,
+          lastVisitedAt: null,
+          deletedAt: null,
+        },
+      });
+
       const targetBeforeRollback = yield* projectionStore.getThreadProjection(targetThreadId);
       assert.deepEqual(
         targetBeforeRollback.visibleTurnItems.map((row) => row.item.type),
@@ -8064,6 +8082,78 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
         ["source one", "one", "fork"],
       );
 
+      const laterForkId = ThreadId.make("thread:projection-fork-source-rollback:later-fork");
+      const sourceThread = (yield* projectionStore.getThreadRecords(sourceThreadId, [])).thread;
+      yield* projectionStore.apply({
+        id: EventId.make("event:projection-fork-source-rollback:later-fork"),
+        type: "thread.created",
+        threadId: laterForkId,
+        occurredAt: now,
+        payload: {
+          ...sourceThread,
+          id: laterForkId,
+          activeProviderThreadId: null,
+          lineage: {
+            parentThreadId: sourceThreadId,
+            relationshipToParent: "fork",
+            rootThreadId: sourceThreadId,
+          },
+          forkedFrom: { type: "run", threadId: sourceThreadId, runId: sourceRun3Id },
+          createdAt: now,
+          updatedAt: now,
+        },
+      });
+      for (const rewindAgain of [false, true]) {
+        if (rewindAgain) {
+          const run = (yield* projectionStore.getThreadRecords(sourceThreadId, ["runs"])).runs.find(
+            (candidate) => candidate.id === sourceRun3Id,
+          )!;
+          yield* projectionStore.apply({
+            id: EventId.make("event:projection-fork-source-rollback:rewind-again"),
+            type: "run.updated",
+            threadId: sourceThreadId,
+            runId: run.id,
+            occurredAt: now,
+            payload: { ...run, status: "rolled_back", completedAt: now },
+          });
+          // A replayed thread.created must keep the membership captured at fork time.
+          const fork = (yield* projectionStore.getThreadRecords(laterForkId, [])).thread;
+          yield* projectionStore.apply({
+            id: EventId.make("event:projection-fork-source-rollback:later-fork"),
+            type: "thread.created",
+            threadId: laterForkId,
+            occurredAt: now,
+            payload: fork,
+          });
+        }
+        const projection = yield* projectionStore.getThreadProjection(laterForkId);
+        assert.deepEqual(
+          projection.visibleTurnItems.flatMap((row) =>
+            row.item.type === "user_message" ? [row.item.text] : [],
+          ),
+          ["source one"],
+          rewindAgain ? "after second rewind" : "before second rewind",
+        );
+        assert.lengthOf(projection.visibleTurnItems, 303);
+        assert.deepEqual(
+          (yield* projectionStore.getTimelinePage(laterForkId, { view: "activity", limit: 500 }))
+            .items,
+          projection.visibleTurnItems,
+        );
+        assert.deepEqual(
+          (yield* projectionStore.getThreadSnapshotWindow(laterForkId, { rowLimit: 500 }))
+            .projection.visibleTurnItems,
+          projection.visibleTurnItems,
+        );
+        assert.equal((yield* projectionStore.getThreadShell(laterForkId))?.visibleItemCount, 303);
+        assert.equal(
+          (yield* projectionStore.getShellSnapshot()).threads.find(
+            (thread) => thread.id === laterForkId,
+          )?.visibleItemCount,
+          303,
+        );
+      }
+
       const emptyBoundaryRunId = RunId.make(
         "run:projection-fork-source-rollback:source:empty-boundary",
       );
@@ -8222,7 +8312,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
           attempt_id, thread_id, run_id, attempt_ordinal, root_node_id, provider,
           provider_instance_id, provider_thread_id, provider_turn_id, status, payload_json
         ) VALUES (
-          'attempt:projection-fork-source-rollback:inherited-superseded',
+          ${"attempt:projection-fork-source-rollback:inherited-superseded"},
           ${sourceThreadId}, ${sourceRun2Id}, 2, ${inheritedSupersededNodeId}, 'codex',
           ${providerInstanceId}, ${sourceProviderThreadId}, NULL, 'superseded',
           ${encodeUnknownJsonString({