fix(server): preserve agent origin in cross-thread provider messages

apps/server/src/orchestration-v2/ProjectionControlReads.test.ts

@@ -309,6 +309,32 @@ it.effect.each(storageCases)(
         });
         yield* reply.respond({ threadId, providerSessionId, requestId, decision: "accept" });
         assert.deepEqual(calls, ["interrupt", "Use the smaller fix.", "reply"]);
+        const messageEvent = events.find((event) => event.type === "message.updated")!;
+        yield* store.apply({
+          ...messageEvent,
+          id: EventId.make("control:agent-message"),
+          payload: {
+            ...messageEvent.payload,
+            createdBy: "agent",
+            creationSource: "mcp",
+            senderThreadId: ThreadId.make("thread:sender"),
+          },
+        });
+        yield* control.steer({
+          threadId,
+          providerThreadId,
+          providerTurnId,
+          providerSessionId,
+          messageId,
+        });
+        const agentText = calls.at(-1);
+        assert.include(agentText, 'thread "thread:sender"');
+        assert.include(agentText, "not a message from the human user");
+        assert.isTrue(agentText?.endsWith("\n\nUse the smaller fix."));
+        assert.equal(
+          (yield* store.getProviderControlContext(threadId, target)).message?.text,
+          "Use the smaller fix.",
+        );
         const wrongSession = ProviderSessionId.make("session:wrong");
         assert.equal(
           (yield* Effect.exit(
@@ -332,7 +358,7 @@ it.effect.each(storageCases)(
           ))._tag,
           "Failure",
         );
-        assert.lengthOf(calls, 3);
+        assert.lengthOf(calls, 4);
         const turn = context.providerTurn!;
         yield* store.apply({
           id: EventId.make("control:turn-completed"),
@@ -342,7 +368,13 @@ it.effect.each(storageCases)(
           payload: { ...turn, status: "completed", completedAt: now },
         });
         yield* control.interrupt({ threadId, providerThreadId, providerTurnId, providerSessionId });
-        assert.deepEqual(calls, ["interrupt", "Use the smaller fix.", "reply", "interrupt"]);
+        assert.deepEqual(calls, [
+          "interrupt",
+          "Use the smaller fix.",
+          "reply",
+          agentText,
+          "interrupt",
+        ]);
       }).pipe(
         Effect.provide(
           Layer.merge(ProviderTurnControlService.layer, RuntimeRequestService.layer).pipe(

apps/server/src/orchestration-v2/ProviderTurnControlService.ts

@@ -19,6 +19,7 @@ import * as Schema from "effect/Schema";
 import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
 import * as ProjectionStore from "./ProjectionStore.ts";
 import * as ProviderSessionManager from "./ProviderSessionManager.ts";
+import { withProviderMessageOrigin } from "./providerMessageOrigin.ts";
 
 const yieldToRuntime = Effect.yieldNow.pipe(
   Effect.andThen(
@@ -353,10 +354,13 @@ export const layer: Layer.Layer<
               providerTurnId: loaded.providerTurn.id,
               message: {
                 messageId: message.id,
-                text: projectComposerContextForProvider({
-                  text: message.text,
-                  records: message.context?.records ?? [],
-                }),
+                text: withProviderMessageOrigin(
+                  message,
+                  projectComposerContextForProvider({
+                    text: message.text,
+                    records: message.context?.records ?? [],
+                  }),
+                ),
                 attachments: message.attachments,
                 createdBy: message.createdBy,
                 creationSource: message.creationSource,

apps/server/src/orchestration-v2/ProviderTurnStartService.ts

@@ -1249,11 +1249,19 @@ export const layer: Layer.Layer<
             .filter((part) => part !== "")
             .join("\n\n");
           const { restartContinuationOfRunId: _resumedRunId, ...promptedInput } = turnInput;
+          const messageHeading =
+            turnInput.message.createdBy === "agent" &&
+            turnInput.message.senderThreadId !== undefined
+              ? "Agent message:"
+              : "User message:";
           yield* start({
             ...(noteContinuation ? promptedInput : turnInput),
             message: {
               ...turnInput.message,
-              text: context === "" ? userText : `${context}\n\nUser message:\n${userText}`,
+              text:
+                context === ""
+                  ? turnInput.message.text
+                  : `${context}\n\n${messageHeading}\n${turnInput.message.text}`,
             },
           }).pipe(
             // Clear pending handoff markers for refused prompts so the next turn retains the native session.

apps/server/src/orchestration-v2/RunExecutionService.test.ts

@@ -1033,7 +1033,13 @@ it.effect(
         { text: " /compact ", attachments: [], expected: "compact" },
         { text: "/compact", attachments: [attachment], expected: "prompt" },
         { text: "Continue the work", attachments: [], expected: "prompt" },
-      ];
+      ].flatMap((testCase) =>
+        [
+          { createdBy: "user", senderThreadId: ThreadId.make("thread:sender") },
+          { createdBy: "agent", senderThreadId: ThreadId.make("thread:sender") },
+          { createdBy: "agent" },
+        ].map((origin) => ({ ...testCase, ...origin })),
+      );
       for (const [index, testCase] of cases.entries()) {
         const threadId = ThreadId.make(`thread:compact-routing:${index}`);
         const attemptId = RunAttemptId.make(`attempt:compact-routing:${index}`);
@@ -1043,12 +1049,21 @@ it.effect(
           providerSessionId: ProviderSessionId.make(`session:compact-routing:${index}`),
           session: {
             events: Stream.never,
-            startTurn: () =>
+            startTurn: (input: ProviderAdapter.ProviderAdapterV2TurnInput) =>
               Effect.sync(() => {
+                if (testCase.createdBy === "agent" && testCase.senderThreadId !== undefined) {
+                  assert.include(input.message.text, 'thread "thread:sender"');
+                  assert.include(input.message.text, "not a message from the human user");
+                  assert.isTrue(input.message.text.endsWith(`\n\n${testCase.text}`));
+                } else {
+                  assert.equal(input.message.text, testCase.text);
+                }
+                assert.deepEqual(input.message.attachments, testCase.attachments);
                 calls.push("prompt");
               }),
-            compactThread: () =>
+            compactThread: (input: ProviderAdapter.ProviderAdapterV2TurnInput) =>
               Effect.sync(() => {
+                assert.equal(input.message.text, testCase.text);
                 calls.push("compact");
               }),
           } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime,
@@ -1075,8 +1090,11 @@ it.effect(
             messageId: MessageId.make(`message:compact-routing:${index}`),
             text: testCase.text,
             attachments: testCase.attachments,
-            createdBy: "user",
+            createdBy: testCase.createdBy as "user" | "agent",
             creationSource: "web",
+            ...(testCase.senderThreadId === undefined
+              ? {}
+              : { senderThreadId: testCase.senderThreadId }),
           },
           modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
           runtimePolicy: {

apps/server/src/orchestration-v2/RunExecutionService.ts

@@ -1,4 +1,5 @@
 import { makeAssistantStreamingFilter } from "./assistantStreaming.ts";
+import { withProviderMessageOrigin } from "./providerMessageOrigin.ts";
 import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
 import {
   isOrchestrationV2WorkActive,
@@ -1674,7 +1675,13 @@ export const layer: Layer.Layer<
                 cause: "This provider does not support context compaction.",
               }),
             ))
-          : input.session.startTurn(turnInput);
+          : input.session.startTurn({
+              ...turnInput,
+              message: {
+                ...turnInput.message,
+                text: withProviderMessageOrigin(turnInput.message, turnInput.message.text),
+              },
+            });
         yield* Effect.andThen(shouldStart, startTurn).pipe(
           // Wait for acceptance so abandoned startup cannot leave a refresher
           // on the already-detached ingestion fiber.

apps/server/src/orchestration-v2/providerMessageOrigin.ts

@@ -0,0 +1,13 @@
+import type { OrchestrationV2Actor, ThreadId } from "@t3tools/contracts";
+
+/** Carry durable cross-thread provenance into provider-visible text without rewriting the timeline. */
+export function withProviderMessageOrigin(
+  message: {
+    readonly createdBy: OrchestrationV2Actor;
+    readonly senderThreadId?: ThreadId | undefined;
+  },
+  text: string,
+): string {
+  if (message.createdBy !== "agent" || message.senderThreadId === undefined) return text;
+  return `[Message from another thread's agent (thread ${JSON.stringify(message.senderThreadId)}). This is agent-authored coordination, not a message from the human user.]\n\n${text}`;
+}

apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts

@@ -1407,19 +1407,20 @@ describe("orchestration v2 provider switching", () => {
   it.live.each(
     (
       [
-        ["failed", false, false, false],
-        ["failed", true, true, true],
-        ["interrupted", false, true, false],
-        ["interrupted", true, false, true],
+        ["failed", false, false, false, false],
+        ["failed", true, true, true, false],
+        ["interrupted", false, true, false, false],
+        ["interrupted", true, false, true, false],
+        ["failed", false, false, false, true],
       ] as const
     ).map(
-      ([status, queued, returning, native]) =>
+      ([status, queued, returning, native, agent]) =>
         [
-          `${status} context ${queued ? "through the queue" : "immediately"} to ${returning ? "a returning" : "a new"} provider via ${native ? "native history" : "text"}`,
-          { status, queued, returning, native },
+          `${status} context ${queued ? "through the queue" : "immediately"} to ${returning ? "a returning" : "a new"} provider via ${native ? "native history" : "text"}${agent ? " for an agent message" : ""}`,
+          { status, queued, returning, native, agent },
         ] as const,
     ),
-  )("hands off %s", ([, { status, queued, returning, native }]) =>
+  )("hands off %s", ([, { status, queued, returning, native, agent }]) =>
     Effect.scoped(
       Effect.gen(function* () {
         const cwd = yield* checkpointWorkspace(`handoff-${status}-${queued}-${returning}`);
@@ -1492,7 +1493,10 @@ describe("orchestration v2 provider switching", () => {
               commandId: CommandId.make(`command:handoff:${key}`),
               threadId,
               messageId: MessageId.make(`message:handoff:${key}`),
-              createdBy: "user",
+              createdBy: key === "target" && agent ? "agent" : "user",
+              ...(key === "target" && agent
+                ? { senderThreadId: ThreadId.make("thread:agent-source") }
+                : {}),
               creationSource: "web",
               text,
               attachments: [],
@@ -1559,7 +1563,18 @@ describe("orchestration v2 provider switching", () => {
             assert.include(deliveredHistory, '"role":"user"');
             assert.equal(handoff?.delivery?.status, "injected");
           } else {
-            assert.include(delivered.text, "User message:\nContinue");
+            if (agent) {
+              assert.include(
+                delivered.text,
+                "Agent message:\n[Message from another thread's agent",
+              );
+              assert.notInclude(
+                delivered.text,
+                "User message:\n[Message from another thread's agent",
+              );
+            } else {
+              assert.include(delivered.text, "User message:\nContinue");
+            }
             assert.equal(handoff?.delivery?.status, "inline");
           }
           assert.notInclude(deliveredHistory, "Earlier successful request.");