fix(agents): observe native provider prompt egress (#119219)

* fix(agents): observe final provider prompt egress * fix(agents): observe native provider prompt egress * test(agents): type native prompt observer models
2026-08-12 21:53:00 -06:00 · 2026-08-04 22:46:07 +08:00
parent e79f358b02
commit a54bec799c
18 changed files with 988 additions and 59 deletions
@@ -0,0 +1,14 @@
+import { describe, expect, it } from "vitest";
+import { readQaScenarioPack } from "./scenario-catalog.js";
+
+describe("no-meta QA catalog", () => {
+  it("keeps context visibility proof on one primary scenario", () => {
+    const primaryOwnerIds = readQaScenarioPack()
+      .scenarios.filter((scenario) =>
+        scenario.coverage?.primary.includes("session-memory.context-visibility-no-meta-leak"),
+      )
+      .map((scenario) => scenario.id);
+
+    expect(primaryOwnerIds).toStrictEqual(["instruction-profile-artifact-followthrough-live"]);
+  });
+});
@@ -69,6 +69,66 @@ async function runWebchatTranscriptWait(
  });
 }

+function readCurrentRunProviderPromptEvidenceFlow(trajectoryEvents: unknown[]): QaScenarioFlow {
+  const scenario = readQaScenarioById("instruction-profile-artifact-followthrough-live");
+  const actions = scenario.execution.flow?.steps[0]?.actions;
+  if (!actions) {
+    throw new Error("instruction profile scenario has no actions");
+  }
+  const evidenceIndex = actions.findIndex(
+    (action) =>
+      typeof action === "object" &&
+      action !== null &&
+      "set" in action &&
+      action.set === "providerPromptEvidence",
+  );
+  const assertionIndex = actions.findIndex(
+    (action, index) =>
+      index > evidenceIndex &&
+      typeof action === "object" &&
+      action !== null &&
+      "assert" in action &&
+      JSON.stringify(action).includes("current-run provider prompt evidence mismatch"),
+  );
+  if (evidenceIndex < 0 || assertionIndex < 0) {
+    throw new Error("instruction profile scenario has no provider prompt evidence assertion");
+  }
+  const instructionContents = scenario.execution.config?.instructionContents;
+  const instructionChars =
+    typeof instructionContents === "string" ? instructionContents.trimEnd().length : 0;
+  return {
+    steps: [
+      {
+        name: "proves current-run provider prompt evidence",
+        actions: [
+          { set: "turn", value: { started: { runId: "current-run" } } },
+          {
+            set: "instructionProfileReport",
+            value: {
+              missing: false,
+              truncated: false,
+              rawChars: instructionChars,
+              injectedChars: instructionChars,
+            },
+          },
+          { set: "trajectoryEvents", value: trajectoryEvents },
+          ...actions
+            .slice(evidenceIndex, assertionIndex + 1)
+            .filter(
+              (action) =>
+                !(
+                  typeof action === "object" &&
+                  action !== null &&
+                  "call" in action &&
+                  action.call === "fs.rm"
+                ),
+            ),
+        ],
+      },
+    ],
+  };
+}
+
 const planningEvidenceCoverageIds = new Set(["runtime.no-meta-leak", "workspace.planning"]);

 type PlanningEvidenceScenario = QaSeedScenarioWithSource & {
@@ -237,6 +297,64 @@ const planningEvidenceFixtures = readQaScenarioPack()
  .map(createPlanningEvidenceFixture);

 describe("scenario-flow-runner", () => {
+  it("ignores stale provider prompt mismatches when the current run matches", async () => {
+    const currentObservation = {
+      egress: "responses-sdk",
+      payloadVariant: "initial",
+      promptSource: "input.developer",
+      expectedChars: 4096,
+      observedChars: 4096,
+      matchesAssembledPrompt: true,
+    };
+    const result = await runLoadedScenarioFlow("instruction-profile-artifact-followthrough-live", {
+      flow: readCurrentRunProviderPromptEvidenceFlow([
+        {
+          type: "provider.prompt.observed",
+          runId: "stale-run",
+          data: {
+            ...currentObservation,
+            promptSource: "missing",
+            observedChars: 0,
+            matchesAssembledPrompt: false,
+          },
+        },
+        { type: "provider.prompt.observed", runId: "current-run", data: currentObservation },
+      ]),
+    });
+
+    expect(result.status).toBe("pass");
+  });
+
+  it("excludes marker-bearing diagnostic trajectory context from bounded no-leak evidence", async () => {
+    const marker = "INSTRUCTION-PROFILE-CONTEXT-MARKER-A6E29D4B";
+    const trajectoryEvents = [
+      {
+        type: "context.compiled",
+        runId: "current-run",
+        data: { systemPrompt: `diagnostic support context ${marker}` },
+      },
+      {
+        type: "provider.prompt.observed",
+        runId: "current-run",
+        data: {
+          egress: "native-codex-websocket",
+          payloadVariant: "initial",
+          promptSource: "instructions",
+          expectedChars: 4096,
+          observedChars: 4096,
+          matchesAssembledPrompt: true,
+        },
+      },
+    ];
+
+    expect(JSON.stringify(trajectoryEvents)).toContain(marker);
+    const result = await runLoadedScenarioFlow("instruction-profile-artifact-followthrough-live", {
+      flow: readCurrentRunProviderPromptEvidenceFlow(trajectoryEvents),
+    });
+
+    expect(result.status).toBe("pass");
+  });
+
  it("keeps live goal followthrough inside the active-goal context limit", async () => {
    const state = createQaBusState();
    const artifactFile = "goal-continuance-live-00000000.txt";
@@ -16,3 +16,5 @@ export * from "../providers/openai-tool-schema-compat.js";
 export * from "../providers/openai-tool-schema.js";
 export * from "../providers/schema-keyword-strip.js";
 export * from "../providers/tool-schema-json-projection.js";
+export { responsesPromptObserver } from "../transports/openai-responses-contracts.js";
+export type { ResponsesPromptObservation } from "../transports/openai-responses-contracts.js";
@@ -136,6 +136,7 @@ const compatibility = {
      "resolveOpenAIProjectedToolsStrictToolFlag",
      "stripUnsupportedSchemaKeywords",
      "projectRuntimeToolInputSchema",
+      "responsesPromptObserver",
    ],
    types: [
      "OpenAICompletionsOptions",
@@ -157,6 +158,7 @@ const compatibility = {
      "OpenAICompletionsToolChoice",
      "RuntimeToolInputSchemaJson",
      "RuntimeToolInputSchemaProjection",
+      "ResponsesPromptObservation",
    ],
  },
 } as const;
@@ -5,6 +5,7 @@ import { zstdDecompressSync } from "node:zlib";
 import { afterEach, describe, expect, it, vi } from "vitest";
 import { WebSocket, WebSocketServer } from "ws";
 import { configureAiTransportHost } from "../host.js";
+import { responsesPromptObserver, type ResponsesPromptObservation } from "../internal/openai.js";
 import { cleanupSessionResources } from "../session-resources.js";
 import { MALFORMED_STREAMING_FRAGMENT_ERROR_MESSAGE } from "../transports/transport-utils.js";
 import type { Context, Model } from "../types.js";
@@ -315,6 +316,9 @@ describe("ChatGPT Responses cached transport", () => {
  });

  it("does not prepare SSE requests or serialize full bodies for cached websocket turns", async () => {
+    const prompt = "PRIVATE-CACHED-WEBSOCKET-PROMPT";
+    const observations: ResponsesPromptObservation[] = [];
+    const order: string[] = [];
    const sentPayloads: string[] = [];

    class CachedWebSocket extends EventTarget {
@@ -326,6 +330,7 @@ describe("ChatGPT Responses cached transport", () => {
      }

      send(payload: string): void {
+        order.push("send");
        sentPayloads.push(payload);
        queueMicrotask(() => {
          this.dispatchEvent(
@@ -352,11 +357,20 @@ describe("ChatGPT Responses cached transport", () => {
      sessionId: "cached-hot-path",
      transport: "websocket-cached" as const,
    };
+    responsesPromptObserver.set(options, (observation) => {
+      order.push("observe");
+      observations.push(observation);
+    });

-    const first = await streamOpenAICodexResponses(model, context, options).result();
+    const first = await streamOpenAICodexResponses(
+      model,
+      { ...context, systemPrompt: prompt },
+      options,
+    ).result();
    const second = await streamOpenAICodexResponses(
      model,
      {
+        systemPrompt: prompt,
        messages: [...context.messages, { role: "user", content: "follow-up", timestamp: 2 }],
      },
      options,
@@ -368,6 +382,11 @@ describe("ChatGPT Responses cached transport", () => {
    expect(headerSet).not.toHaveBeenCalledWith("accept", "text/event-stream");
    expect(headerSet).not.toHaveBeenCalledWith("content-type", "application/json");
    expect(sentPayloads).toHaveLength(2);
+    expect(order).toEqual(["observe", "send", "observe", "send"]);
+    expect(observations).toHaveLength(2);
+    expect(observations.every((entry) => entry.egress === "native-codex-websocket")).toBe(true);
+    expect(observations.every((entry) => entry.matchesAssembledPrompt)).toBe(true);
+    expect(JSON.stringify(observations)).not.toContain(prompt);

    const continuation = JSON.parse(sentPayloads[1] as string) as {
      input?: unknown[];
@@ -1,6 +1,7 @@
 // Covers which ChatGPT Responses failures the SSE transport retries.
 import { afterEach, describe, expect, it, vi } from "vitest";
 import { configureAiTransportHost } from "../host.js";
+import { responsesPromptObserver, type ResponsesPromptObservation } from "../internal/openai.js";
 import type { Context, Model } from "../types.js";
 import {
  closeOpenAICodexWebSocketSessions,
@@ -73,6 +74,8 @@ describe("streamOpenAICodexResponses retry classification", () => {
  );

  it("still retries retryable ChatGPT responses", async () => {
+    const prompt = "PRIVATE-NATIVE-SSE-RETRY-PROMPT";
+    const observations: ResponsesPromptObservation[] = [];
    const fetchMock = vi
      .fn<typeof fetch>()
      .mockResolvedValueOnce(new Response("overloaded", { status: 503 }))
@@ -89,13 +92,25 @@ describe("streamOpenAICodexResponses retry classification", () => {
      return 0 as unknown as ReturnType<typeof setTimeout>;
    });

-    const result = await streamOpenAICodexResponses(model, context, {
+    const options = {
      apiKey: jwt,
-      transport: "sse",
-    }).result();
+      transport: "sse" as const,
+    };
+    responsesPromptObserver.set(options, (observation) => observations.push(observation));
+
+    const result = await streamOpenAICodexResponses(
+      model,
+      { ...context, systemPrompt: prompt },
+      options,
+    ).result();

    expect(result.stopReason).toBe("error");
    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(observations).toHaveLength(2);
+    expect(observations.every((entry) => entry.egress === "native-codex-sse")).toBe(true);
+    expect(observations.every((entry) => entry.payloadVariant === "initial")).toBe(true);
+    expect(observations.every((entry) => entry.matchesAssembledPrompt)).toBe(true);
+    expect(JSON.stringify(observations)).not.toContain(prompt);
  });

  it.each([
@@ -36,6 +36,8 @@ import { getAiTransportHost, resolveAiTransportHeaderSentinels } from "../host.j
 import { parseRetryAfterHttpDateMs } from "../internal/retry-after.js";
 import { sleepWithAbort } from "../internal/retry-sleep.js";
 import { registerSessionResourceCleanup } from "../session-resources.js";
+import { responsesPromptObserver } from "../transports/openai-responses-contracts.js";
+import { createResponsesPromptEgressObserver } from "../transports/openai-responses-prompt-observer-internal.js";
 import {
  processResponsesStream,
  ResponsesStreamFailure,
@@ -138,6 +140,10 @@ interface RequestBody {
  [key: string]: unknown;
 }

+type ObserveResponsesPromptEgress = NonNullable<
+  ReturnType<typeof createResponsesPromptEgressObserver>
+>;
+
 // ============================================================================
 // Retry Helpers
 // ============================================================================
@@ -284,6 +290,10 @@ export const streamOpenAICodexResponses: StreamFunction<
      if (nextBody !== undefined) {
        body = nextBody as RequestBody;
      }
+      const observePromptEgress = createResponsesPromptEgressObserver(
+        options,
+        context.systemPrompt,
+      );
      // NOTE: when options.sessionId is absent, this falls back to a fresh random id
      // per request, which forfeits session-affinity routing on the WS transport (the
      // backend routes by session_id/x-client-request-id). Left as-is for this fix;
@@ -324,6 +334,7 @@ export const streamOpenAICodexResponses: StreamFunction<
              },
              requestOptions,
              firstEventAbort.abort,
+              observePromptEgress,
            );

            if (activeSignal?.aborted) {
@@ -395,6 +406,10 @@ export const streamOpenAICodexResponses: StreamFunction<
        let attemptResponse: Response;
        let errorText: string;
        try {
+          observePromptEgress?.(body, {
+            egress: "native-codex-sse",
+            payloadVariant: "initial",
+          });
          attemptResponse = await fetch(resolveCodexUrl(model.baseUrl), {
            method: "POST",
            headers: sseHeaders,
@@ -516,11 +531,12 @@ export const streamSimpleOpenAICodexResponses: StreamFunction<
    throw new Error(`No API key for provider: ${model.provider}`);
  }

-  const base = buildBaseOptions(model, options, apiKey);
-  return streamOpenAICodexResponses(model, context, {
-    ...base,
+  const resolvedOptions = {
+    ...buildBaseOptions(model, options, apiKey),
    reasoningEffort: resolveResponsesReasoningEffort(model, options?.reasoning),
-  } satisfies OpenAICodexResponsesOptions);
+  } satisfies OpenAICodexResponsesOptions;
+  responsesPromptObserver.copy(options, resolvedOptions);
+  return streamOpenAICodexResponses(model, context, resolvedOptions);
 };

 // ============================================================================
@@ -1461,6 +1477,7 @@ async function processWebSocketStream(
  onStart: () => void,
  options?: OpenAICodexResponsesOptions,
  abortFirstEventStream?: (reason: Error) => void,
+  observePromptEgress?: ObserveResponsesPromptEgress,
 ): Promise<void> {
  const { socket, entry, release } = await acquireWebSocket(
    url,
@@ -1480,6 +1497,10 @@ async function processWebSocketStream(
    if (options?.signal?.aborted) {
      throw transportAbortError(options.signal);
    }
+    observePromptEgress?.(requestBody, {
+      egress: "native-codex-websocket",
+      payloadVariant: "initial",
+    });
    socket.send(JSON.stringify({ type: "response.create", ...requestBody }));
    await processResponsesStream(
      startWebSocketOutputOnFirstEvent(
@@ -30,6 +30,7 @@ import {
  buildOpenAIResponsesParams,
  sanitizeOpenAICodexResponsesParams,
 } from "./openai-responses-params-internal.js";
+import { createResponsesPromptEgressObserver } from "./openai-responses-prompt-observer-internal.js";
 import {
  buildOpenAIResponsesReasoningReplayMetadata,
  createResponsesStreamWithEncryptedContentRetry,
@@ -193,6 +194,10 @@ function createResponsesTransportExecutor(config: ResponsesTransportExecutorOpti
          enforceCodeModeResponsesToolSurface(params, visibleToolNames);
          assertCodeModeResponsesToolSurface(params, visibleToolNames);
        }
+        const observePrompt = createResponsesPromptEgressObserver(
+          responsesOptions,
+          context.systemPrompt,
+        );
        const requestStartedAt = Date.now();
        firstEventAbort = createFirstStreamEventAbortController(options?.signal);
        const requestOptions = buildOpenAISdkRequestOptions(model, firstEventAbort.signal, {
@@ -211,6 +216,7 @@ function createResponsesTransportExecutor(config: ResponsesTransportExecutorOpti
          request: params,
          requestOptions,
          model,
+          observePrompt,
        });
        await options?.onResponse?.(
          { status: response.status, headers: headersToRecord(response.headers) },
@@ -290,7 +296,8 @@ export function createAzureOpenAIResponsesTransportStreamFn(): StreamFn {
        resolveAzureDeploymentName(model),
        metadata,
      ),
-    createResponseStream: async ({ client, request, requestOptions }) => {
+    createResponseStream: async ({ client, request, requestOptions, observePrompt }) => {
+      observePrompt?.(request, { egress: "responses-sdk", payloadVariant: "initial" });
      const { data, response } = await client.responses
        .create(request as never, requestOptions)
        .withResponse();
@@ -46,6 +46,32 @@ export type OpenAIResponsesOptions = BaseOpenAIStreamOptions & {
  toolChoice?: ResponseCreateParamsStreaming["tool_choice"];
 };

+const PROMPT_OBSERVER = Symbol("openaiResponsesPromptObserver");
+export type ResponsesPromptObservation = {
+  egress: "responses-sdk" | "native-codex-websocket" | "native-codex-sse";
+  payloadVariant: "initial" | "encrypted-content-retry";
+  promptSource: "instructions" | "input.developer" | "input.system" | "missing";
+  expectedChars: number;
+  observedChars: number;
+  matchesAssembledPrompt: boolean;
+};
+type ResponsesPromptObserver = (observation: ResponsesPromptObservation) => void;
+
+export const responsesPromptObserver = {
+  set(options: object, observer: ResponsesPromptObserver): void {
+    Reflect.set(options, PROMPT_OBSERVER, observer);
+  },
+  get(options: object) {
+    return Reflect.get(options, PROMPT_OBSERVER) as ResponsesPromptObserver | undefined;
+  },
+  copy(source: object | undefined, target: object): void {
+    const observer = source && responsesPromptObserver.get(source);
+    if (observer) {
+      responsesPromptObserver.set(target, observer);
+    }
+  },
+};
+
 export type OpenAIResponsesReplayContext = {
  provider: string;
  api: Api;
@@ -0,0 +1,60 @@
+import type { EasyInputMessage } from "openai/resources/responses/responses.js";
+import { stripSystemPromptCacheBoundary } from "../utils/system-prompt-cache-boundary.js";
+import {
+  responsesPromptObserver,
+  type ResponsesPromptObservation,
+} from "./openai-responses-contracts.js";
+import { sanitizeTransportPayloadText } from "./transport-stream-shared.js";
+
+type ResponsesPromptRequest = { instructions?: unknown; input?: unknown };
+type ResponsesPromptMetadata = Pick<ResponsesPromptObservation, "egress" | "payloadVariant">;
+
+function readFinalResponsesPrompt(
+  request: ResponsesPromptRequest,
+): [ResponsesPromptObservation["promptSource"], string] {
+  if (typeof request.instructions === "string") {
+    return ["instructions", request.instructions] as const;
+  }
+  const input = Array.isArray(request.input) ? request.input : [];
+  const message = input.find((item) => {
+    const role = (item as EasyInputMessage).role;
+    return role === "developer" || role === "system";
+  }) as EasyInputMessage | undefined;
+  if (!message) {
+    return ["missing", ""] as const;
+  }
+  const content = message.content;
+  const observedPrompt =
+    typeof content === "string"
+      ? content
+      : Array.isArray(content)
+        ? content.flatMap((part) => (part.type === "input_text" ? [part.text] : [])).join("")
+        : "";
+  return [
+    message.role === "developer" ? "input.developer" : "input.system",
+    observedPrompt,
+  ] as const;
+}
+
+export function createResponsesPromptEgressObserver(
+  options: object | undefined,
+  assembledPrompt: string | undefined,
+) {
+  const observer = options ? responsesPromptObserver.get(options) : undefined;
+  if (!observer) {
+    return undefined;
+  }
+  const expectedPrompt = sanitizeTransportPayloadText(
+    stripSystemPromptCacheBoundary(assembledPrompt ?? ""),
+  );
+  return (request: ResponsesPromptRequest, metadata: ResponsesPromptMetadata) => {
+    const [promptSource, observedPrompt] = readFinalResponsesPrompt(request);
+    observer({
+      ...metadata,
+      promptSource,
+      expectedChars: expectedPrompt.length,
+      observedChars: observedPrompt.length,
+      matchesAssembledPrompt: promptSource !== "missing" && observedPrompt === expectedPrompt,
+    });
+  };
+}
@@ -0,0 +1,543 @@
+import { zstdDecompressSync } from "node:zlib";
+import type { Api, Context, Model } from "@openclaw/llm-core";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { configureAiTransportHost, getAiTransportHost } from "../host.js";
+import { responsesPromptObserver, type ResponsesPromptObservation } from "../internal/openai.js";
+import {
+  closeOpenAICodexWebSocketSessions,
+  resetOpenAICodexWebSocketStateForTest,
+  streamOpenAICodexResponses,
+  streamSimpleOpenAICodexResponses,
+} from "../providers/openai-chatgpt-responses.js";
+import { SYSTEM_PROMPT_CACHE_BOUNDARY } from "../utils/system-prompt-cache-boundary.js";
+
+const sdkState = vi.hoisted(() => ({
+  clients: [] as Array<"openai" | "azure">,
+  errors: [] as Error[],
+  order: [] as string[],
+  requests: [] as Array<Record<string, unknown>>,
+}));
+
+vi.mock("openai", () => {
+  const createClient = (client: "openai" | "azure") =>
+    class MockOpenAI {
+      responses = {
+        create: (request: Record<string, unknown>) => {
+          sdkState.clients.push(client);
+          sdkState.order.push(`${client}.create`);
+          sdkState.requests.push(request);
+          const error = sdkState.errors.shift() ?? new Error("stop after request");
+          return {
+            withResponse: async () => {
+              throw error;
+            },
+          };
+        },
+      };
+    };
+  return { default: createClient("openai"), AzureOpenAI: createClient("azure") };
+});
+
+import {
+  createAzureOpenAIResponsesTransportStreamFn,
+  createOpenAIResponsesTransportStreamFn,
+} from "./openai-responses-client.js";
+
+const initialHost = getAiTransportHost();
+
+function createModel<TApi extends Api = "openai-responses">(
+  overrides: Partial<Model<TApi>> = {},
+): Model<TApi> {
+  return {
+    id: "gpt-5.4",
+    name: "GPT-5.4",
+    api: "openai-responses",
+    provider: "openai",
+    baseUrl: "https://api.openai.com/v1",
+    reasoning: true,
+    input: ["text"],
+    cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
+    contextWindow: 200_000,
+    maxTokens: 8192,
+    ...overrides,
+  } as Model<TApi>;
+}
+
+function createContext(systemPrompt: string, overrides: Partial<Context> = {}): Context {
+  return {
+    systemPrompt,
+    messages: [{ role: "user", content: "hello", timestamp: 1 }],
+    tools: [],
+    ...overrides,
+  } as Context;
+}
+
+function createJwt(): string {
+  const encode = (value: object) => Buffer.from(JSON.stringify(value)).toString("base64url");
+  return `${encode({ alg: "none", typ: "JWT" })}.${encode({
+    "https://api.openai.com/auth": { chatgpt_account_id: "acct-1" },
+  })}.signature`;
+}
+
+function completedSseResponse(responseId = "resp_test"): Response {
+  return new Response(
+    `data: ${JSON.stringify({
+      type: "response.completed",
+      response: {
+        id: responseId,
+        status: "completed",
+        output: [],
+        usage: { input_tokens: 5, output_tokens: 3, total_tokens: 8 },
+      },
+    })}\n\n`,
+    { status: 200, headers: { "content-type": "text/event-stream" } },
+  );
+}
+
+async function runObservedRequest(params: {
+  context: Context;
+  model?: Model;
+  azure?: boolean;
+  errors?: Error[];
+  options?: Record<string, unknown>;
+}) {
+  const observations: ResponsesPromptObservation[] = [];
+  const options = { apiKey: "test-key", ...params.options };
+  const requestStart = sdkState.requests.length;
+  const orderStart = sdkState.order.length;
+  sdkState.errors = params.errors ?? [new Error("stop after request")];
+  responsesPromptObserver.set(options, (observation) => {
+    sdkState.order.push("observe");
+    observations.push(observation);
+  });
+  const streamFn = params.azure
+    ? createAzureOpenAIResponsesTransportStreamFn()
+    : createOpenAIResponsesTransportStreamFn();
+  const stream = await Promise.resolve(
+    streamFn(params.model ?? createModel(), params.context, options as never),
+  );
+  expect((await stream.result()).stopReason).toBe("error");
+  return {
+    observations,
+    order: sdkState.order.slice(orderStart),
+    requests: sdkState.requests.slice(requestStart),
+  };
+}
+
+beforeEach(() => {
+  sdkState.clients = [];
+  sdkState.errors = [];
+  sdkState.order = [];
+  sdkState.requests = [];
+  configureAiTransportHost(initialHost);
+});
+
+afterEach(() => {
+  closeOpenAICodexWebSocketSessions();
+  vi.restoreAllMocks();
+  vi.unstubAllGlobals();
+  resetOpenAICodexWebSocketStateForTest();
+  configureAiTransportHost(initialHost);
+});
+
+describe("OpenAI Responses provider prompt observer", () => {
+  it.each([
+    { reasoning: true, promptSource: "input.developer" },
+    { reasoning: false, promptSource: "input.system" },
+  ] as const)("observes the final $promptSource prompt", async ({ reasoning, promptSource }) => {
+    const prompt = `PRIVATE-${promptSource}-PROMPT`;
+    const run = await runObservedRequest({
+      context: createContext(prompt),
+      model: createModel({ reasoning }),
+    });
+
+    expect(run.observations).toEqual([
+      {
+        egress: "responses-sdk",
+        payloadVariant: "initial",
+        promptSource,
+        expectedChars: prompt.length,
+        observedChars: prompt.length,
+        matchesAssembledPrompt: true,
+      },
+    ]);
+    expect(JSON.stringify(run.observations)).not.toContain(prompt);
+  });
+
+  it("observes Azure Responses egress", async () => {
+    const prompt = "PRIVATE-AZURE-PROMPT";
+    const run = await runObservedRequest({
+      azure: true,
+      context: createContext(prompt),
+      model: createModel({
+        api: "azure-openai-responses",
+        provider: "azure-openai-responses",
+        baseUrl: "https://example.openai.azure.com",
+      }),
+    });
+
+    expect(sdkState.clients).toEqual(["azure"]);
+    expect(run.order).toEqual(["observe", "azure.create"]);
+    expect(run.observations[0]).toMatchObject({
+      egress: "responses-sdk",
+      payloadVariant: "initial",
+      promptSource: "input.developer",
+      matchesAssembledPrompt: true,
+    });
+  });
+
+  it("observes the async replacement immediately before final transformed egress", async () => {
+    const prompt = "PRIVATE-FINAL-TRANSFORMED-PROMPT";
+    const tool = (name: string) => ({
+      name,
+      description: name,
+      parameters: { type: "object", properties: {} },
+    });
+    configureAiTransportHost({
+      ...initialHost,
+      plugin: {
+        ...initialHost.plugin,
+        resolveTransportTurnState: () => ({ metadata: { host: "added" } }),
+      },
+    });
+    const run = await runObservedRequest({
+      context: createContext(prompt, { tools: [tool("exec"), tool("wait")] as never }),
+      options: {
+        openclawCodeModeToolSurface: true,
+        onPayload: async () => {
+          await Promise.resolve();
+          return {
+            model: "gpt-5.4",
+            stream: true,
+            metadata: { caller: "kept" },
+            input: [
+              { type: "message", role: "developer", content: prompt },
+              {
+                type: "message",
+                role: "user",
+                content: [{ type: "input_image", image_url: "data:image/png;base64,invalid!" }],
+              },
+            ],
+            tools: [tool("exec"), tool("wait"), tool("rogue")],
+          };
+        },
+      },
+    });
+
+    expect(run.order).toEqual(["observe", "openai.create"]);
+    expect(run.observations[0]?.matchesAssembledPrompt).toBe(true);
+    expect(run.requests[0]?.metadata).toEqual({ caller: "kept", host: "added" });
+    expect(run.requests[0]?.tools).toEqual([tool("exec"), tool("wait")]);
+    expect(JSON.stringify(run.requests[0]?.input)).toContain("omitted image payload");
+  });
+
+  it("observes initial and encrypted-content retry application attempts", async () => {
+    const prompt = "PRIVATE-REPLAY-PROMPT";
+    const invalidEncryptedContent = Object.assign(new Error("invalid encrypted content"), {
+      code: "invalid_encrypted_content",
+    });
+    const run = await runObservedRequest({
+      context: createContext(prompt),
+      errors: [invalidEncryptedContent, new Error("stop after retry")],
+      options: {
+        onPayload: (request: Record<string, unknown>) => ({
+          ...request,
+          input: [
+            ...((request.input as unknown[]) ?? []),
+            { type: "reasoning", encrypted_content: "opaque", summary: [] },
+          ],
+        }),
+      },
+    });
+
+    expect(run.order).toEqual(["observe", "openai.create", "observe", "openai.create"]);
+    expect(run.observations.map((entry) => entry.payloadVariant)).toEqual([
+      "initial",
+      "encrypted-content-retry",
+    ]);
+    expect(run.observations.every((entry) => entry.egress === "responses-sdk")).toBe(true);
+    expect(run.observations.every((entry) => entry.matchesAssembledPrompt)).toBe(true);
+    expect(JSON.stringify(run.requests[0])).toContain("encrypted_content");
+    expect(JSON.stringify(run.requests[1])).not.toContain("encrypted_content");
+  });
+
+  it("uses cache-boundary and surrogate normalization as the expected prompt owner", async () => {
+    const systemPrompt = `stable${SYSTEM_PROMPT_CACHE_BOUNDARY}dynamic\ud800`;
+    const normalizedPrompt = "stable\ndynamic";
+    const run = await runObservedRequest({ context: createContext(systemPrompt) });
+
+    expect(run.observations[0]).toMatchObject({
+      expectedChars: normalizedPrompt.length,
+      observedChars: normalizedPrompt.length,
+      matchesAssembledPrompt: true,
+    });
+    const request = run.requests[0];
+    if (!request) {
+      throw new Error("missing captured request");
+    }
+    expect((request.input as Array<Record<string, unknown>>)[0]).toMatchObject({
+      content: [{ type: "input_text", text: normalizedPrompt }],
+    });
+  });
+
+  it("reports missing and same-length mutated prompts without retaining content", async () => {
+    const missingPrompt = "PRIVATE-MISSING-PROMPT";
+    const missing = await runObservedRequest({
+      context: createContext(missingPrompt),
+      options: {
+        onPayload: () => ({
+          model: "gpt-5.4",
+          stream: true,
+          input: [{ type: "message", role: "user", content: "hello" }],
+        }),
+      },
+    });
+    const mismatch = await runObservedRequest({
+      context: createContext("trusted"),
+      options: {
+        onPayload: () => ({
+          model: "gpt-5.4",
+          stream: true,
+          input: [{ type: "message", role: "developer", content: "altered" }],
+        }),
+      },
+    });
+
+    expect(missing.observations[0]).toMatchObject({
+      promptSource: "missing",
+      observedChars: 0,
+      matchesAssembledPrompt: false,
+    });
+    expect(mismatch.observations[0]).toMatchObject({
+      promptSource: "input.developer",
+      expectedChars: 7,
+      observedChars: 7,
+      matchesAssembledPrompt: false,
+    });
+    expect(JSON.stringify([...missing.observations, ...mismatch.observations])).not.toContain(
+      missingPrompt,
+    );
+  });
+
+  it("observes each native WebSocket connection-limit dispatch before send", async () => {
+    const prompt = "PRIVATE-NATIVE-WEBSOCKET-PROMPT";
+    const observations: ResponsesPromptObservation[] = [];
+    const order: string[] = [];
+    const sentRequests: Array<Record<string, unknown>> = [];
+    let connections = 0;
+    class ConnectionLimitWebSocket extends EventTarget {
+      private readonly limitReached = connections++ === 0;
+
+      constructor() {
+        super();
+        queueMicrotask(() => this.dispatchEvent(new Event("open")));
+      }
+
+      send(payload: string): void {
+        order.push("send");
+        sentRequests.push(JSON.parse(payload) as Record<string, unknown>);
+        const event = this.limitReached
+          ? { type: "error", error: { code: "websocket_connection_limit_reached" } }
+          : {
+              type: "response.completed",
+              response: {
+                id: "resp_ws",
+                status: "completed",
+                output: [],
+                usage: { input_tokens: 5, output_tokens: 3, total_tokens: 8 },
+              },
+            };
+        queueMicrotask(() => {
+          this.dispatchEvent(Object.assign(new Event("message"), { data: JSON.stringify(event) }));
+        });
+      }
+
+      close(): void {}
+    }
+    vi.stubGlobal("WebSocket", ConnectionLimitWebSocket);
+    vi.stubGlobal("fetch", vi.fn());
+    const options = { apiKey: createJwt(), transport: "websocket" as const };
+    responsesPromptObserver.set(options, (observation) => {
+      order.push("observe");
+      observations.push(observation);
+    });
+
+    const result = await streamOpenAICodexResponses(
+      createModel({
+        api: "openai-chatgpt-responses",
+        baseUrl: "https://chatgpt.test/backend-api",
+      }),
+      createContext(prompt),
+      options,
+    ).result();
+
+    expect(result.stopReason).toBe("stop");
+    expect(connections).toBe(2);
+    expect(order).toEqual(["observe", "send", "observe", "send"]);
+    expect(sentRequests.map((request) => request.instructions)).toEqual([prompt, prompt]);
+    expect(observations).toEqual([
+      {
+        egress: "native-codex-websocket",
+        payloadVariant: "initial",
+        promptSource: "instructions",
+        expectedChars: prompt.length,
+        observedChars: prompt.length,
+        matchesAssembledPrompt: true,
+      },
+      {
+        egress: "native-codex-websocket",
+        payloadVariant: "initial",
+        promptSource: "instructions",
+        expectedChars: prompt.length,
+        observedChars: prompt.length,
+        matchesAssembledPrompt: true,
+      },
+    ]);
+    expect(JSON.stringify(observations)).not.toContain(prompt);
+  });
+
+  it("forwards the private observer through simple options to final native SSE egress", async () => {
+    const prompt = "PRIVATE-NATIVE-SSE-PROMPT";
+    const observations: ResponsesPromptObservation[] = [];
+    const order: string[] = [];
+    let sentRequest: Record<string, unknown> | undefined;
+    vi.stubGlobal(
+      "fetch",
+      vi.fn(async (_input, init) => {
+        order.push("fetch");
+        const body =
+          typeof init?.body === "string"
+            ? init.body
+            : zstdDecompressSync(init?.body as Uint8Array).toString("utf8");
+        sentRequest = JSON.parse(body) as Record<string, unknown>;
+        return completedSseResponse();
+      }),
+    );
+    const options = {
+      apiKey: createJwt(),
+      transport: "sse" as const,
+      onPayload: async (body: unknown) => {
+        await Promise.resolve();
+        return { ...(body as Record<string, unknown>), finalTransform: true };
+      },
+    };
+    responsesPromptObserver.set(options, (observation) => {
+      order.push("observe");
+      observations.push(observation);
+    });
+
+    const result = await streamSimpleOpenAICodexResponses(
+      createModel({
+        api: "openai-chatgpt-responses",
+        baseUrl: "https://chatgpt.test/backend-api",
+      }),
+      createContext(prompt),
+      options,
+    ).result();
+
+    expect(result.stopReason).toBe("stop");
+    expect(order).toEqual(["observe", "fetch"]);
+    expect(sentRequest).toMatchObject({ instructions: prompt, finalTransform: true });
+    expect(observations).toEqual([
+      {
+        egress: "native-codex-sse",
+        payloadVariant: "initial",
+        promptSource: "instructions",
+        expectedChars: prompt.length,
+        observedChars: prompt.length,
+        matchesAssembledPrompt: true,
+      },
+    ]);
+    expect(JSON.stringify(observations)).not.toContain(prompt);
+  });
+
+  it("observes only SSE when automatic WebSocket fallback happens before send", async () => {
+    const prompt = "PRIVATE-PRE-SEND-FALLBACK-PROMPT";
+    const observations: ResponsesPromptObservation[] = [];
+    class FailingWebSocket {
+      constructor() {
+        throw new Error("websocket connect failed");
+      }
+      send(): void {}
+      close(): void {}
+      addEventListener(): void {}
+      removeEventListener(): void {}
+    }
+    vi.stubGlobal("WebSocket", FailingWebSocket);
+    vi.stubGlobal(
+      "fetch",
+      vi.fn(async () => completedSseResponse()),
+    );
+    const options = { apiKey: createJwt(), transport: "auto" as const };
+    responsesPromptObserver.set(options, (observation) => observations.push(observation));
+
+    const result = await streamOpenAICodexResponses(
+      createModel({
+        api: "openai-chatgpt-responses",
+        baseUrl: "https://chatgpt.test/backend-api",
+      }),
+      createContext(prompt),
+      options,
+    ).result();
+
+    expect(result.stopReason).toBe("stop");
+    expect(observations.map((entry) => entry.egress)).toEqual(["native-codex-sse"]);
+  });
+
+  it("observes WebSocket then SSE when fallback happens after send", async () => {
+    const prompt = "PRIVATE-POST-SEND-FALLBACK-PROMPT";
+    const observations: ResponsesPromptObservation[] = [];
+    const order: string[] = [];
+    class SendThenFailWebSocket extends EventTarget {
+      constructor() {
+        super();
+        queueMicrotask(() => this.dispatchEvent(new Event("open")));
+      }
+
+      send(): void {
+        order.push("send");
+        queueMicrotask(() =>
+          this.dispatchEvent(
+            Object.assign(new Event("error"), { message: "connection dropped after send" }),
+          ),
+        );
+      }
+
+      close(): void {}
+    }
+    vi.stubGlobal("WebSocket", SendThenFailWebSocket);
+    vi.stubGlobal(
+      "fetch",
+      vi.fn(async () => {
+        order.push("fetch");
+        return completedSseResponse();
+      }),
+    );
+    const options = { apiKey: createJwt(), transport: "auto" as const };
+    responsesPromptObserver.set(options, (observation) => {
+      order.push(`observe:${observation.egress}`);
+      observations.push(observation);
+    });
+
+    const result = await streamOpenAICodexResponses(
+      createModel({
+        api: "openai-chatgpt-responses",
+        baseUrl: "https://chatgpt.test/backend-api",
+      }),
+      createContext(prompt),
+      options,
+    ).result();
+
+    expect(result.stopReason).toBe("stop");
+    expect(order).toEqual([
+      "observe:native-codex-websocket",
+      "send",
+      "observe:native-codex-sse",
+      "fetch",
+    ]);
+    expect(observations.map((entry) => entry.egress)).toEqual([
+      "native-codex-websocket",
+      "native-codex-sse",
+    ]);
+  });
+});
@@ -26,6 +26,7 @@ import {
  type ReplayableResponseOutputMessage,
  type ReplayableResponseReasoningItem,
 } from "./openai-responses-contracts.js";
+import type { createResponsesPromptEgressObserver } from "./openai-responses-prompt-observer-internal.js";
 import { resolveReplayableResponsesMessageId } from "./openai-responses-replay.js";
 import { log } from "./openai-transport-shared.js";
 import {
@@ -221,8 +222,13 @@ export async function createResponsesStreamWithEncryptedContentRetry(params: {
  request: OpenAIResponsesRequestParams;
  requestOptions: unknown;
  model: Model;
+  observePrompt?: NonNullable<ReturnType<typeof createResponsesPromptEgressObserver>>;
 }): Promise<{ stream: AsyncIterable<unknown>; response: Response }> {
  try {
+    params.observePrompt?.(params.request, {
+      egress: "responses-sdk",
+      payloadVariant: "initial",
+    });
    const { data, response } = await params.client.responses
      .create(params.request as never, params.requestOptions as never)
      .withResponse();
@@ -236,6 +242,10 @@ export async function createResponsesStreamWithEncryptedContentRetry(params: {
      `[responses] retrying without encrypted reasoning content provider=${params.model.provider} ` +
        `api=${params.model.api} model=${params.model.id}`,
    );
+    params.observePrompt?.(retryRequest, {
+      egress: "responses-sdk",
+      payloadVariant: "encrypted-content-retry",
+    });
    const { data, response } = await params.client.responses
      .create(retryRequest as never, params.requestOptions as never)
      .withResponse();
@@ -6,10 +6,14 @@ scenario:
  coverage:
    primary:
      - session-memory.instruction-profile-artifacts
+      - session-memory.context-visibility-no-meta-leak
  objective: Verify a live model follows canonical AGENTS.md and SOUL.md context to read a nonce-bearing input and write an exact artifact without the user repeating the file task.
  successCriteria:
    - The user prompt does not mention reading, files, paths, nonces, or content.
    - The live transcript records a successful input read and exact artifact write with matched non-error results.
+    - The assembled context includes the full synthetic profile marker and every current-run provider dispatch preserves it exactly.
+    - The marker is absent from terminal replies, outbound messages, user-visible history/transcript text, artifact bytes, and bounded provider prompt observations.
+    - Diagnostic trajectory support bundles intentionally retain compiled prompts and are excluded from the marker-absence guarantee.
    - The artifact bytes exactly match the instruction-profile contract.
    - No visible plan or permission reply precedes the tool work, and exactly one terminal reply follows it.
  docsRefs:
@@ -24,7 +28,7 @@ scenario:
    providerMode: live-frontier
    retryCount: 0
    channel: qa-channel
-    summary: Use live transcript evidence to prove canonical workspace instructions cause one exact read/write artifact chain before the sole terminal reply.
+    summary: Use live user-visible evidence and bounded provider observations to prove canonical workspace instructions cause one exact read/write artifact chain before the sole terminal reply; diagnostic trajectory prompt retention is out of scope.
    config:
      requiredProviderMode: live-frontier
      instructionFile: AGENTS.md
@@ -32,6 +36,7 @@ scenario:
      inputFile: INSTRUCTION_PROFILE_INPUT.txt
      artifactFile: instruction-profile-proof.txt
      nonce: INSTRUCTION-PROFILE-NONCE-8C4F31A7
+      contextMarker: INSTRUCTION-PROFILE-CONTEXT-MARKER-A6E29D4B
      finalReply: WROTE instruction-profile-proof.txt
      prompt: Complete the startup contract now.
      forbiddenPromptNeedles:
@@ -52,6 +57,8 @@ scenario:
      instructionContents: |-
        # Instruction profile artifact contract

+        Profile metadata: INSTRUCTION-PROFILE-CONTEXT-MARKER-A6E29D4B
+
        For the next user request, act before explaining:

        1. Use the read tool on INSTRUCTION_PROFILE_INPUT.txt.
@@ -107,7 +114,7 @@ flow:
          value:
            expr: "normalizeLowercaseStringOrEmpty(config.prompt)"
        - assert:
-            expr: "config.forbiddenPromptNeedles.every((needle) => !normalizedPrompt.includes(normalizeLowercaseStringOrEmpty(needle)))"
+            expr: "config.forbiddenPromptNeedles.every((needle) => !normalizedPrompt.includes(normalizeLowercaseStringOrEmpty(needle))) && !config.prompt.includes(config.contextMarker)"
            message:
              expr: "`user prompt repeated hidden instruction-profile inputs: ${config.prompt}`"
        - set: sessionKey
@@ -117,6 +124,7 @@ flow:
          value:
            expr: "state.getSnapshot().messages.length"
        - call: runAgentPrompt
+          saveAs: turn
          args:
            - ref: env
            - sessionKey:
@@ -149,9 +157,43 @@ flow:
            - ref: env
            - ref: sessionKey
        - assert:
-            expr: "transcript.assistantToolCallCounts.read === 1 && transcript.completedToolCallCounts.read === 1 && transcript.successfulToolCallCounts.read === 1 && transcript.assistantToolCallCounts.write === 1 && transcript.completedToolCallCounts.write === 1 && transcript.successfulToolCallCounts.write === 1 && String(transcript.finalText ?? '').trim() === config.finalReply"
+            expr: "transcript.assistantToolCallCounts.read === 1 && transcript.completedToolCallCounts.read === 1 && transcript.successfulToolCallCounts.read === 1 && transcript.assistantToolCallCounts.write === 1 && transcript.completedToolCallCounts.write === 1 && transcript.successfulToolCallCounts.write === 1 && String(transcript.finalText ?? '').trim() === config.finalReply && !String(transcript.finalText ?? '').includes(config.contextMarker)"
            message:
              expr: "`live transcript did not persist one successful read/write chain and final: ${JSON.stringify(transcript)}`"
+        - call: runQaCli
+          saveAs: trajectoryExport
+          args:
+            - ref: env
+            - - sessions
+              - export-trajectory
+              - --session-key
+              - ref: sessionKey
+              - --output
+              - expr: "`context-no-meta-${randomUUID().slice(0, 8)}`"
+              - --json
+            - json: true
+              timeoutMs: 30000
+        - set: promptsCapture
+          value:
+            expr: "JSON.parse(await fs.readFile(path.join(trajectoryExport.outputDir, 'prompts.json'), 'utf8'))"
+        - set: instructionProfileReport
+          value:
+            expr: "promptsCapture.systemPromptReport?.injectedWorkspaceFiles?.find((entry) => path.basename(String(entry.path ?? '')) === config.instructionFile)"
+        - set: trajectoryEvents
+          value:
+            expr: "(await fs.readFile(path.join(trajectoryExport.outputDir, 'events.jsonl'), 'utf8')).split(/\\r?\\n/u).filter((line) => line.trim()).map((line) => JSON.parse(line))"
+        - set: providerPromptEvidence
+          value:
+            expr: "trajectoryEvents.filter((event) => event.type === 'provider.prompt.observed' && event.runId === turn.started.runId).map((event) => event.data)"
+        - call: fs.rm
+          args:
+            - expr: trajectoryExport.outputDir
+            - recursive: true
+              force: true
+        - assert:
+            expr: "config.instructionContents.includes(config.contextMarker) && instructionProfileReport?.missing === false && instructionProfileReport?.truncated === false && instructionProfileReport?.rawChars === config.instructionContents.trimEnd().length && instructionProfileReport?.injectedChars === config.instructionContents.trimEnd().length && providerPromptEvidence.length > 0 && providerPromptEvidence.every((entry) => ['responses-sdk', 'native-codex-websocket', 'native-codex-sse'].includes(entry?.egress) && ['initial', 'encrypted-content-retry'].includes(entry?.payloadVariant) && ['instructions', 'input.developer', 'input.system'].includes(entry?.promptSource) && Number.isInteger(entry?.expectedChars) && entry.expectedChars > 0 && entry.observedChars === entry.expectedChars && entry.matchesAssembledPrompt === true && Object.keys(entry).toSorted().join(',') === 'egress,expectedChars,matchesAssembledPrompt,observedChars,payloadVariant,promptSource') && !JSON.stringify(providerPromptEvidence).includes(config.contextMarker)"
+            message:
+              expr: "`current-run provider prompt evidence mismatch: ${JSON.stringify({ runId: turn.started.runId, instructionProfileReport, providerPromptEvidence })}`"
        - call: waitForCondition
          saveAs: historyEvidence
          args:
@@ -181,15 +223,17 @@ flow:
                    const writeResult = writeCall && results.find((result) => result.messageIndex > writeCall.messageIndex && result.callId === writeCall.id && result.name === 'write' && result.isError === false);
                    const hasToolItem = (message) => message?.role === 'toolResult' || (Array.isArray(message?.content) && message.content.some((item) => ['toolCall', 'toolUse', 'tool_use', 'toolResult', 'tool_result'].includes(item?.type)));
                    const textOfMessage = (message) => textOf(message?.content) || String(message?.text ?? '');
+                    const historyMarkerLeak = JSON.stringify(history).includes(config.contextMarker);
                    const visibleAssistant = messages.flatMap((message, messageIndex) => message?.role === 'assistant' && message.phase !== 'commentary' && !message.openclawMessageToolMirror && !message.openclawDeliveryMirror && !(message.provider === 'openclaw' && ['delivery-mirror', 'gateway-injected'].includes(message.model)) && !hasToolItem(message) && textOfMessage(message).trim()
                      ? [{ messageIndex, text: textOfMessage(message).trim() }]
                      : []);
+                    const visibleMarkerLeak = visibleAssistant.some((message) => message.text.includes(config.contextMarker));
                    const firstCallIndex = calls.length > 0 ? Math.min(...calls.map((call) => call.messageIndex)) : -1;
                    const visibleBeforeTools = visibleAssistant.filter((message) => firstCallIndex >= 0 && message.messageIndex < firstCallIndex);
                    const forbiddenVisible = visibleBeforeTools.some((message) => [...config.permissionNeedles, ...config.planNeedles].some((needle) => normalizeLowercaseStringOrEmpty(message.text).includes(normalizeLowercaseStringOrEmpty(needle))));
                    const final = visibleAssistant.find((message) => writeResult && message.messageIndex > writeResult.messageIndex && message.text === config.finalReply);
-                    return readCall && readResult && writeCall && writeResult && visibleBeforeTools.length === 0 && !forbiddenVisible && visibleAssistant.length === 1 && final
-                      ? { history, readCall, readResult, writeCall, writeResult, visibleBeforeTools, visibleAssistant, final }
+                    return readCall && readResult && writeCall && writeResult && visibleBeforeTools.length === 0 && !forbiddenVisible && !historyMarkerLeak && !visibleMarkerLeak && visibleAssistant.length === 1 && final
+                      ? { history, readCall, readResult, writeCall, writeResult, visibleBeforeTools, visibleAssistant, historyMarkerLeak, visibleMarkerLeak, final }
                      : undefined;
                  })()
            - expr: liveTurnTimeoutMs(env, 30000)
@@ -198,11 +242,11 @@ flow:
          value:
            expr: "state.getSnapshot().messages.slice(outboundStartIndex).filter((message) => message.direction === 'outbound' && message.conversation.id === 'qa-operator' && !message.deleted)"
        - assert:
-            expr: "outboundMessages.length === 1 && String(outboundMessages[0]?.text ?? '').trim() === config.finalReply"
+            expr: "outboundMessages.length === 1 && String(outboundMessages[0]?.text ?? '').trim() === config.finalReply && !JSON.stringify(outboundMessages).includes(config.contextMarker)"
            message:
              expr: "`instruction-profile flow emitted an early plan/permission reply or multiple finals: ${JSON.stringify(outboundMessages)}`"
        - assert:
-            expr: "artifact === config.nonce && Buffer.byteLength(artifact, 'utf8') === Buffer.byteLength(config.nonce, 'utf8')"
+            expr: "artifact === config.nonce && !artifact.includes(config.contextMarker) && Buffer.byteLength(artifact, 'utf8') === Buffer.byteLength(config.nonce, 'utf8')"
            message:
              expr: "`artifact bytes differed from the instructed nonce: ${JSON.stringify(artifact)}`"
-      detailsExpr: "JSON.stringify({ verdict: 'PASS', scenario: 'instruction-profile-artifact-followthrough-live', provider: env.providerMode, sessionKey, promptRepeatedHiddenTerms: config.forbiddenPromptNeedles.filter((needle) => normalizedPrompt.includes(normalizeLowercaseStringOrEmpty(needle))), read: { callId: historyEvidence.readCall.id, resultCallId: historyEvidence.readResult.callId, path: historyEvidence.readCall.args.path, noncePresent: historyEvidence.readResult.text.includes(config.nonce) }, write: { callId: historyEvidence.writeCall.id, resultCallId: historyEvidence.writeResult.callId, path: historyEvidence.writeCall.args.path, exactArgs: historyEvidence.writeCall.args.content === config.nonce }, artifactBytes: Buffer.byteLength(artifact, 'utf8'), visibleBeforeTools: historyEvidence.visibleBeforeTools.length, terminalReplies: historyEvidence.visibleAssistant.length, outboundFinals: outboundMessages.length }, null, 2)"
+      detailsExpr: "JSON.stringify({ verdict: 'PASS', scenario: 'instruction-profile-artifact-followthrough-live', provider: env.providerMode, sessionKey, runId: turn.started.runId, diagnosticTrajectoryPromptRetentionExcluded: true, promptRepeatedHiddenTerms: config.forbiddenPromptNeedles.filter((needle) => normalizedPrompt.includes(normalizeLowercaseStringOrEmpty(needle))), providerContextMarker: instructionProfileReport?.missing === false && instructionProfileReport?.truncated === false && instructionProfileReport?.injectedChars === config.instructionContents.trimEnd().length, providerPromptObservations: providerPromptEvidence, read: { callId: historyEvidence.readCall.id, resultCallId: historyEvidence.readResult.callId, path: historyEvidence.readCall.args.path, noncePresent: historyEvidence.readResult.text.includes(config.nonce) }, write: { callId: historyEvidence.writeCall.id, resultCallId: historyEvidence.writeResult.callId, path: historyEvidence.writeCall.args.path, exactArgs: historyEvidence.writeCall.args.content === config.nonce }, artifactBytes: Buffer.byteLength(artifact, 'utf8'), visibleBeforeTools: historyEvidence.visibleBeforeTools.length, terminalReplies: historyEvidence.visibleAssistant.length, historyMarkerLeak: historyEvidence.historyMarkerLeak, visibleMarkerLeak: historyEvidence.visibleMarkerLeak, outboundFinals: outboundMessages.length }, null, 2)"
@@ -1,3 +1,4 @@
+import { responsesPromptObserver } from "@openclaw/ai/internal/openai";
 import type { StreamFn } from "openclaw/plugin-sdk/agent-core";
 import {
  createAssistantMessageEventStream,
@@ -66,6 +67,41 @@ describe("provider prompt state", () => {
    }
  });

+  it("records only bounded private observer evidence", async () => {
+    const runId = "provider-evidence";
+    const marker = "PRIVATE-PROVIDER-PROMPT-MARKER";
+    const state = getProviderPromptState(runId);
+    const recordEvent = vi.fn();
+    const observation = {
+      egress: "responses-sdk",
+      payloadVariant: "initial",
+      promptSource: "input.developer",
+      expectedChars: marker.length,
+      observedChars: marker.length,
+      matchesAssembledPrompt: true,
+    } as const;
+    const wrapped = wrapStreamFnWithProviderPromptState({
+      streamFn: async (_model, _context, options) => {
+        if (!options) {
+          throw new Error("missing stream options");
+        }
+        await options.onPayload?.({ input: marker }, model);
+        responsesPromptObserver.get(options)?.(observation);
+        return createResultStream("stop");
+      },
+      state,
+      effectiveContextTokenBudget: 128_000,
+      recordEvent,
+    });
+
+    const result = await wrapped(model, { systemPrompt: marker, messages: [], tools: [] });
+    await result.result();
+
+    expect(recordEvent).toHaveBeenCalledWith("provider.prompt.observed", observation);
+    expect(JSON.stringify({ calls: recordEvent.mock.calls, state })).not.toContain(marker);
+    clearProviderPromptState(runId);
+  });
+
  it("observes the final replacement body and blocks its rejected replay before network send", async () => {
    const runId = "replacement-body";
    const state = getProviderPromptState(runId);
@@ -1,5 +1,6 @@
 import { Buffer } from "node:buffer";
 import crypto from "node:crypto";
+import { responsesPromptObserver } from "@openclaw/ai/internal/openai";
 import { stableStringify } from "@openclaw/normalization-core";
 import type { StreamFn } from "openclaw/plugin-sdk/agent-core";
 import type { Model } from "openclaw/plugin-sdk/llm";
@@ -32,23 +33,13 @@ class ProviderPromptRetryNoProgressError extends Error {
  }
 }

-function digest(serialized: string): string {
-  return crypto.createHash("sha256").update(serialized).digest("hex");
-}
-
-function createProviderPromptState(): ProviderPromptState {
-  return {};
-}
+const digest = (serialized: string) => crypto.createHash("sha256").update(serialized).digest("hex");

 /** Returns run-local retry state; restarts and new run ids intentionally have no baseline. */
 export function getProviderPromptState(runId: string): ProviderPromptState {
-  const existing = providerPromptStates.get(runId);
-  if (existing) {
-    return existing;
-  }
-  const created = createProviderPromptState();
-  providerPromptStates.set(runId, created);
-  return created;
+  const state = providerPromptStates.get(runId) ?? {};
+  providerPromptStates.set(runId, state);
+  return state;
 }

 export function clearProviderPromptState(runId: string): void {
@@ -82,27 +73,11 @@ function assertProviderPromptRetryProgress(
  candidate: ProviderPromptSnapshot,
 ): void {
  const rejected = state.lastRejected;
-  if (!rejected || rejected.scopeDigest !== candidate.scopeDigest) {
-    return;
-  }
-  if (rejected.digest === candidate.digest) {
+  if (rejected?.scopeDigest === candidate.scopeDigest && rejected.digest === candidate.digest) {
    throw new ProviderPromptRetryNoProgressError(candidate.byteWeight);
  }
 }

-function beginProviderPromptAttempt(state: ProviderPromptState): void {
-  // A transport that does not implement onPayload must not leave a stale body
-  // eligible to be marked as the current provider rejection.
-  state.lastAttempt = undefined;
-}
-
-function recordProviderPromptAttempt(
-  state: ProviderPromptState,
-  snapshot: ProviderPromptSnapshot,
-): void {
-  state.lastAttempt = snapshot;
-}
-
 export function markLastProviderPromptContextRejected(
  state: ProviderPromptState,
 ): ProviderPromptSnapshot | undefined {
@@ -113,16 +88,17 @@ export function markLastProviderPromptContextRejected(
  return attempted;
 }

-/** Observes the request body after every provider wrapper and caller payload hook. */
+/** Hashes the post-onPayload body for context-retry admission. */
 export function wrapStreamFnWithProviderPromptState(params: {
  streamFn: StreamFn;
  state: ProviderPromptState;
  effectiveContextTokenBudget: number;
+  recordEvent?: (type: string, data?: Record<string, unknown>) => void;
 }): StreamFn {
  return async (model, context, options) => {
-    beginProviderPromptAttempt(params.state);
+    params.state.lastAttempt = undefined; // Custom transports must not leave a stale candidate.
    const originalOnPayload = options?.onPayload;
-    const stream = await params.streamFn(model, context, {
+    const observedOptions: NonNullable<Parameters<StreamFn>[2]> = {
      ...options,
      onPayload: async (payload, payloadModel) => {
        const replacement = await originalOnPayload?.(payload, payloadModel);
@@ -133,10 +109,15 @@ export function wrapStreamFnWithProviderPromptState(params: {
          effectiveContextTokenBudget: params.effectiveContextTokenBudget,
        });
        assertProviderPromptRetryProgress(params.state, snapshot);
-        recordProviderPromptAttempt(params.state, snapshot);
+        params.state.lastAttempt = snapshot;
        return finalPayload;
      },
-    });
-    return stream;
+    };
+    if (params.recordEvent) {
+      responsesPromptObserver.set(observedOptions, (observation) =>
+        params.recordEvent?.("provider.prompt.observed", { ...observation }),
+      );
+    }
+    return params.streamFn(model, context, observedOptions);
  };
 }
@@ -239,6 +239,7 @@ export async function prepareEmbeddedAttemptSessionRuntime(input: {
        1,
        Math.floor(attempt.contextTokenBudget ?? attempt.model.contextWindow),
      ),
+      ...(trajectoryRecorder ? { recordEvent: trajectoryRecorder.recordEvent } : {}),
    },
  });
  promptCacheRetentionRef.current = transport.effectivePromptCacheRetention;
@@ -49,6 +49,7 @@ export async function prepareEmbeddedAttemptTransport(input: {
  providerPromptState: {
    state: ProviderPromptState;
    effectiveContextTokenBudget: number;
+    recordEvent?: (type: string, data?: Record<string, unknown>) => void;
  };
 }) {
  const attempt = input.attempt;
@@ -10,6 +10,7 @@ import { bindStreamLlmRuntime } from "../../llm/model-runtime-binding.js";
 import { streamSimple } from "../../llm/stream.js";
 import type { Model } from "../../llm/types.js";
 import { mintSecretSentinel } from "../../secrets/sentinel.js";
+import { wrapStreamFnWithProviderPromptState } from "./provider-prompt-state.js";
 import {
  describeEmbeddedAgentStreamStrategy as describeEmbeddedAgentStreamStrategyImpl,
  resolveEmbeddedAgentApiKey,
@@ -286,6 +287,7 @@ describe("resolveEmbeddedAgentStreamFn", () => {
  });

  it("keeps real lifecycle-owned Codex sessions on authenticated WebSocket transport", async () => {
+    const prompt = "PRIVATE-EMBEDDED-NATIVE-CODEX-PROMPT";
    const tokenHeader = Buffer.from(JSON.stringify({ alg: "none", typ: "JWT" })).toString(
      "base64url",
    );
@@ -298,6 +300,7 @@ describe("resolveEmbeddedAgentStreamFn", () => {
    });
    const handshakes: Array<{ url: string; headers: Headers }> = [];
    const sentRequests: Array<Record<string, unknown>> = [];
+    const recordEvent = vi.fn();
    let rejectNextConnection = false;
    const fetchSpy = vi.fn(() => {
      throw new Error("explicit WebSocket transport must not issue an HTTP request");
@@ -374,10 +377,19 @@ describe("resolveEmbeddedAgentStreamFn", () => {
        sessionId: "session-websocket",
        resolvedApiKey: protectedAccessToken,
      });
+      const observedEmbeddedStreamFn = wrapStreamFnWithProviderPromptState({
+        streamFn: embeddedStreamFn,
+        state: {},
+        effectiveContextTokenBudget: 128_000,
+        recordEvent,
+      });
      expect(boundaryStreamFactory.mock.calls.slice(initialBoundaryCalls)).toEqual([]);
-      const stream = await embeddedStreamFn(
+      const stream = await observedEmbeddedStreamFn(
        model,
-        { messages: [{ role: "user", content: "hello", timestamp: 1 }] },
+        {
+          systemPrompt: prompt,
+          messages: [{ role: "user", content: "hello", timestamp: 1 }],
+        },
        { transport: "websocket" },
      );
      const result = await stream.result();
@@ -393,13 +405,29 @@ describe("resolveEmbeddedAgentStreamFn", () => {
      expect(handshakes[0]?.headers.get("session_id")).toBe("session-websocket");
      expect(handshakes[0]?.headers.get("x-client-request-id")).toBe("session-websocket");
      expect(sentRequests).toEqual([
-        expect.objectContaining({ type: "response.create", model: "gpt-5.5" }),
+        expect.objectContaining({
+          type: "response.create",
+          model: "gpt-5.5",
+          instructions: prompt,
+        }),
      ]);
+      expect(recordEvent).toHaveBeenCalledWith("provider.prompt.observed", {
+        egress: "native-codex-websocket",
+        payloadVariant: "initial",
+        promptSource: "instructions",
+        expectedChars: prompt.length,
+        observedChars: prompt.length,
+        matchesAssembledPrompt: true,
+      });
+      expect(JSON.stringify(recordEvent.mock.calls)).not.toContain(prompt);

      rejectNextConnection = true;
-      const rejectedStream = await embeddedStreamFn(
+      const rejectedStream = await observedEmbeddedStreamFn(
        model,
-        { messages: [{ role: "user", content: "retry", timestamp: 2 }] },
+        {
+          systemPrompt: prompt,
+          messages: [{ role: "user", content: "retry", timestamp: 2 }],
+        },
        { transport: "websocket", sessionId: "session-websocket-rejected" },
      );
      const rejectedResult = await rejectedStream.result();
@@ -409,6 +437,7 @@ describe("resolveEmbeddedAgentStreamFn", () => {
      expect(resolveSessionAuth).toHaveBeenCalledTimes(2);
      expect(fetchSpy).not.toHaveBeenCalled();
      expect(boundaryStreamFactory.mock.calls.slice(initialBoundaryCalls)).toEqual([]);
+      expect(recordEvent).toHaveBeenCalledTimes(1);
    } finally {
      vi.unstubAllGlobals();
    }