diff --git a/apps/mobile/src/features/threads/ThreadFeed.tsx b/apps/mobile/src/features/threads/ThreadFeed.tsx index a0cb6ee33962..02d56a475b26 100644 --- a/apps/mobile/src/features/threads/ThreadFeed.tsx +++ b/apps/mobile/src/features/threads/ThreadFeed.tsx @@ -18,7 +18,10 @@ import { useViewabilityAmount, type LegendListRef } from "@legendapp/list/react- import { scopeThreadRef } from "@t3tools/client-runtime/environment"; import type { HandoffEndpoints } from "@t3tools/client-runtime/handoff"; import { resolveUserMessagePresentation } from "@t3tools/client-runtime/user-message"; -import { canForkProjectedAssistantItem } from "@t3tools/client-runtime/state/thread-workflows"; +import { + canForkProjectedAssistantItem, + forkCutTurnItemId, +} from "@t3tools/client-runtime/state/thread-workflows"; import { type OrchestrationMessageContext, ThreadId, @@ -171,6 +174,7 @@ import { } from "../../lib/markdownLinks"; import { failedFeedRunIds, + terminalFeedAssistantMessageIds, deriveThreadFeedPresentation, threadFeedRunIsUnsettled, isContextCompactionActivityGroup, @@ -357,8 +361,12 @@ function UserMessageCopyButton(props: { function AssistantForkButton(props: { readonly environmentId: EnvironmentId; + /** The thread the response shows in, which may be a fork of its source. */ + readonly threadId: ThreadId; 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({ @@ -384,6 +392,17 @@ function AssistantForkButton(props: { disabled={busy} onPress={() => { const targetThreadId = ThreadId.make(uuidv4()); + const turnItemId = forkCutTurnItemId({ + projectedItem: props.projectedItem, + midRun: props.midRun, + threadId: props.threadId, + readForkedFrom: (threadId) => + appAtomRegistry.get( + environmentThreadShells.threadShellAtom( + scopeThreadRef(props.environmentId, threadId), + ), + )?.forkedFrom, + }); setBusy(true); void Haptics.selectionAsync(); void forkFromRun({ @@ -392,6 +411,7 @@ function AssistantForkButton(props: { sourceThreadId: props.projectedItem.sourceThreadId, targetThreadId, runId, + ...(turnItemId === undefined ? {} : { turnItemId }), title: `${props.sourceTitle} fork`, creationSource: "mobile", }, @@ -978,7 +998,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; @@ -1146,7 +1166,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; @@ -1421,8 +1441,10 @@ function renderFeedEntry( {message.projectedItem ? ( ) : null} @@ -2195,15 +2217,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 ca1638b7fbed..37e3ce8d7258 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, @@ -2058,6 +2059,67 @@ 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("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"), + message("commentary", "assistant"), + message("final", "assistant"), + ]); + + expect([...ids.terminalIds]).toEqual(["final"]); + expect(ids.midRunIds.size).toBe(0); + }); +}); + describe("subagent continued by a later turn", () => { const run2 = RunId.make("run-2"); const at = (second: number) => `2026-10-09T10:00:${String(second).padStart(2, "0")}.000Z`; diff --git a/apps/mobile/src/lib/threadActivity.ts b/apps/mobile/src/lib/threadActivity.ts index fb8bbe29aaa9..68e1af7768a3 100644 --- a/apps/mobile/src/lib/threadActivity.ts +++ b/apps/mobile/src/lib/threadActivity.ts @@ -1038,6 +1038,38 @@ 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). `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 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, segment + 1); + continue; + } + lastBySegment.set(`${runId}:${segment}`, { runId, segment, id: entry.message.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 }; +} + function deriveThreadFeedRunFolds( feed: ReadonlyArray, latestRun: ThreadFeedLatestRun | null, 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 b4991241e166..da38100e7f81 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -44,6 +44,7 @@ import { RunAttemptId, RunId, ThreadId, + TurnItemId, } from "@t3tools/contracts"; import { assert, describe, it } from "@effect/vitest"; import * as Clock from "effect/Clock"; @@ -1867,6 +1868,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", @@ -2342,6 +2380,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { readonly subagentLaunchToolUseId?: string; readonly nativeSessionId?: string; readonly settings?: ClaudeSettings; + readonly forkSession?: ClaudeAdapterV2.ClaudeAgentSdkQueryRunnerShape["forkSession"]; readonly freshQueueOnReopen?: boolean; }) => Effect.gen(function* () { @@ -2461,7 +2500,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { close: options?.close?.(sdkMessages) ?? Effect.void, }; }), - forkSession: () => Effect.die("unused forkSession"), + forkSession: options?.forkSession ?? (() => Effect.die("unused forkSession")), subagentLaunchToolUseId: () => Effect.succeed(options?.subagentLaunchToolUseId ?? null), assertComplete: Effect.void, }, @@ -2933,6 +2972,75 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + // The update reuses the Thought's item id, which is no SDK message uuid; a + // fork cut at it must still pass forkSession a uuid from the transcript. + it.effect("forks at a thinking update through the snapshot that held it", () => + Effect.gen(function* () { + const forkCalls: Array = []; + const harness = yield* makeWakeHarnessWithOptions({ + // Forking closes the live CLI process first. + close: (sdkMessages) => Queue.shutdown(sdkMessages), + forkSession: (input) => + Effect.sync(() => { + forkCalls.push(input.options); + return { sessionId: "forked-thinking-session" }; + }), + }); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("thinking-update-fork-attempt"), + text: "Check the memory file", + attachments: [], + }), + ); + const snapshot = (uuid: string, content: ReadonlyArray) => + claudeSdkFrame({ + type: "assistant", + uuid, + session_id: WAKE_NATIVE_SESSION, + parent_tool_use_id: null, + message: { id: "reply-message", model: "claude-opus-5-5", content }, + }); + for (const frame of [ + snapshot("thinking-snapshot-0", [{ type: "thinking", thinking: plan, signature: "sig" }]), + snapshot("thinking-snapshot-1", [{ type: "thinking", thinking: update, signature: "sig" }]), + snapshot("tool-snapshot", [ + { type: "tool_use", id: "tool-1", name: "Bash", input: { command: "ls" } }, + ]), + makeResultFrame({ uuid: "thinking-update-fork-result", result: "" }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* Queue.take(harness.terminalReceipts); + const updateItem = harness.events.findLast( + (event) => + event.type === "turn_item.updated" && event.turnItem.type === "assistant_message", + ); + assert( + updateItem?.type === "turn_item.updated" && + updateItem.turnItem.type === "assistant_message", + ); + assert.equal(updateItem.turnItem.text, update); + assert.equal(updateItem.turnItem.nativeItemRef?.nativeId, "thinking-snapshot-1"); + + yield* harness.runtime.forkThread({ + sourceProviderThread: harness.providerThread, + sourceProviderTurns: [], + throughTurnItem: updateItem.turnItem, + targetThreadId: ThreadId.make("thread-claude-thinking-update-fork"), + }); + assert.deepEqual(forkCalls, [{ dir: "/workspace", upToMessageId: "thinking-snapshot-1" }]); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); + it.effect.each(["cancelled", "denied", "permission_denied", undefined])( "preserves native tool non-execution metadata %s without inferring a denial from text", (kind) => diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 897ac19221df..0c2c42b36f36 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -269,6 +269,7 @@ export const ClaudeProviderCapabilitiesV2 = { canRollbackThread: true, canForkThread: true, canForkFromTurn: true, + canForkFromItem: true, canForkFromSubagentThread: false, exposesNativeThreadId: true, }, @@ -1439,6 +1440,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; } @@ -2828,6 +2848,8 @@ function buildAssistantArtifacts(input: { readonly turnInput: ProviderAdapter.ProviderAdapterV2TurnInput; readonly providerTurnId: OrchestrationV2ProviderTurn["id"]; readonly nativeItemId: string; + /** The SDK message uuid, when the item id is derived from something else. */ + readonly nativeMessageUuid?: string | undefined; readonly text: string; readonly ordinal: number; readonly startedAt: DateTime.Utc; @@ -2849,9 +2871,11 @@ function buildAssistantArtifacts(input: { driver: CLAUDE_PROVIDER, nativeItemId: input.nativeItemId, }); + // A fork cut at this item passes the native id to forkSession as + // upToMessageId, so it must be an SDK message uuid. const nativeItemRef = { driver: CLAUDE_PROVIDER, - nativeId: input.nativeItemId, + nativeId: input.nativeMessageUuid ?? input.nativeItemId, strength: "strong" as const, }; @@ -2975,6 +2999,8 @@ interface ActiveClaudeTurnContext { lastThinking: { readonly messageId: string; readonly itemId: string; + // The SDK snapshot that held the block, a valid forkSession cut. + readonly snapshotUuid: string; readonly text: string; readonly followsThinking: boolean; } | null; @@ -5881,6 +5907,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const emitAssistantTextArtifacts = Effect.fnUntraced(function* (input: { readonly context: ActiveClaudeTurnContext; readonly nativeItemId: string; + readonly nativeMessageUuid?: string; readonly text: string; }) { if (input.context.assistant.emittedNativeItemIds.has(input.nativeItemId)) { @@ -5894,6 +5921,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( turnInput: input.context.input, providerTurnId: input.context.providerTurnId, nativeItemId: input.nativeItemId, + nativeMessageUuid: input.nativeMessageUuid, text: input.text, ordinal, startedAt: now, @@ -6628,9 +6656,12 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( update.followsThinking && claudeWritesUpdatesAsThinking(message.message.model) ) { + // Cutting a fork at its snapshot keeps the thinking and drops + // the tool call, which has no result yet. yield* emitAssistantTextArtifacts({ context, nativeItemId: update.itemId, + nativeMessageUuid: update.snapshotUuid, text: update.text, }); } @@ -6652,6 +6683,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ? { messageId: message.message.id, itemId, + snapshotUuid: message.uuid, text: completedText, followsThinking: previous !== null && previous.messageId === message.message.id, } diff --git a/apps/server/src/orchestration-v2/CommandPolicy.ts b/apps/server/src/orchestration-v2/CommandPolicy.ts index 1098c72c3072..401c12a16b37 100644 --- a/apps/server/src/orchestration-v2/CommandPolicy.ts +++ b/apps/server/src/orchestration-v2/CommandPolicy.ts @@ -196,6 +196,7 @@ export interface CommandPolicyV2Shape { readonly ensureNativeFork: ( input: CapabilityCheckInput & { readonly fromSpecificTurn: boolean; + readonly fromSpecificItem?: boolean; }, ) => Effect.Effect; readonly decideForkExecution: ( @@ -204,6 +205,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: ( @@ -290,6 +293,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( @@ -370,6 +378,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 8369af77fdb6..9351ec427638 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -623,6 +623,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( @@ -683,6 +684,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 { @@ -3586,6 +3619,28 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: forkableSourceRunStatusError(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 ${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); const emitEvent = emit(events, command); @@ -3601,7 +3656,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 }), @@ -3683,6 +3741,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({ @@ -5704,6 +5769,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"; @@ -5717,6 +5783,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio providerInstanceId: modelSelection.instanceId, capabilities, fromSpecificTurn: sourceRun !== null, + fromSpecificItem: pendingForkTransfer?.sourcePoint.turnItemId !== undefined, }), ); } @@ -5760,12 +5827,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 1a1cdff54b23..089a4c9c69f7 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -981,6 +981,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; @@ -1306,14 +1307,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 @@ -1337,12 +1367,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 }) + ); }), ); @@ -1361,6 +1393,7 @@ function buildVisibleTurnItems(input: { const inherited = visibleTurnItemsThroughRun({ sourceProjection: input.sourceProjection, sourceRunId: forkedFrom.runId, + throughTurnItemId: forkedFrom.throughTurnItemId, }); const markerItem = makeForkMarkerTurnItem({ targetProjection: input.projection, @@ -1616,6 +1649,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; @@ -1711,7 +1745,8 @@ function visibleItemCountForShell(input: { return ( inheritedPrefixCount + - itemCountThroughRun({ state: sourceState, runId: forkedFrom.runId }) + + itemCountThroughRun({ state: sourceState, runId: forkedFrom.runId }) - + state.forkedFromItemsAfterCut + 1 + state.itemCount ); @@ -3414,14 +3449,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) @@ -4942,12 +4983,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, @@ -4958,6 +5004,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" })); @@ -5226,6 +5277,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, @@ -5818,6 +5881,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 36354f8c1f4e..f117ecde5ebe 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -611,6 +611,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, @@ -621,6 +636,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 98a16831b3d7..4b4c6f057db6 100644 --- a/apps/server/src/orchestration-v2/ThreadFork.execution.test.ts +++ b/apps/server/src/orchestration-v2/ThreadFork.execution.test.ts @@ -247,3 +247,252 @@ 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"); + + // 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); + 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 a7e711075904..fe8ff62b31b8 100644 --- a/apps/server/src/orchestration-v2/ThreadForkService.ts +++ b/apps/server/src/orchestration-v2/ThreadForkService.ts @@ -97,6 +97,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 d5599f8c46a3..76393c030d79 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -75,6 +75,7 @@ import { type ScopedThreadRef, type ThreadId, type ThreadLinkedPullRequest, + type TurnItemId, type RunId, type RuntimeRequestId, type KeybindingCommand, @@ -8226,6 +8227,7 @@ export default function ChatView(props: ChatViewProps) { const onForkFromRun = async (input: { readonly sourceThreadId: ThreadId; readonly runId: RunId; + readonly turnItemId?: TurnItemId; }) => { if (!activeThread || activeEnvironmentUnavailable) return; const targetThreadId = newThreadId(); @@ -8236,6 +8238,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 a8f77bc0efd2..d52ef6ce2c09 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts @@ -4515,6 +4515,141 @@ 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("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( + [ + 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", + ]); + }); + + 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", () => { 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 a602e954de23..cff2eb18c12a 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.ts @@ -565,6 +565,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; @@ -576,6 +578,7 @@ type MessagesTimelineRowContent = createdAt: string; message: ChatMessage; showAssistantCopyButton: boolean; + assistantMidRun?: boolean | undefined; assistantCopyStreaming: boolean; } | { @@ -670,8 +673,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 runSegmentByMessageId = new Map(); + const segmentByRunId = new Map(); let nullTurnResponseIndex = 0; for (const timelineEntry of timelineEntries) { @@ -681,19 +692,36 @@ function deriveTerminalAssistantMessageIds(timelineEntries: ReadonlyArray(); + 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 }; } interface TurnFold { @@ -1212,6 +1240,7 @@ function attachTrailingToolGroupsToAssistant( createdAt: rows[lastTrailingWorkIndex]?.createdAt ?? row.message.updatedAt, message: row.message, showAssistantCopyButton: row.showAssistantCopyButton, + assistantMidRun: row.assistantMidRun, assistantCopyStreaming: row.assistantCopyStreaming, }); } @@ -1304,7 +1333,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, @@ -1349,7 +1378,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); @@ -1803,6 +1833,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" @@ -1925,7 +1956,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); @@ -1951,7 +1989,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) => ({ @@ -2143,6 +2186,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 ); } @@ -2233,6 +2277,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 e2a3b9dfd600..0c9a2a112b9b 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -32,13 +32,17 @@ 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"; import { environmentThreadDetails } from "../../state/threads"; import { resolveUserMessagePresentation } from "@t3tools/client-runtime/user-message"; import { Link } from "@tanstack/react-router"; -import { canForkProjectedAssistantItem } from "@t3tools/client-runtime/state/thread-workflows"; +import { + canForkProjectedAssistantItem, + forkCutTurnItemId, +} from "@t3tools/client-runtime/state/thread-workflows"; import { notificationChildThreadId } from "@t3tools/client-runtime/state/thread-execution"; import { replaceComposerContextReferences } from "@t3tools/shared/composerContextReferences"; import { @@ -178,7 +182,7 @@ import { HtmlRenderFrame } from "./HtmlRenderFrame"; import { McpAppFrame } from "./McpAppFrame"; import { ChangedFilesCard } from "./ChangedFilesTree"; import { useFileContextMenuHandler } from "../../fileContextMenu"; -import { useProject, useThreadShell } from "../../state/entities"; +import { readThreadShell, useProject, useThreadShell } from "../../state/entities"; import { CHAT_TIMELINE_ANCHOR_OFFSET, readTimelinePosition, @@ -331,6 +335,7 @@ interface TimelineRowSharedState { onForkFromRun: (input: { readonly sourceThreadId: ThreadId; readonly runId: RunId; + readonly turnItemId?: TurnItemId; }) => Promise; onRollbackCheckpoint: (input: { readonly checkpointId: string; @@ -457,6 +462,7 @@ interface MessagesTimelineProps { onForkFromRun: (input: { readonly sourceThreadId: ThreadId; readonly runId: RunId; + readonly turnItemId?: TurnItemId; }) => Promise; onRollbackCheckpoint: (input: { readonly checkpointId: string; @@ -2793,6 +2799,7 @@ function AssistantTimelineRow({ row }: { row: Extract @@ -2804,8 +2811,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); @@ -2833,8 +2843,26 @@ function AssistantForkButton({ disabled={busy} onClick={() => { setBusy(true); + const threadRef = ctx.threadRef; + const turnItemId = + threadRef === null + ? midRun + ? projectedItem.item.id + : undefined + : forkCutTurnItemId({ + projectedItem, + midRun, + threadId: threadRef.threadId, + readForkedFrom: (threadId) => + readThreadShell(scopeThreadRef(threadRef.environmentId, threadId)) + ?.forkedFrom, + }); void ctx - .onForkFromRun({ sourceThreadId: projectedItem.sourceThreadId, runId }) + .onForkFromRun({ + sourceThreadId: projectedItem.sourceThreadId, + runId, + ...(turnItemId === undefined ? {} : { turnItemId }), + }) .finally(() => setBusy(false)); }} aria-label="Fork from this response" @@ -2859,6 +2887,7 @@ function AssistantMetaTimelineRow({ className="mt-0.5" projectedItem={row.projectedItem} message={row.message} + midRun={row.assistantMidRun === true} showCopyButton={row.showAssistantCopyButton} copyStreaming={row.assistantCopyStreaming} alwaysVisible @@ -2871,6 +2900,7 @@ function AssistantMessageMeta({ className, projectedItem, message, + midRun, showCopyButton, copyStreaming, alwaysVisible = false, @@ -2878,6 +2908,7 @@ function AssistantMessageMeta({ className?: string; projectedItem?: Extract["projectedItem"]; message: ChatMessage; + midRun: boolean; showCopyButton: boolean; copyStreaming: boolean; alwaysVisible?: boolean; @@ -2905,7 +2936,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 b16a2f5b4895..3ad23924f8b5 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -25,6 +25,7 @@ import { type RuntimeRequestId, type ThreadId, type ThreadEnvMode, + type TurnItemId, type UploadChatAttachment, } from "@t3tools/contracts"; import { modelSelectionCommandType } from "@t3tools/shared/model"; @@ -210,6 +211,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; } @@ -945,7 +951,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..353189e2c9ae 100644 --- a/packages/client-runtime/src/state/threadWorkflows.test.ts +++ b/packages/client-runtime/src/state/threadWorkflows.test.ts @@ -5,6 +5,7 @@ import { canDetachThreadProviderSession, canForkProjectedAssistantItem, deriveThreadQueueWorkflowState, + forkCutTurnItemId, resolveLatestMergeBackRun, threadSupportsProviderHandoff, } from "./threadWorkflows.ts"; @@ -385,6 +386,60 @@ describe("thread workflows", () => { ).toBe(false); }); + it("forks a fork's cut response at that response, at every fork depth", () => { + // Source S's run R: prompt, A, steer, B. F1 forks at A; F2 forks F1's own run. + const forkedFrom: Record = { + S: null, + F1: { type: "run", threadId: "S", runId: "R", throughTurnItemId: "A" }, + F2: { type: "run", threadId: "F1", runId: "R1" }, + WHOLE: { type: "run", threadId: "S", runId: "R" }, + }; + const response = (sourceItemId: string) => + ({ + sourceThreadId: "S", + sourceItemId, + item: { id: sourceItemId, type: "assistant_message", runId: "R", status: "completed" }, + }) as never; + const cut = (threadId: string, sourceItemId: string, midRun = false) => + forkCutTurnItemId({ + projectedItem: response(sourceItemId), + midRun, + threadId: threadId as never, + readForkedFrom: (id) => forkedFrom[id] as never, + }); + + // In a fork, A ends its run, but forking there must not bring back the steer. + expect(cut("F1", "A")).toBe("A"); + expect(cut("F2", "A")).toBe("A"); + // The source's own last response and a whole-run fork's copy fork the run. + expect(cut("S", "B")).toBeUndefined(); + expect(cut("WHOLE", "B")).toBeUndefined(); + // A response a steer cut off cuts at itself wherever it shows. + expect(cut("S", "A", true)).toBe("A"); + }); + + 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 69e691840b89..df32b08a061a 100644 --- a/packages/client-runtime/src/state/threadWorkflows.ts +++ b/packages/client-runtime/src/state/threadWorkflows.ts @@ -1,8 +1,11 @@ import type { ChatAttachment, + OrchestrationV2AppThread, OrchestrationV2ProjectedTurnItem, OrchestrationV2ProviderCapabilities, OrchestrationV2ThreadProjection, + ThreadId, + TurnItemId, } from "@t3tools/contracts"; import { copySorted } from "@t3tools/shared/Array"; @@ -165,7 +168,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) { @@ -182,6 +190,38 @@ export function canForkProjectedAssistantItem(input: { return canForkNatively || capabilities.context.supportsFullThreadHandoff; } +/** + * The item a Fork on this response cuts at, or undefined to fork its whole run. + * A response a steer cut off cuts at itself. So does the response a fork was + * cut at: in the fork, and in forks of it, it looks like its run's last + * response, but the source run goes on past it. `readForkedFrom` looks up a + * thread's `forkedFrom`; the walk starts at the thread the response shows in. + */ +export function forkCutTurnItemId(input: { + readonly projectedItem: OrchestrationV2ProjectedTurnItem; + readonly midRun: boolean; + readonly threadId: ThreadId; + readonly readForkedFrom: ( + threadId: ThreadId, + ) => OrchestrationV2AppThread["forkedFrom"] | undefined; +}): TurnItemId | undefined { + if (input.midRun) return input.projectedItem.item.id; + const seen = new Set(); + let threadId = input.threadId; + while (!seen.has(threadId)) { + seen.add(threadId); + const forkedFrom = input.readForkedFrom(threadId); + if (forkedFrom?.type !== "run") return undefined; + if (forkedFrom.threadId === input.projectedItem.sourceThreadId) { + return forkedFrom.throughTurnItemId === input.projectedItem.sourceItemId + ? input.projectedItem.item.id + : undefined; + } + threadId = forkedFrom.threadId; + } + return undefined; +} + export function canDetachThreadProviderSession(projection: Projection): boolean { const session = resolveThreadProviderSession(projection); return session !== null && session.status !== "stopped" && session.status !== "error"; diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index fa4b7738eaae..5dfb5a10238a 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, }); @@ -372,7 +376,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 1cc66c77f9cd..3653bd3a5d4a 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -479,6 +479,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;