fix(codex): isolate unowned provider thread events

apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts

@@ -29,6 +29,10 @@ import { HostProcessPlatform } from "@t3tools/shared/hostProcess";
 const ROOT = wireFixture.rootThreadId;
 const [CHILD_A, CHILD_B] = wireFixture.childThreadIds as [string, string];
 const MEMORY = "memory-consolidation-thread";
+const FOREIGN = "unregistered-foreign-thread";
+const FOREIGN_RECEIVER = "unregistered-foreign-receiver";
+const FOREIGN_ACTIVITY_CHILD = "unregistered-foreign-activity-child";
+const NESTED_ACTIVITY_CHILD = "owned-nested-activity-child";
 const decodeMcpElicitationResponse = Schema.decodeUnknownEffect(
   Schema.fromJsonString(
     Schema.Struct({
@@ -62,6 +66,85 @@ function buildScript() {
         },
       },
     },
+    // An already-owned child can establish ownership for descendants.
+    {
+      method: "item/completed",
+      params: {
+        completedAtMs: 5,
+        threadId: CHILD_A,
+        turnId: `${CHILD_A}-turn-1`,
+        item: {
+          id: "nested-sub-agent-activity",
+          type: "subAgentActivity",
+          kind: "started",
+          agentThreadId: NESTED_ACTIVITY_CHILD,
+          agentPath: "/root/alpha/nested",
+        },
+      },
+    },
+    // Foreign collaboration announcements cannot establish ownership for
+    // receiver IDs or subAgentActivity children in this runtime.
+    {
+      method: "item/completed",
+      params: {
+        completedAtMs: 2,
+        threadId: FOREIGN,
+        turnId: "foreign-turn",
+        item: {
+          id: "foreign-collab-call",
+          type: "collabAgentToolCall",
+          tool: "wait",
+          status: "completed",
+          senderThreadId: FOREIGN,
+          receiverThreadIds: [FOREIGN_RECEIVER],
+        },
+      },
+    },
+    {
+      method: "item/completed",
+      params: {
+        completedAtMs: 3,
+        threadId: FOREIGN_RECEIVER,
+        turnId: "foreign-receiver-turn",
+        item: {
+          id: "foreign-receiver-message",
+          type: "agentMessage",
+          phase: "final_answer",
+          text: "foreign receiver report",
+        },
+      },
+    },
+    {
+      method: "item/completed",
+      params: {
+        completedAtMs: 4,
+        threadId: FOREIGN,
+        turnId: "foreign-turn",
+        item: {
+          id: "foreign-sub-agent-activity",
+          type: "subAgentActivity",
+          kind: "started",
+          agentThreadId: FOREIGN_ACTIVITY_CHILD,
+          agentPath: "/root/foreign",
+        },
+      },
+    },
+    // A different app-server session can emit an assistant item without a
+    // preceding thread/started notification. It must not become parent chat.
+    {
+      method: "item/completed",
+      params: {
+        completedAtMs: 1,
+        threadId: FOREIGN,
+        turnId: "foreign-turn",
+        item: {
+          id: "foreign-message",
+          type: "agentMessage",
+          phase: "final_answer",
+          text: "unrelated background report",
+        },
+      },
+    },
     // Child terminal lifecycle AFTER the receiver map knows the children —
     // pre-fix, the legacy suppressor dropped these before interception saw
     // them, so no synthetic agent events were emitted.
@@ -166,6 +249,90 @@ const peerPath = NodePath.join(
 );
 
 describe("CodexSessionRuntime collab integration", () => {
+  it.effect("keeps startup ownership on the authoritative thread response", () =>
+    Effect.gen(function* () {
+      const script = {
+        rootThreadId: ROOT,
+        startupResponseDelayMs: 25,
+        startupNotifications: [
+          capturedSpawnedThread(FOREIGN),
+          {
+            method: "item/agentMessage/delta",
+            params: {
+              delta: "foreign startup content",
+              itemId: "foreign-startup-message",
+              threadId: FOREIGN,
+              turnId: "foreign-startup-turn",
+            },
+          },
+        ],
+        notifications: [
+          {
+            method: "item/agentMessage/delta",
+            params: {
+              delta: "owned root content",
+              itemId: "root-message",
+              threadId: ROOT,
+              turnId: `${ROOT}-turn`,
+            },
+          },
+          { method: "warning", params: { message: "foreign warning", threadId: FOREIGN } },
+          { method: "warning", params: { message: "root warning", threadId: ROOT } },
+        ],
+      };
+      // @effect-diagnostics-next-line preferSchemaOverJson:off
+      NodeFS.writeFileSync(scriptPath, JSON.stringify(script), "utf8");
+      yield* Effect.addFinalizer(() =>
+        Effect.sync(() => NodeFS.rmSync(scriptPath, { force: true })),
+      );
+
+      const runtime = yield* makeCodexSessionRuntime({
+        threadId: ThreadId.make("thread-startup-ownership"),
+        binaryPath: peerPath,
+        cwd: NodeOS.tmpdir(),
+        runtimeMode: "full-access",
+        environment: { ...process.env, T3_CODEX_COLLAB_SCRIPT: scriptPath },
+      });
+      const eventsFiber = yield* runtime.events.pipe(
+        Stream.takeUntil((event) => event.method === "turn/completed"),
+        Stream.runCollect,
+        Effect.forkScoped,
+      );
+
+      const session = yield* runtime.start();
+      assert.deepEqual(session.resumeCursor, { threadId: ROOT });
+      yield* runtime.sendTurn({ input: "continue on the owned root" });
+      const events = Array.from(yield* Fiber.join(eventsFiber));
+
+      assert.isFalse(
+        events.some((event) => event.textDelta === "foreign startup content"),
+        "foreign content emitted before the root response must stay out of chat",
+      );
+      assert.isTrue(
+        events.some((event) => event.textDelta === "owned root content"),
+        "root content remains visible after authoritative ownership is established",
+      );
+      assert.isFalse(
+        events.some(
+          (event) =>
+            event.method === "warning" &&
+            (event.payload as { message?: string } | undefined)?.message === "foreign warning",
+        ),
+        "foreign thread warnings must stay out of the root conversation",
+      );
+      assert.isTrue(
+        events.some(
+          (event) =>
+            event.method === "warning" &&
+            (event.payload as { message?: string } | undefined)?.message === "root warning",
+        ),
+        "root thread warnings remain visible",
+      );
+
+      yield* runtime.close;
+    }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
+  );
+
   it.effect("looks up child model metadata once after activity registration", () =>
     Effect.gen(function* () {
       const script = {
@@ -468,6 +635,32 @@ describe("CodexSessionRuntime collab integration", () => {
         [],
         "child thread/* lifecycle must not appear as parent events",
       );
+      assert.isFalse(
+        events.some(
+          (event) => (event.payload as { threadId?: string } | undefined)?.threadId === FOREIGN,
+        ),
+        "unregistered foreign assistant items must not appear as parent events",
+      );
+      assert.isFalse(
+        events.some((event) => {
+          const payload = event.payload as
+            | { threadId?: string; agentThreadId?: string }
+            | undefined;
+          return (
+            payload?.threadId === FOREIGN_RECEIVER ||
+            payload?.agentThreadId === FOREIGN_ACTIVITY_CHILD
+          );
+        }),
+        "foreign collaboration announcements must not establish child ownership",
+      );
+      assert.isTrue(
+        events.some(
+          (event) =>
+            (event.payload as { agentThreadId?: string } | undefined)?.agentThreadId ===
+            NESTED_ACTIVITY_CHILD,
+        ),
+        "owned children can register nested subAgentActivity descendants",
+      );
 
       yield* runtime.close;
     }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),

apps/server/src/provider/Layers/CodexSessionRuntime.test.ts

@@ -18,6 +18,7 @@ import {
   isRecoverableThreadResumeError,
   makeMemoryConsolidationNotificationFilter,
   openCodexThread,
+  shouldSuppressUnownedCodexNotification,
   readCodexThread,
   rollbackCodexThread,
   toMcpElicitationResponse,
@@ -802,6 +803,121 @@ describe("makeMemoryConsolidationNotificationFilter", () => {
   });
 });
 
+describe("shouldSuppressUnownedCodexNotification", () => {
+  const agentMessageDelta = (threadId: string, delta: string) => ({
+    method: "item/agentMessage/delta" as const,
+    params: {
+      delta,
+      itemId: `${threadId}-message`,
+      threadId,
+      turnId: `${threadId}-turn`,
+    },
+  });
+
+  it("suppresses unregistered foreign assistant text without hiding root or child replies", () => {
+    const foreignNotifications: ReadonlyArray<
+      Parameters<typeof shouldSuppressUnownedCodexNotification>[0]
+    > = [
+      agentMessageDelta("foreign-thread", "unrelated background report"),
+      {
+        method: "item/started" as const,
+        params: {
+          startedAtMs: 1,
+          threadId: "foreign-thread",
+          turnId: "foreign-turn",
+          item: {
+            id: "foreign-message",
+            type: "agentMessage" as const,
+            text: "",
+          },
+        },
+      },
+      {
+        method: "item/completed" as const,
+        params: {
+          completedAtMs: 2,
+          threadId: "foreign-thread",
+          turnId: "foreign-turn",
+          item: {
+            id: "foreign-message",
+            type: "agentMessage" as const,
+            text: "unrelated background report",
+          },
+        },
+      },
+    ];
+    for (const notification of foreignNotifications) {
+      NodeAssert.equal(
+        shouldSuppressUnownedCodexNotification(notification, "root-thread", false),
+        true,
+      );
+    }
+    NodeAssert.equal(
+      shouldSuppressUnownedCodexNotification(
+        agentMessageDelta("root-thread", "root commentary"),
+        "root-thread",
+        false,
+      ),
+      false,
+    );
+    NodeAssert.equal(
+      shouldSuppressUnownedCodexNotification(
+        agentMessageDelta("registered-child", "child reply"),
+        "root-thread",
+        true,
+      ),
+      false,
+    );
+    NodeAssert.equal(
+      shouldSuppressUnownedCodexNotification(
+        { method: "warning", params: { message: "foreign warning", threadId: "foreign-thread" } },
+        "root-thread",
+        false,
+      ),
+      true,
+    );
+    NodeAssert.equal(
+      shouldSuppressUnownedCodexNotification(
+        { method: "warning", params: { message: "root warning", threadId: "root-thread" } },
+        "root-thread",
+        false,
+      ),
+      false,
+    );
+  });
+
+  it("keeps foreign request resolution on the parent correlation path", () => {
+    NodeAssert.equal(
+      shouldSuppressUnownedCodexNotification(
+        {
+          method: "serverRequest/resolved",
+          params: { requestId: "request-1", threadId: "foreign-thread" },
+        },
+        "root-thread",
+        false,
+      ),
+      false,
+    );
+  });
+
+  it("suppresses thread-addressed startup traffic until the root response establishes ownership", () => {
+    const threadStarted = makeThreadStartedNotification("foreign-thread", "appServer");
+    NodeAssert.equal(shouldSuppressUnownedCodexNotification(threadStarted, undefined, false), true);
+    NodeAssert.equal(
+      shouldSuppressUnownedCodexNotification(threadStarted, "root-thread", false),
+      true,
+    );
+    NodeAssert.equal(
+      shouldSuppressUnownedCodexNotification(
+        { method: "warning", params: { message: "global warning" } },
+        undefined,
+        false,
+      ),
+      false,
+    );
+  });
+});
+
 describe("codexSessionAppServerArgs", () => {
   it("keeps the app-server subcommand when explicit args are provided", () => {
     NodeAssert.deepStrictEqual(codexSessionAppServerArgs(["-c", "model=gpt-5"], undefined), [