fix(server): agents learn when the user moves their thread to another account

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({

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>()(
   "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,25 +858,61 @@ 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,
       });
       const restartNote =
         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);

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" } },
+            },
+          }),
+        ),
+      );
     }),
 );

apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts

@@ -288,6 +288,7 @@ export function layerWithRegistry<Error>(
       store: ProjectionStore.ProjectionStoreV2Shape,
     ) => ProjectionStore.ProjectionStoreV2Shape;
     readonly managedPolicy?: ManagedPolicyDocument;
+    readonly serverSettings?: Parameters<typeof ServerSettings.layerTest>[0];
   } = {},
 ): Layer.Layer<
   | Orchestrator.OrchestratorV2
@@ -321,6 +322,7 @@ export function layerWithRegistry<Error>(
       ServerSettings.layerTest(
         {
           responseStreamingMode: "turn",
+          ...options.serverSettings,
           ...(options.continueThreadsAfterServerUpdate === undefined
             ? {}
             : { continueThreadsAfterServerUpdate: options.continueThreadsAfterServerUpdate }),

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<ReadonlyArray<CapturedTurn>>([]);
       const started = yield* Deferred.make<void>();
+      const release = yield* Deferred.make<void>();
       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",
+        ],
       );
     }),
   );