fix(codex): warn and continue on projector drift (#99287)

* fix(codex): diagnose projector protocol drift

Co-authored-by: luyifan <al3060388206@gmail.com>

* fix(codex): warn and continue on projector drift

Co-authored-by: luyifan <al3060388206@gmail.com>

* test(codex): publish materialized auth snapshot

Co-authored-by: luyifan <al3060388206@gmail.com>

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>

extensions/codex/src/app-server/event-projector-diagnostics.ts

@@ -0,0 +1,89 @@
+import { embeddedAgentLog } from "openclaw/plugin-sdk/agent-harness-runtime";
+import { redactSensitiveText } from "openclaw/plugin-sdk/logging-core";
+import { sanitizeTerminalText } from "openclaw/plugin-sdk/text-chunking";
+import { unknownItemStatus } from "./event-projector-items.js";
+import {
+  readCodexNotificationThreadId,
+  readCodexNotificationTurnId,
+} from "./notification-correlation.js";
+import type { CodexServerNotification, CodexThreadItem, JsonObject } from "./protocol.js";
+
+const KNOWN_NOOP_NOTIFICATION_METHODS = new Set([
+  "thread/compacted",
+  "turn/started",
+  "turn/diff/updated",
+  "item/reasoning/summaryPartAdded",
+  "item/commandExecution/terminalInteraction",
+  "item/fileChange/outputDelta",
+  "item/fileChange/patchUpdated",
+  "item/mcpToolCall/progress",
+  "model/rerouted",
+  "model/verification",
+  "turn/moderationMetadata",
+  "model/safetyBuffering/updated",
+]);
+
+export function isKnownNoopCodexNotificationMethod(method: string): boolean {
+  return KNOWN_NOOP_NOTIFICATION_METHODS.has(method);
+}
+
+export function redactCodexEventKind(method: string): string {
+  return redactSensitiveText(sanitizeTerminalText(method));
+}
+
+export class CodexProjectionDiagnostics {
+  private readonly warningKeys = new Set<string>();
+
+  constructor(
+    private readonly threadId: string,
+    private readonly turnId: string,
+  ) {}
+
+  warnUnknownItemStatus(item: CodexThreadItem | undefined): void {
+    if (!item) {
+      return;
+    }
+    const status = unknownItemStatus(item);
+    if (!status) {
+      return;
+    }
+    const safeStatus = redactCodexEventKind(status);
+    const safeItemType = redactCodexEventKind(item.type);
+    this.warnOnce(
+      JSON.stringify(["status", item.type, status]),
+      "codex app-server item reported unknown status; continuing projection",
+      {
+        itemId: item.id,
+        itemType: safeItemType,
+        status: safeStatus,
+      },
+    );
+  }
+
+  warnUnknownEvent(notification: CodexServerNotification, params: JsonObject): void {
+    const notificationThreadId = readCodexNotificationThreadId(params);
+    const notificationTurnId = readCodexNotificationTurnId(params);
+    const eventKind = redactCodexEventKind(notification.method);
+    this.warnOnce(
+      JSON.stringify(["method", notification.method]),
+      `codex app-server projector received unknown event kind; continuing: ${eventKind}`,
+      {
+        eventKind,
+        activeThreadId: this.threadId,
+        activeTurnId: this.turnId,
+        threadId: notificationThreadId,
+        turnId: notificationTurnId,
+        matchesActiveThread: notificationThreadId === this.threadId,
+        matchesActiveTurn: notificationTurnId === this.turnId,
+      },
+    );
+  }
+
+  private warnOnce(key: string, message: string, context: Record<string, unknown>): void {
+    if (this.warningKeys.has(key)) {
+      return;
+    }
+    this.warningKeys.add(key);
+    embeddedAgentLog.warn(message, context);
+  }
+}

extensions/codex/src/app-server/event-projector-items.ts

@@ -63,6 +63,23 @@ export function itemStatus(item: CodexThreadItem): "completed" | "failed" | "run
   return "completed";
 }
 
+export function unknownItemStatus(item: CodexThreadItem): string | undefined {
+  const status = readItemString(item, "status");
+  switch (status) {
+    case undefined:
+    case "completed":
+    case "failed":
+    case "error":
+    case "declined":
+    case "inProgress":
+    case "in_progress":
+    case "running":
+      return undefined;
+    default:
+      return status;
+  }
+}
+
 export function auditNativeToolTerminalStatus(item: CodexThreadItem): CodexNativeToolAuditStatus {
   if (item.type === "imageView" || item.type === "sleep") {
     return "completed";

extensions/codex/src/app-server/event-projector.test.ts

@@ -6269,6 +6269,119 @@ describe("CodexAppServerEventProjector", () => {
     expect(toolResult.isError).toBe(true);
   });
 
+  it("warns once and preserves projection for an unknown Codex-native item status", async () => {
+    const warn = vi.spyOn(embeddedAgentLog, "warn").mockImplementation(() => undefined);
+    const onAgentEvent = vi.fn();
+    const projector = await createProjector({ ...(await createParams()), onAgentEvent });
+    const notification = forCurrentTurn("item/completed", {
+      item: {
+        type: "commandExecution",
+        id: "cmd-future-status",
+        command: "pnpm test extensions/codex",
+        cwd: "/workspace",
+        processId: null,
+        source: "agent",
+        status: "pausedByProtocol",
+        commandActions: [],
+        aggregatedOutput: null,
+        exitCode: null,
+        durationMs: null,
+      },
+    });
+
+    await projector.handleNotification(notification);
+    await projector.handleNotification(notification);
+
+    expect(
+      findAgentEvent(onAgentEvent, {
+        stream: "item",
+        phase: "end",
+        itemId: "cmd-future-status",
+      }).data.status,
+    ).toBe("completed");
+    const toolResult = findAgentEvent(onAgentEvent, {
+      stream: "tool",
+      phase: "result",
+      itemId: "cmd-future-status",
+      name: "bash",
+    }).data;
+    expect(toolResult).toMatchObject({ status: "completed", isError: false });
+    expect(warn).toHaveBeenCalledTimes(1);
+    expect(warn).toHaveBeenCalledWith(
+      "codex app-server item reported unknown status; continuing projection",
+      {
+        itemId: "cmd-future-status",
+        itemType: "commandExecution",
+        status: "pausedByProtocol",
+      },
+    );
+  });
+
+  it("warns once per raw unknown event kind and continues projecting known events", async () => {
+    const warn = vi.spyOn(embeddedAgentLog, "warn").mockImplementation(() => undefined);
+    const params = await createParams();
+    const onPartialReply = vi.fn();
+    const projector = await createProjector({ ...params, onPartialReply });
+    const rawEventKind = "item/futureStatus/updated\nforged";
+    const collidingSanitizedEventKind = "item/futureStatus/updated\\nforged";
+    const notification = forCurrentTurn(rawEventKind, {
+      itemId: "future-1",
+    });
+
+    await projector.handleNotification(notification);
+    await projector.handleNotification(notification);
+    await projector.handleNotification(
+      forCurrentTurn(collidingSanitizedEventKind, { itemId: "future-2" }),
+    );
+    await projector.handleNotification(
+      forCurrentTurn("item/started", {
+        item: { type: "agentMessage", id: "msg-after-unknown", phase: "final_answer", text: "" },
+      }),
+    );
+    await projector.handleNotification(agentMessageDelta("still projects", "msg-after-unknown"));
+    await projector.handleNotification(
+      forCurrentTurn("item/completed", {
+        item: {
+          type: "agentMessage",
+          id: "msg-after-unknown",
+          phase: "final_answer",
+          text: "still projects",
+        },
+      }),
+    );
+    await projector.handleNotification(
+      turnCompleted([
+        {
+          type: "agentMessage",
+          id: "msg-after-unknown",
+          phase: "final_answer",
+          text: "still projects",
+        },
+      ]),
+    );
+
+    expect(projector.buildResult(buildEmptyToolTelemetry()).assistantTexts).toEqual([
+      "still projects",
+    ]);
+    expect(onPartialReply).toHaveBeenCalledWith({
+      text: "still projects",
+      delta: "still projects",
+    });
+    expect(warn).toHaveBeenCalledTimes(2);
+    expect(warn).toHaveBeenCalledWith(
+      "codex app-server projector received unknown event kind; continuing: item/futureStatus/updated\\nforged",
+      {
+        eventKind: "item/futureStatus/updated\\nforged",
+        activeThreadId: THREAD_ID,
+        activeTurnId: TURN_ID,
+        threadId: THREAD_ID,
+        turnId: TURN_ID,
+        matchesActiveThread: true,
+        matchesActiveTurn: true,
+      },
+    );
+  });
+
   it("leaves Codex dynamic tool item progress to item/tool/call normalization", async () => {
     const onAgentEvent = vi.fn();
     const projector = await createProjector({ ...(await createParams()), onAgentEvent });

extensions/codex/src/app-server/event-projector.ts

@@ -13,6 +13,10 @@ import {
   type MessagingToolSourceReplyPayload,
 } from "openclaw/plugin-sdk/agent-harness-runtime";
 import { CodexAssistantProjection } from "./event-projector-assistant.js";
+import {
+  CodexProjectionDiagnostics,
+  isKnownNoopCodexNotificationMethod,
+} from "./event-projector-diagnostics.js";
 import { CodexEventProjection } from "./event-projector-events.js";
 import {
   itemName,
@@ -90,6 +94,7 @@ export class CodexAppServerEventProjector {
   private readonly activeCompactionItemIds = new Set<string>();
   private readonly terminalPresentationClearedItemIds = new Set<string>();
   private readonly nativeToolOutcomeOrdinals = new Map<string, number>();
+  private readonly diagnostics: CodexProjectionDiagnostics;
   private readonly generatedMediaProjection: CodexGeneratedMediaProjection;
   private readonly eventProjection: CodexEventProjection;
   private readonly nativeToolLifecycleProjector: CodexNativeToolLifecycleProjector;
@@ -110,6 +115,7 @@ export class CodexAppServerEventProjector {
     private readonly turnId: string,
     private readonly options: CodexAppServerEventProjectorOptions = {},
   ) {
+    this.diagnostics = new CodexProjectionDiagnostics(threadId, turnId);
     this.nativeToolLifecycleProjector = new CodexNativeToolLifecycleProjector(
       params,
       threadId,
@@ -283,6 +289,9 @@ export class CodexAppServerEventProjector {
         this.promptErrorSource = "prompt";
         break;
       default:
+        if (!isKnownNoopCodexNotificationMethod(notification.method)) {
+          this.diagnostics.warnUnknownEvent(notification, params);
+        }
         break;
     }
   }
@@ -515,6 +524,7 @@ export class CodexAppServerEventProjector {
 
   private async handleItemCompleted(params: JsonObject): Promise<void> {
     const item = readItem(params.item);
+    this.diagnostics.warnUnknownItemStatus(item);
     this.recordNativeToolOutcome(item);
     this.clearTerminalPresentationForNativeItem(item);
     const itemId = item?.id ?? readString(params, "itemId");
@@ -626,6 +636,7 @@ export class CodexAppServerEventProjector {
       }
     }
     for (const item of turnItems) {
+      this.diagnostics.warnUnknownItemStatus(item);
       this.assistantProjection.recordSnapshotItem(item);
       this.reasoningProjection.recordItem(item);
       this.generatedMediaProjection.recordNative(item);

extensions/codex/src/app-server/run-attempt-test-harness.ts

@@ -8,6 +8,7 @@ import {
   resetAgentEventsForTest,
   type EmbeddedRunAttemptParams,
 } from "openclaw/plugin-sdk/agent-harness-runtime";
+import { clearRuntimeAuthProfileStoreSnapshots } from "openclaw/plugin-sdk/agent-runtime";
 import { resetDiagnosticEventsForTest } from "openclaw/plugin-sdk/diagnostic-runtime";
 import { clearInternalHooks, resetGlobalHookRunner } from "openclaw/plugin-sdk/hook-runtime";
 import { clearMemoryPluginState } from "openclaw/plugin-sdk/memory-core-host-runtime-core";
@@ -605,6 +606,7 @@ export function createRuntimeDynamicTool(name: string): RuntimeDynamicToolForTes
 export function setupRunAttemptTestHooks(): void {
   beforeEach(async () => {
     resetCodexTestBindingStore();
+    clearRuntimeAuthProfileStoreSnapshots();
     vi.useRealTimers();
     clearInternalHooks();
     clearMemoryPluginState();
@@ -620,6 +622,7 @@ export function setupRunAttemptTestHooks(): void {
     await drainActiveAppServerAttemptsForTest();
     await sandboxExecServerRegistry.closeAll();
     resetCodexAppServerClientFactoryForTest();
+    clearRuntimeAuthProfileStoreSnapshots();
     dynamicToolBuildState.openClawCodingToolsFactory = undefined;
     codexWorkspaceDirCache.clear();
     nativeHookRelayUnregisterQueue.clear();

extensions/codex/src/app-server/run-attempt.test.ts

@@ -5,6 +5,7 @@ import {
   embeddedAgentLog,
   type EmbeddedRunAttemptParams,
 } from "openclaw/plugin-sdk/agent-harness-runtime";
+import { replaceRuntimeAuthProfileStoreSnapshots } from "openclaw/plugin-sdk/agent-runtime";
 import { SessionManager } from "openclaw/plugin-sdk/agent-sessions";
 import {
   onInternalDiagnosticEvent,
@@ -597,7 +598,22 @@ describe("runCodexAppServerAttempt", () => {
     const config = {
       auth: { profiles: { [authProfileId]: { provider: "openai", mode: "api_key" as const } } },
     };
-    vi.stubEnv("OPENAI_WORK_KEY", "work-key");
+    replaceRuntimeAuthProfileStoreSnapshots([
+      {
+        agentDir,
+        store: {
+          version: 1,
+          profiles: {
+            [authProfileId]: {
+              type: "api_key",
+              provider: "openai",
+              keyRef: { source: "env", provider: "default", id: "OPENAI_WORK_KEY" },
+              key: "work-key",
+            },
+          },
+        },
+      },
+    ]);
     let clientOptions: CodexAppServerClientOptions | undefined;
     const harness = createStartedThreadHarness(async () => undefined, {
       onStart: (_profileId, _agentDir, options) => {

extensions/codex/src/app-server/turn-router.test.ts

@@ -115,6 +115,55 @@ describe("CodexAppServerTurnRouter", () => {
     expect(secondResponse).toEqual({ id: "request-2", result: { owner: "second" } });
   });
 
+  it("warns once only for a thread-correlated stale turn notification", async () => {
+    const warn = vi.spyOn(embeddedAgentLog, "warn").mockImplementation(() => undefined);
+    const harness = createHarness();
+    const notifications = vi.fn();
+    const route = getCodexAppServerTurnRouter(harness.client).reserveThread({
+      threadId: "thread-1",
+      onNotification: notifications,
+    });
+    route.armTurn();
+    await route.bindTurn("turn-current");
+    const staleNotification = {
+      method: "item/agentMessage/delta",
+      params: {
+        threadId: "thread-1",
+        turnId: "turn-stale",
+        itemId: "msg-stale",
+        delta: "ignored",
+      },
+    };
+
+    harness.send(staleNotification);
+    harness.send(staleNotification);
+    harness.send({
+      method: "thread/status/changed",
+      params: { threadId: "thread-1", status: { type: "active" } },
+    });
+    harness.send({ method: "configWarning", params: { message: "global" } });
+    await settleInput();
+
+    expect(notifications).toHaveBeenCalledOnce();
+    expect(notifications).toHaveBeenCalledWith(
+      {
+        method: "thread/status/changed",
+        params: { threadId: "thread-1", status: { type: "active" } },
+      },
+      { threadId: "thread-1" },
+    );
+    expect(warn).toHaveBeenCalledTimes(1);
+    expect(warn).toHaveBeenCalledWith("codex app-server notification ignored for inactive turn", {
+      eventKind: "item/agentMessage/delta",
+      activeThreadId: "thread-1",
+      activeTurnId: "turn-current",
+      threadId: "thread-1",
+      turnId: "turn-stale",
+      matchesActiveThread: true,
+      matchesActiveTurn: false,
+    });
+  });
+
   it("buffers pre-bind notifications in order and filters the bound turn", async () => {
     const harness = createHarness();
     const methods: string[] = [];

extensions/codex/src/app-server/turn-router.ts

@@ -1,6 +1,7 @@
 /** Keyed routing for all turn traffic on one shared Codex app-server client. */
 import { embeddedAgentLog } from "openclaw/plugin-sdk/agent-harness-runtime";
 import type { CodexAppServerClient } from "./client.js";
+import { redactCodexEventKind } from "./event-projector-diagnostics.js";
 import {
   readCodexNotificationThreadId,
   readCodexNotificationTurnId,
@@ -91,6 +92,7 @@ type Route = {
   pending: PendingNotification[];
   notificationTail: Promise<void>;
   nativeTurnCompleted: boolean;
+  ignoredTurnNotificationKeys: Set<string>;
   nativeTurnCompletion?: Deferred;
   detachReleaseOn?: () => void;
 };
@@ -144,6 +146,7 @@ class ClientTurnRouter implements CodexAppServerTurnRouter {
       pending: [],
       notificationTail: Promise.resolve(),
       nativeTurnCompleted: false,
+      ignoredTurnNotificationKeys: new Set(),
     };
     this.routes.set(threadId, route);
     if (options.onNotification || options.onRequest) {
@@ -254,6 +257,7 @@ class ClientTurnRouter implements CodexAppServerTurnRouter {
       throw new Error(`codex app-server thread route cannot arm from ${route.gate}`);
     }
     route.gate = "armed";
+    route.ignoredTurnNotificationKeys.clear();
     route.nativeTurnCompleted = false;
     route.binding = deferred();
   }
@@ -337,6 +341,7 @@ class ClientTurnRouter implements CodexAppServerTurnRouter {
       return undefined;
     }
     if (route.gate === "bound" && scope.turnId && scope.turnId !== route.turnId) {
+      this.warnDroppedStaleTurnNotification(route, notification, routeScope);
       return undefined;
     }
     if (route.gate === "armed") {
@@ -410,21 +415,44 @@ class ClientTurnRouter implements CodexAppServerTurnRouter {
       return;
     }
     for (const pending of route.pending.splice(0)) {
-      if (
-        !pending.scope.turnId ||
-        route.gate !== "bound" ||
-        pending.scope.turnId === route.turnId
-      ) {
-        route.handlers?.onNotificationReceived?.(
-          pending.notification,
-          pending.scope,
-          pending.receivedAtMs,
-        );
-        this.enqueueNotification(route, handler, pending.notification, pending.scope);
+      if (route.gate === "bound" && pending.scope.turnId && pending.scope.turnId !== route.turnId) {
+        this.warnDroppedStaleTurnNotification(route, pending.notification, pending.scope);
+        continue;
       }
+      route.handlers?.onNotificationReceived?.(
+        pending.notification,
+        pending.scope,
+        pending.receivedAtMs,
+      );
+      this.enqueueNotification(route, handler, pending.notification, pending.scope);
     }
   }
 
+  private warnDroppedStaleTurnNotification(
+    route: Route,
+    notification: CodexServerNotification,
+    scope: CodexThreadRouteScope,
+  ): void {
+    if (notification.method === "turn/completed" || !scope.turnId || !route.turnId) {
+      return;
+    }
+    const eventKind = redactCodexEventKind(notification.method);
+    const key = JSON.stringify([notification.method, scope.turnId]);
+    if (route.ignoredTurnNotificationKeys.has(key)) {
+      return;
+    }
+    route.ignoredTurnNotificationKeys.add(key);
+    embeddedAgentLog.warn("codex app-server notification ignored for inactive turn", {
+      eventKind,
+      activeThreadId: route.threadId,
+      activeTurnId: route.turnId,
+      threadId: scope.threadId,
+      turnId: scope.turnId,
+      matchesActiveThread: true,
+      matchesActiveTurn: false,
+    });
+  }
+
   private bufferNotification(
     route: Route,
     notification: CodexServerNotification,