diff --git a/extensions/codex/src/app-server/event-projector-diagnostics.ts b/extensions/codex/src/app-server/event-projector-diagnostics.ts new file mode 100644 index 000000000000..a68e0b7db0ed --- /dev/null +++ b/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(); + + 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): void { + if (this.warningKeys.has(key)) { + return; + } + this.warningKeys.add(key); + embeddedAgentLog.warn(message, context); + } +} diff --git a/extensions/codex/src/app-server/event-projector-items.ts b/extensions/codex/src/app-server/event-projector-items.ts index cfe8df2de6d9..dea7ed46cb5e 100644 --- a/extensions/codex/src/app-server/event-projector-items.ts +++ b/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"; diff --git a/extensions/codex/src/app-server/event-projector.test.ts b/extensions/codex/src/app-server/event-projector.test.ts index 9bc17701f863..1e7d88a88ff5 100644 --- a/extensions/codex/src/app-server/event-projector.test.ts +++ b/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 }); diff --git a/extensions/codex/src/app-server/event-projector.ts b/extensions/codex/src/app-server/event-projector.ts index 308d929a71c4..fa49a4a8b4f4 100644 --- a/extensions/codex/src/app-server/event-projector.ts +++ b/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(); private readonly terminalPresentationClearedItemIds = new Set(); private readonly nativeToolOutcomeOrdinals = new Map(); + 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 { 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); diff --git a/extensions/codex/src/app-server/run-attempt-test-harness.ts b/extensions/codex/src/app-server/run-attempt-test-harness.ts index 0e060a41e107..c1760eb6b595 100644 --- a/extensions/codex/src/app-server/run-attempt-test-harness.ts +++ b/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(); diff --git a/extensions/codex/src/app-server/run-attempt.test.ts b/extensions/codex/src/app-server/run-attempt.test.ts index 95ef0af9db36..7b24be2a06a3 100644 --- a/extensions/codex/src/app-server/run-attempt.test.ts +++ b/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) => { diff --git a/extensions/codex/src/app-server/turn-router.test.ts b/extensions/codex/src/app-server/turn-router.test.ts index f2b9ff71a006..c847de8d8e53 100644 --- a/extensions/codex/src/app-server/turn-router.test.ts +++ b/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[] = []; diff --git a/extensions/codex/src/app-server/turn-router.ts b/extensions/codex/src/app-server/turn-router.ts index 48d5ba35a69e..cf45679a30a6 100644 --- a/extensions/codex/src/app-server/turn-router.ts +++ b/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; nativeTurnCompleted: boolean; + ignoredTurnNotificationKeys: Set; 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,