diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index 7b0759ae99b0b..98a7df66e2d7f 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -2307,6 +2307,93 @@ effectIt.effect.each([ }), ); +effectIt.effect.each( + ["/compact", "/goal finish"].flatMap((command) => + [false, true].map((steered) => ({ command, steered })), + ), +)( + "keeps the switch note pending after a resumed $command unless steered: $steered", + ({ command, steered }) => + Effect.gen(function* () { + const harness = makeLocalCommandHarness({ + text: "Next prompt", + sideChat: { contextRun: "current" }, + previousTurn: { runStatus: "completed", turnStatus: "completed" }, + }); + const projection = harness.projection(); + const current = projection.runs.at(-1)!; + const previous = projection.runs[0]!; + const sourceId = RunId.make("unsent-command"); + const resumedId = RunId.make("resumed-command"); + const commandId = MessageId.make("command-message"); + const earlierId = MessageId.make("earlier-message"); + const resumedMessageId = MessageId.make("resume-message"); + const resumedAttemptId = RunAttemptId.make("resumed-command-attempt"); + const previousAttempt = projection.attempts[0]!; + harness.setProjection({ + ...projection, + runs: [ + { + ...previous, + providerInstanceId: harness.oldInstanceId, + userMessageId: earlierId, + ordinal: 1, + }, + { + ...current, + id: sourceId, + providerThreadId: null, + userMessageId: commandId, + status: "interrupted", + ordinal: 2, + }, + { + ...current, + id: resumedId, + userMessageId: resumedMessageId, + manualContinuationOfRunId: sourceId, + status: "completed", + ordinal: 3, + }, + { ...current, ordinal: 4 }, + ], + messages: [ + { ...projection.messages[0]!, id: earlierId, text: "Earlier prompt" }, + { ...projection.messages[0]!, id: commandId, text: command }, + { + ...projection.messages[0]!, + id: resumedMessageId, + text: steered ? "Steered prompt" : "", + }, + ...projection.messages, + ], + attempts: [ + { ...previousAttempt, providerInstanceId: harness.oldInstanceId }, + { + ...previousAttempt, + id: resumedAttemptId, + reason: steered ? "steering_restart" : "initial", + runId: resumedId, + providerInstanceId: harness.newInstanceId, + }, + ...projection.attempts.slice(1), + ], + providerTurns: [ + ...projection.providerTurns, + { + ...projection.providerTurns[0]!, + id: ProviderTurnId.make("resumed-command-turn"), + runAttemptId: resumedAttemptId, + }, + ], + }); + yield* harness.start; + expect( + harness.startTurn.mock.calls[0]?.[0].message.text.includes("switched by the user from"), + ).toBe(!steered); + }), +); + effectIt.effect("leaves delegated children to their own runs when the parent starts a turn", () => Effect.gen(function* () { const harness = makeLocalCommandHarness({ diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 6587e742c7cd0..bae8ae20f8bc5 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -12,6 +12,7 @@ import { type OrchestrationV2TurnItem, isCompactionCommandMessage, parseGoalCommand, + type ProviderInstanceId, RunId, RunAttemptId, NodeId, @@ -118,6 +119,14 @@ const refusedBeforePrompt = (error: unknown): boolean => error.detail === CLAUDE_QUERY_ENDED_BEFORE_TURN_DETAIL) || (Predicate.hasProperty(error, "cause") && refusedBeforePrompt(error.cause)); +// Same-driver account switches resume the native conversation, so nothing else tells the agent. +type NamedInstance = { readonly id: string; readonly displayName?: string | undefined }; +export const providerSwitchNote = (from: NamedInstance, to: NamedInstance): string => { + const name = ({ id, displayName }: NamedInstance) => + displayName === undefined ? `\`${id}\`` : `${displayName} (\`${id}\`)`; + return `Note: this thread now runs on ${name(to)}, switched by the user from ${name(from)}.`; +}; + /** Missing execution records cannot be repaired by retrying a start. */ export class ProviderTurnStartRunStateMissingError extends Schema.TaggedError()( "ProviderTurnStartRunStateMissingError", @@ -199,6 +208,7 @@ export const layer: Layer.Layer< const runtimePolicy = yield* RuntimePolicy.RuntimePolicyV2; const restoreMissingWorktree = yield* makeRestoreMissingWorktree; const turnItemPositions = yield* TurnItemPositionStore.TurnItemPositionStoreV2; + const serverSettings = yield* ServerSettings.ServerSettingsService; // Build draining callbacks outside startup scope so they cannot retain full thread history. const makeRunControls = (input: { @@ -848,18 +858,19 @@ export const layer: Layer.Layer< projection.providerTurns, projection.attempts, ); + const commandMessageIds = new Set( + projection.messages + .filter( + (candidate) => + isCompactionCommandMessage(candidate) || + (goalDriver && parseGoalCommand(candidate) !== null), + ) + .map((candidate) => candidate.id), + ); const restartCancelledWork = pendingRestartCancelledBackgroundWork({ runs: projection.runs, providerTurns: projection.providerTurns, - compactionMessageIds: new Set( - projection.messages - .filter( - (candidate) => - isCompactionCommandMessage(candidate) || - (goalDriver && parseGoalCommand(candidate) !== null), - ) - .map((candidate) => candidate.id), - ), + compactionMessageIds: commandMessageIds, run, attempts: projection.attempts, }); @@ -867,6 +878,41 @@ export const layer: Layer.Layer< restartCancelledWork.length === 0 ? "" : restartCancelledBackgroundWorkNote(restartCancelledWork); + // Attempts, not runs: restarting the active run moves it to the new account. Only a provider + // turn proves a prompt arrived, so a refused one keeps the note pending. + const acceptedAttemptIds = new Set(projection.providerTurns.map((turn) => turn.runAttemptId)); + const previousAttempt = projection.runs + .filter((source) => source.ordinal <= run.ordinal && source.status !== "rolled_back") + .flatMap((source) => + projection.attempts.filter( + (candidate) => + candidate.runId === source.id && + !commandMessageIds.has( + (candidate.reason === "steering_restart" + ? source + : (unsentManualContinuationSource(source, projection) ?? source) + ).userMessageId, + ), + ), + ) + .findLast( + (candidate) => candidate.id !== attempt.id && acceptedAttemptIds.has(candidate.id), + ); + const instances = yield* serverSettings.getSettings.pipe( + Effect.map((settings) => settings.providerInstances), + Effect.orElseSucceed(() => undefined), + ); + const named = (id: ProviderInstanceId) => ({ id, displayName: instances?.[id]?.displayName }); + const switchNote = + commandMessageIds.has(providerMessage.id) || + previousAttempt === undefined || + previousAttempt.providerInstanceId === run.providerInstanceId + ? "" + : providerSwitchNote( + named(previousAttempt.providerInstanceId), + named(run.providerInstanceId), + ); + const turnNote = [restartNote, switchNote].filter((note) => note !== "").join("\n\n"); const tokenCap = yield* handoffTokenCapConfig.pipe( Effect.orElseSucceed(() => DEFAULT_HANDOFF_TOKEN_CAP), ); @@ -1012,7 +1058,7 @@ export const layer: Layer.Layer< return handoffBudget({ tokenCap, modelContextWindow, - userText: restartNote === "" ? userText : `${restartNote}\n\n${userText}`, + userText: turnNote === "" ? userText : `${turnNote}\n\n${userText}`, attachments: providerMessage.attachments, providerThread: budgetProviderThread, nativeContextEstimate: @@ -1732,7 +1778,7 @@ export const layer: Layer.Layer< sideChatHandoff === undefined || compact || goalCommand ? "" : sideChatHandoff.summaryText; - const context = [delivery.context, sideChatContext, restartNote] + const context = [delivery.context, sideChatContext, turnNote] .filter((part) => part !== "") .join("\n\n"); const { restartContinuationOfRunId: _resumedRunId, ...promptedInput } = turnInput; @@ -1810,7 +1856,7 @@ export const layer: Layer.Layer< effectiveHandoffs.length === 0 && retryHandoff.length === 0 && sideChatHandoff === undefined && - restartNote === "" && + turnNote === "" && !noteContinuation ? session : makeDeliverySession(session, startWithHandoffs); diff --git a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts index 7b12f72f4cead..a5758ef46f72c 100644 --- a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts +++ b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts @@ -1016,7 +1016,9 @@ it.live.each(["active", "idle", "selection-command", "pooled", "separate-home"] assert.deepEqual(resumes, [ { instanceId: targetId, nativeId: originalThread.nativeThreadRef?.nativeId }, ]); - assert.deepEqual(messages, ["first", "second"]); + const secondTurn = + "Note: this thread now runs on Provider 2 (`codex-shadow-account`), switched by the user from `codex-restart-test`.\n\nUser message:\nsecond"; + assert.deepEqual(messages, ["first", secondTurn]); assert.equal((yield* Ref.get(state)).closedSessionCount, mode === "pooled" ? 0 : 1); if (mode === "pooled") { assert.isFalse( @@ -1033,7 +1035,19 @@ it.live.each(["active", "idle", "selection-command", "pooled", "separate-home"] assert.deepEqual(returnedThread.nativeThreadRef, originalThread.nativeThreadRef); assert.equal(returnedThread.providerInstanceId, providerInstanceId); assert.isEmpty(third.contextHandoffs); - assert.deepEqual(messages, ["first", "second", "third"]); - }).pipe(Effect.provide(ProviderReplayHarness.layerWithRegistry({ name }, layerRegistry))); + assert.deepEqual(messages, [ + "first", + secondTurn, + "Note: this thread now runs on `codex-restart-test`, switched by the user from Provider 2 (`codex-shadow-account`).\n\nUser message:\nthird", + ]); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry({ name }, layerRegistry, { + serverSettings: { + providerInstances: { [targetId]: { driver: "codex", displayName: "Provider 2" } }, + }, + }), + ), + ); }), ); diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 0ba9137e45dbc..bf019f59538e0 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -288,6 +288,7 @@ export function layerWithRegistry( store: ProjectionStore.ProjectionStoreV2Shape, ) => ProjectionStore.ProjectionStoreV2Shape; readonly managedPolicy?: ManagedPolicyDocument; + readonly serverSettings?: Parameters[0]; } = {}, ): Layer.Layer< | Orchestrator.OrchestratorV2 @@ -321,6 +322,7 @@ export function layerWithRegistry( ServerSettings.layerTest( { responseStreamingMode: "turn", + ...options.serverSettings, ...(options.continueThreadsAfterServerUpdate === undefined ? {} : { continueThreadsAfterServerUpdate: options.continueThreadsAfterServerUpdate }), diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index f07d942cbac2f..1ffabdd17fcb4 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -65,12 +65,16 @@ import { GROK_MODEL_SELECTION, } from "./fixtures/shared.ts"; import * as ProviderReplayHarness from "./ProviderReplayHarness.ts"; +import { withInlineContext } from "../ContextHandoffDelivery.ts"; +import { providerSwitchNote } from "../ProviderTurnStartService.ts"; import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; const encodeJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown)); const threadId = ThreadId.make("thread:provider-switch"); +const switched = (from: string, to: string, text: string) => + withInlineContext(providerSwitchNote({ id: from }, { id: to }), text); const projectId = ProjectId.make("project:provider-switch"); const CODEX_DRIVER = ProviderDriverKind.make("codex"); const CLAUDE_DRIVER = ProviderDriverKind.make("claudeAgent"); @@ -1066,7 +1070,16 @@ describe("orchestration v2 provider switching", () => { ); } else { assert.equal(handoff.delivery?.status, "injected"); - assert.equal((yield* Ref.get(capturedTurns)).at(-1)!.text, current); + assert.equal( + (yield* Ref.get(capturedTurns)).at(-1)!.text, + [ + "large-current-input", + "screenshot-reported-capacity-native", + "screenshot-option-change-retry-native", + ].includes(scenario) + ? switched("codex", "claudeAgent", current) + : current, + ); assert.include( yield* encodeJson(yield* Ref.get(injectedHistory)), "Original request with constraints", @@ -1213,7 +1226,8 @@ describe("orchestration v2 provider switching", () => { yield* wait(3, "completed"); const retried = yield* orchestrator.getThreadProjection(threadId); const lastTurn = (yield* Ref.get(capturedTurns)).at(-1)!; - assert.equal(lastTurn.text, retryText); + // The failed target turn never reached the provider, so the retry carries the note. + assert.equal(lastTurn.text, switched("codex", "claudeAgent", retryText)); if (failure !== "injection" && failure !== "exhausted-target-retry") { const delta = yield* encodeJson( (yield* Ref.get(injectedHistory)).slice(historyBeforeRetry), @@ -1888,7 +1902,10 @@ describe("orchestration v2 provider switching", () => { assert.include(deliveredHistory, partialResponse); assert.include(deliveredHistory, `run-status=${status}`); if (native) { - assert.equal(delivered.text, "Continue"); + assert.equal( + delivered.text, + switched(sourceSelection.instanceId, targetSelection.instanceId, "Continue"), + ); assert.notInclude(deliveredHistory, '"text":"Continue"'); assert.include(deliveredHistory, '"role":"assistant"'); assert.include(deliveredHistory, '"role":"user"'); @@ -2378,6 +2395,7 @@ describe("orchestration v2 provider switching", () => { const cwd = yield* checkpointWorkspace("queued-account-switch"); const capturedTurns = yield* Ref.make>([]); const started = yield* Deferred.make(); + const release = yield* Deferred.make(); const alternateSelection: ModelSelection = { ...CODEX_MODEL_SELECTION, instanceId: ProviderInstanceId.make("codex-alternate"), @@ -2395,6 +2413,7 @@ describe("orchestration v2 provider switching", () => { responseByRunOrdinal: {}, capturedTurns, holdFirstTurn: started, + releaseFirstTurn: release, }), makeTestAdapter({ instanceId: alternateSelection.instanceId, @@ -2438,7 +2457,6 @@ describe("orchestration v2 provider switching", () => { const queuedThreadId = ThreadId.make("thread:queued-account-switch"); const projection = yield* Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; - const eventSink = yield* EventSink.EventSinkV2; const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; yield* orchestrator.dispatch({ type: "thread.create", @@ -2489,20 +2507,7 @@ describe("orchestration v2 provider switching", () => { (providerThread) => providerThread.id === queued.runs[0]?.providerThreadId, )?.nativeThreadRef; assert.isNotNull(sourceNativeRef); - const now = yield* DateTime.now; - yield* eventSink.write({ - events: [ - { - id: EventId.make("event:queued-account-switch:first-complete"), - type: "run.updated", - threadId: queuedThreadId, - runId: queued.runs[0]!.id, - providerInstanceId: CODEX_MODEL_SELECTION.instanceId, - occurredAt: now, - payload: { ...queued.runs[0]!, status: "completed", completedAt: now }, - }, - ], - }); + yield* Deferred.succeed(release, undefined); yield* orchestrator.resumeQueuedRuns; yield* orchestrator.streamStoredEvents.pipe( Stream.filter( @@ -2549,7 +2554,10 @@ describe("orchestration v2 provider switching", () => { assert.lengthOf(projection.contextHandoffs, 0); assert.deepEqual( (yield* Ref.get(capturedTurns)).map((turn) => turn.text), - ["First account turn", "Alternate account turn"], + [ + "First account turn", + "Note: this thread now runs on `codex-alternate`, switched by the user from `codex`.\n\nUser message:\nAlternate account turn", + ], ); }), ); diff --git a/evals/mcp-tool-use/cases/orchestration.ndjson b/evals/mcp-tool-use/cases/orchestration.ndjson index df13d9cd6313b..ae80bd233121c 100644 --- a/evals/mcp-tool-use/cases/orchestration.ndjson +++ b/evals/mcp-tool-use/cases/orchestration.ndjson @@ -44,4 +44,6 @@ {"id": "settle-after-delayed-merge-waits", "area": "orchestration", "kind": "guard", "pattern": "settle before delayed work finishes", "experience": "Asked to merge a PR once its checks pass and then settle this thread, an agent following the #26749 settle bullet can settle in the first turn; a settled thread refuses watch_pull_request and drops PR-watch wakes, so the merge never happens.", "rule": "Asked to settle ... a thread, including this one: call t3_thread_organize. (the request is to settle after the merge, not now)", "privacy": "commit-safe", "notes": "From the CodeRabbit finding on #26765. Settling this thread applies at turn end, and Orchestrator.ts rejects watch wakes on a settled thread. Good: watch #123 and end the turn without settling. gh pr merge --auto plus a settle would still lose the wake that confirms the merge.", "setup": {"role": "top", "priorTurns": [{"role": "user", "text": "Okay this looks good to me. Any risk merging it?"}, {"role": "assistant", "text": "Low risk. [#123](https://github.com/acme/app/pull/123) only adds an accessible title to the HTML preview frame, and the new test covers it. Two checks are still running: Check and Test Web.", "toolCalls": [{"tool": "Bash", "input": {"command": "gh pr view 123 --json title,files --jq '.title, .files[].path'"}, "result": "fix(web): HTML preview frame gets an accessible title\napps/web/src/HtmlPreview.tsx\napps/web/src/HtmlPreview.test.tsx"}, {"tool": "Bash", "input": {"command": "gh pr checks 123"}, "result": "Lint\tpass\t41s\nTypecheck\tpass\t1m2s\nCheck\tpending\t0\nTest Web\tpending\t0"}, {"tool": "link_pull_request", "input": {"url": "https://github.com/acme/app/pull/123"}, "result": {"host": "github.com", "repository": "acme/app", "number": 123, "url": "https://github.com/acme/app/pull/123", "alreadyLinked": false}}]}], "mocks": {"watch_pull_request": "default", "link_pull_request": "default", "t3_thread_organize": {"settlesWhenTurnEnds": true}}, "world": {"branch": "fix/html-preview-title", "commits": [{"sha": "abc1234", "subject": "fix(web): HTML preview frame gets an accessible title"}], "diff": "diff --git a/apps/web/src/HtmlPreview.tsx b/apps/web/src/HtmlPreview.tsx\n--- a/apps/web/src/HtmlPreview.tsx\n+++ b/apps/web/src/HtmlPreview.tsx\n@@ -1,3 +1,3 @@\n export function HtmlPreview({ html }: { html: string }) {\n- return