diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index cef5d9358ca94..7e5ff68977313 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -226,6 +226,7 @@ export const make = Effect.gen(function* () { inlineMessageContext: true, requiredWorktreeBootstrap: true, threadSettlement: true, + threadForkFromItem: true, threadAutoSettlement: true, storageCleanup: true, storageCleanupRun: true, diff --git a/apps/server/src/mcp/toolkits/thread/tools.ts b/apps/server/src/mcp/toolkits/thread/tools.ts index b5fb7d6be3a68..c0d5a5d7bd512 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, checkpoint, or completed/interrupted assistant response using the existing fork command. Omit threadId to fork this thread. The fork inherits the source configuration. Acceptance does not mean a provider turn has completed.", parameters: Schema.Struct({ threadId: Schema.optional(ThreadId), sourcePoint: OrchestrationV2ThreadForkSourcePoint, diff --git a/apps/server/src/orchestration-v2/CommandPolicy.ts b/apps/server/src/orchestration-v2/CommandPolicy.ts index 923542d877dfe..e4a3a488f7e03 100644 --- a/apps/server/src/orchestration-v2/CommandPolicy.ts +++ b/apps/server/src/orchestration-v2/CommandPolicy.ts @@ -240,6 +240,7 @@ export interface CommandPolicyV2Shape { readonly hasStrongNativeSource: boolean; readonly sourceRunStatus: OrchestrationV2Run["status"]; readonly fromSpecificTurn: boolean; + readonly fromSpecificItem?: boolean; }, ) => Effect.Effect; readonly ensureRollback: ( @@ -400,7 +401,9 @@ const ensureContextHandoff: CommandPolicyV2Shape["ensureContextHandoff"] = (inpu const decideForkExecution: CommandPolicyV2Shape["decideForkExecution"] = (input) => { // Unsuccessful runs may have no native turn or assistant cursor. Forking // those at native head can include later turns, so use the bounded transcript. + // Item cuts use portable context because native turn cursors can include a later steer. const canForkNatively = + input.fromSpecificItem !== true && (input.sourceRunStatus === "completed" || input.sourceRunStatus === "waiting") && input.sameProvider && input.hasStrongNativeSource && diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 11689485e6ed2..76b994135b4a4 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -638,6 +638,7 @@ function runForSourcePoint( case "latest_stable": return latestStableRun(projection); case "run": + case "turn_item": return projection.runs.find((run) => run.id === sourcePoint.runId) ?? null; case "checkpoint": { const checkpoint = projection.checkpoints.find( @@ -3681,6 +3682,23 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: forkableSourceRunStatusError(sourceRun), }); } + const throughTurnItemId = + command.sourcePoint.type === "turn_item" ? command.sourcePoint.turnItemId : undefined; + if (throughTurnItemId !== undefined) { + const runItems = yield* readHandoffItems(command.sourceThreadId, [sourceRun.id]); + const item = runItems.find((candidate) => candidate.id === throughTurnItemId); + if ( + item?.type !== "assistant_message" || + item.runId !== sourceRun.id || + (item.status !== "completed" && item.status !== "interrupted") + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Fork source item ${throughTurnItemId} is not a finished assistant response of run ${sourceRun.id}.`, + }); + } + } const sourceProviderThread = providerThreadForRun(sourceProjection, sourceRun); const now = command.createdAt ?? (yield* DateTime.now); const emitEvent = emit(events, command); @@ -3696,7 +3714,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio sourceProjection, sourceRun, sourceProviderThread, - canonicalSourcePoint: contextSourcePointForRun(sourceProjection, sourceRun), + canonicalSourcePoint: { + ...contextSourcePointForRun(sourceProjection, sourceRun), + ...(throughTurnItemId === undefined ? {} : { turnItemId: throughTurnItemId }), + }, transferId, targetThreadId: command.targetThreadId, ...(command.title === undefined ? {} : { title: command.title }), @@ -3778,6 +3799,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); } + if (command.sourcePoint.type === "turn_item") { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: "Merge-back requires a run or checkpoint rather than a single response.", + }); + } const sourceRun = runForSourcePoint(sourceProjection, command.sourcePoint); if (sourceRun === null) { return yield* new OrchestratorDispatchError({ @@ -5803,6 +5831,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio hasStrongNativeSource: sourceProviderThread?.nativeThreadRef?.strength === "strong", sourceRunStatus: sourceRun.status, fromSpecificTurn: sourceRun !== null, + fromSpecificItem: pendingForkTransfer.sourcePoint.turnItemId !== undefined, }), ); const canResolveForkNatively = forkExecution === "native_fork"; @@ -5859,12 +5888,30 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const portableForkItems = !requiresPortableFork || sourceProjection === null || sourceRun === null ? [] - : yield* readHandoffItems(sourceProjection.thread.id, [ - ...sourceProjection.runs - .filter((run) => run.ordinal <= sourceRun.ordinal) - .map((run) => run.id), - ...(sourceProjection.thread.historyOrigin === "v1_import" ? [null] : []), - ]); + : yield* projectionStore.getThreadProjection(command.threadId).pipe( + Effect.map((targetHistory) => + targetHistory.visibleTurnItems + .filter((row) => row.visibility === "inherited") + .map((row) => row.item), + ), + Effect.mapError( + (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }), + ), + ); + const cutItemId = pendingForkTransfer?.sourcePoint.turnItemId; + const cutItem = portableForkItems.find( + (item) => + item.id === cutItemId && + item.threadId === sourceProjection?.thread.id && + item.runId === sourceRun?.id, + ); + if (requiresPortableFork && cutItemId !== undefined && cutItem === undefined) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Fork source item ${cutItemId} is no longer available.`, + }); + } const portableForkHandoff = !requiresPortableFork || pendingForkTransfer === undefined || diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 002ef7bafa950..ce2bf40cd390a 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1008,6 +1008,7 @@ type ShellThreadRow = { readonly thread_id: string; readonly payload_json: string; readonly forked_from_run_source_thread_id: string | null; + readonly forked_from_items_after_cut: number; readonly latest_run_id: string | null; readonly latest_run_status: string | null; readonly latest_run_requested_at: string | null; @@ -1338,14 +1339,43 @@ export function isTurnItemAtOrBeforeRun(input: { return ordinal !== undefined && ordinal <= input.sourceRunOrdinal; } +/** + * A fork that ends at a response inside its source run (one a steer cut off) + * keeps that run's items only through the response. + */ +function isTurnItemWithinForkCut(input: { + readonly item: Pick; + 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 @@ -1369,12 +1399,14 @@ function visibleTurnItemsThroughRun(input: { ) { return false; } - return isTurnItemAtOrBeforeRun({ - historyOrigin: input.sourceProjection.thread.historyOrigin, - itemRunId: item.runId, - runOrdinalById, - sourceRunOrdinal: sourceRun.ordinal, - }); + return ( + isTurnItemAtOrBeforeRun({ + historyOrigin: input.sourceProjection.thread.historyOrigin, + itemRunId: item.runId, + runOrdinalById, + sourceRunOrdinal: sourceRun.ordinal, + }) && isTurnItemWithinForkCut({ item, sourceRunId: sourceRun.id, cutOrdinal }) + ); }), ); @@ -1393,6 +1425,7 @@ function buildVisibleTurnItems(input: { const inherited = visibleTurnItemsThroughRun({ sourceProjection: input.sourceProjection, sourceRunId: forkedFrom.runId, + throughTurnItemId: forkedFrom.throughTurnItemId, }); const markerItem = makeForkMarkerTurnItem({ targetProjection: input.projection, @@ -1648,6 +1681,7 @@ type ShellThreadState = { readonly goal: OrchestrationV2ThreadShell["goal"]; readonly itemCount: number; readonly runlessItemCount: number; + readonly forkedFromItemsAfterCut: number; readonly updatedAt: OrchestrationV2ThreadProjection["updatedAt"]; readonly runOrdinalById: ReadonlyMap; readonly itemCountByRunId: ReadonlyMap; @@ -1743,7 +1777,8 @@ function visibleItemCountForShell(input: { return ( inheritedPrefixCount + - itemCountThroughRun({ state: sourceState, runId: forkedFrom.runId }) + + itemCountThroughRun({ state: sourceState, runId: forkedFrom.runId }) - + state.forkedFromItemsAfterCut + 1 + state.itemCount ); @@ -3451,14 +3486,20 @@ export const layer: Layer.Layer = WHERE thread_id = ${forkedFrom.threadId} AND run_id = ${forkedFrom.runId} LIMIT 1 + ), fork_cut AS ( + SELECT ordinal + FROM orchestration_v2_projection_turn_items + WHERE thread_id = ${forkedFrom.threadId} + AND turn_item_id = ${forkedFrom.throughTurnItemId ?? null} + LIMIT 1 ), fork_boundary AS ( - SELECT ( + SELECT COALESCE((SELECT ordinal FROM fork_cut), ( SELECT item.ordinal FROM orchestration_v2_projection_turn_items AS item WHERE item.run_id = run.run_id ORDER BY item.ordinal DESC LIMIT 1 - ) AS ordinal + )) AS ordinal FROM orchestration_v2_projection_runs AS run WHERE run.thread_id = ${forkedFrom.threadId} AND run.ordinal <= (SELECT ordinal FROM fork_run) @@ -5003,12 +5044,17 @@ export const layer: Layer.Layer = if (sourceRun !== undefined) { const ordinals = new Map(source.records.runs.map((run) => [run.id, run.ordinal])); const sourceItems = source.local.map((row) => row.item); + // Rows are in ordinal order, so the cut is the cut item's index. + const cutIndex = + fork.throughTurnItemId === undefined + ? undefined + : sourceItems.findIndex((item) => item.id === fork.throughTurnItemId); inherited = [ ...source.visible.filter( (row) => row.item.threadId !== fork.threadId || row.item.type === "fork", ), ...source.local.filter( - (row) => + (row, index) => !isOrchestrationV2SupersededInterrupt({ item: row.item, attempts: source.records.attempts, @@ -5019,6 +5065,11 @@ export const layer: Layer.Layer = itemRunId: row.item.runId, runOrdinalById: ordinals, sourceRunOrdinal: sourceRun.ordinal, + }) && + isTurnItemWithinForkCut({ + item: { runId: row.item.runId, ordinal: index }, + sourceRunId: sourceRun.id, + cutOrdinal: cutIndex, }), ), ].map((row) => ({ ...row, visibility: "inherited" })); @@ -5287,6 +5338,18 @@ export const layer: Layer.Layer = THEN json_extract(t.payload_json, '$.forkedFrom.threadId') ELSE NULL END AS forked_from_run_source_thread_id, + -- Source-run items after the response a fork ends at (a steer + -- and the reply to it); the shell leaves them out of its count. + ( + SELECT COUNT(*) + FROM orchestration_v2_projection_turn_items AS cut_item + JOIN orchestration_v2_projection_turn_items AS after_item + ON after_item.thread_id = cut_item.thread_id + AND after_item.run_id = cut_item.run_id + AND after_item.ordinal > cut_item.ordinal + WHERE cut_item.thread_id = json_extract(t.payload_json, '$.forkedFrom.threadId') + AND cut_item.turn_item_id = json_extract(t.payload_json, '$.forkedFrom.throughTurnItemId') + ) AS forked_from_items_after_cut, presented.run_id AS latest_run_id, presented.status AS latest_run_status, presented.requested_at AS latest_run_requested_at, @@ -5892,6 +5955,7 @@ export const layer: Layer.Layer = ), 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/ThreadFork.execution.test.ts b/apps/server/src/orchestration-v2/ThreadFork.execution.test.ts index 920b5fe321b27..131aa1ea43446 100644 --- a/apps/server/src/orchestration-v2/ThreadFork.execution.test.ts +++ b/apps/server/src/orchestration-v2/ThreadFork.execution.test.ts @@ -13,15 +13,21 @@ import { RunId, ThreadId, TurnItemId, + type OrchestrationV2TurnItem, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as Schema from "effect/Schema"; +import { ThreadToolkit } from "../mcp/toolkits/thread/tools.ts"; import * as TestClock from "effect/testing/TestClock"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import { ClaudeProviderCapabilitiesV2 } from "./Adapters/ClaudeAdapterV2.ts"; import * as EventSink from "./EventSink.ts"; import * as Orchestrator from "./Orchestrator.ts"; +import * as ProjectStore from "./ProjectStore.ts"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; @@ -396,3 +402,431 @@ it.effect.each(forkCases)( assert.isNull(target.providerThreads[0]?.forkedFrom); }).pipe(Effect.provide(layer)), ); + +const decodeForkInput = Schema.decodeUnknownEffect( + ThreadToolkit.tools.t3_thread_fork.parametersSchema, +); + +const responseForkCases = [ + { ...forkCases[0]!, replyStatus: "completed" as const, ancestry: "root" }, + { + ...forkCases[0]!, + replyStatus: "completed" as const, + ancestry: "root after source rollback", + }, + { ...forkCases[0]!, replyStatus: "completed" as const, ancestry: "fork with a local run" }, + { ...forkCases[3]!, replyStatus: "completed" as const, ancestry: "root" }, + { ...forkCases[0]!, replyStatus: "interrupted" as const, ancestry: "root" }, + { ...forkCases[0]!, replyStatus: "failed" as const, ancestry: "root" }, +]; + +it.effect.each(responseForkCases)( + "bounds a $driver fork at a $replyStatus response before a steer ($ancestry)", + ({ driver, instanceId, modelSelection, layer, replyStatus, ancestry }) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const sourceThreadId = ThreadId.make("response-source"); + const runId = RunId.make("response-run"); + const providerThreadId = ProviderThreadId.make("response-provider-thread"); + const laterRunId = RunId.make("later-run"); + const cutId = TurnItemId.make("response-cut"); + const projectId = ProjectId.make("response-project"); + const projects = yield* ProjectStore.ProjectStoreV2; + yield* projects.apply({ + sequence: 0, + eventId: EventId.make("response-project-created"), + aggregateKind: "project", + aggregateId: projectId, + occurredAt: DateTime.formatIso(now), + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "project.created", + payload: { + projectId, + title: "Response forks", + workspaceRoot: "/work/response-project", + defaultModelSelection: null, + scripts: [], + createdAt: DateTime.formatIso(now), + updatedAt: DateTime.formatIso(now), + }, + }); + + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create-response-source"), + threadId: sourceThreadId, + projectId, + title: "Response before steer", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + const base = { + threadId: sourceThreadId, + runId, + nodeId: null, + providerThreadId, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + } as const; + const items = [ + { + ...base, + id: TurnItemId.make("prompt"), + ordinal: 1, + type: "user_message", + messageId: MessageId.make("prompt"), + text: "INCLUDED_PROMPT", + attachments: [], + inputIntent: "turn_start", + createdBy: "user", + creationSource: "web", + }, + { + ...base, + id: cutId, + ordinal: 2, + type: "assistant_message", + messageId: MessageId.make("cut"), + text: "INCLUDED_RESPONSE", + streaming: false, + status: replyStatus, + }, + { + ...base, + id: TurnItemId.make("steer"), + ordinal: 3, + type: "user_message", + messageId: MessageId.make("steer"), + text: "EXCLUDED_STEER", + attachments: [], + inputIntent: "steer", + createdBy: "user", + creationSource: "web", + }, + { + ...base, + id: TurnItemId.make("reply"), + ordinal: 4, + type: "assistant_message", + messageId: MessageId.make("reply"), + text: "EXCLUDED_REPLY", + streaming: false, + }, + { + ...base, + id: TurnItemId.make("later-response"), + runId: laterRunId, + ordinal: 5, + type: "assistant_message", + messageId: MessageId.make("later-response"), + text: "EXCLUDED_LATER_RUN", + streaming: false, + }, + ] satisfies ReadonlyArray; + yield* eventSink.write({ + events: [ + { + id: EventId.make("response-provider-thread"), + type: "provider-thread.updated", + threadId: sourceThreadId, + occurredAt: now, + payload: { + id: providerThreadId, + driver, + providerInstanceId: instanceId, + providerSessionId: null, + appThreadId: sourceThreadId, + ownerNodeId: null, + nativeThreadRef: null, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: 1, + lastRunOrdinal: 2, + handoffIds: [], + forkedFrom: null, + createdAt: now, + updatedAt: now, + }, + }, + ], + }); + for (const [ordinal, id] of [runId, laterRunId].entries()) { + yield* eventSink.write({ + events: [ + { + id: EventId.make(id), + type: "run.created", + threadId: sourceThreadId, + runId: id, + occurredAt: now, + payload: { + id, + threadId: sourceThreadId, + ordinal: ordinal + 1, + providerInstanceId: instanceId, + modelSelection, + providerThreadId, + userMessageId: MessageId.make(ordinal === 0 ? "prompt" : "later-prompt"), + rootNodeId: null, + activeAttemptId: null, + status: + replyStatus === "interrupted" && ordinal === 0 ? "interrupted" : "completed", + requestedAt: now, + startedAt: now, + completedAt: now, + checkpointId: null, + contextHandoffId: null, + }, + }, + ], + }); + } + yield* eventSink.write({ + events: items.map((item) => ({ + id: EventId.make(item.id), + type: "turn-item.updated" as const, + threadId: sourceThreadId, + runId: item.runId, + occurredAt: now, + payload: item, + })), + }); + + const continueFork = (targetThreadId: ThreadId) => + orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`continue-${targetThreadId}`), + threadId: targetThreadId, + messageId: MessageId.make(`continue-${targetThreadId}`), + text: "Continue", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + + const runEndId = ThreadId.make("whole-run-fork"); + yield* orchestrator.dispatch({ + type: "thread.fork", + commandId: CommandId.make("fork-whole-run"), + sourceThreadId, + targetThreadId: runEndId, + sourcePoint: { type: "run", runId }, + createdBy: "user", + creationSource: "web", + }); + assert.deepEqual( + (yield* orchestrator.getThreadProjection(runEndId)).visibleTurnItems + .filter((row) => row.visibility === "inherited") + .map((row) => row.sourceItemId), + items.slice(0, 4).map((item) => item.id), + ); + + yield* continueFork(runEndId); + const wholeRun = yield* orchestrator.getThreadProjection(runEndId); + assert.include(wholeRun.contextHandoffs[0]?.summaryText ?? "", "EXCLUDED_STEER"); + assert.include(wholeRun.contextHandoffs[0]?.summaryText ?? "", "EXCLUDED_REPLY"); + + // Decode the public MCP input before dispatching its point through the real service. + const forkAt = (turnItemId: TurnItemId, target: string) => + Effect.gen(function* () { + const input = yield* decodeForkInput({ + sourcePoint: { type: "turn_item", runId, turnItemId }, + }); + return yield* orchestrator.dispatch({ + type: "thread.fork", + commandId: CommandId.make(target), + sourceThreadId, + targetThreadId: ThreadId.make(target), + sourcePoint: input.sourcePoint, + createdBy: "user", + creationSource: "mcp", + }); + }); + + // A valid point must decode, so an invalid item's failure cannot pass on schema rejection. + yield* decodeForkInput({ + sourcePoint: { type: "turn_item", runId, turnItemId: cutId }, + }); + for (const invalidId of [ + "steer", + "missing", + "later-response", + ...(replyStatus === "failed" ? [cutId] : []), + ]) { + const targetId = ThreadId.make(`invalid-${invalidId}`); + const error = yield* forkAt(TurnItemId.make(invalidId), targetId).pipe(Effect.flip); + assert.equal(error._tag, "OrchestratorDispatchError"); + assert.isNull(yield* orchestrator.getThreadShell(targetId)); + } + if (replyStatus === "failed") return; + + const targetThreadId = ThreadId.make("response-fork"); + yield* forkAt(cutId, targetThreadId); + if (ancestry === "root after source rollback") { + const source = yield* orchestrator.getThreadProjection(sourceThreadId); + const sourceRun = source.runs.find((run) => run.id === runId)!; + yield* eventSink.write({ + events: [ + { + id: EventId.make("source-run-rolled-back"), + type: "run.updated", + threadId: sourceThreadId, + runId, + occurredAt: now, + payload: { ...sourceRun, status: "rolled_back" }, + }, + ], + }); + } + const expectedIds = items.slice(0, 2).map((item) => item.id); + const inherited = yield* orchestrator.getThreadProjection(targetThreadId); + assert.deepEqual( + inherited.visibleTurnItems + .filter((row) => row.visibility === "inherited") + .map((row) => row.sourceItemId), + expectedIds, + ); + const window = yield* orchestrator.getThreadSnapshotWindow(targetThreadId, { rowLimit: 1 }); + assert.deepEqual( + window.projection.visibleTurnItems + .filter((row) => row.visibility === "inherited") + .map((row) => row.sourceItemId), + [cutId], + ); + const olderWindow = yield* orchestrator.getThreadSnapshotWindow(targetThreadId, { + rowLimit: 1, + anchorItemId: items[0]!.id, + }); + assert.deepEqual( + olderWindow.projection.visibleTurnItems + .filter((row) => row.visibility === "inherited") + .map((row) => row.sourceItemId), + [items[0]!.id], + ); + const page = yield* orchestrator.getTimelinePage(targetThreadId, { limit: 1 }); + assert.deepEqual( + page.items.map((row) => row.sourceItemId), + [items[0]!.id], + ); + assert.equal(page.totalItems, inherited.visibleTurnItems.length); + assert.isTrue(page.hasMore); + const nextPage = yield* orchestrator.getTimelinePage(targetThreadId, { + afterPosition: page.items[0]!.position, + limit: 10, + }); + assert.deepEqual( + nextPage.items + .filter((row) => row.visibility === "inherited") + .map((row) => row.sourceItemId), + [cutId], + ); + assert.isFalse(nextPage.hasMore); + const shell = yield* orchestrator.getThreadShell(targetThreadId); + assert.equal(shell?.visibleItemCount, inherited.visibleTurnItems.length); + assert.equal( + (yield* orchestrator.searchThread({ threadId: targetThreadId, query: "INCLUDED_RESPONSE" })) + .totalMatches, + 1, + ); + assert.equal( + (yield* orchestrator.searchThread({ threadId: targetThreadId, query: "EXCLUDED" })) + .totalMatches, + 0, + ); + + yield* continueFork(targetThreadId); + const continued = yield* orchestrator.getThreadProjection(targetThreadId); + // This source has no native session. Check the context payload, not the chosen strategy. + assert.lengthOf(continued.contextHandoffs, 1); + const handoff = continued.contextHandoffs[0]!; + const context = [ + handoff.summaryText, + ...(handoff.history?.messages.map((message) => message.text) ?? []), + ].join("\n"); + assert.include(context, "INCLUDED_PROMPT"); + assert.include(context, "INCLUDED_RESPONSE"); + assert.notInclude(context, "EXCLUDED"); + + if (ancestry !== "fork with a local run") return; + + const localRun = continued.runs[0]!; + const localItems = items.slice(0, 4).map((item) => ({ + ...item, + id: TurnItemId.make(`local-${item.id}`), + threadId: targetThreadId, + runId: localRun.id, + providerThreadId: localRun.providerThreadId, + messageId: MessageId.make(`local-${item.messageId}`), + text: item.text.replace("INCLUDED", "LOCAL"), + })); + yield* eventSink.write({ + events: [ + { + id: EventId.make("local-run-completed"), + type: "run.updated", + threadId: targetThreadId, + runId: localRun.id, + occurredAt: now, + payload: { ...localRun, status: "completed", completedAt: now }, + }, + ...localItems.map((item) => ({ + id: EventId.make(item.id), + type: "turn-item.updated" as const, + threadId: targetThreadId, + runId: localRun.id, + occurredAt: now, + payload: item, + })), + ], + }); + const grandchildId = ThreadId.make("local-response-fork"); + yield* orchestrator.dispatch({ + type: "thread.fork", + commandId: CommandId.make("fork-local-response"), + sourceThreadId: targetThreadId, + targetThreadId: grandchildId, + sourcePoint: { + type: "turn_item", + runId: localRun.id, + turnItemId: localItems[1]!.id, + }, + createdBy: "user", + creationSource: "mcp", + }); + yield* continueFork(grandchildId); + const grandchild = yield* orchestrator.getThreadProjection(grandchildId); + assert.lengthOf(grandchild.contextHandoffs, 1); + const grandchildHandoff = grandchild.contextHandoffs[0]!; + const grandchildContext = [ + grandchildHandoff.summaryText, + ...(grandchildHandoff.history?.messages.map((message) => message.text) ?? []), + ].join("\n"); + assert.include(grandchildContext, "LOCAL_PROMPT"); + assert.include(grandchildContext, "LOCAL_RESPONSE"); + assert.notInclude(grandchildContext, "EXCLUDED"); + assert.include(grandchildContext, "INCLUDED_PROMPT"); + assert.include(grandchildContext, "INCLUDED_RESPONSE"); + }).pipe( + Effect.provide( + Layer.merge(layer, ProjectStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory))), + ), + ), +); diff --git a/apps/server/src/orchestration-v2/ThreadForkService.ts b/apps/server/src/orchestration-v2/ThreadForkService.ts index b652ea1b14bee..1508892d471bc 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/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index b78cf0f999718..9ec74b790f8a4 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -88,6 +88,8 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ inlineMessageContext: Schema.optionalKey(Schema.Boolean), requiredWorktreeBootstrap: Schema.optionalKey(Schema.Boolean), threadSettlement: Schema.optionalKey(Schema.Boolean), + /** Item fork points require a compatible server decoder; absent means unsupported. */ + threadForkFromItem: Schema.optionalKey(Schema.Boolean), threadAutoSettlement: Schema.optionalKey(Schema.Boolean), storageCleanup: Schema.optionalKey(Schema.Boolean), storageCleanupRun: Schema.optionalKey(Schema.Boolean), diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 94a9eca331d13..eb84725f541b5 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; @@ -372,7 +374,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"),