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

apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts

@@ -9693,7 +9693,7 @@ describe("CodexAdapterV2 post-settle continuation", () => {
       ),
     );
 
-    it.effect("sends a /goal objective at the 4000 character limit unchanged", () =>
+    it.effect("sends an agent /goal objective at the 4000 character limit unchanged", () =>
       Effect.gen(function* () {
         const atLimit = "y".repeat(4000);
         const goal = (status: string) => ({ ...codexGoal(status, 0), objective: atLimit });
@@ -9713,7 +9713,7 @@ describe("CodexAdapterV2 post-settle continuation", () => {
                 nativeThreadId,
                 nativeTurnId: "goal-turn-limit",
                 prompt: atLimit,
-                sentPrompt: `This thread's goal is already set. Do not call create_goal. Start working on it now: ${atLimit}`,
+                sentPrompt: `[Message from another thread's agent (thread "thread:sender"). This is agent-authored coordination, not a message from the human user.]\n\nThis thread's goal is already set. Do not call create_goal. Start working on it now: ${atLimit}`,
               }).slice(5),
               5,
             ),
@@ -9735,7 +9735,15 @@ describe("CodexAdapterV2 post-settle continuation", () => {
           ],
         });
         const harness = yield* makeCodexReplayHarness(transcript);
-        yield* harness.runtime.startTurn(yield* goalTurnInput(harness, `/goal ${atLimit}`));
+        const input = yield* goalTurnInput(harness, `/goal ${atLimit}`);
+        yield* harness.runtime.startTurn({
+          ...input,
+          message: {
+            ...input.message,
+            createdBy: "agent",
+            senderThreadId: ThreadId.make("thread:sender"),
+          },
+        });
         yield* harness.firstTerminal;
 
         assert.deepEqual(

apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts

@@ -104,6 +104,7 @@ import * as Stream from "effect/Stream";
 import { ChildProcess, ChildProcessSpawner } from "effect/process";
 
 import { resolveAttachmentPath, resolveAttachmentPathById } from "../../attachmentStore.ts";
+import { withProviderMessageOrigin } from "../providerMessageOrigin.ts";
 import * as PortScanner from "../../preview/PortScanner.ts";
 import { getCodexServiceTierOptionValue } from "../../codexModelOptions.ts";
 import { ServerConfig } from "../../config.ts";
@@ -7440,7 +7441,11 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
                     ...turnInput,
                     message: {
                       ...turnInput.message,
-                      text: `This thread's goal is already set. Do not call create_goal. Start working on it now: ${command.objective}`,
+                      text: withProviderMessageOrigin(
+                        turnInput.message,
+                        `This thread's goal is already set. Do not call create_goal. Start working on it now: ${command.objective}`,
+                        CODEX_PROVIDER,
+                      ),
                     },
                   })
                 : Effect.succeed([]),

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

@@ -13,8 +13,11 @@ import {
 } from "@t3tools/provider-core/server/handoffBudget";
 
 /** The provider prompt for a turn that carries inline context ahead of the user's text. */
-export const withInlineContext = (context: string, userText: string): string =>
-  context === "" ? userText : `${context}\n\nUser message:\n${userText}`;
+export const withInlineContext = (
+  context: string,
+  userText: string,
+  messageHeading = "User message:",
+): string => (context === "" ? userText : `${context}\n\n${messageHeading}\n${userText}`);
 
 /** Persist around injection; ambiguous delivery requires a fresh native thread. Call unsent if refused before prompt consumption. */
 export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextHandoffs")(

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

@@ -275,6 +275,7 @@ it.effect.each(storageCases)(
       const layerSessions = Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({
         get: () =>
           Effect.succeedSome({
+            driver,
             interruptTurn: () =>
               Effect.sync(() => {
                 calls.push("interrupt");
@@ -308,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(
@@ -331,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"),
@@ -341,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

@@ -20,6 +20,7 @@ import * as Schema from "effect/Schema";
 import * 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(
@@ -412,10 +413,14 @@ export const layer: Layer.Layer<
                 : { modelSelection: input.modelSelection }),
               message: {
                 messageId: message.id,
-                text: projectComposerContextForProvider({
-                  text: message.text,
-                  records: message.context?.records ?? [],
-                }),
+                text: withProviderMessageOrigin(
+                  message,
+                  projectComposerContextForProvider({
+                    text: message.text,
+                    records: message.context?.records ?? [],
+                  }),
+                  loaded.session.value.driver,
+                ),
                 attachments: message.attachments,
                 createdBy: message.createdBy,
                 creationSource: message.creationSource,

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

@@ -61,6 +61,7 @@ import {
   makeProviderFailure,
 } from "@t3tools/provider-core/server/failure";
 import * as RunExecutionService from "./RunExecutionService.ts";
+import { isCrossThreadAgentMessage } from "./providerMessageOrigin.ts";
 import * as RuntimePolicy from "./RuntimePolicy.ts";
 import * as TurnItemPositionStore from "./TurnItemPositionStore.ts";
 import {
@@ -870,10 +871,7 @@ export const layer: Layer.Layer<
         records: message.context?.records ?? [],
       });
       // Native goal commands go verbatim; handoff history waits a turn.
-      const goalDriver =
-        session.driver === "codex" ||
-        session.driver === "claudeAgent" ||
-        session.providerSession.capabilities.turns.supportsNativeGoals === true;
+      const goalDriver = RunExecutionService.acceptsNativeGoalCommands(session);
       const goalCommand = goalDriver && parseGoalCommand(message) !== null;
       // The provider turn marks cancellation work delivered; restart continuations use their own text or resume natively.
       const noteContinuation = isRestartNoteContinuation(
@@ -1756,11 +1754,16 @@ export const layer: Layer.Layer<
             .filter((part) => part !== "")
             .join("\n\n");
           const { restartContinuationOfRunId: _resumedRunId, ...promptedInput } = turnInput;
+          const messageHeading = isCrossThreadAgentMessage(turnInput.message)
+            ? "Agent message:"
+            : "User message:";
           yield* start({
             ...(noteContinuation ? promptedInput : turnInput),
             message: {
               ...turnInput.message,
-              text: goalCommand ? userText : withInlineContext(context, userText),
+              text: goalCommand
+                ? turnInput.message.text
+                : withInlineContext(context, turnInput.message.text, messageHeading),
             },
           }).pipe(
             // Clear pending handoff markers for refused prompts so the next turn retains the native session.