From d1f4fa4ac3929e29bb531716a1964c39a5839793 Mon Sep 17 00:00:00 2001 From: Apple Date: Fri, 28 Aug 2026 22:06:42 -0500 Subject: [PATCH] fix(proxy): quarantine poisoned bridge anchors at circuit open Source sidechat: 01a04ac6-27a1-74b3-88d3-f1ede3901184 Recovered from: 8c6ff6b6b4e76e453a2cfed29bd3b0ef0141a778 --- .../proxy/_service/http_bridge/helpers.py | 19 +- .../proxy/_service/http_bridge/quarantine.py | 20 +- .../_service/http_bridge/request_submit.py | 8 +- .../_service/http_bridge/retry_circuit.py | 62 +++- .../proxy/_service/http_bridge/streaming.py | 10 +- .../_service/http_bridge/upstream_events.py | 87 ++++- .../proposal.md | 33 ++ .../specs/responses-api-compat/spec.md | 75 ++++ .../tasks.md | 34 ++ tests/unit/test_proxy_http_bridge.py | 346 ++++++++++++++++++ 10 files changed, 675 insertions(+), 19 deletions(-) create mode 100644 openspec/changes/recover-poisoned-anchor-on-circuit-open/proposal.md create mode 100644 openspec/changes/recover-poisoned-anchor-on-circuit-open/specs/responses-api-compat/spec.md create mode 100644 openspec/changes/recover-poisoned-anchor-on-circuit-open/tasks.md diff --git a/app/modules/proxy/_service/http_bridge/helpers.py b/app/modules/proxy/_service/http_bridge/helpers.py index 6e9277f2a3..5635b74e34 100644 --- a/app/modules/proxy/_service/http_bridge/helpers.py +++ b/app/modules/proxy/_service/http_bridge/helpers.py @@ -5,7 +5,7 @@ import logging import sys import time -from collections.abc import Callable, Sequence +from collections.abc import Callable, Iterable, Sequence from dataclasses import dataclass, replace from hashlib import sha256 from ipaddress import ip_address @@ -863,6 +863,23 @@ def _http_bridge_retry_circuit_attempt_selection_for_pending_requests( return _HTTPBridgeRetryCircuitAttemptSelection(kind="ineligible" if attempt_seen else "absent") +def _http_bridge_continuity_bound_without_safe_replay(request_state: _WebSocketRequestState) -> bool: + """Return whether retrying would require replaying an unsafe continuation.""" + if request_state.previous_response_id is not None: + return not (request_state.fresh_upstream_request_is_retry_safe and request_state.fresh_upstream_request_text) + return request_state.hard_continuity_anchor and not ( + request_state.fresh_upstream_request_is_retry_safe and request_state.fresh_upstream_request_text + ) + + +def _http_bridge_abandonment_strands_requests(request_states: Iterable[Any]) -> bool: + """Return whether every request covered by an abandonment is stranded.""" + states = [state for state in request_states if state is not None] + if not states: + return False + return all(_http_bridge_continuity_bound_without_safe_replay(state) for state in states) + + def _http_bridge_session_has_admission_waiter(session: object | None) -> bool: """Keep a closed bridge registered while an unsent request owns its handoff.""" return session is not None and bool(getattr(session, "admission_waiter_count", 0)) diff --git a/app/modules/proxy/_service/http_bridge/quarantine.py b/app/modules/proxy/_service/http_bridge/quarantine.py index 64306533c1..b1e66157e7 100644 --- a/app/modules/proxy/_service/http_bridge/quarantine.py +++ b/app/modules/proxy/_service/http_bridge/quarantine.py @@ -35,6 +35,9 @@ _HTTP_BRIDGE_QUARANTINE_WEDGED_REATTACH_REASON = "reattach_missing_response_created" _HTTP_BRIDGE_QUARANTINE_REPEATED_EVENTLESS_REASON = "repeated_eventless_timeout" +# A retry circuit opened on an eventless poison-class failure. The anchor that +# caused the opening must not be re-injected into the next half-open probe. +_HTTP_BRIDGE_QUARANTINE_POISONED_ANCHOR_REASON = "retry_circuit_poisoned_anchor" @dataclass(slots=True) @@ -112,18 +115,29 @@ def _http_bridge_quarantine_generation(service: Any, key: _HTTPBridgeSessionKey) return entry.generation -def _quarantine_http_bridge_session(service: Any, session: _HTTPBridgeSession, *, reason: str) -> None: +def _quarantine_http_bridge_session( + service: Any, + session: _HTTPBridgeSession, + *, + reason: str, + minimum_seconds: float | None = None, +) -> None: """Quarantine a bridge session that has proven silent/wedged. Session-scoped only: no account-health writes happen here, and the entry is bounded by TTL, a registry size cap, and the healthy-completion clear. + ``minimum_seconds`` lets the retry circuit keep a poison quarantine alive + through its cooldown and half-open probe lease. """ now = time.monotonic() registry = _http_bridge_quarantine_registry(service) entry = registry.setdefault(session.key, _HTTPBridgeQuarantineEntry()) already_quarantined = entry.quarantined_until > now entry.generation += 1 - entry.quarantined_until = max(entry.quarantined_until, now + _HTTP_BRIDGE_QUARANTINE_TTL_SECONDS) + ttl_seconds = _HTTP_BRIDGE_QUARANTINE_TTL_SECONDS + if minimum_seconds is not None: + ttl_seconds = max(ttl_seconds, max(0.0, minimum_seconds)) + entry.quarantined_until = max(entry.quarantined_until, now + ttl_seconds) entry.last_touched_monotonic = now entry.reason = reason _prune_http_bridge_quarantine_registry(registry, now) @@ -135,7 +149,7 @@ def _quarantine_http_bridge_session(service: Any, session: _HTTPBridgeSession, * session.key, account_id=session.account.id, model=session.request_model, - detail=f"reason={reason}, ttl_seconds={_HTTP_BRIDGE_QUARANTINE_TTL_SECONDS:.0f}", + detail=f"reason={reason}, ttl_seconds={ttl_seconds:.0f}", cache_key_family=session.key.affinity_kind, model_class=_extract_model_class(session.request_model) if session.request_model else None, ) diff --git a/app/modules/proxy/_service/http_bridge/request_submit.py b/app/modules/proxy/_service/http_bridge/request_submit.py index 842f5cd4fd..263cc98370 100644 --- a/app/modules/proxy/_service/http_bridge/request_submit.py +++ b/app/modules/proxy/_service/http_bridge/request_submit.py @@ -78,6 +78,7 @@ from app.modules.proxy._service.http_bridge.helpers import ( _await_task_deferring_cancellation, _build_http_bridge_prewarm_text, + _http_bridge_abandonment_strands_requests, _http_bridge_durable_lease_ttl_seconds, _http_bridge_is_previous_response_owner_unavailable, _http_bridge_key_strength, @@ -2946,7 +2947,12 @@ async def _retire_stale_pending_http_bridge_session( # failure. Without this, only the admission-waiter reader # path could ever poison an anchor, and an anchored session # failing without waiters cooled down forever (issue #1830). - durable_cleared = await _abandon_durable_http_bridge_continuity(self, session, detail=poison_detail) + durable_cleared = await _abandon_durable_http_bridge_continuity( + self, + session, + detail=poison_detail, + settle_circuit=_http_bridge_abandonment_strands_requests(retired_request_states), + ) if not durable_cleared and session.durable_session_id is not None: # Keep failed waiterless clears visible in the same # poison-clear telemetry the admission-waiter path emits; diff --git a/app/modules/proxy/_service/http_bridge/retry_circuit.py b/app/modules/proxy/_service/http_bridge/retry_circuit.py index 1d0143cb95..332a0c9fa0 100644 --- a/app/modules/proxy/_service/http_bridge/retry_circuit.py +++ b/app/modules/proxy/_service/http_bridge/retry_circuit.py @@ -9,6 +9,10 @@ import anyio from app.core.metrics.prometheus import PROMETHEUS_AVAILABLE, http_bridge_retry_circuit_total +from app.modules.proxy._service.http_bridge.quarantine import ( + _HTTP_BRIDGE_QUARANTINE_POISONED_ANCHOR_REASON, + _quarantine_http_bridge_session, +) from app.modules.proxy._service.observability import _hash_identifier from app.modules.proxy._service.support import ( _HTTPBridgeResponseCreateAttempt, @@ -61,6 +65,11 @@ def _http_bridge_anchor_poison_detail(detail: str | None) -> str | None: return _HTTP_BRIDGE_ANCHOR_POISON_DETAILS.get(aliased) +def _http_bridge_poison_quarantine_minimum_seconds(cooldown_remaining: float) -> float: + """Keep poison quarantine live through cooldown and the half-open probe.""" + return max(0.0, cooldown_remaining) + _HTTP_BRIDGE_RETRY_CIRCUIT_HALF_OPEN_LEASE_SECONDS + + @dataclass(slots=True) class _HTTPBridgeRetryCircuitState: consecutive_failures: int = 0 @@ -610,6 +619,7 @@ async def _record_http_bridge_retry_circuit_failure( *, detail: str, attempt: _HTTPBridgeResponseCreateAttempt | None = None, + terminal_pre_response_frame: bool = False, ) -> 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: @@ -623,7 +633,7 @@ async def _record_http_bridge_retry_circuit_failure( attempt=scoped_attempt, detail=detail, ) - if scoped_attempt.disarmed or scoped_attempt.response_observed: + if scoped_attempt.disarmed or (scoped_attempt.response_observed and not terminal_pre_response_frame): return None await self._load_http_bridge_retry_circuit(session) @@ -634,10 +644,15 @@ async def _record_http_bridge_retry_circuit_failure( now = time.monotonic() duplicate_attempt: _HTTPBridgeResponseCreateAttempt | None = None state: _HTTPBridgeRetryCircuitState | None = None + poison_class_failure = _http_bridge_anchor_poison_detail(detail) is not None + quarantine_poisoned_anchor = False + quarantine_cooldown_remaining = 0.0 async with self._http_bridge_retry_circuit_lock: if scoped_attempt is not None and scoped_attempt.retry_circuit_failure_recorded: duplicate_attempt = scoped_attempt - elif scoped_attempt is not None and (scoped_attempt.disarmed or scoped_attempt.response_observed): + elif scoped_attempt is not None and ( + scoped_attempt.disarmed or (scoped_attempt.response_observed and not terminal_pre_response_frame) + ): return None else: state = self._http_bridge_retry_circuits.setdefault( @@ -660,6 +675,9 @@ async def _record_http_bridge_retry_circuit_failure( if detail == "clean_close": backoff = min(backoff, clean_close_max_backoff) state.cooldown_until = max(state.cooldown_until, now + backoff) + if poison_class_failure: + quarantine_poisoned_anchor = True + quarantine_cooldown_remaining = max(0.0, state.cooldown_until - now) if PROMETHEUS_AVAILABLE and http_bridge_retry_circuit_total is not None: http_bridge_retry_circuit_total.labels(outcome="opened").inc() logger.warning( @@ -678,18 +696,54 @@ async def _record_http_bridge_retry_circuit_failure( detail=detail, ) assert state is not None + if quarantine_poisoned_anchor: + _quarantine_http_bridge_session( + self, + session, + reason=_HTTP_BRIDGE_QUARANTINE_POISONED_ANCHOR_REASON, + minimum_seconds=_http_bridge_poison_quarantine_minimum_seconds(quarantine_cooldown_remaining), + ) try: await self._persist_http_bridge_retry_circuit(session, state) + merged_quarantine_cooldown_remaining = 0.0 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) consecutive_failures = state.consecutive_failures + if poison_class_failure and consecutive_failures >= threshold and not quarantine_poisoned_anchor: + # A durable conflict merge can open the circuit even when + # this replica's local count was below the threshold. + merged_quarantine_cooldown_remaining = max(0.0, state.cooldown_until - time.monotonic()) + if merged_quarantine_cooldown_remaining or ( + poison_class_failure and consecutive_failures >= threshold and not quarantine_poisoned_anchor + ): + _quarantine_http_bridge_session( + self, + session, + reason=_HTTP_BRIDGE_QUARANTINE_POISONED_ANCHOR_REASON, + minimum_seconds=_http_bridge_poison_quarantine_minimum_seconds( + merged_quarantine_cooldown_remaining + ), + ) return consecutive_failures finally: if scoped_attempt is not None and scoped_attempt.retry_circuit_failure_settled is not None: scoped_attempt.retry_circuit_failure_settled.set() - async def _clear_http_bridge_retry_circuit(self: Any, session: _HTTPBridgeSession) -> None: + async def _clear_http_bridge_retry_circuit( + self: Any, + session: _HTTPBridgeSession, + *, + settle_unfenced: bool = False, + ) -> None: + """Settle a hard key's retry circuit. + + ``settle_unfenced`` is asserted by callers that have removed the cause + of the circuit rather than merely outlived it. The normal version + fence protects a durable row another replica may have created, while + cause-removal paths may also clear a circuit opened before its first + durable write completed. + """ if session.key.strength != "hard": return @@ -705,7 +759,7 @@ async def _clear_http_bridge_retry_circuit(self: Any, session: _HTTPBridgeSessio # concurrently, so leave the durable row untouched when no state was # observed. Preserve the existing best-effort clear on read failures, # which is still useful for settling a row after a transient outage. - if durable_load_succeeded and (state is None or expected_updated_at_epoch is None): + if durable_load_succeeded and (state is None or (expected_updated_at_epoch is None and not settle_unfenced)): return try: # Clearing is idempotent and must be attempted even when the diff --git a/app/modules/proxy/_service/http_bridge/streaming.py b/app/modules/proxy/_service/http_bridge/streaming.py index 90170b136e..81d12bd530 100644 --- a/app/modules/proxy/_service/http_bridge/streaming.py +++ b/app/modules/proxy/_service/http_bridge/streaming.py @@ -73,6 +73,7 @@ ) from app.modules.proxy._service.http_bridge.helpers import ( _effective_http_bridge_idle_ttl_seconds, + _http_bridge_continuity_bound_without_safe_replay, _http_bridge_durable_lease_ttl_seconds, _http_bridge_durable_lookup_allows_turn_state_takeover, _http_bridge_is_context_overflow_error, @@ -246,15 +247,6 @@ _RESPONSE_CREATE_GATE_RETRY_SLEEP_SECONDS = 10.0 -def _http_bridge_continuity_bound_without_safe_replay(request_state: _WebSocketRequestState) -> bool: - """Return whether retrying would require replaying an unsafe continuation.""" - if request_state.previous_response_id is not None: - return not (request_state.fresh_upstream_request_is_retry_safe and request_state.fresh_upstream_request_text) - return request_state.hard_continuity_anchor and not ( - request_state.fresh_upstream_request_is_retry_safe and request_state.fresh_upstream_request_text - ) - - def _http_bridge_durable_recovery_predecessor_proven(request_state: _WebSocketRequestState) -> bool: """Return whether the operation has a durable predecessor anchor.""" return request_state.previous_response_id is not None or request_state.operation_parent_response_id is not None diff --git a/app/modules/proxy/_service/http_bridge/upstream_events.py b/app/modules/proxy/_service/http_bridge/upstream_events.py index 0137e4da19..452cc85a16 100644 --- a/app/modules/proxy/_service/http_bridge/upstream_events.py +++ b/app/modules/proxy/_service/http_bridge/upstream_events.py @@ -61,6 +61,8 @@ from app.modules.proxy._service.http_bridge.helpers import ( _HTTP_BRIDGE_MISSING_RESPONSE_CREATED_TIMEOUT_DETAIL, _await_task_deferring_cancellation, + _http_bridge_abandonment_strands_requests, + _http_bridge_continuity_bound_without_safe_replay, _http_bridge_durable_lease_ttl_seconds, _http_bridge_eventless_precreated_deadline, _http_bridge_request_budget_seconds, @@ -76,6 +78,7 @@ _record_http_bridge_quarantine_wedged_pending, ) from app.modules.proxy._service.http_bridge.retry_circuit import ( + _HTTP_BRIDGE_RETRY_CIRCUIT_FAILURE_THRESHOLD, _http_bridge_anchor_poison_detail, ) from app.modules.proxy._service.http_bridge.service_stubs import ( @@ -968,6 +971,7 @@ async def _abandon_durable_http_bridge_continuity( session: "_HTTPBridgeSession", *, detail: str = "repeated_zero_event_idle_timeout", + settle_circuit: bool = False, ) -> bool: """Clear durable continuity before retiring a repeatedly poisoned bridge. @@ -1008,6 +1012,13 @@ async def _abandon_durable_http_bridge_continuity( cache_key_family=session.key.affinity_kind, model_class=_extract_model_class(session.request_model) if session.request_model else None, ) + # Retirement and close funnels may reach this helper while a verified + # stale-anchor replay is still in flight. Those callers opt in only when + # every covered request is stranded; a replay that still holds a safe + # anchor-free body needs the circuit generation fence until dispatch. + if not settle_circuit: + return True + await service._clear_http_bridge_retry_circuit(session, settle_unfenced=True) return True @@ -1194,7 +1205,12 @@ async def _fail_http_bridge_reader_and_maybe_retire( ): poison_detail = poison_candidate_detail if poison_detail is not None: - durable_cleared = await _abandon_durable_http_bridge_continuity(self, session, detail=poison_detail) + durable_cleared = await _abandon_durable_http_bridge_continuity( + self, + session, + detail=poison_detail, + settle_circuit=_http_bridge_abandonment_strands_requests(pending_request_states), + ) if durable_cleared: await self._retire_stale_pending_http_bridge_session( session, @@ -2021,7 +2037,9 @@ async def _process_parsed_http_bridge_upstream_event( if is_missing_tool_output_event else "stream_incomplete" ) + grouped_retry_detail = "stream_incomplete" if is_previous_response_not_found_event else grouped_error_reason grouped_terminal_events = [] + grouped_terminal_strike_failures: list[int] = [] for grouped_request_state in grouped_previous_response_request_states: grouped_request_state.error_http_status_override = 502 ( @@ -2045,6 +2063,15 @@ async def _process_parsed_http_bridge_upstream_event( grouped_operation_state, ) ) + if grouped_request_state.response_event_count == 0: + grouped_failure_count = await self._record_http_bridge_retry_circuit_failure( + session, + detail=grouped_retry_detail, + attempt=grouped_request_state.response_create_attempt, + terminal_pre_response_frame=True, + ) + if grouped_failure_count is not None: + grouped_terminal_strike_failures.append(grouped_failure_count) append_terminal_batch = getattr( getattr(self, "_http_bridge_operation_event_batcher", None), @@ -2192,6 +2219,16 @@ async def persist_grouped_terminal_events() -> Exception | None: raise grouped_cancellation if grouped_error is not None: raise grouped_error + if any( + failures >= _HTTP_BRIDGE_RETRY_CIRCUIT_FAILURE_THRESHOLD + for failures in grouped_terminal_strike_failures + ): + await _abandon_durable_http_bridge_continuity( + self, + session, + detail="repeated_zero_event_stream_incomplete", + settle_circuit=True, + ) return if len(grouped_previous_response_request_states) == 1 and terminal_request_state is None: @@ -2949,6 +2986,39 @@ async def persist_grouped_terminal_events() -> Exception | None: if retried: return + terminal_strike_failures: int | None = None + terminal_poison_detail: str | None = None + if settlement_event_type in {"response.failed", "response.incomplete", "error"}: + terminal_error = ( + settlement_event.error + if settlement_event_type == "error" and settlement_event is not None + else settlement_event.response.error + if settlement_event is not None and settlement_event.response is not None + else None + ) + terminal_error_code = _normalize_error_code( + terminal_error.code if terminal_error else None, + terminal_error.type if terminal_error else None, + ) + terminal_retry_detail = terminal_error_code + if is_previous_response_not_found_event and terminal_retry_detail not in { + "stream_incomplete", + "stream_idle_timeout", + }: + terminal_retry_detail = "stream_incomplete" + if ( + terminal_request_state is not None + and terminal_request_state.response_event_count == 0 + and _http_bridge_continuity_bound_without_safe_replay(terminal_request_state) + ): + terminal_strike_failures = await self._record_http_bridge_retry_circuit_failure( + session, + detail=terminal_retry_detail or "stream_incomplete", + attempt=terminal_request_state.response_create_attempt, + terminal_pre_response_frame=True, + ) + terminal_poison_detail = _http_bridge_anchor_poison_detail(terminal_retry_detail) + matched_event_queue = ( completed_event_queue if completed_event_queue_claimed and matched_request_state is terminal_request_state @@ -3047,6 +3117,21 @@ async def persist_grouped_terminal_events() -> Exception | None: # awaited recovery work before it rechecks this scope. completed_delivery_scope.terminal_enqueued = True + if ( + terminal_poison_detail is not None + and terminal_strike_failures is not None + and terminal_strike_failures >= _HTTP_BRIDGE_RETRY_CIRCUIT_FAILURE_THRESHOLD + ): + # Terminal frames settle without entering the retirement funnel. + # Clear the durable anchor after publishing the failure so a client + # resend is covered by the quarantine already armed by the strike. + await _abandon_durable_http_bridge_continuity( + self, + session, + detail=terminal_poison_detail, + settle_circuit=True, + ) + if settlement_event_type in {"response.failed", "response.incomplete", "error"}: error_code = None if settlement_event_type == "error": diff --git a/openspec/changes/recover-poisoned-anchor-on-circuit-open/proposal.md b/openspec/changes/recover-poisoned-anchor-on-circuit-open/proposal.md new file mode 100644 index 0000000000..f54749cb04 --- /dev/null +++ b/openspec/changes/recover-poisoned-anchor-on-circuit-open/proposal.md @@ -0,0 +1,33 @@ +# Recover poisoned HTTP bridge anchors when the retry circuit opens + +Source sidechat: `01a04ac6-27a1-74b3-88d3-f1ede3901184` + +Recovered from: `8c6ff6b6b4e76e453a2cfed29bd3b0ef0141a778` + +## Why + +The HTTP Responses bridge currently opens its hard-affinity retry circuit after +two eventless `stream_incomplete` or idle failures, while durable-anchor +poisoning waits for the independent default threshold of seven. Once cooldown +suppresses new submissions, the seventh failure is unreachable. The same dead +`previous_response_id` is therefore re-used by the next half-open probe and the +conversation remains wedged. + +## What changes + +- Add a circuit-open quarantine reason for eventless poison-class failures. +- Keep that quarantine active through the remaining cooldown and half-open + probe lease, including when another replica's durable merge opens the circuit. +- When a full-conversation resend is planned under that quarantine, suppress + every proxy-owned durable-anchor injection so the probe is sent unanchored; + delta-only requests retain their anchor because it is their only context. +- Count eventless terminal error frames through the same attempt-scoped retry + circuit path and clear the durable anchor once the circuit threshold is met, + without charging requests that already emitted response events. + +## Impact + +The change is default-on and requires no setting or migration. It is scoped to +hard-affinity HTTP bridge keys and preserves account selection, reservation +settlement, and the existing 503 cooldown envelope. A successful unanchored +probe can complete normally and clear the temporary quarantine. diff --git a/openspec/changes/recover-poisoned-anchor-on-circuit-open/specs/responses-api-compat/spec.md b/openspec/changes/recover-poisoned-anchor-on-circuit-open/specs/responses-api-compat/spec.md new file mode 100644 index 0000000000..9352ba181e --- /dev/null +++ b/openspec/changes/recover-poisoned-anchor-on-circuit-open/specs/responses-api-compat/spec.md @@ -0,0 +1,75 @@ +# responses-api-compat Delta + +## ADDED Requirements + +### Requirement: Circuit-open poison quarantine makes the half-open probe unanchored + +When a hard-affinity HTTP bridge retry circuit opens on an eventless +`stream_incomplete` or `stream_idle_timeout` failure, the proxy MUST quarantine +the exact affinity key with reason `retry_circuit_poisoned_anchor`. The +quarantine MUST remain active for at least the circuit's remaining cooldown plus +the half-open probe lease, and MUST be re-evaluated when a durable merge raises +the observed failure count to the circuit threshold. + +While this quarantine is active, a full-conversation resend MUST NOT receive a +proxy-injected durable `previous_response_id` through fresh-reattach, +session-state, or recovery injection. The request MUST be sent upstream with +the client's complete payload and no anchor. A delta-only continuation MUST +retain the durable anchor so prior context is not lost. + +An eventless upstream terminal error (`response.failed`, `response.incomplete`, +or `error`) MUST consume one attempt-scoped circuit strike when no response +event was observed. A request that already emitted response events MUST NOT +consume this pre-response strike. When the circuit threshold is reached for a +poison-class terminal error, the proxy MUST clear the durable continuity anchor +under the current owner fence; failed or fenced clears MUST leave the failure +state observable for retry. + +A terminal failure that still has a verified safe full-resend body MUST NOT +consume a circuit strike: the request remains recoverable in band and the +replay claims the captured circuit generation at dispatch. A confirmed durable +anchor abandonment MUST settle the retry circuit only when every request it +covers is stranded without a safe replay; a replayable request keeps the +generation fence alive. Fenced or failed abandonment MUST leave the circuit +cooling. + +#### Scenario: second eventless failure quarantines the hard key + +- **GIVEN** a hard-affinity key has one eventless `stream_incomplete` failure +- **WHEN** a second eventless `stream_incomplete` failure is recorded +- **THEN** the retry circuit opens and persists its cooldown +- **AND** the key is quarantined with reason `retry_circuit_poisoned_anchor` +- **AND** a full-resend probe is planned without `previous_response_id` + +#### Scenario: clean close does not poison an anchor + +- **GIVEN** a hard-affinity key has two pre-response `clean_close` failures +- **WHEN** the circuit opens +- **THEN** the key is not quarantined as a poisoned anchor + +#### Scenario: quarantine covers cooldown and probe lease + +- **GIVEN** a poison quarantine is opened at the maximum circuit cooldown +- **WHEN** the cooldown expires and the half-open probe is admitted +- **THEN** the quarantine is still active while that probe is planned + +#### Scenario: eventless terminal errors count once and clear the anchor + +- **GIVEN** an anchored request receives a terminal error before any response event +- **WHEN** two such terminal failures open the circuit +- **THEN** each physical attempt contributes exactly one strike +- **AND** the durable anchor is cleared under the owner fence +- **AND** a midstream terminal error contributes no pre-response strike + +#### Scenario: safe full resend does not consume a terminal strike + +- **GIVEN** an anchored request has a verified complete full-resend body +- **WHEN** its upstream terminal frame rejects the stale anchor before any response event +- **THEN** the request is replayed in band without advancing the source circuit + +#### Scenario: abandonment settles only stranded requests + +- **GIVEN** a poison-class circuit is cooling and its durable anchor is abandoned +- **WHEN** no covered request retains a safe replay +- **THEN** the local and durable circuit state is cleared +- **BUT** a covered request with a safe replay keeps the circuit generation fence diff --git a/openspec/changes/recover-poisoned-anchor-on-circuit-open/tasks.md b/openspec/changes/recover-poisoned-anchor-on-circuit-open/tasks.md new file mode 100644 index 0000000000..edba399307 --- /dev/null +++ b/openspec/changes/recover-poisoned-anchor-on-circuit-open/tasks.md @@ -0,0 +1,34 @@ +# Tasks + +## Specification + +- [x] Add the circuit-open poison quarantine and eventless terminal settlement + requirements to the `responses-api-compat` delta. +- [x] Record the cooldown/half-open, clean-close, full-resend, delta-only, and + durable-anchor-clear scenarios. + +## Implementation + +- [x] Add the `retry_circuit_poisoned_anchor` quarantine reason and a TTL floor + covering cooldown plus half-open lease. +- [x] Re-arm quarantine after a durable conflict merge opens the circuit. +- [x] Count eventless terminal frames through the attempt-scoped recorder and + clear the poisoned durable anchor at the circuit threshold. +- [x] Do not charge a terminal frame when a verified safe full resend remains + available; preserve the circuit generation until that replay dispatches. +- [x] Settle the circuit after confirmed abandonment only when all covered + requests are stranded; retain it for an in-flight safe replay. + +## Regression coverage + +- [x] Cover two eventless failures opening/quarantining the key. +- [x] Cover clean-close non-quarantine and full-resend unanchored planning. +- [x] Cover quarantine lifetime, terminal strike accounting, and fenced anchor + clearing. +- [x] Cover safe full-resend terminal recovery without an extra circuit strike. + +## Verification + +- [x] Run focused HTTP bridge unit/integration tests, Ruff, type checks, and + OpenSpec validation. +- [ ] Record release image, Argo revision, health, and rollback evidence. diff --git a/tests/unit/test_proxy_http_bridge.py b/tests/unit/test_proxy_http_bridge.py index 143425a14b..c676c7e522 100644 --- a/tests/unit/test_proxy_http_bridge.py +++ b/tests/unit/test_proxy_http_bridge.py @@ -32230,3 +32230,349 @@ async def _wait_once() -> Any: remaining = await asyncio.gather(*waiters[25:], return_exceptions=True) assert all(isinstance(result, ProxyResponseError) and result.status_code == 429 for result in remaining) assert key not in service._http_bridge_inflight_sessions + + +@pytest.mark.asyncio +async def test_retry_circuit_open_on_poison_detail_quarantines_key() -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + hard_session = _make_bridge_session(key_value="bridge-circuit-poison-quarantine") + service._durable_bridge = SimpleNamespace( + lookup_retry_circuit=AsyncMock(return_value=None), + persist_retry_circuit=AsyncMock(return_value=None), + ) + + await service._record_http_bridge_retry_circuit_failure(hard_session, detail="stream_incomplete") + assert http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, hard_session.key) is False + + await service._record_http_bridge_retry_circuit_failure(hard_session, detail="stream_incomplete") + + assert http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, hard_session.key) is True + entry = http_bridge_quarantine_module._http_bridge_quarantine_registry(service)[hard_session.key] + assert entry.reason == "retry_circuit_poisoned_anchor" + + +@pytest.mark.asyncio +async def test_retry_circuit_open_on_clean_close_does_not_quarantine_key() -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + hard_session = _make_bridge_session(key_value="bridge-circuit-clean-no-quarantine") + service._durable_bridge = SimpleNamespace( + lookup_retry_circuit=AsyncMock(return_value=None), + persist_retry_circuit=AsyncMock(return_value=None), + ) + + await service._record_http_bridge_retry_circuit_failure(hard_session, detail="clean_close") + await service._record_http_bridge_retry_circuit_failure(hard_session, detail="clean_close") + + assert http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, hard_session.key) is False + + +@pytest.mark.asyncio +async def test_poison_quarantine_outlives_maximum_retry_cooldown() -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + hard_session = _make_bridge_session(key_value="bridge-circuit-quarantine-ttl") + service._durable_bridge = SimpleNamespace( + lookup_retry_circuit=AsyncMock(return_value=None), + persist_retry_circuit=AsyncMock(return_value=None), + ) + + for _ in range(8): + await service._record_http_bridge_retry_circuit_failure(hard_session, detail="stream_incomplete") + + state = cast(Any, service)._http_bridge_retry_circuits[hard_session.key] + entry = http_bridge_quarantine_module._http_bridge_quarantine_registry(service)[hard_session.key] + assert entry.quarantined_until > state.cooldown_until + assert entry.quarantined_until - time.monotonic() >= ( + max(0.0, state.cooldown_until - time.monotonic()) + + http_bridge_retry_circuit_module._HTTP_BRIDGE_RETRY_CIRCUIT_HALF_OPEN_LEASE_SECONDS + - 1.0 + ) + + +@pytest.mark.asyncio +async def test_http_bridge_full_resend_probe_skips_poisoned_anchor( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Exercise the product planning path, not just the circuit helper.""" + service = proxy_service.ProxyService(cast(Any, nullcontext())) + session = _make_bridge_session(key_value="sid-poison-probe") + session.key = proxy_service._HTTPBridgeSessionKey("session_header", "sid-poison-probe", None) + service._durable_bridge = SimpleNamespace( + lookup_retry_circuit=AsyncMock(return_value=None), + persist_retry_circuit=AsyncMock(return_value=None), + lookup_request_targets=AsyncMock( + return_value=DurableBridgeLookup( + session_id="sess-poison", + canonical_kind="session_header", + canonical_key="sid-poison-probe", + api_key_scope="anonymous", + account_id="acc-bridge", + owner_instance_id="instance-a", + owner_epoch=1, + lease_expires_at=datetime.now(timezone.utc), + state=HttpBridgeSessionState.ACTIVE, + latest_turn_state="http_turn_1", + latest_response_id="resp-poisoned", + ) + ), + ) + await service._record_http_bridge_retry_circuit_failure(session, detail="stream_incomplete") + await service._record_http_bridge_retry_circuit_failure(session, detail="stream_incomplete") + + payload = proxy_service.ResponsesRequest.model_validate( + { + "model": "gpt-5.4", + "instructions": "hi", + "input": [ + {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "first"}]}, + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "reply"}], + }, + {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "second"}]}, + ], + } + ) + request_state = proxy_service._WebSocketRequestState( + request_id="req-poison-probe", + model="gpt-5.4", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=1.0, + event_queue=asyncio.Queue(), + transport="http", + ) + assert request_state.event_queue is not None + await request_state.event_queue.put(None) + captured: dict[str, object] = {} + + def fake_prepare( + prepared_payload: proxy_service.ResponsesRequest, + _headers: dict[str, str] | Any, + *, + api_key: proxy_service.ApiKeyData | None, + api_key_reservation: proxy_service.ApiKeyUsageReservationData | None, + request_id: str, + client_ip: str | None = None, + ) -> tuple[proxy_service._WebSocketRequestState, str]: + del api_key, api_key_reservation, request_id, client_ip + captured["previous_response_id"] = prepared_payload.previous_response_id + return request_state, '{"type":"response.create"}' + + 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: _make_app_settings()) + monkeypatch.setattr(service, "_prepare_http_bridge_request", fake_prepare) + monkeypatch.setattr(service, "_get_or_create_http_bridge_session", AsyncMock(return_value=session)) + monkeypatch.setattr(service, "_submit_http_bridge_request", AsyncMock()) + monkeypatch.setattr(service, "_detach_http_bridge_request", AsyncMock()) + + chunks = [ + chunk + async for chunk in service._stream_via_http_bridge( + payload, + headers={"x-codex-session-id": "sid-poison-probe"}, + codex_session_affinity=True, + propagate_http_errors=False, + openai_cache_affinity=False, + 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 == [] + assert captured["previous_response_id"] is None + + +def _make_terminal_error_bridge_fixture( + *, + request_id: str, + key_value: str, + response_event_count: int, +) -> tuple[proxy_service.ProxyService, proxy_service._HTTPBridgeSession, proxy_service._WebSocketRequestState]: + from app.modules.proxy._service.support import _HTTPBridgeResponseCreateAttempt + + service = proxy_service.ProxyService(cast(Any, nullcontext())) + request_state = proxy_service._WebSocketRequestState( + request_id=request_id, + model="gpt-5.4", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=1.0, + previous_response_id="resp_dead_anchor", + event_queue=asyncio.Queue(), + transport="http", + skip_request_log=True, + ) + request_state.response_create_attempt = _HTTPBridgeResponseCreateAttempt(ordinal=1) + request_state.response_event_count = response_event_count + session = proxy_service._HTTPBridgeSession( + key=proxy_service._HTTPBridgeSessionKey("session_header", key_value, None), + headers={"x-codex-session-id": key_value}, + affinity=proxy_service._AffinityPolicy( + key=key_value, + kind=proxy_service.StickySessionKind.CODEX_SESSION, + ), + request_model="gpt-5.4", + account=cast(Any, SimpleNamespace(id="acc-1", status=AccountStatus.ACTIVE)), + upstream=cast(UpstreamWebSocket, SimpleNamespace(close=AsyncMock())), + upstream_control=proxy_service._WebSocketUpstreamControl(), + pending_requests=deque([request_state]), + pending_lock=anyio.Lock(), + response_create_gate=asyncio.Semaphore(1), + queued_request_count=1, + last_used_at=1.0, + idle_ttl_seconds=120.0, + ) + return service, session, request_state + + +_PREVIOUS_RESPONSE_NOT_FOUND_FRAME = json.dumps( + { + "type": "error", + "status": 400, + "error": { + "type": "invalid_request_error", + "code": "previous_response_not_found", + "message": "Previous response with id 'resp_dead_anchor' not found.", + "param": "previous_response_id", + }, + }, + separators=(",", ":"), +) + + +@pytest.mark.asyncio +async def test_eventless_terminal_error_records_attempt_scoped_circuit_strike( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service, session, request_state = _make_terminal_error_bridge_fixture( + request_id="req-terminal-strike", + key_value="sid-terminal-strike", + response_event_count=0, + ) + record_failure = AsyncMock(return_value=None) + monkeypatch.setattr(service, "_record_http_bridge_retry_circuit_failure", record_failure) + monkeypatch.setattr(service, "_handle_stream_error", AsyncMock()) + + await service._process_http_bridge_upstream_text(session, _PREVIOUS_RESPONSE_NOT_FOUND_FRAME) + + record_failure.assert_awaited_once() + await_args = record_failure.await_args + assert await_args is not None + assert await_args.args[0] is session + assert await_args.kwargs["detail"] == "stream_incomplete" + assert await_args.kwargs["attempt"] is request_state.response_create_attempt + assert await_args.kwargs["terminal_pre_response_frame"] is True + + +@pytest.mark.asyncio +async def test_midstream_terminal_error_does_not_record_circuit_strike( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service, session, _request_state = _make_terminal_error_bridge_fixture( + request_id="req-terminal-midstream", + key_value="sid-terminal-midstream", + response_event_count=3, + ) + record_failure = AsyncMock(return_value=None) + monkeypatch.setattr(service, "_record_http_bridge_retry_circuit_failure", record_failure) + monkeypatch.setattr(service, "_handle_stream_error", AsyncMock()) + + await service._process_http_bridge_upstream_text(session, _PREVIOUS_RESPONSE_NOT_FOUND_FRAME) + + record_failure.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_safe_full_resend_terminal_error_does_not_record_circuit_strike( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # A verified full resend still has an in-band anchor-free recovery. The + # terminal rejection is therefore not a stranded lifecycle and must not + # charge the source circuit before the replay claims its generation. + service, session, request_state = _make_terminal_error_bridge_fixture( + request_id="req-terminal-safe-replay", + key_value="sid-terminal-safe-replay", + response_event_count=0, + ) + request_state.fresh_upstream_request_is_retry_safe = True + request_state.fresh_upstream_request_text = '{"type":"response.create"}' + record_failure = AsyncMock(return_value=None) + monkeypatch.setattr(service, "_record_http_bridge_retry_circuit_failure", record_failure) + monkeypatch.setattr(service, "_handle_stream_error", AsyncMock()) + + await service._process_http_bridge_upstream_text(session, _PREVIOUS_RESPONSE_NOT_FOUND_FRAME) + + record_failure.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_two_eventless_terminal_errors_open_circuit_and_clear_anchor() -> None: + service, session, _first = _make_terminal_error_bridge_fixture( + request_id="req-terminal-open-1", + key_value="sid-terminal-open", + response_event_count=0, + ) + session.durable_session_id = "durable-terminal-open" + session.durable_owner_epoch = 1 + rebind = AsyncMock(return_value=True) + clear_retry_circuit = AsyncMock(return_value=None) + service._durable_bridge = SimpleNamespace( + lookup_retry_circuit=AsyncMock(return_value=None), + persist_retry_circuit=AsyncMock(return_value=None), + clear_retry_circuit=clear_retry_circuit, + rebind_session_account=rebind, + ) + service._handle_stream_error = AsyncMock() # type: ignore[method-assign] + + await service._process_http_bridge_upstream_text(session, _PREVIOUS_RESPONSE_NOT_FOUND_FRAME) + + from app.modules.proxy._service.support import _HTTPBridgeResponseCreateAttempt + + second = proxy_service._WebSocketRequestState( + request_id="req-terminal-open-2", + model="gpt-5.4", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=2.0, + previous_response_id="resp_dead_anchor", + event_queue=asyncio.Queue(), + transport="http", + skip_request_log=True, + ) + second.response_create_attempt = _HTTPBridgeResponseCreateAttempt(ordinal=2) + session.pending_requests.append(second) + session.queued_request_count = 1 + + await service._process_http_bridge_upstream_text(session, _PREVIOUS_RESPONSE_NOT_FOUND_FRAME) + + assert rebind.await_count == 1 + assert rebind.await_args is not None + assert rebind.await_args.kwargs["clear_continuity"] is True + clear_retry_circuit.assert_awaited_once() + assert session.key not in cast(Any, service)._http_bridge_retry_circuits + assert http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, session.key) is True