diff --git a/app/core/config/settings.py b/app/core/config/settings.py index 56934d95a..e38723400 100644 --- a/app/core/config/settings.py +++ b/app/core/config/settings.py @@ -297,6 +297,7 @@ class Settings(BaseSettings): http_responses_session_bridge_codex_idle_ttl_seconds: float = Field(default=900.0, gt=0) http_responses_session_bridge_codex_prewarm_enabled: bool = False http_responses_session_bridge_stuck_gate_retire_after_seconds: float = Field(default=300.0, gt=0) + http_responses_session_bridge_anchor_poison_failure_threshold: int = Field(default=7, ge=1, le=100) http_responses_session_bridge_max_sessions: int = Field(default=256, gt=0) http_responses_session_bridge_queue_limit: int = Field(default=8, gt=0) http_responses_session_bridge_clean_close_retry_jitter_max_seconds: float = Field( diff --git a/app/db/alembic/versions/20260806_120000_add_http_bridge_owner_process_epoch.py b/app/db/alembic/versions/20260806_120000_add_http_bridge_owner_process_epoch.py new file mode 100644 index 000000000..a6fa9e3fe --- /dev/null +++ b/app/db/alembic/versions/20260806_120000_add_http_bridge_owner_process_epoch.py @@ -0,0 +1,57 @@ +"""add owner process epoch to durable HTTP bridge sessions + +Revision ID: 20260806_120000_add_http_bridge_owner_process_epoch +Revises: 20260808_000000_tune_usage_history_autovacuum +Create Date: 2026-08-06 12:00:00.000000 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.engine import Connection + +revision = "20260806_120000_add_http_bridge_owner_process_epoch" +down_revision = "20260808_000000_tune_usage_history_autovacuum" +branch_labels = None +depends_on = None + +_TABLE = "http_bridge_sessions" +_COLUMN = "owner_process_epoch" + + +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=64), nullable=True)) + op.drop_index("idx_http_bridge_sessions_owner_state", table_name=_TABLE, if_exists=True) + op.create_index( + "idx_http_bridge_sessions_owner_state", + _TABLE, + ["owner_instance_id", _COLUMN, "state"], + if_not_exists=True, + ) + + +def downgrade() -> None: + bind = op.get_bind() + if _COLUMN not in _columns(bind): + return + op.drop_index("idx_http_bridge_sessions_owner_state", table_name=_TABLE, if_exists=True) + op.create_index( + "idx_http_bridge_sessions_owner_state", + _TABLE, + ["owner_instance_id", "state"], + if_not_exists=True, + ) + with op.batch_alter_table(_TABLE) as batch_op: + batch_op.drop_column(_COLUMN) diff --git a/app/db/models.py b/app/db/models.py index b2a90bd1c..496d63385 100644 --- a/app/db/models.py +++ b/app/db/models.py @@ -1791,6 +1791,7 @@ class HttpBridgeSessionRecord(Base): session_key_hash: Mapped[str] = mapped_column(String(64), nullable=False) api_key_scope: Mapped[str] = mapped_column(String(255), nullable=False) owner_instance_id: Mapped[str | None] = mapped_column(String(255), nullable=True) + owner_process_epoch: Mapped[str | None] = mapped_column(String(64), nullable=True) owner_epoch: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default=text("0")) lease_expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) state: Mapped[HttpBridgeSessionState] = mapped_column( @@ -2126,7 +2127,12 @@ class HttpBridgeRetryCircuit(Base): Index("idx_automation_runs_status_started_at", AutomationRun.status, AutomationRun.started_at) Index("idx_automation_runs_scheduled_for", AutomationRun.scheduled_for) Index("idx_automation_runs_cycle_key_started_at", AutomationRun.cycle_key, AutomationRun.started_at) -Index("idx_http_bridge_sessions_owner_state", HttpBridgeSessionRecord.owner_instance_id, HttpBridgeSessionRecord.state) +Index( + "idx_http_bridge_sessions_owner_state", + HttpBridgeSessionRecord.owner_instance_id, + HttpBridgeSessionRecord.owner_process_epoch, + HttpBridgeSessionRecord.state, +) Index("idx_http_bridge_sessions_lease", HttpBridgeSessionRecord.lease_expires_at) Index("idx_http_bridge_sessions_last_seen", HttpBridgeSessionRecord.last_seen_at.desc()) Index( diff --git a/app/main.py b/app/main.py index 13151af88..00f3c7878 100644 --- a/app/main.py +++ b/app/main.py @@ -81,6 +81,7 @@ from app.modules.proxy.cap_partitioning import refresh_cap_partition from app.modules.proxy.durable_bridge_coordinator import DurableBridgeSessionCoordinator from app.modules.proxy.durable_bridge_repository import missing_durable_bridge_tables +from app.modules.proxy.durable_bridge_runtime import http_bridge_owner_process_epoch from app.modules.proxy.rate_limit_cache import get_rate_limit_headers_cache from app.modules.proxy.ring_membership import ( RING_HEARTBEAT_INTERVAL_SECONDS, @@ -276,6 +277,7 @@ async def lifespan(app: FastAPI): ) deleted_bridge_rows = await DurableBridgeSessionCoordinator(SessionLocal).purge_owned_sessions_on_startup( instance_id=settings.http_responses_session_bridge_instance_id, + owner_process_epoch=http_bridge_owner_process_epoch(), ownerless_cutoff=ownerless_cutoff, ) if deleted_bridge_rows > 0: diff --git a/app/modules/proxy/_service/http_bridge/retry_circuit.py b/app/modules/proxy/_service/http_bridge/retry_circuit.py index 5a21c1033..51047cc16 100644 --- a/app/modules/proxy/_service/http_bridge/retry_circuit.py +++ b/app/modules/proxy/_service/http_bridge/retry_circuit.py @@ -341,10 +341,10 @@ async def _record_http_bridge_retry_circuit_failure( session: _HTTPBridgeSession, *, detail: str, - ) -> None: + ) -> int | None: detail = _HTTP_BRIDGE_RETRY_CIRCUIT_DETAIL_ALIASES.get(detail, detail) if session.key.strength != "hard" or detail not in _HTTP_BRIDGE_RETRY_CIRCUIT_FAILURE_DETAILS: - return + return None await self._load_http_bridge_retry_circuit(session) threshold = max(1, _HTTP_BRIDGE_RETRY_CIRCUIT_FAILURE_THRESHOLD) @@ -385,6 +385,7 @@ async def _record_http_bridge_retry_circuit_failure( async with self._http_bridge_retry_circuit_lock: if self._http_bridge_retry_circuits.get(session.key) is state: self._http_bridge_retry_circuit_loaded_keys.add(session.key) + return state.consecutive_failures async def _clear_http_bridge_retry_circuit(self: Any, session: _HTTPBridgeSession) -> None: if session.key.strength != "hard": diff --git a/app/modules/proxy/_service/http_bridge/session_registry.py b/app/modules/proxy/_service/http_bridge/session_registry.py index a3d6616e2..58cd3bacd 100644 --- a/app/modules/proxy/_service/http_bridge/session_registry.py +++ b/app/modules/proxy/_service/http_bridge/session_registry.py @@ -47,6 +47,7 @@ DurableBridgeAliasRegistration, DurableBridgeAliasRegistrationReceipt, ) +from app.modules.proxy.durable_bridge_runtime import http_bridge_owner_process_epoch logger = logging.getLogger("app.modules.proxy.service") @@ -433,6 +434,7 @@ async def _claim_durable_http_bridge_session( clear_latest_turn_state: bool = False, ) -> None: current_instance = _service_get_settings().http_responses_session_bridge_instance_id + current_process_epoch = http_bridge_owner_process_epoch() try: lookup: DurableBridgeLookup | None = None for claim_attempt in range(2): @@ -441,6 +443,7 @@ async def _claim_durable_http_bridge_session( session_key_value=session.key.affinity_key, api_key_id=session.key.api_key_id, instance_id=current_instance, + owner_process_epoch=current_process_epoch, lease_ttl_seconds=_http_bridge_durable_lease_ttl_seconds(), account_id=claim_account_id or session.account.id, model=session.request_model, diff --git a/app/modules/proxy/_service/http_bridge/streaming.py b/app/modules/proxy/_service/http_bridge/streaming.py index ff3054e3c..b513193b9 100644 --- a/app/modules/proxy/_service/http_bridge/streaming.py +++ b/app/modules/proxy/_service/http_bridge/streaming.py @@ -80,6 +80,7 @@ _http_bridge_owner_lookup_unavailable_error_envelope, _http_bridge_payload_looks_like_full_resend, _http_bridge_payload_without_previous_response_id, + _http_bridge_previous_response_error_envelope, _http_bridge_request_budget_seconds, _http_bridge_request_needs_unanchored_handoff, _http_bridge_request_stage, @@ -217,6 +218,7 @@ ) from app.modules.proxy.durable_bridge_coordinator import DurableBridgeLookup from app.modules.proxy.durable_bridge_repository import durable_bridge_hash +from app.modules.proxy.durable_bridge_runtime import http_bridge_owner_process_epoch from app.modules.proxy.helpers import ( _normalize_error_code, ) @@ -397,6 +399,66 @@ def _verify_durable_full_resend( return _VerifiedDurableFullResend._verify(payload, durable_lookup) +_HTTP_BRIDGE_DEAD_OWNER_NOT_FOUND_DETAIL = "The previous bridge owner is no longer available." + + +def _http_bridge_dead_owner_previous_response_not_found_terminal( + *, + previous_response_id: str, + response_id: str, +) -> dict[str, JsonValue]: + """Return the standard not-found terminal for an unreplayable dead owner.""" + + error = cast( + dict[str, JsonValue], + _http_bridge_previous_response_error_envelope( + previous_response_id, + _HTTP_BRIDGE_DEAD_OWNER_NOT_FOUND_DETAIL, + )["error"], + ) + return cast( + dict[str, JsonValue], + response_failed_event( + cast(str, error["code"]), + cast(str, error["message"]), + error_type=cast(str, error["type"]), + response_id=response_id, + error_param=cast(str, error["param"]), + ), + ) + + +def _http_bridge_dead_owner_previous_response_not_found_proxy_error( + *, + previous_response_id: str, +) -> ProxyResponseError: + return ProxyResponseError( + 400, + _http_bridge_previous_response_error_envelope( + previous_response_id, + _HTTP_BRIDGE_DEAD_OWNER_NOT_FOUND_DETAIL, + ), + ) + + +def _http_bridge_dead_owner_previous_response_id(request_state: _WebSocketRequestState) -> str: + return request_state.previous_response_id or _websocket_downstream_response_id(request_state) + + +def _http_bridge_durable_owner_is_dead( + lookup: DurableBridgeLookup, + *, + current_instance: str, + current_process_epoch: str, +) -> bool: + """Classify an anchored durable lookup whose owner cannot be live now.""" + + previous_process_epoch = lookup.owner_process_epoch + owner_instance_is_dead = lookup.owner_instance_id is not None and lookup.owner_instance_id != current_instance + process_epoch_is_dead = previous_process_epoch is not None and previous_process_epoch != current_process_epoch + return owner_instance_is_dead or process_epoch_is_dead or not lookup.lease_is_active(now=utcnow()) + + def _http_bridge_payload_is_account_neutral_fresh_replay(payload: ResponsesRequest) -> bool: return responses_payload_is_account_neutral_fresh_replay(payload.to_replay_safety_payload()) @@ -958,6 +1020,8 @@ async def _stream_via_http_bridge( deferred_account_backoff_tracker: _DeferredAccountBackoffTracker | None = None, ) -> AsyncIterator[str]: del suppress_text_done_events + dead_owner_anchor = False + dead_owner_process_epoch_mismatch = False request_id = ensure_request_id() dashboard_settings = await _service_get_settings_cache().get() runtime_config = _http_bridge_runtime_config(dashboard_settings, _service_get_settings()) @@ -1032,6 +1096,7 @@ def prepare_bridge_request( request_state.deferred_account_error_backoffs = lifecycle.pending_backoffs request_state.deferred_account_backoff_tracker = deferred_account_backoff_tracker request_state.deferred_account_backoff_lifecycle = lifecycle + request_state.durable_owner_dead = dead_owner_anchor return request_state, text_data async def release_unowned_bridge_lifecycle( @@ -1175,6 +1240,18 @@ async def release_unowned_bridge_lifecycle( exc_info=True, ) durable_lookup = None + if durable_lookup is not None and durable_lookup.latest_response_id is not None: + current_instance = _service_get_settings().http_responses_session_bridge_instance_id + current_process_epoch = http_bridge_owner_process_epoch() + dead_owner_process_epoch_mismatch = ( + durable_lookup.owner_process_epoch is not None + and durable_lookup.owner_process_epoch != current_process_epoch + ) + dead_owner_anchor = _http_bridge_durable_owner_is_dead( + durable_lookup, + current_instance=current_instance, + current_process_epoch=current_process_epoch, + ) effective_payload = payload untrimmed_effective_payload = payload proxy_injected_previous_response_id = False @@ -1305,6 +1382,7 @@ def classify_durable_full_resend( service_tier=None, latest_turn_state=durable_lookup.latest_turn_state, latest_response_id=None, + owner_process_epoch=http_bridge_owner_process_epoch(), # Revalidate the stale lookup under the # row lock; an active owner that appeared # after the lookup must not be displaced. @@ -1398,6 +1476,7 @@ def classify_durable_full_resend( ) force_local_recovery_creation = True durable_lookup = None + dead_owner_anchor = False if durable_lookup is not None: bridge_session_key = _HTTPBridgeSessionKey( durable_lookup.canonical_kind, @@ -1680,14 +1759,13 @@ def classify_durable_full_resend( fresh_replay_excluded_account_ids: set[str] = set() unanchored_fork_spill_attempted = False - def owner_unavailable_allows_account_neutral_replay(exc: ProxyResponseError) -> bool: + def durable_full_resend_allows_account_neutral_replay() -> bool: nonlocal durable_full_resend_fresh_payload nonlocal durable_full_resend_is_account_neutral nonlocal durable_full_resend_retains_prior_output if ( - not _http_bridge_is_previous_response_owner_unavailable(exc) - or forwarded_request + forwarded_request or rewritten_file_account_id is not None or durable_full_resend_anchor_count is None or durable_full_resend_anchor_fingerprint is None @@ -1727,10 +1805,18 @@ def owner_unavailable_allows_account_neutral_replay(exc: ProxyResponseError) -> ) return durable_full_resend_is_account_neutral + def owner_unavailable_allows_account_neutral_replay(exc: ProxyResponseError) -> bool: + return ( + _http_bridge_is_previous_response_owner_unavailable(exc) + and durable_full_resend_allows_account_neutral_replay() + ) + def switch_to_account_neutral_replay() -> None: nonlocal account_neutral_recovery nonlocal affinity nonlocal bridge_session_key + nonlocal dead_owner_anchor + nonlocal dead_owner_process_epoch_mismatch nonlocal durable_full_resend_anchor_count nonlocal durable_full_resend_anchor_fingerprint nonlocal durable_full_resend_fresh_payload @@ -1799,6 +1885,8 @@ def switch_to_account_neutral_replay() -> None: durable_full_resend_fresh_payload = None durable_full_resend_is_account_neutral = None durable_lookup = None + dead_owner_anchor = False + dead_owner_process_epoch_mismatch = False file_required_preferred_account = False if durable_recovery_attempt_claimed: @@ -1807,6 +1895,24 @@ def switch_to_account_neutral_replay() -> None: request_state.recovery_attempt_session_id = durable_recovery_attempt_session_id request_state.recovery_attempt_owner_epoch = durable_recovery_attempt_owner_epoch request_state.recovery_attempt_claimed = True + elif ( + dead_owner_anchor + and durable_lookup is not None + and durable_lookup.state == HttpBridgeSessionState.ACTIVE + and dead_owner_process_epoch_mismatch + and durable_full_resend_allows_account_neutral_replay() + ): + _log_http_bridge_event( + "dead_owner_fresh_resend", + bridge_session_key, + account_id=request_state.preferred_account_id, + model=payload.model, + detail="outcome=projected_plaintext_full_resend_without_anchor", + cache_key_family=bridge_session_key.affinity_kind, + model_class=_extract_model_class(payload.model) if payload.model else None, + owner_check_applied=True, + ) + switch_to_account_neutral_replay() if required_continuity_owner_missing: owner_unavailable = ProxyResponseError( @@ -3331,6 +3437,10 @@ async def startup_continuity_cooldown_terminal_event() -> str | None: await self._release_websocket_request_state_reservation(request_state) request_state.api_key_reservation = None if propagate_http_errors: + if request_state.durable_owner_dead: + raise _http_bridge_dead_owner_previous_response_not_found_proxy_error( + previous_response_id=_http_bridge_dead_owner_previous_response_id(request_state), + ) raise ProxyResponseError( 503, openai_error( @@ -3344,7 +3454,12 @@ async def startup_continuity_cooldown_terminal_event() -> str | None: return format_sse_event( cast( Mapping[str, JsonValue], - response_failed_event( + _http_bridge_dead_owner_previous_response_not_found_terminal( + previous_response_id=_http_bridge_dead_owner_previous_response_id(request_state), + response_id=_websocket_downstream_response_id(request_state), + ) + if request_state.durable_owner_dead + else response_failed_event( "stream_idle_timeout", "Upstream did not respond within the keepalive window", response_id=_websocket_downstream_response_id(request_state), @@ -3463,7 +3578,12 @@ async def startup_continuity_cooldown_terminal_event() -> str | None: terminal_event = format_sse_event( cast( Mapping[str, JsonValue], - response_failed_event( + _http_bridge_dead_owner_previous_response_not_found_terminal( + previous_response_id=_http_bridge_dead_owner_previous_response_id(request_state), + response_id=_websocket_downstream_response_id(request_state), + ) + if request_state.durable_owner_dead + else response_failed_event( "stream_idle_timeout", "Upstream did not respond within the keepalive window", response_id=_websocket_downstream_response_id(request_state), @@ -3476,6 +3596,10 @@ async def startup_continuity_cooldown_terminal_event() -> str | None: # upstream handoff for retirement. await self._detach_http_bridge_request(session, request_state=request_state) if propagate_http_errors: + if request_state.durable_owner_dead: + raise _http_bridge_dead_owner_previous_response_not_found_proxy_error( + previous_response_id=_http_bridge_dead_owner_previous_response_id(request_state), + ) raise ProxyResponseError( 503, openai_error( @@ -3712,7 +3836,14 @@ def stream_idle_keepalive(*, downstream_response_id: str) -> str | None: yield format_sse_event( cast( Mapping[str, JsonValue], - response_failed_event( + _http_bridge_dead_owner_previous_response_not_found_terminal( + previous_response_id=_http_bridge_dead_owner_previous_response_id( + request_state + ), + response_id=downstream_response_id, + ) + if request_state.durable_owner_dead + else response_failed_event( "stream_idle_timeout", "Upstream did not respond within the keepalive window", response_id=downstream_response_id, diff --git a/app/modules/proxy/_service/http_bridge/upstream_events.py b/app/modules/proxy/_service/http_bridge/upstream_events.py index aac8dc36e..aaa205116 100644 --- a/app/modules/proxy/_service/http_bridge/upstream_events.py +++ b/app/modules/proxy/_service/http_bridge/upstream_events.py @@ -656,6 +656,52 @@ async def _clear_durable_http_bridge_response_anchor( ) +async def _abandon_durable_http_bridge_continuity( + service: Any, + session: "_HTTPBridgeSession", +) -> bool: + """Clear durable continuity before retiring a repeatedly poisoned bridge. + + ``rebind_session_account(clear_continuity=True)`` is an existing fenced + write that clears the durable response/turn anchor and its alias rows while + this worker still owns the session. The ordinary retirement path then + closes the row and removes the process-local registrations. + """ + if session.durable_session_id is None or session.durable_owner_epoch is None: + return False + try: + cleared = await service._durable_bridge.rebind_session_account( + session_id=session.durable_session_id, + api_key_id=session.key.api_key_id, + instance_id=_service_get_settings().http_responses_session_bridge_instance_id, + owner_epoch=session.durable_owner_epoch, + account_id=session.account.id, + clear_continuity=True, + ) + except Exception: + logger.warning("Failed to abandon poisoned HTTP bridge continuity", exc_info=True) + return False + if not cleared: + logger.warning( + "Durable bridge continuity clear was fenced before poisoned anchor retirement", + extra={ + "session_id": session.durable_session_id, + "account_id": session.account.id, + }, + ) + return False + _log_http_bridge_event( + "durable_anchor_poisoned", + session.key, + account_id=session.account.id, + model=session.request_model, + detail="repeated_zero_event_idle_timeout", + cache_key_family=session.key.affinity_kind, + model_class=_extract_model_class(session.request_model) if session.request_model else None, + ) + return True + + class _HTTPBridgeUpstreamEventsMixin: async def _fail_http_bridge_reader_and_maybe_retire( self: Any, @@ -743,6 +789,7 @@ async def _fail_http_bridge_reader_and_maybe_retire( penalize_account=penalize_account, ) finally: + poison_after_deferred_failures = False if session.admission_waiter_count > 0 and not force_retire: retry_circuit_detail = None if close_classification == "clean": @@ -757,20 +804,48 @@ async def _fail_http_bridge_reader_and_maybe_retire( None, ) if failed_pending_count > 0 and retry_circuit_detail is not None: - await self._record_http_bridge_retry_circuit_failure( + consecutive_failures = await self._record_http_bridge_retry_circuit_failure( session, detail=retry_circuit_detail, ) - _log_http_bridge_event( - "retire_deferred_for_admission_waiter", - session.key, - account_id=session.account.id, - model=session.request_model, - pending_count=session.admission_waiter_count, - detail=retire_detail or error_code, - cache_key_family=session.key.affinity_kind, - model_class=_extract_model_class(session.request_model) if session.request_model else None, - ) + poison_after_deferred_failures = bool( + retry_circuit_detail == "stream_idle_timeout" + and observed_response_events == 0 + and consecutive_failures is not None + and consecutive_failures + >= _service_get_settings().http_responses_session_bridge_anchor_poison_failure_threshold + ) + if poison_after_deferred_failures: + durable_cleared = await _abandon_durable_http_bridge_continuity(self, session) + if durable_cleared: + await self._retire_stale_pending_http_bridge_session( + session, + detail="repeated_zero_event_idle_timeout", + response_events_seen=observed_response_events, + ) + force_retire = True + else: + _log_http_bridge_event( + "durable_anchor_poison_clear_failed", + session.key, + account_id=session.account.id, + model=session.request_model, + pending_count=session.admission_waiter_count, + detail="repeated_zero_event_idle_timeout", + cache_key_family=session.key.affinity_kind, + model_class=_extract_model_class(session.request_model) if session.request_model else None, + ) + else: + _log_http_bridge_event( + "retire_deferred_for_admission_waiter", + session.key, + account_id=session.account.id, + model=session.request_model, + pending_count=session.admission_waiter_count, + detail=retire_detail or error_code, + cache_key_family=session.key.affinity_kind, + model_class=_extract_model_class(session.request_model) if session.request_model else None, + ) else: if close_classification == "clean" and failed_pending_count > 0: await self._retire_stale_pending_http_bridge_session( diff --git a/app/modules/proxy/_service/support.py b/app/modules/proxy/_service/support.py index afbde89b8..e61bad4e9 100644 --- a/app/modules/proxy/_service/support.py +++ b/app/modules/proxy/_service/support.py @@ -841,6 +841,10 @@ class _WebSocketRequestState: # explicit turn-state header guarantees continuity for stale recovery. hard_continuity_anchor: bool = False proxy_injected_previous_response_id: bool = False + # The durable lookup carried an anchor, but its owner was already stale, + # ownerless, or lease-expired when this request arrived. Such a request + # must not be presented to the client as a retryable upstream timeout. + durable_owner_dead: bool = False # True only when the client's own incoming payload (before this anchor was # injected or trimmed) already looked like a full conversation resend # (``_http_bridge_payload_looks_like_full_resend``). Deliberately weaker diff --git a/app/modules/proxy/durable_bridge_coordinator.py b/app/modules/proxy/durable_bridge_coordinator.py index 389f47c6f..6ae3703c8 100644 --- a/app/modules/proxy/durable_bridge_coordinator.py +++ b/app/modules/proxy/durable_bridge_coordinator.py @@ -44,6 +44,7 @@ class DurableBridgeLookup: latest_input_full_fingerprint: str | None = None model: str | None = None latest_pending_tool_calls: dict[str, str] | None = None + owner_process_epoch: str | None = None def lease_is_active(self, *, now: datetime) -> bool: if self.owner_instance_id is None: @@ -296,6 +297,7 @@ async def claim_live_session( latest_turn_state: str | None, latest_response_id: str | None, allow_takeover: bool, + owner_process_epoch: str, force_owner_epoch_advance: bool = False, ) -> DurableBridgeLookup: api_key_scope = durable_bridge_api_key_scope(api_key_id) @@ -312,6 +314,7 @@ async def claim_live_session( latest_turn_state=latest_turn_state, latest_response_id=latest_response_id, allow_takeover=allow_takeover, + owner_process_epoch=owner_process_epoch, force_owner_epoch_advance=force_owner_epoch_advance, ) return _to_lookup(snapshot) @@ -489,11 +492,13 @@ async def purge_owned_sessions_on_startup( self, *, instance_id: str, + owner_process_epoch: str | None = None, ownerless_cutoff: datetime | None = None, ) -> int: async with self._session() as session: return await DurableBridgeRepository(session).purge_owned_sessions_on_startup( instance_id=instance_id, + owner_process_epoch=owner_process_epoch, ownerless_cutoff=ownerless_cutoff, ) @@ -613,6 +618,7 @@ def _to_lookup(snapshot: DurableBridgeSessionSnapshot) -> DurableBridgeLookup: api_key_scope=snapshot.api_key_scope, account_id=snapshot.account_id, owner_instance_id=snapshot.owner_instance_id, + owner_process_epoch=snapshot.owner_process_epoch, owner_epoch=snapshot.owner_epoch, lease_expires_at=snapshot.lease_expires_at, state=snapshot.state, diff --git a/app/modules/proxy/durable_bridge_repository.py b/app/modules/proxy/durable_bridge_repository.py index c2af1cc3e..bcea92b62 100644 --- a/app/modules/proxy/durable_bridge_repository.py +++ b/app/modules/proxy/durable_bridge_repository.py @@ -9,7 +9,7 @@ from hashlib import sha256 from typing import Any -from sqlalchemy import Row, and_, case, delete, func, or_, select, text, update +from sqlalchemy import Row, and_, case, delete, func, or_, select, text, true, update from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.dialects.sqlite import insert as sqlite_insert from sqlalchemy.exc import IntegrityError @@ -130,6 +130,7 @@ class DurableBridgeSessionSnapshot: last_seen_at: datetime closed_at: datetime | None latest_pending_tool_calls: dict[str, str] | None = None + owner_process_epoch: str | None = None @dataclass(frozen=True, slots=True) @@ -546,6 +547,7 @@ async def claim_session( latest_turn_state: str | None, latest_response_id: str | None, allow_takeover: bool, + owner_process_epoch: str, force_owner_epoch_advance: bool = False, ) -> DurableBridgeSessionSnapshot: session_key_hash = durable_bridge_hash(session_key_value) @@ -569,6 +571,7 @@ async def claim_session( session_key_hash=session_key_hash, api_key_scope=api_key_scope, owner_instance_id=instance_id, + owner_process_epoch=owner_process_epoch, owner_epoch=1, lease_expires_at=lease_expires_at, state=HttpBridgeSessionState.ACTIVE, @@ -609,6 +612,7 @@ async def claim_session( async with sqlite_writer_section(): existing.owner_instance_id = instance_id + existing.owner_process_epoch = owner_process_epoch existing.owner_epoch = next_epoch existing.lease_expires_at = lease_expires_at existing.state = HttpBridgeSessionState.ACTIVE @@ -1057,6 +1061,7 @@ async def purge_owned_sessions_on_startup( self, *, instance_id: str, + owner_process_epoch: str | None = None, ownerless_cutoff: datetime | None = None, batch_size: int = _PURGE_CLOSED_BATCH_SIZE, ) -> int: @@ -1072,7 +1077,17 @@ async def purge_owned_sessions_on_startup( deleted_count = 0 while True: now = utcnow() - purge_predicates = [HttpBridgeSessionRecord.owner_instance_id == instance_id] + if owner_process_epoch is None: + owned_restart_filter = HttpBridgeSessionRecord.owner_instance_id == instance_id + else: + owned_restart_filter = and_( + HttpBridgeSessionRecord.owner_instance_id == instance_id, + or_( + HttpBridgeSessionRecord.owner_process_epoch.is_(None), + HttpBridgeSessionRecord.owner_process_epoch != owner_process_epoch, + ), + ) + purge_predicates = [owned_restart_filter] if ownerless_cutoff is not None: purge_predicates.append( and_( @@ -1094,6 +1109,7 @@ async def purge_owned_sessions_on_startup( HttpBridgeSessionRecord.session_key_kind, HttpBridgeSessionRecord.session_key_value, HttpBridgeSessionRecord.owner_instance_id, + HttpBridgeSessionRecord.owner_process_epoch, HttpBridgeSessionRecord.last_seen_at, ) .where(startup_purge_filter) @@ -1108,6 +1124,7 @@ async def purge_owned_sessions_on_startup( candidate.id for candidate in candidates if candidate.owner_instance_id == instance_id + and getattr(candidate, "owner_process_epoch", None) == owner_process_epoch and (ownerless_cutoff is None or to_utc_naive(candidate.last_seen_at) >= to_utc_naive(ownerless_cutoff)) and is_http_bridge_account_neutral_replay( kind=candidate.session_key_kind, @@ -1121,6 +1138,7 @@ async def purge_owned_sessions_on_startup( .where( HttpBridgeSessionRecord.id.in_(retained_recovery_ids), HttpBridgeSessionRecord.owner_instance_id == instance_id, + HttpBridgeSessionRecord.owner_process_epoch == owner_process_epoch, ) .values( owner_instance_id=None, @@ -1131,13 +1149,72 @@ async def purge_owned_sessions_on_startup( ) deletable_ids = [session_id for session_id in session_ids if session_id not in retained_recovery_ids] if deletable_ids: - deleted = await self._session.execute( - delete(HttpBridgeSessionRecord) - .where(HttpBridgeSessionRecord.id.in_(deletable_ids)) - .where(startup_purge_filter) - .returning(HttpBridgeSessionRecord.id) - ) - deleted_ids = list(deleted.scalars().all()) + if owner_process_epoch is None: + deleted = await self._session.execute( + delete(HttpBridgeSessionRecord) + .where(HttpBridgeSessionRecord.id.in_(deletable_ids)) + .where(startup_purge_filter) + .returning(HttpBridgeSessionRecord.id) + ) + deleted_ids = list(deleted.scalars().all()) + else: + previous_process_ids = [ + candidate.id for candidate in candidates if candidate.owner_instance_id == instance_id + ] + ownerless_ids = [ + candidate.id + for candidate in candidates + if candidate.owner_instance_id is None and candidate.id not in retained_recovery_ids + ] + retired_ids: list[str] = [] + if previous_process_ids: + retired = await self._session.execute( + update(HttpBridgeSessionRecord) + .where(HttpBridgeSessionRecord.id.in_(previous_process_ids)) + .where( + HttpBridgeSessionRecord.owner_instance_id == instance_id, + or_( + HttpBridgeSessionRecord.owner_process_epoch.is_(None), + HttpBridgeSessionRecord.owner_process_epoch != owner_process_epoch, + ), + ) + .values( + owner_instance_id=None, + lease_expires_at=None, + state=HttpBridgeSessionState.CLOSED, + closed_at=now, + last_seen_at=now, + latest_turn_state=None, + latest_response_id=None, + latest_input_item_count=None, + latest_input_full_fingerprint=None, + latest_pending_tool_calls_json=None, + ) + .returning(HttpBridgeSessionRecord.id) + ) + retired_ids = list(retired.scalars().all()) + deleted_ownerless_ids: list[str] = [] + if ownerless_ids: + deleted_ownerless = await self._session.execute( + delete(HttpBridgeSessionRecord) + .where(HttpBridgeSessionRecord.id.in_(ownerless_ids)) + .where( + HttpBridgeSessionRecord.owner_instance_id.is_(None), + HttpBridgeSessionRecord.state.in_( + (HttpBridgeSessionState.ACTIVE, HttpBridgeSessionState.DRAINING), + ), + or_( + HttpBridgeSessionRecord.lease_expires_at.is_(None), + HttpBridgeSessionRecord.lease_expires_at < now, + ), + HttpBridgeSessionRecord.last_seen_at < ownerless_cutoff + if ownerless_cutoff is not None + else true(), + ) + .returning(HttpBridgeSessionRecord.id) + ) + deleted_ownerless_ids = list(deleted_ownerless.scalars().all()) + deleted_ids = retired_ids + deleted_ownerless_ids else: deleted_ids = [] if deleted_ids: @@ -1727,6 +1804,7 @@ async def missing_durable_bridge_tables(session: AsyncSession) -> tuple[str, ... HttpBridgeSessionRecord.session_key_hash, HttpBridgeSessionRecord.api_key_scope, HttpBridgeSessionRecord.owner_instance_id, + HttpBridgeSessionRecord.owner_process_epoch, HttpBridgeSessionRecord.owner_epoch, HttpBridgeSessionRecord.lease_expires_at, HttpBridgeSessionRecord.state, @@ -1752,6 +1830,7 @@ def _returned_row_to_snapshot(row: Row[tuple[object, ...]]) -> DurableBridgeSess session_key_hash=mapping[HttpBridgeSessionRecord.session_key_hash], api_key_scope=mapping[HttpBridgeSessionRecord.api_key_scope], owner_instance_id=mapping[HttpBridgeSessionRecord.owner_instance_id], + owner_process_epoch=mapping[HttpBridgeSessionRecord.owner_process_epoch], owner_epoch=mapping[HttpBridgeSessionRecord.owner_epoch], lease_expires_at=mapping[HttpBridgeSessionRecord.lease_expires_at], state=mapping[HttpBridgeSessionRecord.state], @@ -1781,6 +1860,7 @@ def _to_snapshot(row: HttpBridgeSessionRecord | None) -> DurableBridgeSessionSna session_key_hash=row.session_key_hash, api_key_scope=row.api_key_scope, owner_instance_id=row.owner_instance_id, + owner_process_epoch=row.owner_process_epoch, owner_epoch=row.owner_epoch, lease_expires_at=row.lease_expires_at, state=row.state, diff --git a/app/modules/proxy/durable_bridge_runtime.py b/app/modules/proxy/durable_bridge_runtime.py new file mode 100644 index 000000000..622757e30 --- /dev/null +++ b/app/modules/proxy/durable_bridge_runtime.py @@ -0,0 +1,9 @@ +from __future__ import annotations + +from uuid import uuid4 + +_HTTP_BRIDGE_OWNER_PROCESS_EPOCH = uuid4().hex + + +def http_bridge_owner_process_epoch() -> str: + return _HTTP_BRIDGE_OWNER_PROCESS_EPOCH diff --git a/docs/reference/settings.md b/docs/reference/settings.md index 435237590..e7486b232 100644 --- a/docs/reference/settings.md +++ b/docs/reference/settings.md @@ -7,7 +7,7 @@ Regenerate with `uv run python scripts/generate_settings_reference.py`; `tests/unit/test_settings_reference.py` fails when this page drifts from `app/core/config/settings.py`. -codex-lb currently exposes 117 settings. Every setting is an environment +codex-lb currently exposes 118 settings. Every setting is an environment variable with the `CODEX_LB_` prefix (process environment or `.env` / `.env.local` next to the process). All defaults work with zero configuration — start from [Configuration](../configuration.md) for the handful that matter, @@ -82,6 +82,7 @@ the host side of the compose `ports` mapping instead. | Environment variable | Type | Default | | --- | --- | --- | | `CODEX_LB_HTTP_RESPONSES_SESSION_BRIDGE_ADVERTISE_BASE_URL` | `str \| None` | `None` | +| `CODEX_LB_HTTP_RESPONSES_SESSION_BRIDGE_ANCHOR_POISON_FAILURE_THRESHOLD` | `int` | `7` | | `CODEX_LB_HTTP_RESPONSES_SESSION_BRIDGE_CLEAN_CLOSE_RETRY_JITTER_MAX_SECONDS` | `float` | `2.0` | | `CODEX_LB_HTTP_RESPONSES_SESSION_BRIDGE_CODEX_IDLE_TTL_SECONDS` | `float` | `900.0` | | `CODEX_LB_HTTP_RESPONSES_SESSION_BRIDGE_CODEX_PREWARM_ENABLED` | `bool` | `False` | diff --git a/openspec/changes/recover-bridge-restart-anchors/proposal.md b/openspec/changes/recover-bridge-restart-anchors/proposal.md new file mode 100644 index 000000000..0b3c55a11 --- /dev/null +++ b/openspec/changes/recover-bridge-restart-anchors/proposal.md @@ -0,0 +1,27 @@ +## Why + +A killed codex-lb process can leave durable HTTP bridge rows that still look +owned by the replacement process when Docker restarts the same container id. +Those stale rows keep previous-response and session anchors addressable, so +clients reconnect to dead continuity state and receive retryable +`stream_idle_timeout` semantics instead of a fast recovery boundary. + +## What Changes + +- Persist a per-process bridge owner epoch alongside the existing instance id + and owner fencing epoch. +- On startup, retire rows owned by a previous process epoch for the same + instance id and remove their attachable aliases. +- Treat continuity-bound idle terminals as non-retryable fresh-turn guidance + only when durable ownership is proven dead; ordinary upstream silence keeps + the existing retryable timeout behavior. +- Poison a durable bridge anchor after repeated zero-event idle failures on + the same hard bridge key, using the existing retry-circuit count and capping + admission-waiter retirement deferral. + +## Impact + +- Affected capabilities: `responses-api-compat`. +- Existing durable owner fencing remains intact; the process epoch only + distinguishes process incarnations under a stable instance id. +- No live deployment or live database mutation is part of this change. diff --git a/openspec/changes/recover-bridge-restart-anchors/specs/responses-api-compat/spec.md b/openspec/changes/recover-bridge-restart-anchors/specs/responses-api-compat/spec.md new file mode 100644 index 000000000..f853e0630 --- /dev/null +++ b/openspec/changes/recover-bridge-restart-anchors/specs/responses-api-compat/spec.md @@ -0,0 +1,109 @@ +# responses-api-compat Delta + +## ADDED Requirements + +### Requirement: Durable bridge ownership distinguishes process incarnations + +Durable HTTP bridge ownership MUST include a per-process owner epoch in +addition to the stable bridge instance id and the existing owner fencing epoch. +The process owner epoch MUST be generated when the process starts and MUST be +persisted on newly claimed durable HTTP bridge session rows. + +On startup, an instance MUST retire durable HTTP bridge sessions whose +`owner_instance_id` equals the current instance id but whose process owner epoch +is missing or differs from the current process owner epoch. Retired rows MUST +be closed and MUST NOT remain attachable through session-header, +turn-state, previous-response, latest-turn-state, or latest-response lookup. +Retired rows MUST clear stored previous-response, latest-turn-state, input +fingerprint, and pending-tool continuity anchors before any future claim can +reuse the same canonical session key. + +#### Scenario: Same-container restart retires previous-process rows + +- **GIVEN** a durable HTTP bridge session is ACTIVE under instance + `container-74e8e7cda9fb` and process epoch `boot-a` +- **WHEN** codex-lb starts again in the same container id with process epoch + `boot-b` +- **THEN** startup closes the `boot-a` durable session row +- **AND** request-target lookup for that session header, turn state, or + previous response no longer returns the closed row +- **AND** rows already owned by `boot-b` remain attachable + +### Requirement: Dead durable anchors recover transparently when safe + +The proxy MUST classify proven-dead durable anchors as automatic recovery +candidates before returning any client-visible error. + +When a continuity-bound HTTP bridge request would otherwise return a retryable +`stream_idle_timeout` or cooldown terminal, and the durable lookup that supplied +the request's previous-response anchor is proven dead because its owner +instance, process owner epoch, or lease is no longer current, the proxy MUST +dispatch a fresh turn transparently when the request payload has an existing +safe replay proof, including account-neutral full-context resends and +proxy-injected anchor requests whose captured fresh body is replay-safe. The +client MUST receive the normal upstream stream for that fresh turn and MUST NOT +receive a bridge-specific recovery error. + +When the request is bound to a client-provided anchor that cannot be safely +replayed as a fresh turn, the proxy MUST return the same OpenAI-compatible +`previous_response_not_found` error shape and HTTP status used by the existing +previous-response-not-found path. The proxy MUST NOT expose a +`bridge_continuity_recovery_required` code to clients. The proxy MUST keep the +existing retryable `stream_idle_timeout` semantics when the durable owner is +current and the failure is ordinary transient upstream silence. + +#### Scenario: Previous-process anchor with replayable context recovers automatically + +- **GIVEN** a request is bound to a durable previous-response anchor +- **AND** that durable row belongs to the same instance id but a different + process owner epoch +- **AND** the payload has a safe full-context replay proof +- **WHEN** the bridge hits the pre-submit, startup-cooldown, or retry-circuit + idle terminal path +- **THEN** the proxy dispatches the request as a fresh turn without the dead + previous-response anchor +- **AND** the client receives the normal streaming response +- **AND** the response does not include `stream_idle_timeout` retry guidance or + a bridge-specific recovery error + +#### Scenario: Unreplayable client anchor uses the standard not-found contract + +- **GIVEN** a request is bound to a client-provided durable previous-response + anchor +- **AND** that durable row belongs to a dead owner +- **AND** the payload does not have a safe fresh-turn replay proof +- **WHEN** the bridge must fail closed +- **THEN** the client receives the standard `previous_response_not_found` + error shape for `previous_response_id` +- **AND** HTTP error collection uses the standard previous-response-not-found + status +- **AND** the response does not include a bridge-specific recovery code + +#### Scenario: Current-owner silence remains retryable + +- **GIVEN** a request is bound to a durable owner whose instance id, process + owner epoch, and lease are current +- **WHEN** upstream produces no response events through the existing idle window +- **THEN** the proxy preserves the existing retryable `stream_idle_timeout` + behavior + +### Requirement: Repeated zero-event idle failures poison dead anchors + +For hard HTTP bridge keys, repeated zero-event idle failures MUST use the +existing durable retry-circuit counter to identify an anchor that should no +longer remain addressable. When consecutive failures for the same hard bridge +key reach the configured poison threshold, the proxy MUST abandon durable +continuity for that session and retire the bridge even when admission waiters +exist. The default threshold MUST be no greater than seven failures. + +#### Scenario: Admission waiters cannot defer anchor poisoning forever + +- **GIVEN** a hard durable bridge key has admission waiters +- **AND** repeated zero-event idle failures for that same key reach the poison + threshold +- **WHEN** the reader failure path would normally defer retirement for the + admission waiter +- **THEN** the proxy clears the durable continuity anchors +- **AND** retires the session despite the admission waiter +- **AND** the next attach starts from fresh durable state rather than the + poisoned previous-response anchor diff --git a/openspec/changes/recover-bridge-restart-anchors/tasks.md b/openspec/changes/recover-bridge-restart-anchors/tasks.md new file mode 100644 index 000000000..1ce138802 --- /dev/null +++ b/openspec/changes/recover-bridge-restart-anchors/tasks.md @@ -0,0 +1,10 @@ +- [x] Add a per-process durable bridge owner epoch and additive migration. +- [x] Retire previous-process same-instance durable bridge rows during startup. +- [x] Return non-retryable fresh-turn guidance for proven-dead durable owners. +- [x] Keep genuine transient upstream silence on existing retryable idle semantics. +- [x] Poison repeated zero-event idle anchors using retry-circuit state. +- [x] Add focused regressions for restart retirement, dead-owner semantics, and anchor poisoning. +- [x] Run strict OpenSpec validation and the full test suite; full suite has one unrelated + `test_quota_planner_warm_now_keeps_bootstrap_for_metadata_less_primary_rows` + failure that reproduces on clean `origin/main`. +- [x] Commit and push `bridge-restart-recovery`. diff --git a/tests/integration/test_http_responses_bridge.py b/tests/integration/test_http_responses_bridge.py index 1fed61d10..2007188cd 100644 --- a/tests/integration/test_http_responses_bridge.py +++ b/tests/integration/test_http_responses_bridge.py @@ -5005,6 +5005,7 @@ async def test_forwarded_recovery_uses_durable_owner_and_strips_stale_affinity( session_key_value=recovery_key, api_key_id=None, instance_id=target_settings.http_responses_session_bridge_instance_id, + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id=account.id, model="gpt-5.1", @@ -7552,6 +7553,7 @@ async def test_backend_responses_soft_prompt_cache_follow_up_uses_durable_owner_ session_key_value=prompt_cache_key, api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id=owner_account.id, model="gpt-5.1", @@ -7882,6 +7884,7 @@ async def test_backend_responses_verified_full_resend_ignores_stale_broad_owner_ session_key_value=session_id, api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id=owner_account.id, model="gpt-5.1", diff --git a/tests/integration/test_migrations.py b/tests/integration/test_migrations.py index 77054d2be..b20c38dbb 100644 --- a/tests/integration/test_migrations.py +++ b/tests/integration/test_migrations.py @@ -1674,6 +1674,7 @@ async def test_stamped_merge_rollup_repair_downgrade_preserves_schema(tmp_path): session_key_hash VARCHAR(64) NOT NULL, api_key_scope VARCHAR(255) NOT NULL, owner_instance_id VARCHAR(255), + owner_process_epoch VARCHAR(64), owner_epoch INTEGER NOT NULL DEFAULT 0, lease_expires_at DATETIME, state VARCHAR(16) NOT NULL DEFAULT 'active', diff --git a/tests/integration/test_sticky_sessions_api.py b/tests/integration/test_sticky_sessions_api.py index f30a18fbe..07929a3bc 100644 --- a/tests/integration/test_sticky_sessions_api.py +++ b/tests/integration/test_sticky_sessions_api.py @@ -156,6 +156,7 @@ async def test_durable_bridge_owned_alias_registration_is_epoch_fenced(db_setup) session_key_value="sid-owned-alias-fence", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id=None, model="gpt-5.6-sol", @@ -169,6 +170,7 @@ async def test_durable_bridge_owned_alias_registration_is_epoch_fenced(db_setup) session_key_value="sid-owned-alias-fence", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id=None, model="gpt-5.6-sol", diff --git a/tests/unit/test_bridge_ring_lifecycle.py b/tests/unit/test_bridge_ring_lifecycle.py index c65929d0d..744cd6c51 100644 --- a/tests/unit/test_bridge_ring_lifecycle.py +++ b/tests/unit/test_bridge_ring_lifecycle.py @@ -81,6 +81,7 @@ async def _claim( session_key_value=session_key_value, api_key_scope="__anonymous__", instance_id=instance_id, + owner_process_epoch="test-process", lease_ttl_seconds=lease_ttl_seconds, account_id="acc-1", model="gpt-5.4", diff --git a/tests/unit/test_durable_bridge_sessions.py b/tests/unit/test_durable_bridge_sessions.py index b5d7ed2a8..8977c7d4d 100644 --- a/tests/unit/test_durable_bridge_sessions.py +++ b/tests/unit/test_durable_bridge_sessions.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import inspect from collections.abc import AsyncIterator, Callable from datetime import datetime, timedelta, timezone from types import SimpleNamespace @@ -57,6 +58,12 @@ async def coordinator(async_session_factory: Callable[[], AsyncSession]) -> Dura return DurableBridgeSessionCoordinator(async_session_factory) +def test_durable_bridge_live_claim_requires_process_epoch() -> None: + parameter = inspect.signature(DurableBridgeSessionCoordinator.claim_live_session).parameters["owner_process_epoch"] + + assert parameter.default is inspect.Parameter.empty + + @pytest.mark.asyncio async def test_durable_bridge_lookup_prefers_turn_state_then_previous_response_then_session_header( coordinator: DurableBridgeSessionCoordinator, @@ -66,6 +73,7 @@ async def test_durable_bridge_lookup_prefers_turn_state_then_previous_response_t session_key_value="sid-123", api_key_id="key-1", instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-1", model="gpt-5.4", @@ -140,6 +148,7 @@ async def test_reversible_recovery_turn_state_registration_restores_previous_own session_key_value="sid-recovery-predecessor", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-a", model="gpt-5.6-sol", @@ -166,6 +175,7 @@ async def test_reversible_recovery_turn_state_registration_restores_previous_own session_key_value=recovery_key, api_key_id=None, instance_id="instance-b", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-b", model="gpt-5.6-sol", @@ -217,6 +227,7 @@ async def test_reversible_recovery_rollback_does_not_restore_reclaimed_predecess session_key_value="sid-reclaimed-predecessor", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-a", model="gpt-5.6-sol", @@ -242,6 +253,7 @@ async def test_reversible_recovery_rollback_does_not_restore_reclaimed_predecess session_key_value=recovery_key, api_key_id=None, instance_id="instance-b", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-b", model="gpt-5.6-sol", @@ -265,6 +277,7 @@ async def test_reversible_recovery_rollback_does_not_restore_reclaimed_predecess session_key_value="sid-reclaimed-predecessor", api_key_id=None, instance_id="instance-c", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-c", model="gpt-5.6-sol", @@ -297,6 +310,7 @@ async def test_durable_bridge_lookup_accepts_same_account_alias_session_divergen session_key_value="sid-turn-owner", api_key_id="key-same-account", instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-shared", model="gpt-5.4", @@ -310,6 +324,7 @@ async def test_durable_bridge_lookup_accepts_same_account_alias_session_divergen session_key_value="sid-response-owner", api_key_id="key-same-account", instance_id="instance-b", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-shared", model="gpt-5.4", @@ -367,6 +382,7 @@ async def test_durable_bridge_lookup_prefers_newest_same_account_response_anchor session_key_value="sid-turn-old-anchor", api_key_id="key-newest-anchor", instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-shared", model="gpt-5.4", @@ -380,6 +396,7 @@ async def test_durable_bridge_lookup_prefers_newest_same_account_response_anchor session_key_value="sid-session-new-anchor", api_key_id="key-newest-anchor", instance_id="instance-b", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-shared", model="gpt-5.4", @@ -454,6 +471,7 @@ async def test_durable_bridge_lookup_preserves_requested_response_alias_after_an session_key_value="sid-requested-anchor", api_key_id="key-requested-anchor", instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-shared", model="gpt-5.4", @@ -467,6 +485,7 @@ async def test_durable_bridge_lookup_preserves_requested_response_alias_after_an session_key_value="sid-fresher-turn", api_key_id="key-requested-anchor", instance_id="instance-b", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-shared", model="gpt-5.4", @@ -538,6 +557,7 @@ async def test_durable_bridge_lookup_rejects_ownerless_and_live_alias_divergence session_key_value="sid-ownerless", api_key_id="key-ownerless-conflict", instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id=None, model="gpt-5.4", @@ -551,6 +571,7 @@ async def test_durable_bridge_lookup_rejects_ownerless_and_live_alias_divergence session_key_value="sid-live-owner", api_key_id="key-ownerless-conflict", instance_id="instance-b", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-live", model="gpt-5.4", @@ -598,6 +619,7 @@ async def test_durable_bridge_lookup_rejects_conflicting_turn_and_response_alias session_key_value="sid-turn-owner", api_key_id="key-conflict", instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-turn-owner", model="gpt-5.4", @@ -611,6 +633,7 @@ async def test_durable_bridge_lookup_rejects_conflicting_turn_and_response_alias session_key_value="sid-response-owner", api_key_id="key-conflict", instance_id="instance-b", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-response-owner", model="gpt-5.4", @@ -659,6 +682,7 @@ async def test_durable_bridge_next_turn_prefers_verified_replay_over_shared_sess session_key_value="sid-shared", api_key_id="key-replay", instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-retired", model="gpt-5.4", @@ -677,6 +701,7 @@ async def test_durable_bridge_next_turn_prefers_verified_replay_over_shared_sess session_key_value=replay_key, api_key_id="key-replay", instance_id="instance-b", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-replay", model="gpt-5.4", @@ -732,6 +757,7 @@ async def test_durable_verified_replay_alias_cannot_be_stolen_by_predecessor( session_key_value=f"old-{predecessor_kind}", api_key_id="key-alias-fence", instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-old", model="gpt-5.4", @@ -757,6 +783,7 @@ async def test_durable_verified_replay_alias_cannot_be_stolen_by_predecessor( session_key_value=replay_key, api_key_id="key-alias-fence", instance_id="instance-b", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-replay", model="gpt-5.4", @@ -815,6 +842,7 @@ async def test_concurrent_recovery_lanes_publish_only_one_active_turn_owner( session_key_value=replay_key, api_key_id="key-concurrent-recovery", instance_id=f"instance-{index}", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id=f"acc-{index}", model="gpt-5.4", @@ -860,6 +888,7 @@ async def test_recovery_lane_replaces_alias_with_nonnull_owner_and_null_lease( session_key_value=old_key, api_key_id="key-null-lease", instance_id="instance-old", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-old", model="gpt-5.4", @@ -893,6 +922,7 @@ async def test_recovery_lane_replaces_alias_with_nonnull_owner_and_null_lease( session_key_value=new_key, api_key_id="key-null-lease", instance_id="instance-new", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-new", model="gpt-5.4", @@ -929,6 +959,7 @@ async def test_durable_bare_replay_prefix_does_not_receive_alias_protection( session_key_value=HTTP_BRIDGE_ACCOUNT_NEUTRAL_REPLAY_KEY_PREFIX, api_key_id=None, instance_id="instance-malformed", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-malformed", model="gpt-5.4", @@ -953,6 +984,7 @@ async def test_durable_bare_replay_prefix_does_not_receive_alias_protection( session_key_value="sid-valid-ordinary", api_key_id=None, instance_id="instance-ordinary", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-ordinary", model="gpt-5.4", @@ -991,6 +1023,7 @@ async def test_durable_verified_replay_alias_does_not_replace_unrelated_internal session_key_value="unrelated-internal-lane", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-internal", model="gpt-5.4", @@ -1016,6 +1049,7 @@ async def test_durable_verified_replay_alias_does_not_replace_unrelated_internal session_key_value=replay_key, api_key_id=None, instance_id="instance-b", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-replay", model="gpt-5.4", @@ -1058,6 +1092,7 @@ async def test_durable_replay_alias_policy_is_scoped_to_conflicting_row( session_key_value="sid-unrelated-rebindable", api_key_id="key-row-scope", instance_id="instance-decoy", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-decoy", model="gpt-5.4", @@ -1079,6 +1114,7 @@ async def test_durable_replay_alias_policy_is_scoped_to_conflicting_row( session_key_value="protected-internal-lane", api_key_id="key-row-scope", instance_id="instance-protected", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-protected", model="gpt-5.4", @@ -1101,6 +1137,7 @@ async def test_durable_replay_alias_policy_is_scoped_to_conflicting_row( session_key_value=replay_key, api_key_id="key-row-scope", instance_id="instance-replay", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-replay", model="gpt-5.4", @@ -1138,6 +1175,7 @@ async def test_durable_ordinary_rebind_ignores_unrelated_replay_alias( session_key_value=replay_key, api_key_id="key-row-scope-inverse", instance_id="instance-replay", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-replay", model="gpt-5.4", @@ -1159,6 +1197,7 @@ async def test_durable_ordinary_rebind_ignores_unrelated_replay_alias( session_key_value="first-ordinary-owner", api_key_id="key-row-scope-inverse", instance_id="instance-first", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-first", model="gpt-5.4", @@ -1180,6 +1219,7 @@ async def test_durable_ordinary_rebind_ignores_unrelated_replay_alias( session_key_value="second-ordinary-owner", api_key_id="key-row-scope-inverse", instance_id="instance-second", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-second", model="gpt-5.4", @@ -1216,6 +1256,7 @@ async def test_durable_bridge_ordinary_unanchored_key_does_not_override_shared_s session_key_value="sid-shared-ordinary", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-shared", model="gpt-5.4", @@ -1234,6 +1275,7 @@ async def test_durable_bridge_ordinary_unanchored_key_does_not_override_shared_s session_key_value="a" * 64, api_key_id=None, instance_id="instance-b", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-ordinary", model="gpt-5.4", @@ -1274,6 +1316,7 @@ async def test_durable_bridge_verified_replay_does_not_hide_specific_alias_confl session_key_value=replay_key, api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-replay", model="gpt-5.4", @@ -1287,6 +1330,7 @@ async def test_durable_bridge_verified_replay_does_not_hide_specific_alias_confl session_key_value="http_turn_response_owner", api_key_id=None, instance_id="instance-b", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-response", model="gpt-5.4", @@ -1334,6 +1378,7 @@ async def test_durable_bridge_turn_state_lookup_does_not_fall_back_to_canonical_ session_key_value="sid-123", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-1", model="gpt-5.4", @@ -1375,6 +1420,7 @@ async def test_durable_bridge_turn_state_proof_does_not_accept_latest_state_with session_key_value="sid-latest-only", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-1", model="gpt-5.4", @@ -1402,6 +1448,7 @@ async def test_durable_bridge_stale_owner_cannot_register_turn_state_after_epoch session_key_value="sid-stale-alias", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-1", model="gpt-5.4", @@ -1415,6 +1462,7 @@ async def test_durable_bridge_stale_owner_cannot_register_turn_state_after_epoch session_key_value="sid-stale-alias", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-2", model="gpt-5.4", @@ -1456,6 +1504,7 @@ async def test_durable_bridge_claim_renews_same_owner_epoch( session_key_value="sid-123", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -1470,6 +1519,7 @@ async def test_durable_bridge_claim_renews_same_owner_epoch( session_key_value="sid-123", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -1494,6 +1544,7 @@ async def test_durable_bridge_account_change_advances_epoch_to_fence_stale_relea session_key_value="sid-account-change", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -1508,6 +1559,7 @@ async def test_durable_bridge_account_change_advances_epoch_to_fence_stale_relea session_key_value="sid-account-change", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-2", model="gpt-5.4", @@ -1543,6 +1595,7 @@ async def test_durable_bridge_forced_generation_advance_fences_same_account_stal session_key_value="sid-forced-generation", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -1557,6 +1610,7 @@ async def test_durable_bridge_forced_generation_advance_fences_same_account_stal session_key_value="sid-forced-generation", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -1592,6 +1646,7 @@ async def test_durable_bridge_clear_response_anchor_nulls_anchor_fields_but_keep session_key_value="sid-clear-anchor", api_key_id="key-1", instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.6-sol", @@ -1668,6 +1723,7 @@ async def test_durable_bridge_clear_response_anchor_is_noop_after_epoch_advance( session_key_value="sid-clear-anchor-stale-epoch", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.6-sol", @@ -1684,6 +1740,7 @@ async def test_durable_bridge_clear_response_anchor_is_noop_after_epoch_advance( session_key_value="sid-clear-anchor-stale-epoch", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.6-sol", @@ -1715,6 +1772,7 @@ async def test_durable_bridge_claim_takes_over_after_release( session_key_value="sid-123", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -1735,6 +1793,7 @@ async def test_durable_bridge_claim_takes_over_after_release( session_key_value="sid-123", api_key_id=None, instance_id="instance-b", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -1759,6 +1818,7 @@ async def test_durable_bridge_release_without_draining_marks_session_closed( session_key_value="sid-closed", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -1784,6 +1844,7 @@ async def test_durable_bridge_release_without_draining_marks_session_closed( session_key_value="sid-closed", api_key_id=None, instance_id="instance-b", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -1806,6 +1867,7 @@ async def test_durable_bridge_takeover_clears_stale_recovery_anchor_for_fresh_se session_key_value="sid-reset", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -1826,6 +1888,7 @@ async def test_durable_bridge_takeover_clears_stale_recovery_anchor_for_fresh_se session_key_value="sid-reset", api_key_id=None, instance_id="instance-b", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-2", model="gpt-5.4", @@ -1849,6 +1912,7 @@ async def test_durable_bridge_same_account_closed_takeover_preserves_restart_anc session_key_value="sid-restart", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -1869,6 +1933,7 @@ async def test_durable_bridge_same_account_closed_takeover_preserves_restart_anc session_key_value="sid-restart", api_key_id=None, instance_id="instance-b", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -1892,6 +1957,7 @@ async def test_durable_bridge_takeover_preserves_existing_anchor_when_replacemen session_key_value="sid-preserve", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -1912,6 +1978,7 @@ async def test_durable_bridge_takeover_preserves_existing_anchor_when_replacemen session_key_value="sid-preserve", api_key_id=None, instance_id="instance-b", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -1935,6 +2002,7 @@ async def test_durable_bridge_previous_response_records_completed_input_prefix( session_key_value="sid-prefix", api_key_id="key-1", instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -1985,6 +2053,7 @@ async def test_durable_bridge_pending_tool_calls_are_bound_to_response_id( session_key_value="sid-manifest-response", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.6-sol", @@ -2037,6 +2106,7 @@ async def test_durable_bridge_takeover_with_account_change_clears_stale_aliases( session_key_value="sid-alias-reset", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -2074,6 +2144,7 @@ async def test_durable_bridge_takeover_with_account_change_clears_stale_aliases( session_key_value="sid-alias-reset", api_key_id=None, instance_id="instance-b", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-2", model="gpt-5.4", @@ -2128,6 +2199,7 @@ async def test_durable_bridge_lookup_active_lease_survives_request_lookup( session_key_value="http_turn_1", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -2164,6 +2236,7 @@ async def test_durable_bridge_lookup_falls_back_to_latest_turn_state_when_alias_ session_key_value="thread-123", api_key_id="key-1", instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -2220,6 +2293,7 @@ async def test_durable_bridge_lookup_falls_back_to_latest_response_id_when_alias session_key_value="thread-123", api_key_id="key-1", instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -2275,6 +2349,7 @@ async def test_mark_instance_draining_keeps_current_owner_lease_active( session_key_value="sid-draining", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -2323,6 +2398,7 @@ async def test_startup_purges_owned_bridge_rows( session_key_value="sid-restart", api_key_id=None, instance_id="instance-a", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", @@ -2358,6 +2434,156 @@ async def test_startup_purges_owned_bridge_rows( assert sticky is not None +@pytest.mark.asyncio +async def test_startup_closes_same_instance_previous_process_epoch_rows( + coordinator: DurableBridgeSessionCoordinator, + async_session_factory: Callable[[], AsyncSession], +) -> None: + previous_process = await coordinator.claim_live_session( + session_key_kind="session_header", + session_key_value="sid-reused-container", + api_key_id=None, + instance_id="container-74e8e7cda9fb", + owner_process_epoch="boot-a", + lease_ttl_seconds=120.0, + account_id="acc-1", + model="gpt-5.6-luna", + service_tier=None, + latest_turn_state="http_turn_reused_container", + latest_response_id="resp_reused_container", + allow_takeover=True, + ) + await coordinator.register_session_header( + session_id=previous_process.session_id, + api_key_id=None, + session_header="sid-reused-container", + ) + current_process = await coordinator.claim_live_session( + session_key_kind="session_header", + session_key_value="sid-current-process", + api_key_id=None, + instance_id="container-74e8e7cda9fb", + owner_process_epoch="boot-b", + lease_ttl_seconds=120.0, + account_id="acc-1", + model="gpt-5.6-luna", + service_tier=None, + latest_turn_state="http_turn_current_process", + latest_response_id="resp_current_process", + allow_takeover=True, + ) + + retired = await coordinator.purge_owned_sessions_on_startup( + instance_id="container-74e8e7cda9fb", + owner_process_epoch="boot-b", + ownerless_cutoff=utcnow() - timedelta(seconds=60), + ) + + assert retired == 1 + current_lookup = await coordinator.lookup_request_targets( + session_key_kind="session_header", + session_key_value="sid-current-process", + api_key_id=None, + turn_state="http_turn_current_process", + session_header="sid-current-process", + previous_response_id="resp_current_process", + ) + assert current_lookup is not None + assert current_lookup.session_id == current_process.session_id + assert current_lookup.owner_process_epoch == "boot-b" + async with async_session_factory() as session: + retired_row = await session.get(HttpBridgeSessionRecord, previous_process.session_id) + retired_aliases = list( + ( + await session.execute( + select(HttpBridgeSessionAlias).where( + HttpBridgeSessionAlias.session_id == previous_process.session_id, + ) + ) + ) + .scalars() + .all() + ) + assert retired_row is not None + assert retired_row.state == HttpBridgeSessionState.CLOSED + assert retired_row.owner_instance_id is None + assert retired_row.closed_at is not None + assert retired_aliases == [] + + +@pytest.mark.asyncio +async def test_startup_retirement_reclaim_does_not_resurrect_dead_anchors( + coordinator: DurableBridgeSessionCoordinator, + async_session_factory: Callable[[], AsyncSession], +) -> None: + previous_process = await coordinator.claim_live_session( + session_key_kind="session_header", + session_key_value="sid-stable-reclaim", + api_key_id=None, + instance_id="container-stable", + owner_process_epoch="boot-a", + lease_ttl_seconds=120.0, + account_id="acc-stable", + model="gpt-5.6-luna", + service_tier=None, + latest_turn_state="http_turn_dead_anchor", + latest_response_id="resp_dead_anchor", + allow_takeover=True, + ) + await coordinator.register_session_header( + session_id=previous_process.session_id, + api_key_id=None, + session_header="sid-stable-reclaim", + ) + await coordinator.register_previous_response_id( + session_id=previous_process.session_id, + api_key_id=None, + instance_id="container-stable", + owner_epoch=previous_process.owner_epoch, + response_id="resp_dead_anchor", + lease_ttl_seconds=120.0, + input_item_count=7, + input_full_fingerprint="d" * 64, + pending_tool_calls={"call_dead": "function_call"}, + ) + + retired = await coordinator.purge_owned_sessions_on_startup( + instance_id="container-stable", + owner_process_epoch="boot-b", + ownerless_cutoff=utcnow() - timedelta(seconds=60), + ) + reclaimed = await coordinator.claim_live_session( + session_key_kind="session_header", + session_key_value="sid-stable-reclaim", + api_key_id=None, + instance_id="container-stable", + owner_process_epoch="boot-b", + lease_ttl_seconds=120.0, + account_id="acc-stable", + model="gpt-5.6-luna", + service_tier=None, + latest_turn_state=None, + latest_response_id=None, + allow_takeover=True, + ) + + assert retired == 1 + assert reclaimed.session_id == previous_process.session_id + assert reclaimed.latest_turn_state is None + assert reclaimed.latest_response_id is None + assert reclaimed.latest_input_item_count is None + assert reclaimed.latest_input_full_fingerprint is None + assert reclaimed.latest_pending_tool_calls is None + async with async_session_factory() as session: + row = await session.get(HttpBridgeSessionRecord, previous_process.session_id) + assert row is not None + assert row.latest_turn_state is None + assert row.latest_response_id is None + assert row.latest_input_item_count is None + assert row.latest_input_full_fingerprint is None + assert row.latest_pending_tool_calls_json is None + + @pytest.mark.asyncio async def test_startup_retains_verified_replay_alias_as_ownerless_restart_proof( coordinator: DurableBridgeSessionCoordinator, @@ -2368,6 +2594,7 @@ async def test_startup_retains_verified_replay_alias_as_ownerless_restart_proof( session_key_value="sid-shared-restart", api_key_id=None, instance_id="instance-shared", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-retired", model="gpt-5.4", @@ -2387,6 +2614,7 @@ async def test_startup_retains_verified_replay_alias_as_ownerless_restart_proof( session_key_value=replay_key, api_key_id=None, instance_id="instance-restarting", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-recovered", model="gpt-5.4", @@ -2408,7 +2636,7 @@ async def test_startup_retains_verified_replay_alias_as_ownerless_restart_proof( await session.execute( update(HttpBridgeSessionRecord) .where(HttpBridgeSessionRecord.id == replay.session_id) - .values(last_seen_at=retained_time) + .values(last_seen_at=retained_time, owner_process_epoch=None) ) await session.commit() @@ -2418,6 +2646,7 @@ async def test_startup_retains_verified_replay_alias_as_ownerless_restart_proof( session_key_value=stale_key, api_key_id=None, instance_id="instance-restarting", + owner_process_epoch="test-process", lease_ttl_seconds=120.0, account_id="acc-stale-recovered", model="gpt-5.4", @@ -2431,7 +2660,7 @@ async def test_startup_retains_verified_replay_alias_as_ownerless_restart_proof( await session.execute( update(HttpBridgeSessionRecord) .where(HttpBridgeSessionRecord.id == stale_replay.session_id) - .values(last_seen_at=stale_time, lease_expires_at=stale_time) + .values(last_seen_at=stale_time, lease_expires_at=stale_time, owner_process_epoch=None) ) await session.commit() @@ -2665,6 +2894,7 @@ async def test_startup_preserves_recent_ownerless_drain_rows( session_key_value="sid-fresh-drain", api_key_id=None, instance_id="instance-draining", + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-1", model="gpt-5.4", diff --git a/tests/unit/test_http_bridge_cancel_drain.py b/tests/unit/test_http_bridge_cancel_drain.py index b69124d3a..3cb9b0b77 100644 --- a/tests/unit/test_http_bridge_cancel_drain.py +++ b/tests/unit/test_http_bridge_cancel_drain.py @@ -963,6 +963,7 @@ async def test_response_created_does_not_promote_in_progress_durable_anchor() -> session_key_value="thread-undo-edit", api_key_id=None, instance_id=instance_id, + owner_process_epoch="test-process", lease_ttl_seconds=60.0, account_id="acc-undo-edit", model="gpt-5.5", diff --git a/tests/unit/test_proxy_http_bridge.py b/tests/unit/test_proxy_http_bridge.py index c69c14076..64a051791 100644 --- a/tests/unit/test_proxy_http_bridge.py +++ b/tests/unit/test_proxy_http_bridge.py @@ -6,11 +6,12 @@ import json import logging import pickle +import subprocess import time from collections import deque from contextlib import nullcontext from dataclasses import replace -from datetime import datetime, timedelta, timezone +from datetime import UTC, datetime, timedelta, timezone from types import SimpleNamespace from typing import Any, cast from unittest.mock import AsyncMock, Mock @@ -50,7 +51,7 @@ is_http_bridge_account_neutral_replay, make_http_bridge_account_neutral_replay_key, ) -from app.modules.proxy.durable_bridge_coordinator import DurableBridgeSessionCoordinator +from app.modules.proxy.durable_bridge_coordinator import DurableBridgeLookup, DurableBridgeSessionCoordinator from app.modules.proxy.durable_bridge_repository import ( DurableBridgeAliasRegistration, DurableBridgeAliasRegistrationReceipt, @@ -61,6 +62,70 @@ pytestmark = pytest.mark.unit +def _durable_owner_lookup(*, process_epoch: str, lease_expires_at: datetime) -> DurableBridgeLookup: + return DurableBridgeLookup( + session_id="dead-owner-session", + canonical_kind="session_header", + canonical_key="dead-owner-key", + api_key_scope="anonymous", + account_id="acc-bridge", + owner_instance_id="bridge-instance", + owner_process_epoch=process_epoch, + owner_epoch=2, + lease_expires_at=lease_expires_at, + state=HttpBridgeSessionState.ACTIVE, + latest_turn_state="turn-state", + latest_response_id="resp-dead-owner", + ) + + +def test_http_bridge_dead_owner_epoch_uses_standard_previous_response_not_found_contract() -> None: + now = datetime.now(UTC).replace(tzinfo=None) + stale = http_bridge_streaming_module._http_bridge_durable_owner_is_dead( + _durable_owner_lookup(process_epoch="boot-a", lease_expires_at=now + timedelta(minutes=5)), + current_instance="bridge-instance", + current_process_epoch="boot-b", + ) + transient = http_bridge_streaming_module._http_bridge_durable_owner_is_dead( + _durable_owner_lookup(process_epoch="boot-b", lease_expires_at=now + timedelta(minutes=5)), + current_instance="bridge-instance", + current_process_epoch="boot-b", + ) + + assert stale is True + assert transient is False + + terminal = cast( + dict[str, Any], + http_bridge_streaming_module._http_bridge_dead_owner_previous_response_not_found_terminal( + previous_response_id="resp-dead-owner", + response_id="resp-dead-owner", + ), + ) + terminal_error = cast(dict[str, Any], cast(dict[str, Any], terminal["response"])["error"]) + assert terminal_error["type"] == "invalid_request_error" + assert terminal_error["code"] == "previous_response_not_found" + assert terminal_error["param"] == "previous_response_id" + proxy_error = http_bridge_streaming_module._http_bridge_dead_owner_previous_response_not_found_proxy_error( + previous_response_id="resp-dead-owner", + ) + assert proxy_error.status_code == 400 + assert proxy_error.payload["error"]["type"] == "invalid_request_error" + assert proxy_error.payload["error"]["code"] == "previous_response_not_found" + assert proxy_error.payload["error"]["param"] == "previous_response_id" + + +def test_http_bridge_rejected_dead_owner_recovery_code_is_not_emitted_from_app() -> None: + result = subprocess.run( + ["git", "grep", "-n", "bridge_continuity_recovery_required", "--", "app/"], + check=False, + capture_output=True, + text=True, + ) + + assert result.returncode == 1, result.stdout + + @pytest.fixture(autouse=True) def _share_proxy_dashboard_settings(monkeypatch: pytest.MonkeyPatch) -> None: class _SettingsCache: @@ -20089,6 +20154,166 @@ async def fake_stream_events( account_neutral_classifier.assert_called_once() +@pytest.mark.asyncio +async def test_stream_via_http_bridge_recovers_dead_owner_with_replayable_full_resend( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + settings = _make_app_settings(http_responses_session_bridge_instance_id="bridge-instance") + owner_metadata: proxy_service.JsonValue = {"turn_id": "turn-owner"} + historical_input: list[proxy_service.JsonValue] = [ + { + "role": "user", + "content": [{"type": "input_text", "text": "old question"}], + "internal_chat_message_metadata_passthrough": owner_metadata, + }, + { + "type": "function_call", + "id": "fc_owner", + "call_id": "call_old", + "name": "lookup", + "arguments": "{}", + "internal_chat_message_metadata_passthrough": owner_metadata, + }, + ] + payload = proxy_service.ResponsesRequest.model_validate( + { + "model": "gpt-5.4", + "instructions": "hi", + "previous_response_id": "resp_completed_anchor", + "input": [ + *historical_input, + { + "type": "function_call_output", + "call_id": "call_old", + "output": "old output", + "internal_chat_message_metadata_passthrough": owner_metadata, + }, + { + "type": "message", + "id": "msg_owner", + "role": "assistant", + "status": "completed", + "phase": "final_answer", + "content": [{"type": "output_text", "text": "old answer"}], + "internal_chat_message_metadata_passthrough": owner_metadata, + }, + { + "role": "user", + "content": [{"type": "input_text", "text": "next question"}], + "internal_chat_message_metadata_passthrough": {"turn_id": "turn-next"}, + }, + ], + } + ) + durable_lookup = proxy_service.DurableBridgeLookup( + session_id="durable-dead-owner", + canonical_kind="session_header", + canonical_key="sid-dead-owner", + api_key_scope="__anonymous__", + account_id="acc-owner", + owner_instance_id="bridge-instance", + owner_process_epoch="boot-old", + owner_epoch=1, + lease_expires_at=datetime.now(timezone.utc) + timedelta(seconds=60), + state=HttpBridgeSessionState.ACTIVE, + latest_turn_state="sid-dead-owner", + latest_response_id="resp_completed_anchor", + latest_input_item_count=len(historical_input), + latest_input_full_fingerprint=proxy_service._fingerprint_input_items(historical_input), + model="gpt-5.4", + ) + captured_request_states: list[proxy_service._WebSocketRequestState] = [] + captured_text_data: list[str] = [] + captured_keys: list[proxy_service._HTTPBridgeSessionKey] = [] + captured_kwargs: list[dict[str, Any]] = [] + + async def fake_get_or_create( + key: proxy_service._HTTPBridgeSessionKey, + **kwargs: Any, + ) -> proxy_service._HTTPBridgeSession: + captured_keys.append(key) + captured_kwargs.append(kwargs) + session = _make_bridge_session(key=key, key_value=key.affinity_key) + session.account = cast(Any, SimpleNamespace(id="acc-fallback", status=AccountStatus.ACTIVE)) + session.request_model = payload.model + return session + + async def fake_stream_events( + _session: proxy_service._HTTPBridgeSession, + *, + request_state: proxy_service._WebSocketRequestState, + text_data: str, + **_kwargs: Any, + ): + captured_request_states.append(request_state) + captured_text_data.append(text_data) + yield 'data: {"type":"response.created","response":{"id":"resp_recovered"}}\n\n' + yield 'data: {"type":"response.completed","response":{"id":"resp_recovered"}}\n\n' + + monkeypatch.setattr( + proxy_service, + "get_settings_cache", + lambda: cast( + Any, + SimpleNamespace( + get=AsyncMock( + return_value=SimpleNamespace( + sticky_threads_enabled=False, + openai_cache_affinity_max_age_seconds=1800, + http_responses_session_bridge_prompt_cache_idle_ttl_seconds=3600, + http_responses_session_bridge_gateway_safe_mode=False, + ) + ) + ), + ), + ) + monkeypatch.setattr(proxy_service, "get_settings", lambda: settings) + monkeypatch.setattr(http_bridge_streaming_module, "http_bridge_owner_process_epoch", lambda: "boot-new") + monkeypatch.setattr(service._durable_bridge, "lookup_request_targets", AsyncMock(return_value=durable_lookup)) + monkeypatch.setattr(service, "_http_bridge_has_live_local_session", AsyncMock(return_value=False)) + monkeypatch.setattr(service, "_http_bridge_can_forward_to_active_owner", AsyncMock(return_value=False)) + monkeypatch.setattr(service, "_resolve_file_account_for_responses", AsyncMock(return_value=None)) + monkeypatch.setattr(service, "_resolve_websocket_previous_response_owner", AsyncMock(return_value="acc-owner")) + monkeypatch.setattr(service, "_get_or_create_http_bridge_session", fake_get_or_create) + monkeypatch.setattr(service, "_stream_http_bridge_session_events", fake_stream_events) + + chunks = [ + chunk + async for chunk in service._stream_via_http_bridge( + payload, + headers={"x-codex-session-id": "sid-dead-owner"}, + codex_session_affinity=True, + propagate_http_errors=True, + openai_cache_affinity=True, + api_key=None, + api_key_reservation=None, + suppress_text_done_events=False, + idle_ttl_seconds=120.0, + codex_idle_ttl_seconds=1800.0, + max_sessions=8, + queue_limit=4, + ) + ] + + assert chunks == [ + 'data: {"type":"response.created","response":{"id":"resp_recovered"}}\n\n', + 'data: {"type":"response.completed","response":{"id":"resp_recovered"}}\n\n', + ] + assert all("response.failed" not in chunk for chunk in chunks) + assert len(captured_keys) == 1 + assert is_http_bridge_account_neutral_replay( + kind=captured_keys[0].affinity_kind, + key=captured_keys[0].affinity_key, + ) + assert captured_kwargs[0]["previous_response_id"] is None + assert captured_kwargs[0]["durable_lookup"] is None + assert captured_request_states[0].previous_response_id is None + replay_payload = json.loads(captured_text_data[0]) + assert "previous_response_id" not in replay_payload + assert replay_payload["input"][-1]["content"] == [{"type": "input_text", "text": "next question"}] + + @pytest.mark.asyncio async def test_durable_model_transition_preserves_owner_provenance_when_replacing_retired_session( monkeypatch: pytest.MonkeyPatch, @@ -20157,6 +20382,7 @@ async def fail_first_session_before_output( request_state.awaiting_response_created = False request_state.response_create_gate = None request_state.response_create_gate_acquired = False + assert request_state.durable_owner_dead is False raise gate_timeout_error yield "" @@ -22489,6 +22715,106 @@ async def test_http_bridge_reader_failure_keeps_waiter_count_when_draining_reque record_failure.assert_awaited_once_with(session, detail="stream_incomplete") +@pytest.mark.asyncio +async def test_http_bridge_repeated_zero_event_idle_timeouts_poison_anchor_with_waiter( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + session = _make_bridge_session( + key_value="bridge-anchor-poison", + pending_requests=deque([_make_eventless_http_bridge_owner()]), + queued_request_count=1, + ) + session.admission_waiter_count = 1 + session.durable_session_id = "durable-anchor-poison" + session.durable_owner_epoch = 3 + durable_bridge = SimpleNamespace( + lookup_retry_circuit=AsyncMock(return_value=None), + persist_retry_circuit=AsyncMock(), + rebind_session_account=AsyncMock(return_value=True), + ) + service._durable_bridge = durable_bridge + fail_pending = AsyncMock() + retire = AsyncMock() + monkeypatch.setattr(service, "_fail_pending_websocket_requests", fail_pending) + monkeypatch.setattr(service, "_retire_stale_pending_http_bridge_session", retire) + + for failure_number in range(1, 8): + retired = await service._fail_http_bridge_reader_and_maybe_retire( + session, + error_code="stream_idle_timeout", + error_message="idle timeout", + ) + assert retired is (failure_number == 7) + + durable_bridge.rebind_session_account.assert_awaited_once_with( + session_id="durable-anchor-poison", + api_key_id=None, + instance_id=proxy_service.get_settings().http_responses_session_bridge_instance_id, + owner_epoch=3, + account_id="acc-bridge", + clear_continuity=True, + ) + retire.assert_awaited_once_with( + session, + detail="repeated_zero_event_idle_timeout", + response_events_seen=0, + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("clear_outcome", [False, RuntimeError("clear failed")]) +async def test_http_bridge_anchor_poisoning_waits_when_durable_clear_fails( + clear_outcome: bool | RuntimeError, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + session = _make_bridge_session( + key_value="bridge-anchor-poison-clear-fails", + pending_requests=deque([_make_eventless_http_bridge_owner()]), + queued_request_count=1, + ) + session.admission_waiter_count = 1 + session.durable_session_id = "durable-anchor-poison-clear-fails" + session.durable_owner_epoch = 4 + rebind = ( + AsyncMock(side_effect=clear_outcome) + if isinstance(clear_outcome, RuntimeError) + else AsyncMock(return_value=clear_outcome) + ) + durable_bridge = SimpleNamespace( + lookup_retry_circuit=AsyncMock(return_value=None), + persist_retry_circuit=AsyncMock(), + rebind_session_account=rebind, + ) + service._durable_bridge = durable_bridge + fail_pending = AsyncMock() + retire = AsyncMock() + monkeypatch.setattr(service, "_fail_pending_websocket_requests", fail_pending) + monkeypatch.setattr(service, "_retire_stale_pending_http_bridge_session", retire) + + with caplog.at_level("WARNING"): + for _failure_number in range(1, 8): + retired = await service._fail_http_bridge_reader_and_maybe_retire( + session, + error_code="stream_idle_timeout", + error_message="idle timeout", + ) + + assert retired is False + durable_bridge.rebind_session_account.assert_awaited_once_with( + session_id="durable-anchor-poison-clear-fails", + api_key_id=None, + instance_id=proxy_service.get_settings().http_responses_session_bridge_instance_id, + owner_epoch=4, + account_id="acc-bridge", + clear_continuity=True, + ) + retire.assert_not_awaited() + assert "poisoned anchor" in caplog.text or "poisoned HTTP bridge continuity" in caplog.text + + @pytest.mark.asyncio async def test_http_bridge_eventless_timeout_force_retires_with_admission_waiter( monkeypatch: pytest.MonkeyPatch, diff --git a/tests/unit/test_settings_reference.py b/tests/unit/test_settings_reference.py index 37bedfa33..428b5057b 100644 --- a/tests/unit/test_settings_reference.py +++ b/tests/unit/test_settings_reference.py @@ -51,7 +51,10 @@ def _isolated_settings(**overrides: Any) -> Settings: # gate, issue #1535). Not a hardcoded default because the right congestion # threshold depends on pool size and workload mix, and 0-means-off is the P1 # default-off switch; the companion min-guarantee constant stayed hardcoded. -MAX_SETTINGS_FIELDS = 117 +# 117 -> 118: http_responses_session_bridge_anchor_poison_failure_threshold +# (bridge restart anchor poisoning). Not hardcoded because operators need a +# bounded deployment-specific poison threshold while recovery telemetry matures. +MAX_SETTINGS_FIELDS = 118 def test_generated_settings_reference_matches_code() -> None: