diff --git a/apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts b/apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts index c4068702e3a57..a998ee4455119 100644 --- a/apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts +++ b/apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts @@ -640,4 +640,32 @@ describe("handoff delivery", () => { assert.equal(calls, 0); }), ); + + it.effect("leaves a verified native-thread handoff pending when its wrapper cannot fit", () => + Effect.gen(function* () { + let injections = 0; + let persists = 0; + const result = yield* deliverContextHandoffs({ + handoffs: [handoff], + providerThread, + budget: 0, + deferIfOverBudget: true, + alreadyDeliveredItemIds: new Set(), + inject: () => + Effect.sync(() => { + injections++; + return true; + }), + persist: () => + Effect.sync(() => { + persists++; + }), + }); + + yield* result.delivered; + assert.equal(result.context, ""); + assert.equal(injections, 0); + assert.equal(persists, 0); + }), + ); }); diff --git a/apps/server/src/orchestration-v2/ContextHandoffDelivery.ts b/apps/server/src/orchestration-v2/ContextHandoffDelivery.ts index ff2ca1fbec441..b0d328583d0f0 100644 --- a/apps/server/src/orchestration-v2/ContextHandoffDelivery.ts +++ b/apps/server/src/orchestration-v2/ContextHandoffDelivery.ts @@ -19,6 +19,7 @@ export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextH readonly providerThread: OrchestrationV2ProviderThread; readonly budget: number | Effect.Effect; readonly deferInline?: boolean; + readonly deferIfOverBudget?: boolean; readonly alreadyDeliveredItemIds: ReadonlySet; readonly inject?: ( history: ProviderAdapter.ProviderAdapterV2HistoricalContext, @@ -72,6 +73,10 @@ export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextH }); if (historyCost(selected.messages, selected.context) > budget) { if (input.deferInline) return { context: "", delivered: Effect.void, unsent: Effect.void }; + // The target transcript already has this history, so an oversized wrapper + // must not block the turn. Keep the handoff pending for later delivery. + if (input.deferIfOverBudget) + return { context: "", delivered: Effect.void, unsent: Effect.void }; return yield* new ContextHandoffBudgetError(); } const omittedItemIds = new Set(selected.omittedItemIds); diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index 7a922d79c7317..3c3a4beb44e7e 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -3,6 +3,7 @@ import { expect, it, vi } from "vite-plus/test"; import { it as effectIt } from "@effect/vitest"; import { CheckpointScopeId, + ContextHandoffId, GitCommandError, MessageId, NodeId, @@ -15,6 +16,7 @@ import { RunAttemptId, RunId, ThreadId, + TurnItemId, ProjectId, type OrchestrationV2ThreadProjection, OrchestrationV2DomainEvent, @@ -52,6 +54,54 @@ import * as RuntimePolicy from "./RuntimePolicy.ts"; const isDomainEvent = Schema.is(OrchestrationV2DomainEvent); +it("excludes rolled-back runs from native-thread history", () => { + const providerThreadId = ProviderThreadId.make("provider_thread_native_history"); + const keptRunId = RunId.make("run_native_history_kept"); + const rolledBackRunId = RunId.make("run_native_history_rolled_back"); + const acceptedAttempts = [keptRunId, rolledBackRunId].map((runId, index) => ({ + id: RunAttemptId.make(`attempt_native_history_${index}`), + runId, + providerThreadId, + nativeThreadId: "native-thread", + })); + const nativeInputRunIds = ProviderTurnStart.nativeInputRunIdsForThread({ + acceptedAttempts, + runs: [ + { id: keptRunId, status: "completed" }, + { id: rolledBackRunId, status: "rolled_back" }, + ] as const, + providerThreadId, + nativeThreadId: "native-thread", + }); + + expect(nativeInputRunIds).toEqual(new Set([keptRunId])); +}); + +it("does not treat runs accepted on multiple native threads as native history", () => { + const providerThreadId = ProviderThreadId.make("provider_thread_mixed_native_history"); + const mixedRunId = RunId.make("run_mixed_native_history"); + const unknownRunId = RunId.make("run_unknown_native_history"); + const singleThreadRunId = RunId.make("run_single_native_history"); + const attempts = [ + { runId: mixedRunId, providerThreadId, nativeThreadId: "current-native-thread" }, + { runId: mixedRunId, providerThreadId, nativeThreadId: "older-native-thread" }, + { runId: unknownRunId, providerThreadId, nativeThreadId: undefined }, + { runId: unknownRunId, providerThreadId, nativeThreadId: "current-native-thread" }, + { runId: singleThreadRunId, providerThreadId, nativeThreadId: "current-native-thread" }, + ]; + + expect(ProviderTurnStart.mixedNativeThreadRunIdsForThread(attempts, providerThreadId)).toEqual( + new Set([mixedRunId, unknownRunId]), + ); +}); + +it("only treats provider transcript items as native history", () => { + expect(ProviderTurnStart.isNativeTranscriptHistoryKind("user_message")).toBe(true); + expect(ProviderTurnStart.isNativeTranscriptHistoryKind("command_execution")).toBe(true); + expect(ProviderTurnStart.isNativeTranscriptHistoryKind("run_interrupt_result")).toBe(false); + expect(ProviderTurnStart.isNativeTranscriptHistoryKind("file_change")).toBe(false); +}); + const managedWorktreesDir = "/tmp/provider-turn-start-worktrees"; // fs.realpath applies `..` and `.` before following symlinks. const normalize = (target: string) => { @@ -274,6 +324,7 @@ function makeLocalCommandHarness(input: { readonly openFailure?: unknown; readonly ensureThreadFailure?: unknown; readonly historyReadFailureAfterFallback?: unknown; + readonly allowRecoveryHandoff?: boolean; readonly interruptOpen?: boolean; readonly interruptRunBeforeOpenFailure?: boolean; readonly writeFailure?: unknown; @@ -458,6 +509,7 @@ function makeLocalCommandHarness(input: { checkpoints: [], updatedAt: now, }; + let turnStartHistory: OrchestrationV2ThreadProjection["turnItems"] | undefined; if ("historyReadFailureAfterFallback" in input) { const nativeThreadRef = { driver: providerThread.driver, @@ -565,6 +617,11 @@ function makeLocalCommandHarness(input: { ), ), ); + const startedTurns: Array = []; + const startTurn = vi.fn((turnInput: ProviderAdapter.ProviderAdapterV2TurnInput) => { + startedTurns.push(turnInput); + return Effect.void; + }); const resumeFallbackSession = { driver: providerThread.driver, resumeThread: () => @@ -615,6 +672,7 @@ function makeLocalCommandHarness(input: { }, ensureThread: () => Effect.succeed(providerThread), resumeThread: () => Effect.succeed(projection.providerThreads[0]), + startTurn, } as never) : Effect.die("A local command must not open a native session."), ); @@ -694,9 +752,31 @@ function makeLocalCommandHarness(input: { Layer.provide( Layer.mergeAll( Layer.mock(ContextHandoffService.ContextHandoffServiceV2)({ - prepareProviderHandoff: () => Effect.die("history read must fail first"), + prepareProviderHandoff: (handoff) => + input.allowRecoveryHandoff + ? Effect.succeed({ + ...handoff, + id: ContextHandoffId.make("provider-turn-start-recovery-handoff"), + status: "ready", + summaryMessageId: null, + summaryText: "", + history: { + messages: [], + coverage: "", + omittedItems: 0, + omittedItemIds: [], + }, + createdByProviderInstanceId: null, + updatedAt: handoff.createdAt, + } as never) + : Effect.die("history read must fail first"), + }), + Layer.mock(EventSink.EventSinkV2)({ + writeIfRunCurrent, + ...(input.allowRecoveryHandoff + ? { write: () => Effect.succeed({ storedEvents: [] } as never) } + : {}), }), - Layer.mock(EventSink.EventSinkV2)({ writeIfRunCurrent }), IdAllocator.layer, layerWorktreeLocation(input.worktreesDir), input.fileSystem === undefined @@ -728,12 +808,14 @@ function makeLocalCommandHarness(input: { ), }), getTurnStartHistory: () => - Effect.fail( - new ProjectionStore.ProjectionStoreReadError({ - threadId, - cause: input.historyReadFailureAfterFallback, - }), - ), + turnStartHistory === undefined + ? Effect.fail( + new ProjectionStore.ProjectionStoreReadError({ + threadId, + cause: input.historyReadFailureAfterFallback, + }), + ) + : Effect.succeed(turnStartHistory), }), Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ open, @@ -765,6 +847,8 @@ function makeLocalCommandHarness(input: { resolveRuntimePolicy, createWorktree, open, + startTurn, + startedTurns, writeIfRunCurrent, startRootRun, interruptTurn, @@ -774,6 +858,9 @@ function makeLocalCommandHarness(input: { newInstanceId, attemptId, projection: () => projection, + setTurnStartHistory: (items: OrchestrationV2ThreadProjection["turnItems"]) => { + turnStartHistory = items; + }, start: Effect.gen(function* () { yield* (yield* ProviderTurnStart.ProviderTurnStartServiceV2).start({ threadId, runId }); }).pipe(Effect.provide(layer)), @@ -970,6 +1057,131 @@ effectIt.effect("recovers an abandoned failed turn before continuing its native }), ); +effectIt.effect( + "recovers an over-budget handoff from mixed native threads before starting the turn", + () => + Effect.gen(function* () { + const harness = makeLocalCommandHarness({ + text: "Continue", + previousTurn: { runStatus: "failed", turnStatus: "completed" }, + allowRecoveryHandoff: true, + }); + harness.setTurnStartHistory([]); + const projection = harness.projection(); + const providerThreadId = projection.providerThreads[0]!.id; + const currentNativeThreadId = "existing-native-conversation"; + const previousRunId = RunId.make("previous-failed-run"); + const currentRunId = projection.runs.at(-1)!.id; + const olderAttemptId = RunAttemptId.make("older-native-thread-attempt"); + const currentAttemptId = RunAttemptId.make("current-native-thread-attempt"); + const historicalItemId = TurnItemId.make("mixed-native-thread-history-item"); + const previousTurn = projection.providerTurns[0]!; + harness.setProjection({ + ...projection, + providerThreads: projection.providerThreads.map((thread) => + thread.id === providerThreadId + ? { + ...thread, + contextUsage: { + usedTokens: 39_999, + maxTokens: 40_000, + autoCompactThreshold: 40_000, + }, + } + : thread, + ), + attempts: [ + ...projection.attempts, + { + ...projection.attempts[0]!, + id: olderAttemptId, + runId: previousRunId, + providerThreadId, + nativeThreadId: "older-native-conversation", + }, + { + ...projection.attempts[0]!, + id: currentAttemptId, + runId: previousRunId, + providerThreadId, + nativeThreadId: currentNativeThreadId, + }, + ], + providerTurns: [ + previousTurn, + { + ...previousTurn, + id: ProviderTurnId.make("older-native-turn"), + runAttemptId: olderAttemptId, + }, + { + ...previousTurn, + id: ProviderTurnId.make("current-native-turn"), + runAttemptId: currentAttemptId, + }, + ], + contextHandoffs: [ + { + id: ContextHandoffId.make("mixed-native-thread-handoff"), + threadId: projection.thread.id, + targetRunId: currentRunId, + fromProviderThreadIds: [providerThreadId], + toProviderThreadId: providerThreadId, + coveredRunOrdinals: { from: 1, to: 1 }, + strategy: "delta_since_target_last_seen", + status: "ready", + summaryMessageId: null, + summaryText: "", + history: { + messages: [ + { + role: "user", + text: "Earlier request", + threadId: projection.thread.id, + runId: previousRunId, + itemId: historicalItemId, + providerThreadId, + status: "completed", + kind: "user_message", + }, + ], + coverage: "Earlier request", + omittedItems: 0, + }, + createdByProviderInstanceId: null, + createdAt: DateTime.makeUnsafe("2026-09-04T12:00:00Z"), + updatedAt: DateTime.makeUnsafe("2026-09-04T12:00:00Z"), + }, + ], + }); + + yield* harness.start; + + const startInput = harness.startRootRun.mock.calls[0]?.[0]; + expect(startInput).toBeDefined(); + const delivery = yield* Effect.result( + startInput!.session.startTurn({ + appThread: startInput!.appThread, + threadId: startInput!.run.threadId, + runId: startInput!.run.id, + runOrdinal: startInput!.run.ordinal, + providerTurnOrdinal: startInput!.providerTurnOrdinal, + nativeThreadHasTurns: startInput!.nativeThreadHasTurns ?? false, + attemptId: startInput!.attemptId, + rootNodeId: startInput!.rootNode.id, + providerThread: startInput!.providerThread, + message: startInput!.message, + modelSelection: startInput!.modelSelection, + runtimePolicy: startInput!.runtimePolicy, + }), + ); + + expect(delivery._tag).toBe("Success"); + expect(harness.startTurn).toHaveBeenCalledOnce(); + expect(harness.startedTurns[0]?.message.text).toContain("Earlier request"); + }), +); + effectIt.effect.each([ { runStatus: "failed", turnStatus: "completed" }, { runStatus: "completed", turnStatus: "completed" }, diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 94ba406b106f5..1931788a1a708 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -93,6 +93,61 @@ class WorktreeRecreateNoBaseError extends Schema.TaggedError + >; + readonly runs: ReadonlyArray>; + readonly providerThreadId: OrchestrationV2ProviderThread["id"]; + readonly nativeThreadId: string | undefined; +}) => { + const rolledBackRunIds = new Set( + input.runs.filter((run) => run.status === "rolled_back").map((run) => run.id), + ); + return new Set( + input.acceptedAttempts + .filter( + (attempt) => + attempt.providerThreadId === input.providerThreadId && + attempt.nativeThreadId !== undefined && + attempt.nativeThreadId === input.nativeThreadId && + !rolledBackRunIds.has(attempt.runId), + ) + .map((attempt) => attempt.runId), + ); +}; + +export const mixedNativeThreadRunIdsForThread = ( + acceptedAttempts: ReadonlyArray< + Pick + >, + providerThreadId: OrchestrationV2ProviderThread["id"], +) => { + const nativeThreadIdsByRunId = new Map>(); + const runsWithUnknownNativeThread = new Set(); + for (const attempt of acceptedAttempts) { + if (attempt.providerThreadId !== providerThreadId) continue; + if (attempt.nativeThreadId === undefined) { + runsWithUnknownNativeThread.add(attempt.runId); + continue; + } + const nativeThreadIds = nativeThreadIdsByRunId.get(attempt.runId) ?? new Set(); + nativeThreadIds.add(attempt.nativeThreadId); + nativeThreadIdsByRunId.set(attempt.runId, nativeThreadIds); + } + return new Set( + [...nativeThreadIdsByRunId] + .filter( + ([runId, nativeThreadIds]) => + nativeThreadIds.size > 1 || runsWithUnknownNativeThread.has(runId), + ) + .map(([runId]) => runId), + ); +}; + +export const isNativeTranscriptHistoryKind = (kind: string) => + kind === "user_message" || kind === "assistant_message" || kind === "command_execution"; + /** * Claude refuses these before it reads the prompt: replacing a process running background work, * launch arguments whose appended system prompt or settings cannot be read, or MCP registration @@ -345,7 +400,8 @@ export const layer: Layer.Layer< message.id === source.userMessageId && (isCompactionCommandMessage(message) || parseGoalCommand(message) !== null), - ))), + )) || + (source.status === "completed" && handoff.delivery === undefined)), ))), ); const nativeForkTransfer = projection.contextTransfers.find( @@ -848,20 +904,26 @@ export const layer: Layer.Layer< (source) => source.providerThreadId === providerThread.id && deliveredAttemptIds.has(source.id), ); - const nativeInputRunIds = new Set( - acceptedAttempts - .filter( - (source) => - source.nativeThreadId !== undefined && - source.nativeThreadId === loadedProviderThread.nativeThreadRef?.nativeId, - ) - .map((source) => source.runId), + const nativeInputRunIds = nativeInputRunIdsForThread({ + acceptedAttempts, + runs: projection.runs, + providerThreadId: providerThread.id, + nativeThreadId: loadedProviderThread.nativeThreadRef?.nativeId ?? undefined, + }); + const mixedNativeThreadRunIds = mixedNativeThreadRunIdsForThread( + acceptedAttempts, + providerThread.id, ); const legacyInputRunIds = new Set( acceptedAttempts .filter((source) => source.nativeThreadId === undefined) .map((source) => source.runId), ); + const nativeThreadHasTurns = + nativeInputRunIds.size > 0 || + (legacyInputRunIds.size > 0 && + sameNativeThread && + !acceptedAttempts.some((source) => source.nativeThreadId !== undefined)); const legacyRecoveredRunIds = new Set( projection.runs .filter( @@ -928,6 +990,8 @@ export const layer: Layer.Layer< coveredItemIds, acceptedAttempts, nativeInputRunIds, + mixedNativeThreadRunIds, + nativeThreadHasTurns, legacyInputRunIds, missedRuns, missedRunIds, @@ -1240,9 +1304,9 @@ export const layer: Layer.Layer< sameNativeThread, handoffUsage, deliveredItemIds, - acceptedAttempts, nativeInputRunIds, - legacyInputRunIds, + mixedNativeThreadRunIds, + nativeThreadHasTurns, budget, } = handoffContext; const retryHandoff = resumedRetryHandoff ?? (yield* prepareRetryHandoff(handoffContext)); @@ -1375,9 +1439,68 @@ export const layer: Layer.Layer< ); const startWithHandoffs = Effect.fn("ProviderTurnStartService.startWithHandoffs")( function* (turnInput: Parameters[0], compact = false) { + const handoffsToDeliver = [...effectiveHandoffs, ...retryHandoff]; + const canDeferHandoff = + sameNativeThread && + nativeThreadHasTurns && + retryHandoff.length === 0 && + handoffsToDeliver.length > 0; + const omittedSourceRunIds = new Set( + handoffsToDeliver.flatMap((handoff) => + (handoff.history?.omittedItems ?? 0) === 0 + ? [] + : projection.runs + .filter( + (source) => + source.providerThreadId !== null && + handoff.fromProviderThreadIds.includes(source.providerThreadId) && + source.ordinal >= handoff.coveredRunOrdinals.from && + source.ordinal <= handoff.coveredRunOrdinals.to, + ) + .map((source) => source.id), + ), + ); + const sourceItemsById = + canDeferHandoff && omittedSourceRunIds.size > 0 + ? new Map( + (yield* projectionStore.getTurnStartHistory(input.threadId, [ + ...omittedSourceRunIds, + ])).map((item) => [item.id, item] as const), + ) + : new Map(); + const itemIsInNativeThread = (item: { + readonly threadId: ThreadId; + readonly providerThreadId?: OrchestrationV2TurnItem["providerThreadId"]; + readonly runId: OrchestrationV2TurnItem["runId"]; + readonly type?: string; + readonly kind?: string; + }) => + item.threadId === projection.thread.id && + item.providerThreadId === providerThread.id && + item.runId !== null && + isNativeTranscriptHistoryKind(item.kind ?? item.type ?? "") && + nativeInputRunIds.has(item.runId) && + !mixedNativeThreadRunIds.has(item.runId); + const handoffIsAlreadyInNativeThread = (handoff: OrchestrationV2ContextHandoff) => { + const history = handoff.history; + if (!handoff.fromProviderThreadIds.includes(providerThread.id) || history === undefined) + return false; + const omittedItemIds = history.omittedItemIds ?? []; + return ( + (history.messages.length > 0 || omittedItemIds.length > 0) && + history.omittedItems === omittedItemIds.length && + history.messages.every(itemIsInNativeThread) && + omittedItemIds.every((itemId) => { + const item = sourceItemsById.get(itemId); + return item !== undefined && itemIsInNativeThread(item); + }) + ); + }; const delivery = yield* deliverContextHandoffs({ - handoffs: [...effectiveHandoffs, ...retryHandoff], + handoffs: handoffsToDeliver, deferInline: compact || goalCommand, + deferIfOverBudget: + canDeferHandoff && handoffsToDeliver.every(handoffIsAlreadyInNativeThread), providerThread: runningProviderThread, budget, alreadyDeliveredItemIds: deliveredItemIds, @@ -1492,11 +1615,7 @@ export const layer: Layer.Layer< .filter((turn) => turn.providerThreadId === providerThread.id) .map((turn) => turn.ordinal), ) + 1, - nativeThreadHasTurns: - nativeInputRunIds.size > 0 || - (legacyInputRunIds.size > 0 && - sameNativeThread && - !acceptedAttempts.some((source) => source.nativeThreadId !== undefined)), + nativeThreadHasTurns, shouldStartProviderTurn: runControls.shouldStartProviderTurn, shouldFinalizeRun: runControls.shouldFinalizeRun, hasUnpairedRunInterruptRequest: runControls.hasUnpairedRunInterruptRequest,