fix: preserve response continuity on physical WebSockets

docker-healthcheck.sh

@@ -1,4 +1,7 @@
 #!/bin/sh
-# Read server port from config, fallback to 8080
-PORT=$(grep -A5 '^server:' /app/config/default.yaml 2>/dev/null | grep 'port:' | head -1 | awk '{print $2}')
-curl -fs "http://localhost:${PORT:-8080}/health" || exit 1
+# Prefer the runtime port override, then config, then the default.
+CONFIG_PORT=$(grep -A5 '^server:' /app/config/default.yaml 2>/dev/null | grep 'port:' | head -1 | awk '{print $2}')
+HEALTHCHECK_PORT=${PORT:-${CONFIG_PORT:-8080}}
+# Local health checks must never inherit an upstream HTTP proxy.
+curl --noproxy '*' --fail --silent --show-error --max-time 3 \
+  "http://127.0.0.1:${HEALTHCHECK_PORT}/health" >/dev/null || exit 1

src/auth/session-affinity.ts

@@ -9,10 +9,24 @@
 
 import { createHash } from "crypto";
 
+export interface ChainAdvanceTicket {
+  conversationId: string;
+  variantHash?: string;
+  generation: number;
+  expectedParentResponseId: string | null;
+}
+
+interface ChainHead {
+  conversationId: string;
+  variantHash?: string;
+  responseId: string | null;
+  generation: number;
+  updatedAt: number;
+}
+
 interface AffinityEntry {
   entryId: string;
   conversationId: string;
-  turnState?: string;
   /** SHA-256 hex of the instructions string. Stored as hash to bound memory usage. */
   instructionsHash?: string;
   inputTokens?: number;
@@ -31,6 +45,7 @@ const CLEANUP_INTERVAL_MS = 10 * 60 * 1000; // 10 minutes
 
 export class SessionAffinityMap {
   private map = new Map<string, AffinityEntry>();
+  private chainHeads = new Map<string, ChainHead>();
   private ttlMs: number;
   private cleanupTimer: ReturnType<typeof setInterval> | null = null;
 
@@ -39,7 +54,25 @@ export class SessionAffinityMap {
     this.cleanupTimer = setInterval(() => this.cleanup(), CLEANUP_INTERVAL_MS);
   }
 
-  /** Record that a response was created by a specific account in a conversation. */
+  /** Capture the current chain generation before dispatching an upstream request. */
+  captureChainAdvance(
+    conversationId: string,
+    variantHash?: string,
+    expectedParentResponseId?: string | null,
+  ): ChainAdvanceTicket {
+    const head = this.getChainHead(conversationId, variantHash);
+    return {
+      conversationId,
+      variantHash,
+      generation: head?.generation ?? 0,
+      expectedParentResponseId:
+        expectedParentResponseId === undefined
+          ? head?.responseId ?? null
+          : expectedParentResponseId,
+    };
+  }
+
+  /** Record response metadata and conditionally advance the implicit chain head. */
   record(
     responseId: string,
     entryId: string,
@@ -49,19 +82,45 @@ export class SessionAffinityMap {
     inputTokens?: number,
     functionCallIds?: string[],
     variantHash?: string,
-  ): void {
+    chainTicket?: ChainAdvanceTicket,
+  ): boolean {
+    void turnState;
+    const now = Date.now();
     this.map.set(responseId, {
       entryId,
       conversationId,
-      turnState,
       instructionsHash: instructions !== undefined
         ? createHash("sha256").update(instructions ?? "").digest("hex")
         : undefined,
       inputTokens,
       functionCallIds: functionCallIds ? [...functionCallIds] : undefined,
       variantHash,
-      createdAt: Date.now(),
+      createdAt: now,
+    });
+
+    const key = this.chainKey(conversationId, variantHash);
+    const current = this.getChainHead(conversationId, variantHash);
+    if (current?.responseId === responseId) return true;
+
+    if (chainTicket) {
+      if (
+        chainTicket.conversationId !== conversationId ||
+        chainTicket.variantHash !== variantHash ||
+        (current?.generation ?? 0) !== chainTicket.generation ||
+        (current?.responseId ?? null) !== chainTicket.expectedParentResponseId
+      ) {
+        return false;
+      }
+    }
+
+    this.chainHeads.set(key, {
+      conversationId,
+      variantHash,
+      responseId,
+      generation: (current?.generation ?? 0) + 1,
+      updatedAt: now,
     });
+    return true;
   }
 
   /** Look up which account created a given response. */
@@ -88,6 +147,20 @@ export class SessionAffinityMap {
     variantHash?: string,
   ): string | null {
     const now = Date.now();
+    if (variantHash !== undefined) {
+      const head = this.getChainHead(conversationId, variantHash);
+      if (head) {
+        if (!head.responseId) return null;
+        const entry = this.getEntry(head.responseId);
+        if (!entry) {
+          this.invalidateHead(head);
+          return null;
+        }
+        if (maxAgeMs !== undefined && now - entry.createdAt > maxAgeMs) return null;
+        return head.responseId;
+      }
+    }
+
     let latestResponseId: string | null = null;
     let latestCreatedAt = -1;
     for (const [responseId, entry] of this.map) {
@@ -104,10 +177,12 @@ export class SessionAffinityMap {
     return latestResponseId;
   }
 
-  /** Look up the upstream turn-state token for a given response. */
-  lookupTurnState(responseId: string): string | null {
-    const entry = this.getEntry(responseId);
-    return entry?.turnState ?? null;
+  /**
+   * turnState is scoped to a single Codex turn and must never be restored from
+   * cross-turn affinity. Kept for compatibility with older callers/tests.
+   */
+  lookupTurnState(_responseId: string): null {
+    return null;
   }
 
   lookupInstructionsHash(responseId: string): string | null {
@@ -134,6 +209,9 @@ export class SessionAffinityMap {
   /** Drop a response ID — called after upstream rejects it as not-found. */
   forget(responseId: string): void {
     this.map.delete(responseId);
+    for (const head of this.chainHeads.values()) {
+      if (head.responseId === responseId) this.invalidateHead(head);
+    }
   }
 
   /** Drop every response recorded for a conversation, optionally scoped to a
@@ -150,9 +228,29 @@ export class SessionAffinityMap {
       this.map.delete(responseId);
       dropped++;
     }
+    for (const head of this.chainHeads.values()) {
+      if (head.conversationId !== conversationId) continue;
+      if (variantHash !== undefined && head.variantHash !== variantHash) continue;
+      this.invalidateHead(head);
+    }
     return dropped;
   }
 
+
+  private chainKey(conversationId: string, variantHash?: string): string {
+    return JSON.stringify([conversationId, variantHash ?? null]);
+  }
+
+  private getChainHead(conversationId: string, variantHash?: string): ChainHead | null {
+    return this.chainHeads.get(this.chainKey(conversationId, variantHash)) ?? null;
+  }
+
+  private invalidateHead(head: ChainHead): void {
+    head.responseId = null;
+    head.generation += 1;
+    head.updatedAt = Date.now();
+  }
+
   private getEntry(responseId: string): AffinityEntry | null {
     const entry = this.map.get(responseId);
     if (!entry) return null;
@@ -171,6 +269,11 @@ export class SessionAffinityMap {
         this.map.delete(key);
       }
     }
+    for (const [key, head] of this.chainHeads) {
+      if (now - head.updatedAt > this.ttlMs) {
+        this.chainHeads.delete(key);
+      }
+    }
   }
 
   get size(): number {
@@ -183,6 +286,7 @@ export class SessionAffinityMap {
       this.cleanupTimer = null;
     }
     this.map.clear();
+    this.chainHeads.clear();
   }
 }
 

src/proxy/codex-types.ts

@@ -216,9 +216,23 @@ export class CodexApiError extends Error {
   }
 }
 
-/** previous_response_id 只能通过 WebSocket 安全续链，失败后不能降级为 HTTP delta-only。 */
+export type PreviousResponseContinuityReason =
+  | "busy"
+  | "dead"
+  | "expired"
+  | "missing_owner"
+  | "account_mismatch"
+  | "disabled"
+  | "no_key"
+  | "no_context"
+  | "transport";
+
+/** previous_response_id can only continue on its owning physical WebSocket. */
 export class PreviousResponseWebSocketError extends CodexApiError {
-  constructor(public readonly causeMessage: string) {
+  constructor(
+    public readonly causeMessage: string,
+    public readonly continuityReason?: PreviousResponseContinuityReason,
+  ) {
     super(
       0,
       JSON.stringify({