fix(proxy): reconcile durable full resend owner

app/modules/proxy/_service/http_bridge/streaming.py

@@ -220,6 +220,155 @@
 _RESPONSE_CREATE_GATE_RETRY_SLEEP_SECONDS = 10.0
 
 
+class _VerifiedDurableFullResend:
+    """Immutable proof that one payload contains a durable turn's complete context."""
+
+    _durable_session_id: str
+    _full_input_fingerprint: str
+    _latest_response_id: str
+    _owner_account_id: str
+    _pending_tool_calls: tuple[tuple[str, str], ...] | None
+    _stored_input_fingerprint: str
+    _stored_input_item_count: int
+    __slots__ = (
+        "_durable_session_id",
+        "_full_input_fingerprint",
+        "_latest_response_id",
+        "_owner_account_id",
+        "_pending_tool_calls",
+        "_stored_input_fingerprint",
+        "_stored_input_item_count",
+    )
+    __construction_token = object()
+
+    def __init__(
+        self,
+        *,
+        _token: object,
+        durable_session_id: str,
+        owner_account_id: str,
+        latest_response_id: str,
+        stored_input_item_count: int,
+        stored_input_fingerprint: str,
+        full_input_fingerprint: str,
+        pending_tool_calls: tuple[tuple[str, str], ...] | None,
+    ) -> None:
+        if _token is not self.__construction_token:
+            raise TypeError("verified durable full resend proofs are created only by the verifier")
+        object.__setattr__(self, "_durable_session_id", durable_session_id)
+        object.__setattr__(self, "_owner_account_id", owner_account_id)
+        object.__setattr__(self, "_latest_response_id", latest_response_id)
+        object.__setattr__(self, "_stored_input_item_count", stored_input_item_count)
+        object.__setattr__(self, "_stored_input_fingerprint", stored_input_fingerprint)
+        object.__setattr__(self, "_full_input_fingerprint", full_input_fingerprint)
+        object.__setattr__(self, "_pending_tool_calls", pending_tool_calls)
+
+    def __setattr__(self, _name: str, _value: object) -> None:
+        raise AttributeError("verified durable full resend proofs are immutable")
+
+    def __copy__(self) -> "_VerifiedDurableFullResend":
+        return self
+
+    def __deepcopy__(self, _memo: dict[int, object]) -> "_VerifiedDurableFullResend":
+        return self
+
+    def __reduce_ex__(self, _protocol: object) -> str | tuple[Any, ...]:
+        raise TypeError("verified durable full resend proofs cannot be serialized")
+
+    @property
+    def stored_input_item_count(self) -> int:
+        return self._stored_input_item_count
+
+    def matches(
+        self,
+        payload: ResponsesRequest,
+        durable_lookup: DurableBridgeLookup | None,
+    ) -> bool:
+        input_items = payload.input
+        return (
+            isinstance(input_items, list)
+            and durable_lookup is not None
+            and durable_lookup.session_id == self._durable_session_id
+            and durable_lookup.account_id == self._owner_account_id
+            and durable_lookup.latest_response_id == self._latest_response_id
+            and durable_lookup.latest_input_item_count == self._stored_input_item_count
+            and durable_lookup.latest_input_full_fingerprint == self._stored_input_fingerprint
+            and _pending_tool_calls_identity(durable_lookup.latest_pending_tool_calls) == self._pending_tool_calls
+            and _fingerprint_input_items(cast(list[JsonValue], input_items)) == self._full_input_fingerprint
+        )
+
+    @classmethod
+    def _verify(
+        cls,
+        payload: ResponsesRequest,
+        durable_lookup: DurableBridgeLookup,
+    ) -> "_VerifiedDurableFullResend | None":
+        owner_account_id = durable_lookup.account_id
+        latest_response_id = durable_lookup.latest_response_id
+        stored_count = durable_lookup.latest_input_item_count
+        stored_fingerprint = durable_lookup.latest_input_full_fingerprint
+        if (
+            owner_account_id is None
+            or latest_response_id is None
+            or stored_count is None
+            or stored_fingerprint is None
+            or not _http_bridge_payload_looks_like_full_resend(payload)
+            or not isinstance(payload.input, list)
+            or not _input_prefix_matches_stored_context(
+                payload.input,
+                stored_count=stored_count,
+                stored_fingerprint=stored_fingerprint,
+            )
+        ):
+            return None
+        input_items = cast(list[JsonValue], payload.input)
+        replay_projection = project_responses_input_for_account_neutral_fresh_replay(
+            input_items,
+            stored_count=stored_count,
+        )
+        pending_tool_calls = durable_lookup.latest_pending_tool_calls
+        if replay_projection is None:
+            return None
+        safe_fresh_context = responses_input_suffix_retains_prior_output(
+            replay_projection.input_items,
+            stored_count=replay_projection.stored_prefix_count,
+        ) or (
+            pending_tool_calls is not None
+            and responses_input_suffix_matches_pending_tool_calls(
+                replay_projection.input_items,
+                stored_count=replay_projection.stored_prefix_count,
+                pending_tool_calls=pending_tool_calls,
+            )
+        )
+        if not safe_fresh_context:
+            return None
+        return cls(
+            _token=cls.__construction_token,
+            durable_session_id=durable_lookup.session_id,
+            owner_account_id=owner_account_id,
+            latest_response_id=latest_response_id,
+            stored_input_item_count=stored_count,
+            stored_input_fingerprint=stored_fingerprint,
+            full_input_fingerprint=_fingerprint_input_items(input_items),
+            pending_tool_calls=_pending_tool_calls_identity(pending_tool_calls),
+        )
+
+
+def _pending_tool_calls_identity(
+    pending_tool_calls: Mapping[str, str] | None,
+) -> tuple[tuple[str, str], ...] | None:
+    return None if pending_tool_calls is None else tuple(sorted(pending_tool_calls.items()))
+
+
+def _verify_durable_full_resend(
+    payload: ResponsesRequest,
+    durable_lookup: DurableBridgeLookup | None,
+) -> _VerifiedDurableFullResend | None:
+    if durable_lookup is None or durable_lookup.account_id is None or durable_lookup.latest_response_id is None:
+        return None
+    return _VerifiedDurableFullResend._verify(payload, durable_lookup)
+
+
 def _http_bridge_payload_is_account_neutral_fresh_replay(payload: ResponsesRequest) -> bool:
     return responses_payload_is_account_neutral_fresh_replay(payload.to_payload())
 
@@ -907,6 +1056,8 @@ def prepare_bridge_request(
         durable_full_resend_is_account_neutral: bool | None = None
         durable_full_resend_has_safe_fresh_context = False
         durable_full_resend_retains_prior_output = False
+        durable_full_resend_proof = _verify_durable_full_resend(payload, durable_lookup)
+        durable_full_resend_fresh_bridge_proof: _VerifiedDurableFullResend | None = None
         force_local_recovery_creation = False
         payload_looks_like_full_resend = _http_bridge_payload_looks_like_full_resend(payload)
 
@@ -1020,17 +1171,36 @@ def classify_durable_full_resend(
                 and payload_looks_like_full_resend
                 and durable_full_resend_has_safe_fresh_context
             ):
-                # The client already supplied a complete fresh request. Adding
-                # a durable anchor here can strand it on the new WebSocket.
-                _log_http_bridge_event(
-                    "fresh_reattach_full_resend_preserved",
-                    bridge_session_key,
-                    account_id=durable_lookup.account_id,
-                    model=payload.model,
-                    detail="outcome=client_unanchored_full_resend",
-                    cache_key_family=bridge_session_key.affinity_kind,
-                    model_class=_extract_model_class(payload.model) if payload.model else None,
-                )
+                if durable_full_resend_proof is not None and durable_full_resend_proof.matches(payload, durable_lookup):
+                    durable_full_resend_fresh_bridge_proof = durable_full_resend_proof
+                    # The client already supplied a proved complete fresh
+                    # request. Adding a durable anchor here can strand it on
+                    # the new WebSocket.
+                    _log_http_bridge_event(
+                        "fresh_reattach_full_resend_preserved",
+                        bridge_session_key,
+                        account_id=durable_lookup.account_id,
+                        model=payload.model,
+                        detail="outcome=client_unanchored_full_resend",
+                        cache_key_family=bridge_session_key.affinity_kind,
+                        model_class=_extract_model_class(payload.model) if payload.model else None,
+                    )
+                else:
+                    effective_payload = payload.model_copy(
+                        update={"previous_response_id": durable_lookup.latest_response_id}
+                    )
+                    proxy_injected_previous_response_id = True
+                    _fresh_request_state, fresh_upstream_request_text = prepare_bridge_request(payload)
+                    del _fresh_request_state
+                    _log_http_bridge_event(
+                        "fresh_reattach_anchor_injected",
+                        bridge_session_key,
+                        account_id=None,
+                        model=payload.model,
+                        detail=f"response_id={durable_lookup.latest_response_id}",
+                        cache_key_family=bridge_session_key.affinity_kind,
+                        model_class=_extract_model_class(payload.model) if payload.model else None,
+                    )
             elif fresh_reattach_can_use_durable_anchor:
                 effective_payload = payload.model_copy(
                     update={"previous_response_id": durable_lookup.latest_response_id}
@@ -1055,6 +1225,30 @@ def classify_durable_full_resend(
             affinity = _AffinityPolicy()
             incoming_turn_state_header = None
             session_header_fallback_key = None
+        owner_bound_full_resend_ignores_broad_session = (
+            not forwarded_request
+            and durable_full_resend_fresh_bridge_proof is not None
+            and durable_full_resend_fresh_bridge_proof.matches(payload, durable_lookup)
+            and affinity.codex_session_source == "session_header"
+        )
+        if owner_bound_full_resend_ignores_broad_session:
+            # The durable owner remains required through request_state below.
+            # Remove only the broad client alias that can resolve a stale raw
+            # compatibility row; keep CODEX_SESSION semantics so the new
+            # bridge can anchor later incremental turns to its fresh response.
+            affinity = _AffinityPolicy(kind=StickySessionKind.CODEX_SESSION)
+            incoming_session_header = None
+            session_header_fallback_key = None
+            _log_http_bridge_event(
+                "fresh_reattach_broad_session_owner_ignored",
+                bridge_session_key,
+                account_id=durable_lookup.account_id if durable_lookup is not None else None,
+                model=payload.model,
+                detail=(f"stored_items={durable_full_resend_fresh_bridge_proof.stored_input_item_count}"),
+                cache_key_family=bridge_session_key.affinity_kind,
+                model_class=_extract_model_class(payload.model) if payload.model else None,
+                owner_check_applied=True,
+            )
         if effective_payload.previous_response_id is not None and isinstance(effective_payload.input, list):
             previous_response_input_items = cast(list[JsonValue], effective_payload.input)
             trimmed_input_items = _trim_http_bridge_previous_response_input_items(previous_response_input_items)
@@ -1179,7 +1373,9 @@ def classify_durable_full_resend(
         settings = _service_get_settings()
         request_deadline = request_state.started_at + _http_bridge_request_budget_seconds(settings)
         session_creation_headers = (
-            without_http_bridge_session_affinity_headers(headers) if account_neutral_recovery else dict(headers)
+            without_http_bridge_session_affinity_headers(headers)
+            if account_neutral_recovery or owner_bound_full_resend_ignores_broad_session
+            else dict(headers)
         )
         fresh_replay_excluded_account_ids: set[str] = set()
         unanchored_fork_spill_attempted = False

openspec/changes/reconcile-durable-full-resend-owner/.openspec.yaml

@@ -0,0 +1,2 @@
+schema: spec-driven
+created: 2026-07-25

openspec/changes/reconcile-durable-full-resend-owner/design.md

@@ -0,0 +1,98 @@
+## Context
+
+PR #1486 makes a fingerprint-verified complete resend, including an exact
+response-bound pending-tool-call settlement, start a fresh upstream bridge
+without the stale durable `previous_response_id`. Account selection still
+receives the original session-header affinity, though. During rolling upgrades,
+a raw legacy `CODEX_SESSION` row may represent older hard turn-state ownership
+on account A while the more specific durable bridge row owns this task on
+account B. The load balancer correctly refuses to choose between those sources,
+but the resulting retryable error cannot converge because neither persisted row
+changes.
+
+## Goals / Non-Goals
+
+**Goals:**
+
+- Let a verified complete resend establish a fresh bridge on its durable owner.
+- Remove only the broad session-header source that causes the deterministic
+  selection conflict.
+- Keep subsequent incremental continuity on the newly established bridge.
+- Make the eligibility proof immutable, request-bound, and unavailable through
+  ordinary construction or hydration.
+
+**Non-Goals:**
+
+- Prefer one conflicting turn-state, previous-response, file, or other specific
+  owner over another.
+- Add or widen an account-movement path. Existing separately proved
+  account-neutral replay after a genuine owner-unavailable result remains
+  unchanged.
+- Delete or rebind the broad legacy row, which may still own sibling work.
+- Retry anything after upstream dispatch may have started.
+
+## Decisions
+
+### 1. Bind the bypass to a sealed local proof
+
+The verifier requires a durable owner, latest response ID, positive stored input
+count, full fingerprint, exact raw-prefix match, and either retained prior
+assistant output followed by fresh input or an exact call/output settlement of
+the response-bound pending-tool-call manifest. The proof records the durable
+session, owner, response, stored metadata, pending-tool-call manifest identity,
+and full input fingerprint. Its normal constructor is sealed inside the
+verifier closure, its fields are immutable, and serialization is rejected.
+Before use, it is matched again against the current payload and durable lookup
+so mutation or state substitution invalidates the proof.
+
+The proof is request-local. It is never accepted from a caller, serialized,
+persisted, cloned into another request, or hydrated from the database.
+
+### 2. Remove only broad legacy selection provenance
+
+When the proved full resend is about to create a fresh owner-bound bridge from a
+session header, the service removes downstream session and turn aliases from
+the new upstream connection and replaces selection affinity with a
+`CODEX_SESSION` policy that has no client key or legacy source. The durable
+canonical bridge key and durable owner account remain unchanged, so selection
+cannot move to another account.
+
+The broad sticky row is left intact for sibling traffic. Specific durable
+turn-state, previous-response, and file-owner checks occur before this step and
+remain hard conflicts.
+
+This reconciliation does not itself rebind the request to another account.
+Existing account-neutral full-resend recovery after a genuine
+owner-unavailable result remains separately gated by its own projection and
+replay-safety checks.
+
+### 3. Preserve Codex bridge semantics after creation
+
+The selection policy retains `CODEX_SESSION` kind even though it drops the
+stale client key. The created session therefore remains a Codex continuity
+session and can inject the response ID established by the successful fresh
+request for later incremental turns.
+
+## Risks / Trade-offs
+
+- The new upstream connection no longer receives the stale session header on
+  this one recovery path. The durable canonical key still owns internal routing,
+  and the complete request supplies the upstream context.
+- Python cannot prevent hostile reflection through `object.__new__`, but
+  ordinary construction, mutation, copying with altered fields, and
+  serialization are closed; every use also revalidates the payload and durable
+  identity.
+- An incomplete resend may still surface a continuity conflict. It remains
+  fail-closed because dropping either owner would risk context or account-bound
+  state.
+
+## Example
+
+Durable session `S` records owner B, response `resp_old`, two stored input items,
+their fingerprint, and any pending tool calls bound to that response. A raw
+legacy sticky row for the shared session header still points at A. The client
+resends the two stored items plus either retained completed assistant output and
+a new user message or the exact pending call/output settlement. codex-lb proves
+the complete resend, opens the fresh bridge on B without `resp_old`, and omits
+the stale broad header from selection and the upstream handshake. The raw row
+remains on A for unrelated sibling traffic.

openspec/changes/reconcile-durable-full-resend-owner/proposal.md

@@ -0,0 +1,41 @@
+## Why
+
+The fresh durable full-resend path can still fail before upstream dispatch when
+a broad legacy session-header sticky row points at a different account than the
+durable bridge owner. codex-lb returns `continuity_owner_conflict` as a
+retryable 503, while the client repeats the same request and the two persisted
+owners remain unchanged. This creates a deterministic retry loop even though
+the request already contains fingerprint-verified complete context and can
+safely start a fresh bridge on the durable owner.
+
+## What Changes
+
+- Represent complete durable full-resend eligibility with an immutable,
+  request-bound internal proof created only by the count, fingerprint, and
+  retained-output or response-bound pending-tool-call checks.
+- For that proved fresh reattach only, stop consulting and forwarding the broad
+  legacy session-header alias while retaining the durable canonical key and
+  owner account as hard constraints.
+- Preserve normal Codex session behavior on the replacement bridge so later
+  incremental turns can use its newly established response anchor.
+- Keep incomplete resends, conflicting specific aliases, and file-owner
+  conflicts fail-closed; do not add or widen an account-movement path.
+
+## Capabilities
+
+### New Capabilities
+
+None.
+
+### Modified Capabilities
+
+- `responses-api-compat`: Reconcile a stale broad session mapping during a
+  verified owner-bound fresh reattach without moving accounts.
+
+## Impact
+
+- Affected code: HTTP bridge durable full-resend verification and fresh session
+  affinity preparation.
+- Affected surface: hard Codex session reattach after the live upstream bridge
+  is gone and a legacy raw session row disagrees with the durable owner.
+- No new cross-account replay, schema, setting, dependency, or post-send retry.