From a7fcd1697a2796462f800f120364abde344991fa Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Sat, 10 Oct 2026 01:03:15 -0700 Subject: [PATCH 1/3] fix(orchestration): fork from a response a steer cut off Active steering writes the steer into the same run as the response it interrupts, and a fork point was always a whole run. So a response a steer cut off never got a Fork button, and forking its run would have carried the steer and the reply to it into the fork. A fork can now end at one assistant response. `thread.fork` takes a `turn_item` source point (run plus item). The server checks that the item is a finished assistant message of that run and records the cut on the transfer and as `forkedFrom.throughTurnItemId`. The inherited timeline, the windowed history read, search and the shell item count all stop at that item. Provider context follows the cut. Claude forks natively: an assistant item's native id is the SDK message uuid, which `forkSession` accepts as an inclusive `upToMessageId`. A new optional `canForkFromItem` capability says so. Codex `thread/fork` only cuts at turn ends, so it and every provider without the capability get the portable handoff cut at the item, never the run end. Web and mobile show the response meta row and Fork on the last assistant message before each steer as well as the run's last one. Forks from a mid-run response send the item point; run-end forks still send the run point. A reply the user stopped mid-stream (interrupted) can be forked too, matching the server's existing run status rule. Co-Authored-By: Claude Opus 5.5 --- .../src/features/threads/ThreadFeed.tsx | 22 +- apps/mobile/src/lib/threadActivity.test.ts | 51 +++++ apps/mobile/src/lib/threadActivity.ts | 25 ++ apps/server/src/mcp/toolkits/thread/tools.ts | 2 +- .../Adapters/ClaudeAdapterV2.test.ts | 38 ++++ .../Adapters/ClaudeAdapterV2.ts | 20 ++ .../src/orchestration-v2/CommandPolicy.ts | 10 + .../src/orchestration-v2/Orchestrator.ts | 81 ++++++- .../src/orchestration-v2/ProjectionStore.ts | 84 ++++++- .../ProviderTurnStartService.ts | 16 ++ .../ThreadFork.execution.test.ts | 214 ++++++++++++++++++ .../src/orchestration-v2/ThreadForkService.ts | 3 + apps/web/src/components/ChatView.tsx | 8 +- .../chat/MessagesTimeline.logic.test.ts | 106 +++++++++ .../components/chat/MessagesTimeline.logic.ts | 43 +++- .../src/components/chat/MessagesTimeline.tsx | 18 +- .../provider-capability-system.md | 3 + .../client-runtime/src/operations/commands.ts | 11 +- .../src/state/threadWorkflows.test.ts | 22 ++ .../src/state/threadWorkflows.ts | 7 +- packages/contracts/src/orchestrationV2.ts | 12 +- .../src/server/ProviderAdapter.ts | 2 + 22 files changed, 758 insertions(+), 40 deletions(-) diff --git a/apps/mobile/src/features/threads/ThreadFeed.tsx b/apps/mobile/src/features/threads/ThreadFeed.tsx index ee0c4711b03b..6b1f0eebc33e 100644 --- a/apps/mobile/src/features/threads/ThreadFeed.tsx +++ b/apps/mobile/src/features/threads/ThreadFeed.tsx @@ -161,6 +161,7 @@ import { } from "@t3tools/mobile-markdown-text/links"; import { failedFeedRunIds, + terminalFeedAssistantMessageIds, deriveThreadFeedPresentation, threadFeedRunIsUnsettled, isContextCompactionActivityGroup, @@ -321,6 +322,8 @@ function AssistantForkButton(props: { readonly environmentId: EnvironmentId; readonly iconColor: ColorValue; readonly projectedItem: OrchestrationV2ProjectedTurnItem; + /** A later response in the run follows, so the fork cuts at this item. */ + readonly midRun: boolean; readonly sourceTitle: string; }) { const support = useV2ItemSupport({ @@ -354,6 +357,7 @@ function AssistantForkButton(props: { sourceThreadId: props.projectedItem.sourceThreadId, targetThreadId, runId, + ...(props.midRun ? { turnItemId: props.projectedItem.item.id } : {}), title: `${props.sourceTitle} fork`, creationSource: "mobile", }, @@ -1525,7 +1529,7 @@ function renderFeedEntry( readonly expandedWorkRows: Record; readonly workRowSizing: ReturnType; readonly workGroupScrollPositions: Map; - readonly terminalAssistantMessageIds: ReadonlySet; + readonly terminalAssistantMessageIds: ReturnType; readonly unsettledTurnId: RunId | null; readonly failedRunIds: ReadonlySet; readonly onCopyWorkRow: (rowId: string, value: string) => void; @@ -1694,7 +1698,7 @@ function renderFeedEntry( message.runId === props.unsettledTurnId; const showAssistantMeta = message.role === "assistant" && - props.terminalAssistantMessageIds.has(message.id) && + props.terminalAssistantMessageIds.terminalIds.has(message.id) && !assistantTurnStillInProgress && !message.streaming; @@ -1966,6 +1970,7 @@ function renderFeedEntry( environmentId={props.environmentId} iconColor={iconSubtleColor} projectedItem={message.projectedItem} + midRun={props.terminalAssistantMessageIds.midRunIds.has(message.id)} sourceTitle={props.threadTitle} /> ) : null} @@ -2761,15 +2766,10 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) { () => failedFeedRunIds(props.feed, props.latestRun), [props.feed, props.latestRun], ); - const terminalAssistantMessageIds = useMemo(() => { - const terminalIdsByTurn = new Map(); - for (const entry of props.feed) { - if (entry.type === "message" && entry.message.role === "assistant" && entry.message.runId) { - terminalIdsByTurn.set(entry.message.runId, entry.message.id); - } - } - return new Set(terminalIdsByTurn.values()); - }, [props.feed]); + const terminalAssistantMessageIds = useMemo( + () => terminalFeedAssistantMessageIds(props.feed), + [props.feed], + ); useEffect(() => { const previous = previousLatestTurnRef.current; previousLatestTurnRef.current = props.latestRun; diff --git a/apps/mobile/src/lib/threadActivity.test.ts b/apps/mobile/src/lib/threadActivity.test.ts index fca9430b3df3..8aa80923e52e 100644 --- a/apps/mobile/src/lib/threadActivity.test.ts +++ b/apps/mobile/src/lib/threadActivity.test.ts @@ -31,6 +31,7 @@ import { deriveThreadFeedPresentation, threadFeedActivityIsVisible, threadFeedRunIsUnsettled, + terminalFeedAssistantMessageIds, type ThreadFeedActivity, type ThreadFeedEntry, togglePendingUserInputOptionSelection, @@ -2028,6 +2029,56 @@ const multiSelectQuestion = { multiSelect: true, } as const; +describe("terminalFeedAssistantMessageIds", () => { + const runId = RunId.make("steered-run"); + const message = ( + id: string, + role: "user" | "assistant", + inputIntent?: "turn_start" | "steer", + ): ThreadFeedEntry => ({ + type: "message", + id, + createdAt: "2026-04-01T00:00:00.000Z", + message: { + id: MessageId.make(id), + role, + text: id, + attachments: [], + runId, + streaming: false, + ...(inputIntent === undefined ? {} : { inputIntent }), + visibility: "local", + sourceThreadId: ThreadId.make("thread"), + createdAt: "2026-04-01T00:00:00.000Z", + updatedAt: "2026-04-01T00:00:00.000Z", + }, + }); + + it("ends a response at each steer and marks only the cut-off one as mid-run", () => { + const ids = terminalFeedAssistantMessageIds([ + message("prompt", "user", "turn_start"), + message("commentary", "assistant"), + message("cut-off", "assistant"), + message("steer", "user", "steer"), + message("final", "assistant"), + ]); + + expect([...ids.terminalIds]).toEqual(["cut-off", "final"]); + expect([...ids.midRunIds]).toEqual(["cut-off"]); + }); + + it("keeps one terminal message for an unsteered run", () => { + const ids = terminalFeedAssistantMessageIds([ + message("prompt", "user", "turn_start"), + message("commentary", "assistant"), + message("final", "assistant"), + ]); + + expect([...ids.terminalIds]).toEqual(["final"]); + expect(ids.midRunIds.size).toBe(0); + }); +}); + describe("pending user input answers", () => { it("preserves exact editor text, including a deliberately cleared answer", () => { const question = { ...singleSelectQuestion, initialAnswer: " Proposed message\n" }; diff --git a/apps/mobile/src/lib/threadActivity.ts b/apps/mobile/src/lib/threadActivity.ts index edaaf2b0fab8..5f7288dc14b2 100644 --- a/apps/mobile/src/lib/threadActivity.ts +++ b/apps/mobile/src/lib/threadActivity.ts @@ -1048,6 +1048,31 @@ export function failedFeedRunIds( return failed; } +/** + * The assistant messages that end a stretch of output and so carry metadata: + * each run's last one, plus the last one before a later user message in the + * same run (a steer). Forking from one in `midRunIds` must cut inside its run. + */ +export function terminalFeedAssistantMessageIds(feed: ReadonlyArray) { + const lastIdBySegment = new Map(); + const lastIdByRun = new Map(); + const segmentByRun = new Map(); + for (const entry of feed) { + if (entry.type !== "message" || !entry.message.runId) continue; + const runId = entry.message.runId; + if (entry.message.role === "user") { + segmentByRun.set(runId, (segmentByRun.get(runId) ?? 0) + 1); + continue; + } + lastIdByRun.set(runId, entry.message.id); + lastIdBySegment.set(`${runId}:${segmentByRun.get(runId) ?? 0}`, entry.message.id); + } + const terminalIds = new Set(lastIdBySegment.values()); + const runEndIds = new Set(lastIdByRun.values()); + const midRunIds = new Set([...terminalIds].filter((id) => !runEndIds.has(id))); + return { terminalIds, midRunIds }; +} + /** * A prompt without a run (a provider-native subagent, or a turn imported from * V1) folds its response like a run. `runlessWorkActive` keeps the latest diff --git a/apps/server/src/mcp/toolkits/thread/tools.ts b/apps/server/src/mcp/toolkits/thread/tools.ts index b5fb7d6be3a6..79b20c405e7b 100644 --- a/apps/server/src/mcp/toolkits/thread/tools.ts +++ b/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, a checkpoint, or one finished assistant response (turn_item, which leaves out the rest of its run such as a steer) 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, diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 6777f0a2902a..98803ab35e82 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -27,6 +27,7 @@ import { RunAttemptId, RunId, ThreadId, + TurnItemId, } from "@t3tools/contracts"; import { assert, describe, it } from "@effect/vitest"; import * as Context from "effect/Context"; @@ -1759,6 +1760,43 @@ describe("ClaudeAdapterV2 native fork", () => { assert.equal(forkedProviderThread.forkedFrom?.providerThreadId, sourceProviderThread.id); assert.equal(forkedProviderThread.forkedFrom?.providerTurnId, providerTurnId); + // A response a steer cut off is mid-turn: its assistant item carries the + // SDK message uuid, which forkSession takes as an inclusive cut. + yield* runtime.forkThread({ + sourceProviderThread, + sourceProviderTurns: [], + providerTurnId, + throughTurnItem: { + id: TurnItemId.make("turn-item-claude-cut-off"), + threadId: sourceThreadId, + runId: RunId.make("run-claude-steered"), + nodeId: null, + providerThreadId: sourceProviderThread.id, + providerTurnId, + nativeItemRef: { + driver: ClaudeAdapterV2.CLAUDE_PROVIDER, + nativeId: "cut-off-assistant-uuid", + strength: "strong", + }, + parentItemId: null, + ordinal: 2, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "assistant_message", + messageId: MessageId.make("message-claude-cut-off"), + text: "cut off by a steer", + streaming: false, + }, + targetThreadId: ThreadId.make("thread-claude-fork-steered-target"), + }); + assert.deepEqual(forkCalls[1]?.options, { + dir: "/workspace", + upToMessageId: "cut-off-assistant-uuid", + }); + yield* runtime.startTurn({ appThread: { createdBy: "user", diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 1451ec9ff62e..fd109a191838 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -196,6 +196,7 @@ export const ClaudeProviderCapabilitiesV2 = { canRollbackThread: true, canForkThread: true, canForkFromTurn: true, + canForkFromItem: true, canForkFromSubagentThread: false, exposesNativeThreadId: true, }, @@ -1226,6 +1227,25 @@ const getNativeConversationHeadId = Effect.fnUntraced(function* ( const resolveClaudeForkUpToMessageId = Effect.fn("ClaudeAdapterV2.resolveForkUpToMessageId")( function* (input: ProviderAdapter.ProviderAdapterV2ForkThreadInput) { + // Assistant items carry the SDK message uuid, which forkSession accepts as + // an inclusive cut, so a fork can end before a steer in the same turn. + const throughItem = input.throughTurnItem; + if (throughItem !== undefined) { + const nativeId = throughItem.nativeItemRef?.nativeId; + if ( + throughItem.type !== "assistant_message" || + throughItem.nativeItemRef?.driver !== CLAUDE_PROVIDER || + nativeId === undefined || + nativeId === null + ) { + return yield* new ProviderAdapter.ProviderAdapterForkThreadError({ + driver: CLAUDE_PROVIDER, + providerThreadId: input.sourceProviderThread.id, + cause: `Cannot fork Claude thread at item ${throughItem.id}: it has no SDK assistant message id.`, + }); + } + return nativeId; + } if (input.providerTurnId === undefined || input.sourceProviderTurns === undefined) { return undefined; } diff --git a/apps/server/src/orchestration-v2/CommandPolicy.ts b/apps/server/src/orchestration-v2/CommandPolicy.ts index 5f7d130c17fd..1159417d03d3 100644 --- a/apps/server/src/orchestration-v2/CommandPolicy.ts +++ b/apps/server/src/orchestration-v2/CommandPolicy.ts @@ -197,6 +197,7 @@ export interface CommandPolicyV2Shape { readonly ensureNativeFork: ( input: CapabilityCheckInput & { readonly fromSpecificTurn: boolean; + readonly fromSpecificItem?: boolean; }, ) => Effect.Effect; readonly decideForkExecution: ( @@ -205,6 +206,8 @@ export interface CommandPolicyV2Shape { readonly hasStrongNativeSource: boolean; readonly sourceRunStatus: OrchestrationV2Run["status"]; readonly fromSpecificTurn: boolean; + /** The fork ends at a response inside the run, e.g. one a steer cut off. */ + readonly fromSpecificItem?: boolean; }, ) => Effect.Effect; readonly ensureRollback: ( @@ -291,6 +294,11 @@ const ensureNativeFork: CommandPolicyV2Shape["ensureNativeFork"] = (input) => { ), ); } + if (input.fromSpecificItem === true && input.capabilities.threads.canForkFromItem !== true) { + return Effect.fail( + unsupported(input, "fork_from_turn", "providerInstanceId cannot fork inside a turn"), + ); + } if (input.capabilities.identity.nativeThreadIds !== "strong") { return Effect.fail( unsupported( @@ -371,6 +379,8 @@ const decideForkExecution: CommandPolicyV2Shape["decideForkExecution"] = (input) input.hasStrongNativeSource && input.capabilities.threads.canForkThread && (!input.fromSpecificTurn || input.capabilities.threads.canForkFromTurn) && + // A native fork that can only cut at turn ends would keep a steer made after the response. + (input.fromSpecificItem !== true || input.capabilities.threads.canForkFromItem === true) && input.capabilities.identity.nativeThreadIds === "strong"; if (canForkNatively) { diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 3f17a6faa55d..e369c42232c3 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -628,6 +628,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( @@ -688,6 +689,38 @@ function contextSourcePointForRun( }; } +/** + * Drops the source run's items after the response a fork ends at, such as a + * steer and the reply to it. Without a cut item the whole run is kept. + */ +function itemsThroughForkCut( + items: ReadonlyArray, + sourceRunId: RunId, + throughTurnItemId: TurnItemId | undefined, +): ReadonlyArray { + if (throughTurnItemId === undefined) return items; + const cutOrdinal = items.find((item) => item.id === throughTurnItemId)?.ordinal; + return items.filter( + (item) => + item.runId !== sourceRunId || (cutOrdinal !== undefined && item.ordinal <= cutOrdinal), + ); +} + +/** + * A response a fork may end at: a finished assistant message of the source run. + * Stopping mid-reply leaves it interrupted, which still ends the conversation. + */ +function isForkableResponseItem( + item: OrchestrationV2TurnItem | null, + run: OrchestrationV2Run, +): boolean { + return ( + item?.type === "assistant_message" && + item.runId === run.id && + (item.status === "completed" || item.status === "interrupted") + ); +} + function pendingForkTransferForThread( projection: Pick, ): OrchestrationV2ContextTransfer | undefined { @@ -3543,6 +3576,24 @@ 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 throughItem = yield* projectionStore + .getTurnItem({ threadId: command.sourceThreadId, itemId: throughTurnItemId }) + .pipe( + Effect.mapError( + (cause) => new OrchestratorProjectionError({ threadId: command.sourceThreadId, cause }), + ), + ); + if (!isForkableResponseItem(throughItem, sourceRun)) { + 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); @@ -3558,7 +3609,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 }), @@ -3640,6 +3694,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 takes a run or checkpoint, not a single response.", + }); + } const sourceRun = runForSourcePoint(sourceProjection, command.sourcePoint); if (sourceRun === null) { return yield* new OrchestratorDispatchError({ @@ -5662,6 +5723,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"; @@ -5675,6 +5737,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio providerInstanceId: modelSelection.instanceId, capabilities, fromSpecificTurn: sourceRun !== null, + fromSpecificItem: pendingForkTransfer?.sourcePoint.turnItemId !== undefined, }), ); } @@ -5718,12 +5781,16 @@ 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] : []), - ]); + : itemsThroughForkCut( + yield* readHandoffItems(sourceProjection.thread.id, [ + ...sourceProjection.runs + .filter((run) => run.ordinal <= sourceRun.ordinal) + .map((run) => run.id), + ...(sourceProjection.thread.historyOrigin === "v1_import" ? [null] : []), + ]), + sourceRun.id, + pendingForkTransfer?.sourcePoint.turnItemId, + ); const portableForkHandoff = !requiresPortableFork || pendingForkTransfer === undefined || diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 51343fd8af91..c9808f0b1d70 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -961,6 +961,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; @@ -1283,14 +1284,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; + 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>, + 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; + readonly throughTurnItemId: TurnItemId | undefined; }): Array> { 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 @@ -1314,12 +1344,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 }) + ); }), ); @@ -1338,6 +1370,7 @@ function buildVisibleTurnItems(input: { const inherited = visibleTurnItemsThroughRun({ sourceProjection: input.sourceProjection, sourceRunId: forkedFrom.runId, + throughTurnItemId: forkedFrom.throughTurnItemId, }); const markerItem = makeForkMarkerTurnItem({ targetProjection: input.projection, @@ -1605,6 +1638,7 @@ type ShellThreadState = { readonly goal: OrchestrationV2ThreadShell["goal"]; readonly itemCount: number; readonly runlessItemCount: number; + readonly forkedFromItemsAfterCut: number; readonly updatedAt: OrchestrationV2ThreadProjection["updatedAt"]; readonly runOrdinalById: ReadonlyMap; readonly itemCountByRunId: ReadonlyMap; @@ -1705,7 +1739,8 @@ function visibleItemCountForShell(input: { return ( inheritedPrefixCount + - itemCountThroughRun({ state: sourceState, runId: forkedFrom.runId }) + + itemCountThroughRun({ state: sourceState, runId: forkedFrom.runId }) - + state.forkedFromItemsAfterCut + 1 + state.itemCount ); @@ -3359,14 +3394,20 @@ export const layer: Layer.Layer = 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) @@ -4856,12 +4897,17 @@ export const layer: Layer.Layer = 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, @@ -4872,6 +4918,11 @@ export const layer: Layer.Layer = 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" })); @@ -5144,6 +5195,18 @@ export const layer: Layer.Layer = 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, @@ -5721,6 +5784,7 @@ export const layer: Layer.Layer = ), 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(), diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index e2c905b1e17a..d6836745d85d 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -630,6 +630,21 @@ export const layer: Layer.Layer< cause: `Native fork transfer ${nativeForkTransfer.id} has no source provider execution.`, }); } + const throughTurnItemId = nativeForkTransfer.sourcePoint.turnItemId; + const throughTurnItem = + throughTurnItemId === undefined + ? undefined + : yield* projectionStore.getTurnItem({ + threadId: nativeForkTransfer.sourceThreadId, + itemId: throughTurnItemId, + }); + if (throughTurnItem === null) { + // Forking at the turn end instead would keep what the user cut off. + return yield* new ProviderTurnStartError({ + runId, + cause: `Native fork transfer ${nativeForkTransfer.id} ends at item ${throughTurnItemId}, which no longer exists.`, + }); + } return yield* loadFromProvider( session.forkThread({ sourceProviderThread, @@ -640,6 +655,7 @@ export const layer: Layer.Layer< ...(sourceProviderTurn === undefined ? {} : { providerTurnId: sourceProviderTurn.id }), + ...(throughTurnItem === undefined ? {} : { throughTurnItem }), }), ); } diff --git a/apps/server/src/orchestration-v2/ThreadFork.execution.test.ts b/apps/server/src/orchestration-v2/ThreadFork.execution.test.ts index a13277418550..7cf81af52d3b 100644 --- a/apps/server/src/orchestration-v2/ThreadFork.execution.test.ts +++ b/apps/server/src/orchestration-v2/ThreadFork.execution.test.ts @@ -249,3 +249,217 @@ it.effect.each(forkCases)( assert.isNull(target.providerThreads[0]?.forkedFrom); }).pipe(Effect.provide(layer)), ); + +const steeredForkCases = (["codex", "claudeAgent"] as const).map((driverName) => { + const driver = ProviderDriverKind.make(driverName); + const instanceId = ProviderInstanceId.make(driver); + const adapter: ProviderAdapter.ProviderAdapterV2["Service"] = { + instanceId, + driver, + getCapabilities: () => + Effect.succeed( + driver === "codex" ? CodexProviderCapabilitiesV2 : ClaudeProviderCapabilitiesV2, + ), + planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), + openSession: () => Effect.die("Execution is paused after dispatch for handoff inspection"), + }; + return { + driver, + instanceId, + modelSelection: { instanceId, model: "test-model" }, + // Claude's forkSession cuts at an SDK message uuid; Codex forks only at turn ends. + expectedStrategy: driver === "codex" ? "portable_context" : "native_fork", + layer: ProviderReplayHarness.layerWithRegistry( + { name: `fork-steered-${driver}` }, + ProviderAdapterRegistry.layerFromAdapters([adapter]), + { runEffectWorker: false }, + ), + }; +}); + +it.effect.each(steeredForkCases)( + "forks $driver from a response a steer cut off, without the steer", + ({ driver, instanceId, modelSelection, expectedStrategy, layer }) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const sourceThreadId = ThreadId.make("fork-steered-source"); + const targetThreadId = ThreadId.make("fork-steered-target"); + const providerThreadId = ProviderThreadId.make("fork-steered-native-thread"); + const runId = RunId.make("fork-steered-run"); + const cutOffItemId = TurnItemId.make("fork-steered-cut-off-response"); + const steerItemId = TurnItemId.make("fork-steered-steer"); + + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create-steered-source"), + threadId: sourceThreadId, + projectId: ProjectId.make("fork-steered-project"), + title: "Steered source", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + const itemBase = { + threadId: sourceThreadId, + runId, + nodeId: null, + providerThreadId, + providerTurnId: null, + parentItemId: null, + ordinal: 0, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + } as const; + const userItem = (id: string, text: string, inputIntent: "turn_start" | "steer") => ({ + ...itemBase, + id: TurnItemId.make(id), + nativeItemRef: null, + type: "user_message" as const, + createdBy: "user" as const, + creationSource: "web" as const, + inputIntent, + messageId: MessageId.make(id), + text, + attachments: [], + }); + const assistantItem = (id: TurnItemId, text: string) => ({ + ...itemBase, + id, + nativeItemRef: { driver, nativeId: `sdk-uuid:${id}`, strength: "strong" as const }, + type: "assistant_message" as const, + messageId: MessageId.make(id), + text, + streaming: false, + }); + // Active steering keeps the steer and its reply in the same run. + const items = [ + userItem("fork-steered-prompt", "INITIAL_PROMPT_MARKER", "turn_start"), + assistantItem(cutOffItemId, "CUT_OFF_RESPONSE_MARKER"), + userItem(steerItemId, "STEER_MARKER", "steer"), + assistantItem(TurnItemId.make("fork-steered-reply"), "AFTER_STEER_REPLY_MARKER"), + ]; + yield* eventSink.write({ + events: [ + { + id: EventId.make("steered-provider-thread"), + type: "provider-thread.updated", + threadId: sourceThreadId, + occurredAt: now, + payload: { + id: providerThreadId, + driver, + providerInstanceId: instanceId, + providerSessionId: null, + appThreadId: sourceThreadId, + ownerNodeId: null, + nativeThreadRef: { driver, nativeId: "native-source", strength: "strong" }, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: 1, + lastRunOrdinal: 1, + handoffIds: [], + forkedFrom: null, + createdAt: now, + updatedAt: now, + }, + }, + { + id: EventId.make("steered-run"), + type: "run.created", + threadId: sourceThreadId, + runId, + occurredAt: now, + payload: { + id: runId, + threadId: sourceThreadId, + ordinal: 1, + providerInstanceId: instanceId, + modelSelection, + providerThreadId, + userMessageId: MessageId.make("fork-steered-prompt"), + rootNodeId: null, + activeAttemptId: null, + status: "completed", + queuePosition: null, + requestedAt: now, + startedAt: now, + completedAt: now, + checkpointId: null, + contextHandoffId: null, + }, + }, + ...items.map((item) => ({ + id: EventId.make(`steered-${item.id}`), + type: "turn-item.updated" as const, + threadId: sourceThreadId, + runId, + occurredAt: now, + payload: item, + })), + ], + }); + + const forkAt = (turnItemId: TurnItemId, commandId: string) => + orchestrator.dispatch({ + type: "thread.fork", + commandId: CommandId.make(commandId), + sourceThreadId, + targetThreadId, + sourcePoint: { type: "turn_item", runId, turnItemId }, + createdBy: "user", + creationSource: "web", + }); + const steerFork = yield* Effect.exit(forkAt(steerItemId, "fork-at-steer")); + assert.isTrue(steerFork._tag === "Failure", "a steer is not a response to fork from"); + yield* forkAt(cutOffItemId, "fork-at-cut-off-response"); + + const forked = yield* orchestrator.getThreadProjection(targetThreadId); + const inheritedText = forked.visibleTurnItems + .flatMap(({ item }) => + item.type === "user_message" || item.type === "assistant_message" ? [item.text] : [], + ) + .join("\n"); + assert.equal(inheritedText, "INITIAL_PROMPT_MARKER\nCUT_OFF_RESPONSE_MARKER"); + const shell = (yield* orchestrator.getShellSnapshot()).threads.find( + (thread) => thread.id === targetThreadId, + ); + assert.equal(shell?.visibleItemCount, forked.visibleTurnItems.length); + + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("continue-steered-fork"), + threadId: targetThreadId, + messageId: MessageId.make("continue-steered-fork"), + text: "Continue from the cut-off response", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + const target = yield* orchestrator.getThreadProjection(targetThreadId); + assert.equal(target.contextTransfers[0]?.sourcePoint.turnItemId, cutOffItemId); + if (expectedStrategy === "native_fork") { + assert.lengthOf(target.contextHandoffs, 0); + assert.equal(target.providerThreads[0]?.forkedFrom?.providerThreadId, providerThreadId); + return; + } + assert.equal(target.contextTransfers[0]?.resolution?.strategy, "portable_context"); + const handoff = target.contextHandoffs[0]!; + const history = `${handoff.summaryText}\n${ + handoff.history?.messages.map((message) => message.text).join("\n") ?? "" + }`; + assert.include(history, "CUT_OFF_RESPONSE_MARKER"); + assert.notInclude(history, "STEER_MARKER"); + assert.notInclude(history, "AFTER_STEER_REPLY_MARKER"); + }).pipe(Effect.provide(layer)), +); diff --git a/apps/server/src/orchestration-v2/ThreadForkService.ts b/apps/server/src/orchestration-v2/ThreadForkService.ts index 56c7aa81693d..3ce684d57b23 100644 --- a/apps/server/src/orchestration-v2/ThreadForkService.ts +++ b/apps/server/src/orchestration-v2/ThreadForkService.ts @@ -101,6 +101,9 @@ export const layer: Layer.Layer = Layer.succeed( type: "run", threadId: input.sourceProjection.thread.id, runId: input.sourceRun.id, + ...(input.canonicalSourcePoint.turnItemId === undefined + ? {} + : { throughTurnItemId: input.canonicalSourcePoint.turnItemId }), }, createdAt: input.createdAt, updatedAt: input.createdAt, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 2aeca85e6aa2..4fce6e3bd072 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -74,6 +74,7 @@ import { type ScopedThreadRef, type ThreadId, type ThreadLinkedPullRequest, + type TurnItemId, type RunId, type RuntimeRequestId, type KeybindingCommand, @@ -8651,7 +8652,11 @@ export default function ChatView(props: ChatViewProps) { ); const onForkFromRun = useCallback( - async (input: { readonly sourceThreadId: ThreadId; readonly runId: RunId }) => { + async (input: { + readonly sourceThreadId: ThreadId; + readonly runId: RunId; + readonly turnItemId?: TurnItemId; + }) => { if (!activeThread || activeEnvironmentUnavailable) return; const targetThreadId = newThreadId(); const targetThreadRef = scopeThreadRef(environmentId, targetThreadId); @@ -8661,6 +8666,7 @@ export default function ChatView(props: ChatViewProps) { sourceThreadId: input.sourceThreadId, targetThreadId, runId: input.runId, + ...(input.turnItemId === undefined ? {} : { turnItemId: input.turnItemId }), title: `${activeThread.title} fork`, }, }); diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts index 8ba1552452a5..f059351744c7 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts @@ -4377,6 +4377,112 @@ describe("streaming v2 row projection", () => { }); }); +describe("assistant meta in steered runs", () => { + const runId = RunId.make("steered-run"); + const time = (second: number) => new Date(Date.UTC(2026, 0, 1, 0, 0, second)).toISOString(); + const message = ( + id: string, + second: number, + role: "user" | "assistant", + inputIntent?: "turn_start" | "steer", + ): TimelineEntry => ({ + kind: "message", + id, + createdAt: time(second), + message: { + id: MessageId.make(id), + role, + text: id, + runId, + ...(inputIntent === undefined ? {} : { inputIntent }), + createdAt: time(second), + updatedAt: time(second), + streaming: false, + }, + }); + const createdThread: TimelineEntry = { + kind: "event", + id: "created", + createdAt: time(3), + projectedItem: { + item: { id: "created", type: "thread_created", runId }, + } as OrchestrationV2ProjectedTurnItem, + }; + const settledRows = (timelineEntries: TimelineEntry[], expanded: boolean) => + deriveMessagesTimelineRows({ + timelineEntries, + latestRun: { runId, status: "completed", startedAt: time(0), completedAt: time(20) }, + isWorking: false, + expandedRunIds: expanded ? new Set([runId]) : new Set(), + turnDiffSummaries: [], + supportsConversationRollback: false, + }); + const metaFor = (rows: MessagesTimelineRow[], id: string) => { + const row = rows.find((candidate) => candidate.id === id); + return row?.kind === "message" + ? { meta: row.showAssistantMeta, midRun: row.assistantMidRun === true } + : null; + }; + + it("gives the cut-off response and the final one meta, marking only the cut-off as mid-run", () => { + const rows = settledRows( + [ + message("prompt", 0, "user", "turn_start"), + message("cut-off", 2, "assistant"), + message("steer", 5, "user", "steer"), + message("final", 20, "assistant"), + ], + true, + ); + + expect(metaFor(rows, "cut-off")).toEqual({ meta: true, midRun: true }); + expect(metaFor(rows, "final")).toEqual({ meta: true, midRun: false }); + }); + + it("keeps meta on only the last assistant message of an unsteered run", () => { + const rows = settledRows( + [ + message("prompt", 0, "user", "turn_start"), + message("commentary", 2, "assistant"), + message("final", 20, "assistant"), + ], + true, + ); + + expect(metaFor(rows, "commentary")).toEqual({ meta: false, midRun: false }); + expect(metaFor(rows, "final")).toEqual({ meta: true, midRun: false }); + }); + + it("attaches created-thread summaries once, after the run's final response", () => { + const timelineEntries = [ + message("prompt", 0, "user", "turn_start"), + message("cut-off", 2, "assistant"), + createdThread, + message("steer", 5, "user", "steer"), + message("final", 20, "assistant"), + ]; + + expect(settledRows(timelineEntries, true).map((row) => row.id)).toEqual([ + "prompt", + `turn-fold:${runId}`, + "cut-off", + "created", + "steer", + "final", + "summary:created", + "assistant-meta:final", + ]); + expect(settledRows(timelineEntries, false).map((row) => row.id)).toEqual([ + "prompt", + `turn-fold:${runId}`, + "steer", + "final", + "summary:created", + "assistant-meta:final", + ]); + }); +}); + describe("linked timeline resources", () => { const runId = RunId.make("resource-run"); const event = (id: string, type: "subagent" | "thread_created", eventRunId = runId) => ({ diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.ts b/apps/web/src/components/chat/MessagesTimeline.logic.ts index b4f879e9a995..0380cf054c6e 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.ts @@ -604,6 +604,8 @@ type MessagesTimelineRowContent = durationStart: string; showAssistantMeta: boolean; showAssistantCopyButton: boolean; + /** A later assistant message in this run follows, so forks cut at this one. */ + assistantMidRun?: boolean | undefined; assistantCopyStreaming: boolean; assistantTurnDiffSummary?: TurnDiffSummary | undefined; revertTurnCount?: number | undefined; @@ -615,6 +617,7 @@ type MessagesTimelineRowContent = createdAt: string; message: ChatMessage; showAssistantCopyButton: boolean; + assistantMidRun?: boolean | undefined; assistantCopyStreaming: boolean; } | { @@ -709,8 +712,16 @@ export function resolveAssistantMessageCopyState({ }; } +/** + * The assistant messages that end a stretch of output and so carry metadata: + * each run's last one, plus the last one before a later user message in the + * same run (a steer). `midRunIds` holds the latter; forking from one of them + * must cut inside its run. + */ function deriveTerminalAssistantMessageIds(timelineEntries: ReadonlyArray) { const lastAssistantMessageIdByResponseKey = new Map(); + const lastAssistantMessageIdByRunId = new Map(); + const segmentByRunId = new Map(); let nullTurnResponseIndex = 0; for (const timelineEntry of timelineEntries) { @@ -720,19 +731,31 @@ function deriveTerminalAssistantMessageIds(timelineEntries: ReadonlyArray(); + for (const [responseKey, messageId] of lastAssistantMessageIdByResponseKey) { + if (responseKey.startsWith("turn:") && !runEndIds.has(messageId)) midRunIds.add(messageId); + } + return { terminalIds, midRunIds }; } interface TurnFold { @@ -1222,6 +1245,7 @@ function attachTrailingToolGroupsToAssistant( createdAt: rows[lastTrailingWorkIndex]?.createdAt ?? row.message.updatedAt, message: row.message, showAssistantCopyButton: row.showAssistantCopyButton, + assistantMidRun: row.assistantMidRun, assistantCopyStreaming: row.assistantCopyStreaming, }); } @@ -1315,7 +1339,7 @@ function deriveTimelineTurnFolds( }); return deriveTurnFolds({ timelineEntries, - terminalAssistantMessageIds: deriveTerminalAssistantMessageIds(timelineEntries), + terminalAssistantMessageIds: deriveTerminalAssistantMessageIds(timelineEntries).terminalIds, latestRun: input.latestRun ?? null, unfoldedRunIds: new Set([...activeVisualResponseRunIds, ...failedRunIds]), runlessWorkActive: input.isWorking && input.runlessWorkActive === true, @@ -1361,7 +1385,8 @@ export function deriveMessagesTimelineRows(input: { const durationStartByMessageId = computeMessageDurationStart( timelineEntries.flatMap((entry) => (entry.kind === "message" ? [entry.message] : [])), ); - const terminalAssistantMessageIds = deriveTerminalAssistantMessageIds(timelineEntries); + const { terminalIds: terminalAssistantMessageIds, midRunIds: midRunAssistantMessageIds } = + deriveTerminalAssistantMessageIds(timelineEntries); const unsettledRunId = deriveUnsettledRunId(input.latestRun ?? null, input.runningRunId ?? null); const failedRunIds = failedTimelineRunIds(timelineEntries, input.latestRun ?? null); const liveSubagentEntryIds = liveSubagentCardEntryIds(timelineEntries); @@ -1822,6 +1847,7 @@ export function deriveMessagesTimelineRows(input: { durationStart, showAssistantMeta, showAssistantCopyButton: showAssistantMeta, + assistantMidRun: midRunAssistantMessageIds.has(timelineEntry.message.id), assistantCopyStreaming: timelineEntry.message.streaming || assistantResponseStillInProgress, assistantTurnDiffSummary: timelineEntry.message.role === "assistant" @@ -1989,7 +2015,12 @@ function attachCreatedThreadSummaries( if (terminalIndex !== undefined && (collapsedRuns.has(runId!) || index > terminalIndex)) return []; } - if (row.kind === "message" && row.showAssistantMeta && row.message.runId) { + // A steered run has a meta row per response; summaries follow only its last. + if ( + row.kind === "message" && + row.message.runId && + terminalIndexes.get(row.message.runId) === index + ) { return [ row, ...(createdByRun.get(row.message.runId) ?? []).map((entry) => ({ @@ -2185,6 +2216,7 @@ function isRowUnchanged(a: MessagesTimelineRow, b: MessagesTimelineRow): boolean a.projectedItem === bm.projectedItem && a.message === bm.message && a.showAssistantCopyButton === bm.showAssistantCopyButton && + a.assistantMidRun === bm.assistantMidRun && a.assistantCopyStreaming === bm.assistantCopyStreaming ); } @@ -2275,6 +2307,7 @@ function isRowUnchanged(a: MessagesTimelineRow, b: MessagesTimelineRow): boolean a.durationStart === bm.durationStart && a.showAssistantMeta === bm.showAssistantMeta && a.showAssistantCopyButton === bm.showAssistantCopyButton && + a.assistantMidRun === bm.assistantMidRun && a.assistantCopyStreaming === bm.assistantCopyStreaming && a.assistantTurnDiffSummary === bm.assistantTurnDiffSummary && a.revertTurnCount === bm.revertTurnCount diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index 0518c96afc58..1d38d40a3223 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -33,6 +33,7 @@ import { type RunId, ThreadId, type ToolActivityIcon, + type TurnItemId, } from "@t3tools/contracts"; import { parseScopedThreadKey, scopeThreadRef } from "@t3tools/client-runtime/environment"; import { useAtomValue } from "@effect/atom-react"; @@ -344,6 +345,7 @@ interface TimelineRowSharedState { onForkFromRun: (input: { readonly sourceThreadId: ThreadId; readonly runId: RunId; + readonly turnItemId?: TurnItemId; }) => Promise; onRollbackCheckpoint: (input: { readonly checkpointId: string; @@ -489,6 +491,7 @@ interface MessagesTimelineProps { onForkFromRun: (input: { readonly sourceThreadId: ThreadId; readonly runId: RunId; + readonly turnItemId?: TurnItemId; }) => Promise; onRollbackCheckpoint: (input: { readonly checkpointId: string; @@ -2790,6 +2793,7 @@ function AssistantTimelineRow({ row }: { row: Extract @@ -2801,8 +2805,11 @@ function AssistantTimelineRow({ row }: { row: Extract["projectedItem"]>; + /** A later response in the run follows, so the fork cuts at this item. */ + readonly midRun: boolean; }) { const ctx = use(TimelineRowCtx); const [busy, setBusy] = useState(false); @@ -2831,7 +2838,11 @@ function AssistantForkButton({ onClick={() => { setBusy(true); void ctx - .onForkFromRun({ sourceThreadId: projectedItem.sourceThreadId, runId }) + .onForkFromRun({ + sourceThreadId: projectedItem.sourceThreadId, + runId, + ...(midRun ? { turnItemId: projectedItem.item.id } : {}), + }) .finally(() => setBusy(false)); }} aria-label="Fork from this response" @@ -2856,6 +2867,7 @@ function AssistantMetaTimelineRow({ className="mt-0.5" projectedItem={row.projectedItem} message={row.message} + midRun={row.assistantMidRun === true} showCopyButton={row.showAssistantCopyButton} copyStreaming={row.assistantCopyStreaming} alwaysVisible @@ -2868,6 +2880,7 @@ function AssistantMessageMeta({ className, projectedItem, message, + midRun, showCopyButton, copyStreaming, alwaysVisible = false, @@ -2875,6 +2888,7 @@ function AssistantMessageMeta({ className?: string; projectedItem?: Extract["projectedItem"]; message: ChatMessage; + midRun: boolean; showCopyButton: boolean; copyStreaming: boolean; alwaysVisible?: boolean; @@ -2902,7 +2916,7 @@ function AssistantMessageMeta({ streaming={copyStreaming} /> {projectedItem?.item.type === "assistant_message" ? ( - + ) : null} {!message.streaming && ( diff --git a/docs/orchestration-v2/provider-capability-system.md b/docs/orchestration-v2/provider-capability-system.md index dc544b1af106..27fa4465c082 100644 --- a/docs/orchestration-v2/provider-capability-system.md +++ b/docs/orchestration-v2/provider-capability-system.md @@ -50,6 +50,7 @@ type ThreadCapabilities = { canRollbackThread: boolean; canForkThread: boolean; canForkFromTurn: boolean; + canForkFromItem?: boolean; canForkFromSubagentThread: boolean; exposesNativeThreadId: boolean; }; @@ -57,6 +58,8 @@ type ThreadCapabilities = { Codex can expose strong thread ids and rollback snapshots. Claude may support native forking through its own model/session primitives. Other providers may only support synthetic app forks. +`canForkFromItem` means a native fork can end at one assistant response inside a turn. Active steering keeps the steer in the same turn, so a fork from a response a steer cut off needs it; without it the fork uses portable context cut at that response. Claude sets it because `forkSession` cuts at an SDK message uuid; Codex `thread/fork` cuts only at turn ends. + ## Turn Capabilities ```ts diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index 84f4a2ad3cff..25b57d9e402f 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -24,6 +24,7 @@ import { type RuntimeRequestId, type ThreadId, type ThreadEnvMode, + type TurnItemId, type UploadChatAttachment, } from "@t3tools/contracts"; import { modelSelectionCommandType } from "@t3tools/shared/model"; @@ -215,6 +216,11 @@ export interface ForkThreadFromRunInput extends CommandMetadata { readonly sourceThreadId: ThreadId; readonly targetThreadId: ThreadId; readonly runId: RunId; + /** + * Fork through this response instead of the whole run. Only for a response + * that is not its run's last assistant message, such as one a steer cut off. + */ + readonly turnItemId?: TurnItemId; readonly title?: string; } @@ -944,7 +950,10 @@ export const forkThreadFromRun = Effect.fn("EnvironmentCommands.forkThreadFromRu creationSource: input.creationSource ?? "web", sourceThreadId: input.sourceThreadId, targetThreadId: input.targetThreadId, - sourcePoint: { type: "run", runId: input.runId }, + sourcePoint: + input.turnItemId === undefined + ? { type: "run", runId: input.runId } + : { type: "turn_item", runId: input.runId, turnItemId: input.turnItemId }, ...(input.title === undefined ? {} : { title: input.title }), }); }); diff --git a/packages/client-runtime/src/state/threadWorkflows.test.ts b/packages/client-runtime/src/state/threadWorkflows.test.ts index 23722260b883..a74a46284bc6 100644 --- a/packages/client-runtime/src/state/threadWorkflows.test.ts +++ b/packages/client-runtime/src/state/threadWorkflows.test.ts @@ -385,6 +385,28 @@ describe("thread workflows", () => { ).toBe(false); }); + it("forks from a reply that was stopped mid-stream, but not from a failed one", () => { + const withStatus = (status: string) => + ({ item: { type: "assistant_message", runId: "run", status } }) as never; + + expect(canForkProjectedAssistantItem({ projectedItem: withStatus("interrupted") })).toBe(true); + expect( + canForkProjectedAssistantItem({ + projectedItem: withStatus("interrupted"), + capabilities: capabilities({ portableFork: true }), + }), + ).toBe(true); + expect( + canForkProjectedAssistantItem({ + projectedItem: withStatus("interrupted"), + capabilities: capabilities(), + }), + ).toBe(false); + for (const status of ["failed", "cancelled", "pending", "waiting", "idle"]) { + expect(canForkProjectedAssistantItem({ projectedItem: withStatus(status) })).toBe(false); + } + }); + it("merges the newest provider-finished run while checkpoint capture is pending", () => { const projection = { runs: [ diff --git a/packages/client-runtime/src/state/threadWorkflows.ts b/packages/client-runtime/src/state/threadWorkflows.ts index cb2b4f1fdb4a..9e72fd8cb266 100644 --- a/packages/client-runtime/src/state/threadWorkflows.ts +++ b/packages/client-runtime/src/state/threadWorkflows.ts @@ -167,7 +167,12 @@ export function canForkProjectedAssistantItem(input: { readonly capabilities?: OrchestrationV2ProviderCapabilities | undefined; }): boolean { const item = input.projectedItem.item; - if (item.type !== "assistant_message" || item.runId === null || item.status !== "completed") { + if ( + item.type !== "assistant_message" || + item.runId === null || + // A reply the user stopped mid-stream is still a settled point to fork from. + (item.status !== "completed" && item.status !== "interrupted") + ) { return false; } if (input.capabilities === undefined) { diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 7aa8c74fd80a..5ecba3c7d872 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -135,6 +135,8 @@ export const OrchestrationV2ThreadForkSourcePoint = Schema.Union([ Schema.Struct({ type: Schema.Literal("latest_stable") }), Schema.Struct({ type: Schema.Literal("run"), runId: RunId }), Schema.Struct({ type: Schema.Literal("checkpoint"), checkpointId: CheckpointId }), + /** A finished assistant response; the fork leaves out the rest of its run, such as a steer. */ + Schema.Struct({ type: Schema.Literal("turn_item"), runId: RunId, turnItemId: TurnItemId }), ]); export type OrchestrationV2ThreadForkSourcePoint = typeof OrchestrationV2ThreadForkSourcePoint.Type; @@ -205,6 +207,8 @@ export const OrchestrationV2ThreadCapabilities = Schema.Struct({ canRollbackThread: Schema.Boolean, canForkThread: Schema.Boolean, canForkFromTurn: Schema.Boolean, + // Whether a native fork can end at one item inside a turn, such as a response a steer cut off. + canForkFromItem: Schema.optional(Schema.Boolean), canForkFromSubagentThread: Schema.Boolean, exposesNativeThreadId: Schema.Boolean, }); @@ -383,7 +387,13 @@ export const OrchestrationV2AppThread = Schema.Struct({ lineage: OrchestrationV2AppThreadLineage, forkedFrom: Schema.NullOr( Schema.Union([ - Schema.Struct({ type: Schema.Literal("run"), threadId: ThreadId, runId: RunId }), + Schema.Struct({ + type: Schema.Literal("run"), + threadId: ThreadId, + runId: RunId, + /** Set when the fork cuts inside the run: history ends at this item. */ + throughTurnItemId: Schema.optional(TurnItemId), + }), Schema.Struct({ type: Schema.Literal("node"), nodeId: NodeId }), Schema.Struct({ type: Schema.Literal("provider_thread"), diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 086300e21878..466b0adf8e5d 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -480,6 +480,8 @@ export interface ProviderAdapterV2ForkThreadInput { readonly sourceProviderThread: OrchestrationV2ProviderThread; readonly sourceProviderTurns?: ReadonlyArray; readonly providerTurnId?: ProviderTurnId; + /** Cut inside the provider turn, through this item. Only sent when `canForkFromItem`. */ + readonly throughTurnItem?: OrchestrationV2TurnItem; readonly targetThreadId: ThreadId; readonly ownerNodeId?: NodeId; readonly modelSelection?: ModelSelection; From 19e2bf3cf9ee48b7be346c3bb67281ef24f9b98c Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Sat, 10 Oct 2026 01:57:55 -0700 Subject: [PATCH 2/3] fix(orchestration): a steer with no reply after it still cuts the fork A response counted as mid-run only when a later assistant message of its run followed. A run of response, steer and then only tool calls (or a stop) left the response as the run's last assistant message, so Fork sent the run point and the fork kept the steer. Web and mobile now mark a response mid-run when any later user message of the same run follows it. A turn_item point at the run's last response cut nothing but still recorded a cut. For Claude that could pass a result message uuid, which is no transcript cursor, as upToMessageId. The server now forks such a point exactly like its run: no cut on forkedFrom or the transfer, and native forks wherever the provider forks at turn ends. Co-Authored-By: Claude Opus 5.5 --- apps/mobile/src/lib/threadActivity.test.ts | 11 ++++++ apps/mobile/src/lib/threadActivity.ts | 25 ++++++++----- .../src/orchestration-v2/Orchestrator.ts | 28 ++++++++------- .../ThreadFork.execution.test.ts | 35 +++++++++++++++++++ .../chat/MessagesTimeline.logic.test.ts | 13 +++++++ .../components/chat/MessagesTimeline.logic.ts | 17 +++++---- 6 files changed, 102 insertions(+), 27 deletions(-) diff --git a/apps/mobile/src/lib/threadActivity.test.ts b/apps/mobile/src/lib/threadActivity.test.ts index 8aa80923e52e..6cd4549d4e24 100644 --- a/apps/mobile/src/lib/threadActivity.test.ts +++ b/apps/mobile/src/lib/threadActivity.test.ts @@ -2067,6 +2067,17 @@ describe("terminalFeedAssistantMessageIds", () => { expect([...ids.midRunIds]).toEqual(["cut-off"]); }); + it("marks a response as mid-run when a steer follows it with no reply yet", () => { + const ids = terminalFeedAssistantMessageIds([ + message("prompt", "user", "turn_start"), + message("cut-off", "assistant"), + message("steer", "user", "steer"), + ]); + + expect([...ids.terminalIds]).toEqual(["cut-off"]); + expect([...ids.midRunIds]).toEqual(["cut-off"]); + }); + it("keeps one terminal message for an unsteered run", () => { const ids = terminalFeedAssistantMessageIds([ message("prompt", "user", "turn_start"), diff --git a/apps/mobile/src/lib/threadActivity.ts b/apps/mobile/src/lib/threadActivity.ts index 5f7288dc14b2..9e2529313a72 100644 --- a/apps/mobile/src/lib/threadActivity.ts +++ b/apps/mobile/src/lib/threadActivity.ts @@ -1051,25 +1051,32 @@ export function failedFeedRunIds( /** * The assistant messages that end a stretch of output and so carry metadata: * each run's last one, plus the last one before a later user message in the - * same run (a steer). Forking from one in `midRunIds` must cut inside its run. + * same run (a steer). `midRunIds` holds those a user message of the same run + * follows, replied to or not; forking from one must cut inside its run. */ export function terminalFeedAssistantMessageIds(feed: ReadonlyArray) { - const lastIdBySegment = new Map(); - const lastIdByRun = new Map(); + const lastBySegment = new Map< + string, + { readonly runId: RunId; readonly segment: number; readonly id: string } + >(); const segmentByRun = new Map(); for (const entry of feed) { if (entry.type !== "message" || !entry.message.runId) continue; const runId = entry.message.runId; + const segment = segmentByRun.get(runId) ?? 0; if (entry.message.role === "user") { - segmentByRun.set(runId, (segmentByRun.get(runId) ?? 0) + 1); + segmentByRun.set(runId, segment + 1); continue; } - lastIdByRun.set(runId, entry.message.id); - lastIdBySegment.set(`${runId}:${segmentByRun.get(runId) ?? 0}`, entry.message.id); + lastBySegment.set(`${runId}:${segment}`, { runId, segment, id: entry.message.id }); } - const terminalIds = new Set(lastIdBySegment.values()); - const runEndIds = new Set(lastIdByRun.values()); - const midRunIds = new Set([...terminalIds].filter((id) => !runEndIds.has(id))); + const terminals = [...lastBySegment.values()]; + const terminalIds = new Set(terminals.map((terminal) => terminal.id)); + const midRunIds = new Set( + terminals + .filter((terminal) => terminal.segment < (segmentByRun.get(terminal.runId) ?? 0)) + .map((terminal) => terminal.id), + ); return { terminalIds, midRunIds }; } diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index e369c42232c3..96b7cff06569 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -3576,23 +3576,27 @@ 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 throughItem = yield* projectionStore - .getTurnItem({ threadId: command.sourceThreadId, itemId: throughTurnItemId }) - .pipe( - Effect.mapError( - (cause) => new OrchestratorProjectionError({ threadId: command.sourceThreadId, cause }), - ), - ); - if (!isForkableResponseItem(throughItem, sourceRun)) { + let throughTurnItemId: TurnItemId | undefined; + if (command.sourcePoint.type === "turn_item") { + const pointItemId = command.sourcePoint.turnItemId; + const runItems = yield* readHandoffItems(command.sourceThreadId, [sourceRun.id]); + const throughItem = runItems.find((item) => item.id === pointItemId) ?? null; + if (throughItem === null || !isForkableResponseItem(throughItem, sourceRun)) { 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}.`, + cause: `Fork source item ${pointItemId} is not a finished assistant response of run ${sourceRun.id}.`, }); } + // Cutting at the run's last response changes nothing, so it forks like + // the run. That also keeps item ids that are no native cursor (a Claude + // turn's result fallback, always its last response) out of native forks. + const conversationContinues = runItems.some( + (item) => + item.ordinal > throughItem.ordinal && + (item.type === "user_message" || item.type === "assistant_message"), + ); + throughTurnItemId = conversationContinues ? pointItemId : undefined; } const sourceProviderThread = providerThreadForRun(sourceProjection, sourceRun); const now = command.createdAt ?? (yield* DateTime.now); diff --git a/apps/server/src/orchestration-v2/ThreadFork.execution.test.ts b/apps/server/src/orchestration-v2/ThreadFork.execution.test.ts index 7cf81af52d3b..92041c65dd04 100644 --- a/apps/server/src/orchestration-v2/ThreadFork.execution.test.ts +++ b/apps/server/src/orchestration-v2/ThreadFork.execution.test.ts @@ -420,6 +420,41 @@ it.effect.each(steeredForkCases)( }); const steerFork = yield* Effect.exit(forkAt(steerItemId, "fork-at-steer")); assert.isTrue(steerFork._tag === "Failure", "a steer is not a response to fork from"); + + // The run's last response cuts nothing, so it forks exactly like the run, + // natively where the provider can fork at a turn end. + const runEndTargetId = ThreadId.make("fork-steered-run-end-target"); + yield* orchestrator.dispatch({ + type: "thread.fork", + commandId: CommandId.make("fork-at-run-end-response"), + sourceThreadId, + targetThreadId: runEndTargetId, + sourcePoint: { type: "turn_item", runId, turnItemId: items[3]!.id }, + createdBy: "user", + creationSource: "web", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("continue-run-end-fork"), + threadId: runEndTargetId, + messageId: MessageId.make("continue-run-end-fork"), + text: "Continue from the run end", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + const runEndFork = yield* orchestrator.getThreadProjection(runEndTargetId); + assert.deepEqual(runEndFork.thread.forkedFrom, { + type: "run", + threadId: sourceThreadId, + runId, + }); + assert.isUndefined(runEndFork.contextTransfers[0]?.sourcePoint.turnItemId); + assert.lengthOf(runEndFork.contextHandoffs, 0); + assert.equal(runEndFork.providerThreads[0]?.forkedFrom?.providerThreadId, providerThreadId); + yield* forkAt(cutOffItemId, "fork-at-cut-off-response"); const forked = yield* orchestrator.getThreadProjection(targetThreadId); diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts index f059351744c7..1ee7d5374c6b 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts @@ -4439,6 +4439,19 @@ describe("assistant meta in steered runs", () => { expect(metaFor(rows, "final")).toEqual({ meta: true, midRun: false }); }); + it("marks a response as mid-run when a steer follows it with no reply yet", () => { + const rows = settledRows( + [ + message("prompt", 0, "user", "turn_start"), + message("cut-off", 2, "assistant"), + message("steer", 5, "user", "steer"), + ], + true, + ); + + expect(metaFor(rows, "cut-off")).toEqual({ meta: true, midRun: true }); + }); + it("keeps meta on only the last assistant message of an unsteered run", () => { const rows = settledRows( [ diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.ts b/apps/web/src/components/chat/MessagesTimeline.logic.ts index 0380cf054c6e..f259ec93a8cb 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.ts @@ -720,7 +720,7 @@ export function resolveAssistantMessageCopyState({ */ function deriveTerminalAssistantMessageIds(timelineEntries: ReadonlyArray) { const lastAssistantMessageIdByResponseKey = new Map(); - const lastAssistantMessageIdByRunId = new Map(); + const runSegmentByMessageId = new Map(); const segmentByRunId = new Map(); let nullTurnResponseIndex = 0; @@ -740,20 +740,25 @@ function deriveTerminalAssistantMessageIds(timelineEntries: ReadonlyArray(); - for (const [responseKey, messageId] of lastAssistantMessageIdByResponseKey) { - if (responseKey.startsWith("turn:") && !runEndIds.has(messageId)) midRunIds.add(messageId); + for (const messageId of terminalIds) { + const position = runSegmentByMessageId.get(messageId); + if (position && position.segment < (segmentByRunId.get(position.runId) ?? 0)) { + midRunIds.add(messageId); + } } return { terminalIds, midRunIds }; } From 59324bbd88c3b97c8188d9c64b8ab0ccfaf444b8 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Sat, 10 Oct 2026 21:53:17 -0700 Subject: [PATCH 3/3] fix(web): a chat created after a steer stays where it happened Co-Authored-By: Claude Opus 5.5 --- .../chat/MessagesTimeline.logic.test.ts | 16 ++++++++++++++++ .../components/chat/MessagesTimeline.logic.ts | 9 ++++++++- 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts index 4c07c29fcc0a..65b684970799 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts @@ -4622,6 +4622,22 @@ describe("assistant meta in steered runs", () => { "assistant-meta:final", ]); }); + + it("leaves a created thread in place when no response follows the steer that made it", () => { + const timelineEntries = [ + message("prompt", 0, "user", "turn_start"), + message("cut-off", 2, "assistant"), + message("steer", 5, "user", "steer"), + { ...createdThread, createdAt: time(8) }, + ]; + + expect(settledRows(timelineEntries, true).map((row) => row.id)).toEqual([ + "prompt", + "cut-off", + "steer", + "created", + ]); + }); }); describe("linked timeline resources", () => { diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.ts b/apps/web/src/components/chat/MessagesTimeline.logic.ts index 899820eb5b26..7bfa9adfceb1 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.ts @@ -2054,7 +2054,14 @@ function attachCreatedThreadSummaries( const collapsedRuns = new Set(); const createdByRun = new Map>>(); for (const [index, row] of rows.entries()) { - if (row.kind === "message" && row.showAssistantMeta && row.message.runId) { + // A response a steer cut off is not the run's answer, even when no reply + // follows the steer yet; created chats then stay where they happened. + if ( + row.kind === "message" && + row.showAssistantMeta && + row.assistantMidRun !== true && + row.message.runId + ) { terminalIndexes.set(row.message.runId, index); } if (row.kind === "turn-fold" && !row.expanded) collapsedRuns.add(row.runId);