fix(server): preserve response boundaries when forking threads

apps/server/src/environment/ServerEnvironment.ts

@@ -226,6 +226,7 @@ export const make = Effect.gen(function* () {
       inlineMessageContext: true,
       requiredWorktreeBootstrap: true,
       threadSettlement: true,
+      threadForkFromItem: true,
       threadAutoSettlement: true,
       storageCleanup: true,
       storageCleanupRun: true,

apps/server/src/mcp/toolkits/thread/tools.ts

@@ -203,7 +203,7 @@ const transferResult = Schema.Struct({ sequence: NonNegativeInt, targetThreadId:
 const ThreadForkTool = Tool.make("t3_thread_fork", {
   ...commandTool,
   description:
-    "Fork a thread from a stable run or checkpoint using the existing fork command. Omit threadId to fork this thread. The fork inherits the source configuration. Acceptance does not mean a provider turn has completed.",
+    "Fork a thread from a stable run, checkpoint, or completed/interrupted assistant response using the existing fork command. Omit threadId to fork this thread. The fork inherits the source configuration. Acceptance does not mean a provider turn has completed.",
   parameters: Schema.Struct({
     threadId: Schema.optional(ThreadId),
     sourcePoint: OrchestrationV2ThreadForkSourcePoint,

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

@@ -240,6 +240,7 @@ export interface CommandPolicyV2Shape {
       readonly hasStrongNativeSource: boolean;
       readonly sourceRunStatus: OrchestrationV2Run["status"];
       readonly fromSpecificTurn: boolean;
+      readonly fromSpecificItem?: boolean;
     },
   ) => Effect.Effect<ForkExecutionPolicyV2, CommandPolicyV2Error>;
   readonly ensureRollback: (
@@ -400,7 +401,9 @@ const ensureContextHandoff: CommandPolicyV2Shape["ensureContextHandoff"] = (inpu
 const decideForkExecution: CommandPolicyV2Shape["decideForkExecution"] = (input) => {
   // Unsuccessful runs may have no native turn or assistant cursor. Forking
   // those at native head can include later turns, so use the bounded transcript.
+  // Item cuts use portable context because native turn cursors can include a later steer.
   const canForkNatively =
+    input.fromSpecificItem !== true &&
     (input.sourceRunStatus === "completed" || input.sourceRunStatus === "waiting") &&
     input.sameProvider &&
     input.hasStrongNativeSource &&

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

@@ -638,6 +638,7 @@ function runForSourcePoint(
     case "latest_stable":
       return latestStableRun(projection);
     case "run":
+    case "turn_item":
       return projection.runs.find((run) => run.id === sourcePoint.runId) ?? null;
     case "checkpoint": {
       const checkpoint = projection.checkpoints.find(
@@ -3681,6 +3682,23 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         cause: forkableSourceRunStatusError(sourceRun),
       });
     }
+    const throughTurnItemId =
+      command.sourcePoint.type === "turn_item" ? command.sourcePoint.turnItemId : undefined;
+    if (throughTurnItemId !== undefined) {
+      const runItems = yield* readHandoffItems(command.sourceThreadId, [sourceRun.id]);
+      const item = runItems.find((candidate) => candidate.id === throughTurnItemId);
+      if (
+        item?.type !== "assistant_message" ||
+        item.runId !== sourceRun.id ||
+        (item.status !== "completed" && item.status !== "interrupted")
+      ) {
+        return yield* new OrchestratorDispatchError({
+          commandId: command.commandId,
+          commandType: command.type,
+          cause: `Fork source item ${throughTurnItemId} is not a finished assistant response of run ${sourceRun.id}.`,
+        });
+      }
+    }
     const sourceProviderThread = providerThreadForRun(sourceProjection, sourceRun);
     const now = command.createdAt ?? (yield* DateTime.now);
     const emitEvent = emit(events, command);
@@ -3696,7 +3714,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         sourceProjection,
         sourceRun,
         sourceProviderThread,
-        canonicalSourcePoint: contextSourcePointForRun(sourceProjection, sourceRun),
+        canonicalSourcePoint: {
+          ...contextSourcePointForRun(sourceProjection, sourceRun),
+          ...(throughTurnItemId === undefined ? {} : { turnItemId: throughTurnItemId }),
+        },
         transferId,
         targetThreadId: command.targetThreadId,
         ...(command.title === undefined ? {} : { title: command.title }),
@@ -3778,6 +3799,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       });
     }
 
+    if (command.sourcePoint.type === "turn_item") {
+      return yield* new OrchestratorDispatchError({
+        commandId: command.commandId,
+        commandType: command.type,
+        cause: "Merge-back requires a run or checkpoint rather than a single response.",
+      });
+    }
     const sourceRun = runForSourcePoint(sourceProjection, command.sourcePoint);
     if (sourceRun === null) {
       return yield* new OrchestratorDispatchError({
@@ -5803,6 +5831,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
                 hasStrongNativeSource: sourceProviderThread?.nativeThreadRef?.strength === "strong",
                 sourceRunStatus: sourceRun.status,
                 fromSpecificTurn: sourceRun !== null,
+                fromSpecificItem: pendingForkTransfer.sourcePoint.turnItemId !== undefined,
               }),
             );
       const canResolveForkNatively = forkExecution === "native_fork";
@@ -5859,12 +5888,30 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       const portableForkItems =
         !requiresPortableFork || sourceProjection === null || sourceRun === null
           ? []
-          : yield* readHandoffItems(sourceProjection.thread.id, [
-              ...sourceProjection.runs
-                .filter((run) => run.ordinal <= sourceRun.ordinal)
-                .map((run) => run.id),
-              ...(sourceProjection.thread.historyOrigin === "v1_import" ? [null] : []),
-            ]);
+          : yield* projectionStore.getThreadProjection(command.threadId).pipe(
+              Effect.map((targetHistory) =>
+                targetHistory.visibleTurnItems
+                  .filter((row) => row.visibility === "inherited")
+                  .map((row) => row.item),
+              ),
+              Effect.mapError(
+                (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }),
+              ),
+            );
+      const cutItemId = pendingForkTransfer?.sourcePoint.turnItemId;
+      const cutItem = portableForkItems.find(
+        (item) =>
+          item.id === cutItemId &&
+          item.threadId === sourceProjection?.thread.id &&
+          item.runId === sourceRun?.id,
+      );
+      if (requiresPortableFork && cutItemId !== undefined && cutItem === undefined) {
+        return yield* new OrchestratorDispatchError({
+          commandId: command.commandId,
+          commandType: command.type,
+          cause: `Fork source item ${cutItemId} is no longer available.`,
+        });
+      }
       const portableForkHandoff =
         !requiresPortableFork ||
         pendingForkTransfer === undefined ||

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

@@ -1008,6 +1008,7 @@ type ShellThreadRow = {
   readonly thread_id: string;
   readonly payload_json: string;
   readonly forked_from_run_source_thread_id: string | null;
+  readonly forked_from_items_after_cut: number;
   readonly latest_run_id: string | null;
   readonly latest_run_status: string | null;
   readonly latest_run_requested_at: string | null;
@@ -1338,14 +1339,43 @@ export function isTurnItemAtOrBeforeRun(input: {
   return ordinal !== undefined && ordinal <= input.sourceRunOrdinal;
 }
 
+/**
+ * A fork that ends at a response inside its source run (one a steer cut off)
+ * keeps that run's items only through the response.
+ */
+function isTurnItemWithinForkCut(input: {
+  readonly item: Pick<OrchestrationV2TurnItem, "runId" | "ordinal">;
+  readonly sourceRunId: RunId;
+  readonly cutOrdinal: number | undefined;
+}): boolean {
+  return (
+    input.cutOrdinal === undefined ||
+    input.item.runId !== input.sourceRunId ||
+    input.item.ordinal <= input.cutOrdinal
+  );
+}
+
+function forkCutOrdinal(
+  items: ReadonlyArray<Pick<OrchestrationV2TurnItem, "id" | "ordinal">>,
+  throughTurnItemId: TurnItemId | undefined,
+): number | undefined {
+  // A windowed read stops at or before the cut, so a page without the cut
+  // item holds nothing after it.
+  return throughTurnItemId === undefined
+    ? undefined
+    : items.find((item) => item.id === throughTurnItemId)?.ordinal;
+}
+
 function visibleTurnItemsThroughRun(input: {
   readonly sourceProjection: OrchestrationV2ThreadProjection;
   readonly sourceRunId: NonNullable<OrchestrationV2TurnItem["runId"]>;
+  readonly throughTurnItemId: TurnItemId | undefined;
 }): Array<Omit<OrchestrationV2ProjectedTurnItem, "position">> {
   const sourceRun = input.sourceProjection.runs.find((run) => run.id === input.sourceRunId);
   if (sourceRun === undefined) {
     return [];
   }
+  const cutOrdinal = forkCutOrdinal(input.sourceProjection.turnItems, input.throughTurnItemId);
 
   const runOrdinalById = new Map(input.sourceProjection.runs.map((run) => [run.id, run.ordinal]));
   const inheritedPrefix = input.sourceProjection.visibleTurnItems
@@ -1369,12 +1399,14 @@ function visibleTurnItemsThroughRun(input: {
       ) {
         return false;
       }
-      return isTurnItemAtOrBeforeRun({
-        historyOrigin: input.sourceProjection.thread.historyOrigin,
-        itemRunId: item.runId,
-        runOrdinalById,
-        sourceRunOrdinal: sourceRun.ordinal,
-      });
+      return (
+        isTurnItemAtOrBeforeRun({
+          historyOrigin: input.sourceProjection.thread.historyOrigin,
+          itemRunId: item.runId,
+          runOrdinalById,
+          sourceRunOrdinal: sourceRun.ordinal,
+        }) && isTurnItemWithinForkCut({ item, sourceRunId: sourceRun.id, cutOrdinal })
+      );
     }),
   );
 
@@ -1393,6 +1425,7 @@ function buildVisibleTurnItems(input: {
   const inherited = visibleTurnItemsThroughRun({
     sourceProjection: input.sourceProjection,
     sourceRunId: forkedFrom.runId,
+    throughTurnItemId: forkedFrom.throughTurnItemId,
   });
   const markerItem = makeForkMarkerTurnItem({
     targetProjection: input.projection,
@@ -1648,6 +1681,7 @@ type ShellThreadState = {
   readonly goal: OrchestrationV2ThreadShell["goal"];
   readonly itemCount: number;
   readonly runlessItemCount: number;
+  readonly forkedFromItemsAfterCut: number;
   readonly updatedAt: OrchestrationV2ThreadProjection["updatedAt"];
   readonly runOrdinalById: ReadonlyMap<RunId, number>;
   readonly itemCountByRunId: ReadonlyMap<RunId, number>;
@@ -1743,7 +1777,8 @@ function visibleItemCountForShell(input: {
 
   return (
     inheritedPrefixCount +
-    itemCountThroughRun({ state: sourceState, runId: forkedFrom.runId }) +
+    itemCountThroughRun({ state: sourceState, runId: forkedFrom.runId }) -
+    state.forkedFromItemsAfterCut +
     1 +
     state.itemCount
   );
@@ -3451,14 +3486,20 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
                     WHERE thread_id = ${forkedFrom.threadId}
                       AND run_id = ${forkedFrom.runId}
                     LIMIT 1
+                  ), fork_cut AS (
+                    SELECT ordinal
+                    FROM orchestration_v2_projection_turn_items
+                    WHERE thread_id = ${forkedFrom.threadId}
+                      AND turn_item_id = ${forkedFrom.throughTurnItemId ?? null}
+                    LIMIT 1
                   ), fork_boundary AS (
-                    SELECT (
+                    SELECT COALESCE((SELECT ordinal FROM fork_cut), (
                       SELECT item.ordinal
                       FROM orchestration_v2_projection_turn_items AS item
                       WHERE item.run_id = run.run_id
                       ORDER BY item.ordinal DESC
                       LIMIT 1
-                    ) AS ordinal
+                    )) AS ordinal
                     FROM orchestration_v2_projection_runs AS run
                     WHERE run.thread_id = ${forkedFrom.threadId}
                       AND run.ordinal <= (SELECT ordinal FROM fork_run)
@@ -5003,12 +5044,17 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
         if (sourceRun !== undefined) {
           const ordinals = new Map(source.records.runs.map((run) => [run.id, run.ordinal]));
           const sourceItems = source.local.map((row) => row.item);
+          // Rows are in ordinal order, so the cut is the cut item's index.
+          const cutIndex =
+            fork.throughTurnItemId === undefined
+              ? undefined
+              : sourceItems.findIndex((item) => item.id === fork.throughTurnItemId);
           inherited = [
             ...source.visible.filter(
               (row) => row.item.threadId !== fork.threadId || row.item.type === "fork",
             ),
             ...source.local.filter(
-              (row) =>
+              (row, index) =>
                 !isOrchestrationV2SupersededInterrupt({
                   item: row.item,
                   attempts: source.records.attempts,
@@ -5019,6 +5065,11 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
                   itemRunId: row.item.runId,
                   runOrdinalById: ordinals,
                   sourceRunOrdinal: sourceRun.ordinal,
+                }) &&
+                isTurnItemWithinForkCut({
+                  item: { runId: row.item.runId, ordinal: index },
+                  sourceRunId: sourceRun.id,
+                  cutOrdinal: cutIndex,
                 }),
             ),
           ].map((row) => ({ ...row, visibility: "inherited" }));
@@ -5287,6 +5338,18 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
                   THEN json_extract(t.payload_json, '$.forkedFrom.threadId')
                 ELSE NULL
               END AS forked_from_run_source_thread_id,
+              -- Source-run items after the response a fork ends at (a steer
+              -- and the reply to it); the shell leaves them out of its count.
+              (
+                SELECT COUNT(*)
+                FROM orchestration_v2_projection_turn_items AS cut_item
+                JOIN orchestration_v2_projection_turn_items AS after_item
+                  ON after_item.thread_id = cut_item.thread_id
+                  AND after_item.run_id = cut_item.run_id
+                  AND after_item.ordinal > cut_item.ordinal
+                WHERE cut_item.thread_id = json_extract(t.payload_json, '$.forkedFrom.threadId')
+                  AND cut_item.turn_item_id = json_extract(t.payload_json, '$.forkedFrom.throughTurnItemId')
+              ) AS forked_from_items_after_cut,
               presented.run_id AS latest_run_id,
               presented.status AS latest_run_status,
               presented.requested_at AS latest_run_requested_at,
@@ -5892,6 +5955,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
           ),
           itemCount: row.item_count,
           runlessItemCount: row.runless_item_count,
+          forkedFromItemsAfterCut: row.forked_from_items_after_cut,
           updatedAt: thread.updatedAt,
           runOrdinalById: runOrdinalsByThreadId.get(ThreadId.make(row.thread_id)) ?? new Map(),
           itemCountByRunId: itemCountsByThreadId.get(ThreadId.make(row.thread_id)) ?? new Map(),