fix(responses): drop account-bound continuation when the serving account changes (#4546)

scripts/test-layout/layout.json

@@ -1452,7 +1452,8 @@
     "chat-media-translation.test.ts": "responses",
     "execution-budget-permits.test.ts": "lib",
     "spend-instrumentation-log.test.ts": "server",
-    "codex-pool-refresh-backoff.test.ts": "codex-integration"
+    "codex-pool-refresh-backoff.test.ts": "codex-integration",
+    "responses-account-change-scrub.test.ts": "responses"
   },
   "migrated": [
     "adapters",

src/codex/routing.ts

@@ -338,6 +338,18 @@ const LEGACY_THREAD_AFFINITY_SCOPE = "legacy" as const;
 const threadAccountMap = new Map<string, Map<ThreadAffinityScope, ThreadAffinityEntry>>();
 let threadAffinityEntryTotal = 0;
 
+/**
+ * Which pool account minted the conversation's carried OpenAI state
+ * (`previous_response_id`, encrypted reasoning, provider conversation/file ids).
+ * Keyed by the same affinity key as {@link threadAccountMap}, bounded the same
+ * way, and process-local — raw account ids never reach a log.
+ */
+type ConversationStateIssuerEntry = {
+  accountId: string;
+  lastUsedAt: number;
+};
+const conversationStateIssuerMap = new Map<string, ConversationStateIssuerEntry>();
+
 function isModelDetourAffinityScope(scope: ThreadAffinityScope): scope is ModelDetourAffinityScope {
   return scope.startsWith("model-detour:");
 }
@@ -438,6 +450,11 @@ export function clearThreadAccountMap(): void {
   // A refresh cooldown is per-account runtime state learned alongside these bindings. Leaving it
   // behind here keeps an account out of selection after the roster it belonged to is gone.
   clearAllCodexPoolRefreshFailures();
+  conversationStateIssuerMap.clear();
+}
+
+export function clearConversationStateIssuerMap(): void {
+  conversationStateIssuerMap.clear();
 }
 
 export function clearThreadAccountMapForAccount(
@@ -455,6 +472,55 @@ export function clearThreadAccountMapForAccount(
   }
 }
 
+function pruneConversationStateIssuers(now: number): void {
+  for (const [key, entry] of conversationStateIssuerMap) {
+    if (now - entry.lastUsedAt > CODEX_THREAD_AFFINITY_IDLE_TTL_MS) {
+      conversationStateIssuerMap.delete(key);
+    }
+  }
+  while (conversationStateIssuerMap.size > CODEX_THREAD_AFFINITY_MAX_ENTRIES) {
+    let oldestKey: string | null = null;
+    let oldestAt = Number.POSITIVE_INFINITY;
+    for (const [key, entry] of conversationStateIssuerMap) {
+      if (entry.lastUsedAt < oldestAt) {
+        oldestAt = entry.lastUsedAt;
+        oldestKey = key;
+      }
+    }
+    if (!oldestKey) break;
+    conversationStateIssuerMap.delete(oldestKey);
+  }
+}
+
+/**
+ * Record the pool account that just issued carried conversation state for this
+ * binding key. In-memory only; the id is never written to a request log.
+ */
+export function rememberConversationStateIssuer(
+  bindingKey: string,
+  accountId: string,
+  now = Date.now(),
+): void {
+  if (!bindingKey.trim() || !accountId.trim()) return;
+  if (!admissibleAffinityComponent(bindingKey) || !admissibleAffinityComponent(accountId)) return;
+  pruneConversationStateIssuers(now);
+  conversationStateIssuerMap.set(bindingKey, { accountId, lastUsedAt: now });
+  pruneConversationStateIssuers(now);
+}
+
+/** Last account that minted carried state for this binding, if still in the TTL window. */
+export function peekConversationStateIssuer(
+  bindingKey: string,
+  now = Date.now(),
+): string | undefined {
+  if (!bindingKey.trim() || !admissibleAffinityComponent(bindingKey)) return undefined;
+  pruneConversationStateIssuers(now);
+  const entry = conversationStateIssuerMap.get(bindingKey);
+  if (!entry) return undefined;
+  entry.lastUsedAt = now;
+  return entry.accountId;
+}
+
 /**
  * Why a binding was released, held until that thread's next resolve can report it (#4546).
  *

src/server/request-log.ts

@@ -175,6 +175,11 @@ export interface RequestLogContext {
   affinity?: CodexAffinityMove;
   /** Why the binding was kept, moved, or released (#4546). */
   affinityReason?: CodexAffinityReason;
+  /**
+   * Set when this request dropped account-bound continuation because the serving
+   * Codex pool account was not the issuer. Never an account identifier.
+   */
+  conversationStateScrub?: "account-change";
   transportPhase?: "pre_headers" | "mid_stream" | "terminal_sse";
   terminalSource?: "upstream" | "synthetic";
   /** Bounded route-decision trace (RI-01); never contains secrets. */
@@ -252,6 +257,11 @@ export interface RequestLogEntry {
   affinity?: CodexAffinityMove;
   /** Why that decision was made (#4546): a move is the expensive event, so it names its cause. */
   affinityReason?: CodexAffinityReason;
+  /**
+   * Set when this request dropped account-bound continuation after a Codex pool
+   * account change. Never an account identifier.
+   */
+  conversationStateScrub?: "account-change";
   /** Where the upstream terminal/failure was observed. */
   transportPhase?: "pre_headers" | "mid_stream" | "terminal_sse";
   /**
@@ -385,6 +395,9 @@ export function requestLogEntryFromPersistedUsage(entry: PersistedUsageEntry): R
     ...(isKnownTerminalSource(entry.terminalSource) ? { terminalSource: entry.terminalSource } : {}),
     ...(routeDecision ? { routeDecision } : {}),
     ...(claudeCompatibility ? { claudeCompatibility } : {}),
+    ...(entry.conversationStateScrub === "account-change"
+      ? { conversationStateScrub: "account-change" }
+      : {}),
   };
 }
 
@@ -530,6 +543,9 @@ export function addRequestLog(entry: RequestLogEntry) {
       ...failureDiagnostics,
       ...(entry.routeDecision ? { routeDecision: entry.routeDecision } : {}),
       ...(entry.claudeCompatibility ? { claudeCompatibility: entry.claudeCompatibility } : {}),
+      ...(entry.conversationStateScrub === "account-change"
+        ? { conversationStateScrub: "account-change" }
+        : {}),
     });
   } catch {
     /* request logging must never fail a user request */
@@ -1311,6 +1327,9 @@ export function addFinalRequestLog(
     ...(loggedUsage || cacheProvenance !== "unknown" ? { cacheProvenance } : {}),
     ...(logCtx.affinity ? { affinity: logCtx.affinity } : {}),
     ...(logCtx.affinityReason ? { affinityReason: logCtx.affinityReason } : {}),
+    ...(logCtx.conversationStateScrub === "account-change"
+      ? { conversationStateScrub: "account-change" }
+      : {}),
     ...(logCtx.transportPhase ? { transportPhase: logCtx.transportPhase } : {}),
     ...(logCtx.terminalSource ? { terminalSource: logCtx.terminalSource } : {}),
     ...(logCtx.routeDecision ? { routeDecision: logCtx.routeDecision } : {}),

src/server/responses/account-change-state.ts

@@ -0,0 +1,233 @@
+/**
+ * Codex pool account-change conversation-state portability (#4546).
+ *
+ * OpenAI `encrypted_content` blobs and `previous_response_id` are bound to the
+ * account that minted them. When pool routing serves a live conversation on a
+ * different account, the next turn must drop that state once before dispatch so
+ * the new account can continue from readable history instead of rejecting the
+ * ciphertext.
+ *
+ * The issuer association lives next to thread affinity in `src/codex/routing.ts`.
+ */
+
+import type { CodexAuthContext } from "../../codex/auth-context";
+import {
+  peekConversationStateIssuer,
+  rememberConversationStateIssuer,
+} from "../../codex/routing";
+import type { OcxParsedRequest } from "../../types";
+import type { RequestLogContext } from "../request-log";
+
+export type ConversationStateScrubReason = "account-change";
+
+export type PortabilityDenial =
+  | "previous-response-id"
+  | "provider-conversation-id"
+  | "uploaded-file-ids"
+  | "encrypted-reasoning";
+
+/**
+ * The parts of a request that bind it to the credential that produced them.
+ * Presence is what matters; the values stay opaque so nothing here logs ids.
+ */
+export interface ConversationStateCarriers {
+  readonly previousResponseId?: string | null;
+  readonly providerConversationId?: string | null;
+  readonly fileIds?: readonly string[];
+  readonly encryptedReasoning?: unknown;
+}
+
+export type PortabilityVerdict =
+  | { readonly portable: true }
+  | { readonly portable: false; readonly reason: PortabilityDenial };
+
+function present(value: unknown): boolean {
+  if (value === undefined || value === null) return false;
+  if (typeof value === "string" || Array.isArray(value)) return value.length > 0;
+  return true;
+}
+
+/**
+ * Whether a request's conversational state can move credentials at all.
+ *
+ * `src/routing/identity-domains.ts` owns this decision once that module lands
+ * on this integration line (#4546). Keep the check in this one function so it
+ * can be swapped for the shared export without hunting call sites.
+ */
+export function canPortConversationState(
+  state: ConversationStateCarriers,
+): PortabilityVerdict {
+  if (present(state.previousResponseId)) {
+    return { portable: false, reason: "previous-response-id" };
+  }
+  if (present(state.providerConversationId)) {
+    return { portable: false, reason: "provider-conversation-id" };
+  }
+  if (present(state.fileIds)) {
+    return { portable: false, reason: "uploaded-file-ids" };
+  }
+  if (present(state.encryptedReasoning)) {
+    return { portable: false, reason: "encrypted-reasoning" };
+  }
+  return { portable: true };
+}
+
+function providerConversationIdFromBody(body: Record<string, unknown>): string | undefined {
+  const conversation = body.conversation;
+  if (typeof conversation === "string" && conversation.trim()) return conversation.trim();
+  if (conversation && typeof conversation === "object" && !Array.isArray(conversation)) {
+    const id = (conversation as { id?: unknown }).id;
+    if (typeof id === "string" && id.trim()) return id.trim();
+  }
+  return undefined;
+}
+
+function collectFileIds(input: unknown): string[] {
+  const ids: string[] = [];
+  if (!Array.isArray(input)) return ids;
+  for (const item of input) {
+    if (!item || typeof item !== "object") continue;
+    const record = item as Record<string, unknown>;
+    if (typeof record.file_id === "string" && record.file_id.trim()) ids.push(record.file_id);
+    if (Array.isArray(record.file_ids)) {
+      for (const id of record.file_ids) {
+        if (typeof id === "string" && id.trim()) ids.push(id);
+      }
+    }
+    for (const key of ["content", "output"]) {
+      const parts = record[key];
+      if (!Array.isArray(parts)) continue;
+      for (const part of parts) {
+        if (!part || typeof part !== "object") continue;
+        const partRecord = part as Record<string, unknown>;
+        if (typeof partRecord.file_id === "string" && partRecord.file_id.trim()) {
+          ids.push(partRecord.file_id);
+        }
+      }
+    }
+  }
+  return ids;
+}
+
+function hasEncryptedReasoning(input: unknown): boolean {
+  if (!Array.isArray(input)) return false;
+  for (const item of input) {
+    if (!item || typeof item !== "object") continue;
+    const record = item as Record<string, unknown>;
+    if (typeof record.encrypted_content === "string" && record.encrypted_content.length > 0) {
+      return true;
+    }
+    for (const key of ["content", "output"]) {
+      const parts = record[key];
+      if (!Array.isArray(parts)) continue;
+      for (const part of parts) {
+        if (!part || typeof part !== "object") continue;
+        const encrypted = (part as { encrypted_content?: unknown }).encrypted_content;
+        if (typeof encrypted === "string" && encrypted.length > 0) return true;
+      }
+    }
+  }
+  return false;
+}
+
+export function collectConversationStateCarriers(body: unknown): ConversationStateCarriers {
+  if (!body || typeof body !== "object" || Array.isArray(body)) return {};
+  const record = body as Record<string, unknown>;
+  const previousResponseId = typeof record.previous_response_id === "string"
+    ? record.previous_response_id
+    : undefined;
+  return {
+    previousResponseId,
+    providerConversationId: providerConversationIdFromBody(record),
+    fileIds: collectFileIds(record.input),
+    encryptedReasoning: hasEncryptedReasoning(record.input) ? true : undefined,
+  };
+}
+
+
+/**
+ * Drop account-bound continuation from a request body in place. Readable user
+ * messages and plaintext survive; ciphertext and continuation ids do not.
+ */
+export function scrubUnportableConversationStateInPlace(body: unknown): boolean {
+  if (!body || typeof body !== "object" || Array.isArray(body)) return false;
+  const record = body as Record<string, unknown>;
+  let changed = false;
+  if (typeof record.previous_response_id === "string") {
+    delete record.previous_response_id;
+    changed = true;
+  }
+  if (record.conversation != null) {
+    delete record.conversation;
+    changed = true;
+  }
+  // Encrypted reasoning and compaction ciphertext are deliberately NOT touched here. #2247
+  // already strips them when a pooled thread moves accounts, and in a specific shape: the
+  // reasoning item keeps its readable summary with an emptied content array, and the compaction
+  // item becomes an operator-readable note. Stripping again from this side produced a different
+  // shape and broke that contract for no gain. What #2247 does not cover, and what this function
+  // owns, is the continuation state naming server-side objects the new account cannot read:
+  // `previous_response_id` and a provider-side conversation id.
+  return changed;
+}
+
+export function conversationStateBindingFromAuth(
+  authCtx: CodexAuthContext,
+  fallbackAffinityKey?: string | null,
+): { accountId: string; bindingKey: string } | null {
+  if (authCtx.kind !== "pool" && authCtx.kind !== "main-pool") return null;
+  const bindingKey = authCtx.affinityKey ?? fallbackAffinityKey ?? undefined;
+  if (!bindingKey || !authCtx.accountId) return null;
+  return { accountId: authCtx.accountId, bindingKey };
+}
+
+export function rememberServingConversationStateIssuer(
+  authCtx: CodexAuthContext,
+  fallbackAffinityKey?: string | null,
+): void {
+  const binding = conversationStateBindingFromAuth(authCtx, fallbackAffinityKey);
+  if (!binding) return;
+  rememberConversationStateIssuer(binding.bindingKey, binding.accountId);
+}
+
+export interface ApplyAccountChangeConversationStateScrubArgs {
+  body: unknown;
+  bindingKey: string;
+  servingAccountId: string;
+  /** Account this request body was prepared for, when this is an in-request move. */
+  priorAccountId?: string | null;
+  parsed?: Pick<OcxParsedRequest, "previousResponseId" | "_stripReasoningEncryptedContent">;
+  logCtx?: RequestLogContext;
+}
+
+/**
+ * If the serving account is not the issuer of the carried state, strip that
+ * state from the outbound body before dispatch. One cold turn, not a permanent
+ * downgrade: the next successful serve records the new issuer.
+ */
+export function applyAccountChangeConversationStateScrub(
+  args: ApplyAccountChangeConversationStateScrubArgs,
+): boolean {
+  const { body, bindingKey, servingAccountId, priorAccountId, parsed, logCtx } = args;
+  if (!servingAccountId || !bindingKey) return false;
+  const issuer = peekConversationStateIssuer(bindingKey);
+  const accountChanged = (issuer != null && issuer !== servingAccountId)
+    || (priorAccountId != null && priorAccountId !== servingAccountId);
+  if (!accountChanged) return false;
+  if (canPortConversationState(collectConversationStateCarriers(body)).portable) return false;
+  const scrubbed = scrubUnportableConversationStateInPlace(body);
+  if (!scrubbed) return false;
+  if (parsed) {
+    delete parsed.previousResponseId;
+    parsed._stripReasoningEncryptedContent = true;
+  }
+  if (logCtx && logCtx.conversationStateScrub !== "account-change") {
+    console.warn(
+      "[opencodex] dropped continuation state after a Codex pool account change; continuing fresh",
+    );
+    logCtx.conversationStateScrub = "account-change";
+  } else if (logCtx) {
+    logCtx.conversationStateScrub = "account-change";
+  }
+  return true;
+}