fix(http-bridge): preserve goal-restart recovery across reconnects

app/core/openai/requests.py

@@ -1642,6 +1642,36 @@ def _compact_item_texts(item: Mapping[str, JsonValue]) -> list[str]:
     return texts
 
 
+def responses_input_contains_goal_continuation_context(input_value: JsonValue) -> bool:
+    """Return whether Responses input carries Codex's goal-continuation marker."""
+
+    if not is_json_list(input_value):
+        return False
+    for item in input_value:
+        if not is_json_mapping(item):
+            continue
+        for text in _compact_item_texts(item):
+            if text.lstrip().startswith(_GOAL_CONTINUATION_CONTEXT_PREFIX):
+                return True
+    return False
+
+
+def responses_request_contains_goal_continuation_context(payload: ResponsesRequest) -> bool:
+    """Return whether a normalized request carries Codex's goal restart marker."""
+
+    # ResponsesRequest normalization lifts developer/system input messages into
+    # ``instructions``. The marker can therefore disappear from ``input`` and
+    # follow pre-existing instruction text by the time affinity is classified.
+    # Keep both locations in this check or a harmless parser refactor can
+    # silently break restart recovery while marker-preservation tests still pass.
+    instructions = payload.instructions
+    if isinstance(instructions, str) and any(
+        line.lstrip().startswith(_GOAL_CONTINUATION_CONTEXT_PREFIX) for line in instructions.splitlines()
+    ):
+        return True
+    return responses_input_contains_goal_continuation_context(payload.input)
+
+
 def _compact_trimmed_input_with_markers(
     input_value: list[JsonValue], token_counts: list[int], selected_indices: set[int]
 ) -> list[JsonValue]:

app/db/alembic/versions/20260812_120000_add_sticky_abandonment_scope.py

@@ -0,0 +1,43 @@
+"""add source scope to sticky continuity abandonment
+
+Revision ID: 20260812_120000_add_sticky_abandonment_scope
+Revises: 20260812_000000_merge_recovery_dispatch_and_hourly_cancelled_heads
+Create Date: 2026-08-12 12:00:00.000000
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.engine import Connection
+
+revision = "20260812_120000_add_sticky_abandonment_scope"
+down_revision = "20260812_000000_merge_recovery_dispatch_and_hourly_cancelled_heads"
+branch_labels = None
+depends_on = None
+
+_TABLE = "sticky_sessions"
+_COLUMN = "continuity_abandonment_scope"
+
+
+def _columns(connection: Connection) -> set[str]:
+    inspector = sa.inspect(connection)
+    if not inspector.has_table(_TABLE):
+        return set()
+    return {str(column["name"]) for column in inspector.get_columns(_TABLE) if column.get("name") is not None}
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    if _COLUMN in _columns(bind):
+        return
+    with op.batch_alter_table(_TABLE) as batch_op:
+        batch_op.add_column(sa.Column(_COLUMN, sa.String(length=32), nullable=True))
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    if _COLUMN not in _columns(bind):
+        return
+    with op.batch_alter_table(_TABLE) as batch_op:
+        batch_op.drop_column(_COLUMN)

app/db/models.py

@@ -777,17 +777,15 @@ class StickySession(Base):
         onupdate=func.now(),
         nullable=False,
     )
-    # Set only by purge_stale_hard_codex_session_mappings's first pass. A hard
-    # codex_session row normally proves ownership for `conversation`-continuity
-    # requests (see affinity.py's require_unambiguous_account), which have no
-    # other owner index. Once the durably-unavailable owner's proof is this
-    # stale, we stop treating the row as a live pin (so a fresh account can be
-    # selected) but keep it around with this marker set instead of deleting it
-    # outright, so selection can tell "this key was deliberately abandoned,
-    # picking a new owner is authorized" apart from "this key was never seen,
-    # ambiguity must fail closed." The row is only ever hard-deleted once it
-    # has sat abandoned past a further grace window with nobody claiming it.
+    # A non-null timestamp marks continuity abandoned either globally (scope
+    # is NULL, as written by stale-hard cleanup) or only for one typed source.
+    # Source-scoped abandonment is what lets a process-session restart stop
+    # consulting an ambiguous raw compatibility row without erasing that
+    # row's retained account_id for an explicit turn-state lookup that happens
+    # to use the same client-controlled text. The row is only hard-deleted by
+    # global stale-hard cleanup after its additional grace window.
     continuity_abandoned_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True, default=None)
+    continuity_abandonment_scope: Mapped[str | None] = mapped_column(String(32), nullable=True, default=None)
 
 
 class CapabilityLineageMarker(Base):

app/modules/proxy/_load_balancer/sticky_selection.py

@@ -77,6 +77,9 @@ class SelectionInputsProtocol(Protocol):
     @property
     def effective_continuity_owner_candidates(self) -> list[Account]: ...
 
+    @property
+    def effective_sticky_mutation_authority_account_ids(self) -> frozenset[str]: ...
+
 
 SelectionInputsT = TypeVar("SelectionInputsT", bound=SelectionInputsProtocol)
 
@@ -204,6 +207,7 @@ class StickySelectionRequest(Generic[SelectionInputsT]):
     legacy_sticky_key: str | None
     legacy_existing_account_id: str | None
     spill_bare_session_on_account_cap: bool
+    abandon_unavailable_legacy_owner: bool
     require_unambiguous_account: bool
     sticky_max_age_seconds: int | None
     prefer_earlier_reset_accounts: bool
@@ -266,6 +270,7 @@ async def run_sticky_selection_path(
     legacy_sticky_key = request.legacy_sticky_key
     legacy_existing_account_id = request.legacy_existing_account_id
     spill_bare_session_on_account_cap = request.spill_bare_session_on_account_cap
+    abandon_unavailable_legacy_owner = request.abandon_unavailable_legacy_owner
     require_unambiguous_account = request.require_unambiguous_account
     sticky_max_age_seconds = request.sticky_max_age_seconds
     prefer_earlier_reset_accounts = request.prefer_earlier_reset_accounts
@@ -313,6 +318,7 @@ def _direct_error(
 
     sticky_existing_account_id: str | None | object = _STICKY_EXISTING_UNSET
     sticky_continuity_abandoned = False
+    retired_legacy_owner_account_ids: set[str] = set()
     attempt = 0
     suppress_recovery_probe_candidates = False
     while True:
@@ -326,6 +332,7 @@ def _direct_error(
                     sticky_key,
                     kind=sticky_kind,
                     max_age_seconds=sticky_max_age_seconds,
+                    continuity_source=sticky_source,
                 )
                 sticky_existing_account_id = sticky_owner_lookup.account_id
                 # `is True` (not a truthy check): an un-configured test double
@@ -348,6 +355,17 @@ def _direct_error(
                 required_account_id=required_account_id,
                 redact_sensitive_details=redact_sensitive_details,
             )
+            if retired_legacy_owner_account_ids:
+                # Retirement is authoritative even when this selector loaded a
+                # pre-retirement account snapshot (or another replica still has
+                # one cached). Never let that stale snapshot immediately repin
+                # the account this request just proved durably unavailable.
+                states = [state for state in states if state.account_id not in retired_legacy_owner_account_ids]
+                account_map = {
+                    account_id: account
+                    for account_id, account in account_map.items()
+                    if account_id not in retired_legacy_owner_account_ids
+                }
             effective_routing_costs = (
                 routing_costs_by_account_id
                 if routing_costs_by_account_id is not None
@@ -462,6 +480,63 @@ def _direct_error(
                     traffic_class=traffic_class,
                 )
             probe_reservation: ProbeReservation | None = None
+        # Raw sticky rows are global, while account-assigned API keys and
+        # other authenticated policies narrow a request's mutation authority.
+        # Keep this check on the pre-health continuity pool: quota exhaustion
+        # may authorize retirement, but being outside policy scope never does.
+        legacy_owner_in_effective_policy_scope = (
+            isinstance(sticky_existing_account_id, str)
+            and sticky_existing_account_id in selection_inputs.effective_sticky_mutation_authority_account_ids
+        )
+        if (
+            abandon_unavailable_legacy_owner
+            and hard_sticky
+            and sticky_existing_is_legacy
+            and sticky_source == "session_header"
+            and legacy_sticky_key is not None
+            and isinstance(sticky_existing_account_id, str)
+            and legacy_owner_in_effective_policy_scope
+        ):
+            async with owner._repo_factory() as repos:
+                owner_retired = await repos.sticky_sessions.abandon_legacy_session_header_owner_if_unavailable(
+                    legacy_sticky_key,
+                    kind=StickySessionKind.CODEX_SESSION,
+                    expected_account_id=sticky_existing_account_id,
+                )
+                authoritative_legacy_owner = None
+                if not owner_retired:
+                    authoritative_legacy_owner = await repos.sticky_sessions.get_account_id_and_abandonment(
+                        legacy_sticky_key,
+                        kind=StickySessionKind.CODEX_SESSION,
+                        continuity_source="session_header",
+                    )
+            # One guarded write is authoritative for this selection. Repeating
+            # it in capacity-wait retries would add write pressure and could
+            # reinterpret a later status transition as restart authorization.
+            abandon_unavailable_legacy_owner = False
+            if owner_retired:
+                # The raw compatibility row is now a tombstone. Drop only the
+                # selection loop's cached legacy owner and run the normal path
+                # again so namespaced affinity, leases, and admission checks
+                # are established through the existing selection path.
+                logger.info(
+                    "Legacy Codex session-header owner abandoned for self-contained goal restart account_id=%s",
+                    "<redacted>" if redact_sensitive_details else sticky_existing_account_id,
+                )
+                retired_legacy_owner_account_ids.add(sticky_existing_account_id)
+                legacy_existing_account_id = None
+                continue
+            # A failed compare-and-set means the cached owner is no longer
+            # authoritative: it may have recovered, another request may have
+            # rebound the raw row, or another worker may already have
+            # tombstoned it. Re-read under a fresh transaction and restart the
+            # loop so each outcome is handled by normal selection. Retaining
+            # the stale owner here would defeat the CAS and can fail a restart
+            # even though a concurrent operation already established a valid
+            # replacement.
+            assert authoritative_legacy_owner is not None
+            legacy_existing_account_id = authoritative_legacy_owner.account_id
+            continue
         sticky_outcome = _StickySelectionOutcome(selection=SelectionResult(None, None))
         if fair_share_denial is not None:
             # Denial parks in the transport capacity-wait loop like a cap

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

@@ -8,6 +8,7 @@
 class _HTTPBridgeAccountSessionsMixin:
     async def close_http_bridge_sessions_for_account(self: _HTTPBridgeServiceProtocol, account_id: str) -> int:
         sessions_to_close: list[_HTTPBridgeSession] = []
+        scheduled_session_ids: set[int] = set()
         async with self._http_bridge_lock:
             for key, session in tuple(self._http_bridge_sessions.items()):
                 if session.account.id != account_id:
@@ -24,6 +25,32 @@ async def close_http_bridge_sessions_for_account(self: _HTTPBridgeServiceProtoco
                     model_class=_extract_model_class(session.request_model) if session.request_model else None,
                 )
                 sessions_to_close.append(detached)
+                scheduled_session_ids.add(id(detached))
+            # Detached predecessors still own authenticated sockets and account
+            # leases. Account invalidation must fence them even though a newer
+            # generation occupies (or has vacated) their canonical key.
+            for session in tuple(self._http_bridge_detached_sessions.values()):
+                if session.account.id != account_id or id(session) in scheduled_session_ids:
+                    continue
+                close_task = session.resource_close_task
+                if close_task is not None and (
+                    not close_task.done() or (not close_task.cancelled() and close_task.exception() is None)
+                ):
+                    # ``closed`` only rejects admission. A live close task (or a
+                    # successfully completed one awaiting registry finalization)
+                    # is the proof that this detached generation is already owned.
+                    continue
+                session.closed = True
+                _log_http_bridge_event(
+                    "evict_account_binding_changed",
+                    session.key,
+                    account_id=session.account.id,
+                    model=session.request_model,
+                    cache_key_family=session.key.affinity_kind,
+                    model_class=_extract_model_class(session.request_model) if session.request_model else None,
+                )
+                sessions_to_close.append(session)
+                scheduled_session_ids.add(id(session))
 
         for session in sessions_to_close:
             await self._close_http_bridge_session_bounded(session, reason="account_binding_changed")

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

@@ -1,9 +1,13 @@
 from __future__ import annotations
 
+import asyncio
 from typing import Any
 
+from app.core.clients.proxy import ProxyResponseError
+from app.core.resilience.overload import local_overload_error
 from app.modules.proxy._service.http_bridge.helpers import (
     _close_http_bridge_session_bounded,
+    _http_bridge_capacity_generation_count,
     _http_bridge_pending_count_nowait,
     _http_bridge_pending_state_is_stale,
     _http_bridge_request_counts_against_queue,
@@ -13,7 +17,11 @@
     http_bridge_activity_snapshot_nowait,
 )
 from app.modules.proxy._service.http_bridge.protocol import _HTTPBridgeServiceProtocol
-from app.modules.proxy._service.support import _http_bridge_session_supports_service_tier, _HTTPBridgeSession
+from app.modules.proxy._service.support import (
+    _http_bridge_session_supports_service_tier,
+    _HTTPBridgeSession,
+    _HTTPBridgeSessionKey,
+)
 from app.modules.proxy.affinity import _extract_model_class
 
 
@@ -77,6 +85,66 @@ async def _close_http_bridge_session_bounded(
     ) -> None:
         await _close_http_bridge_session_bounded(self, session, reason=reason)
 
+    def _http_bridge_active_capacity_error(
+        self: _HTTPBridgeServiceProtocol,
+        *,
+        key: _HTTPBridgeSessionKey,
+        request_model: str | None,
+    ) -> ProxyResponseError:
+        _log_http_bridge_event(
+            "capacity_exhausted_active_sessions",
+            key,
+            account_id=None,
+            model=request_model,
+            pending_count=_http_bridge_capacity_generation_count(self),
+            cache_key_family=key.affinity_kind,
+            model_class=_extract_model_class(request_model) if request_model else None,
+        )
+        return ProxyResponseError(
+            429,
+            local_overload_error(
+                "HTTP responses session bridge has no idle capacity",
+                code="capacity_exhausted_active_sessions",
+            ),
+        )
+
+    async def _enforce_http_bridge_capacity_after_planned_closes(
+        self: _HTTPBridgeServiceProtocol,
+        *,
+        key: _HTTPBridgeSessionKey,
+        inflight_future: asyncio.Future[_HTTPBridgeSession] | None,
+        max_sessions: int,
+        request_model: str | None,
+    ) -> None:
+        assert inflight_future is not None
+        async with self._http_bridge_lock:
+            if (
+                self._http_bridge_inflight_sessions.get(key) is not inflight_future
+                or _http_bridge_capacity_generation_count(self) <= max_sessions
+            ):
+                return
+            # Planned evictions are discounted only to reserve this creation
+            # slot. A bounded close may return on timeout while the detached
+            # socket and leases remain live, so registry ownership wins here.
+            _log_http_bridge_event(
+                "capacity_exhausted_after_lru_close",
+                key,
+                account_id=None,
+                model=request_model,
+                pending_count=_http_bridge_capacity_generation_count(self),
+                cache_key_family=key.affinity_kind,
+                model_class=_extract_model_class(request_model) if request_model else None,
+            )
+            capacity_error = ProxyResponseError(
+                429,
+                local_overload_error(
+                    "HTTP responses session bridge has no idle capacity",
+                    code="capacity_exhausted_active_sessions",
+                ),
+            )
+        await self._fail_http_bridge_inflight_session_creation(key, inflight_future, capacity_error)
+        raise capacity_error
+
     async def _http_bridge_pending_count(
         self: _HTTPBridgeServiceProtocol,
         session: _HTTPBridgeSession,