From 052c16d24098ec190b971ed8fb350bf5a429b3db Mon Sep 17 00:00:00 2001 From: kevinsslin Date: Fri, 14 Aug 2026 00:25:55 +0800 Subject: [PATCH 1/4] fix(proxy): separate websocket scope cleanup budget Keep the one-second generic task cancellation bound while allowing normal direct Responses WebSocket scope finalization to use a bounded five-second observation window. Preserve the shared shutdown deadline and add a regression plus OpenSpec contract. Refs #1711 --- app/modules/proxy/_service/websocket/mixin.py | 19 +++-- .../.openspec.yaml | 2 + .../websocket-scope-cleanup-budget/design.md | 68 +++++++++++++++ .../proposal.md | 41 +++++++++ .../specs/responses-api-compat/spec.md | 39 +++++++++ .../websocket-scope-cleanup-budget/tasks.md | 23 +++++ .../test_websocket_terminal_cancellation.py | 83 +++++++++++++++++++ 7 files changed, 270 insertions(+), 5 deletions(-) create mode 100644 openspec/changes/websocket-scope-cleanup-budget/.openspec.yaml create mode 100644 openspec/changes/websocket-scope-cleanup-budget/design.md create mode 100644 openspec/changes/websocket-scope-cleanup-budget/proposal.md create mode 100644 openspec/changes/websocket-scope-cleanup-budget/specs/responses-api-compat/spec.md create mode 100644 openspec/changes/websocket-scope-cleanup-budget/tasks.md diff --git a/app/modules/proxy/_service/websocket/mixin.py b/app/modules/proxy/_service/websocket/mixin.py index 3b7a1b1e2d..ebf860febc 100644 --- a/app/modules/proxy/_service/websocket/mixin.py +++ b/app/modules/proxy/_service/websocket/mixin.py @@ -492,6 +492,9 @@ def _facade() -> Any: logger = logging.getLogger(__name__) _WEBSOCKET_PINNED_REFRESH_UNAVAILABLE_MESSAGE = "Account refresh is temporarily unavailable; retry later." +# Scope teardown coordinates several request/lease finalizers; keep its normal +# observation budget separate from the short generic child-task cancel bound. +_WEBSOCKET_SCOPE_CLEANUP_TIMEOUT_SECONDS = 5.0 _CAPABILITY_REQUIRED_NO_AUTHORIZED_ACCOUNTS_MESSAGE = ( "This request requires Trusted Access for Cyber, but no eligible account is marked as " "security-work-authorized. codex-lb did not fall back to an ordinary account." @@ -2510,9 +2513,15 @@ def take_reader_replay_request_state() -> _WebSocketRequestState | None: scope_cancelled = True raise finally: - cleanup_timeout = shutdown_state.remaining_drain_timeout_seconds() - if cleanup_timeout is None: - cleanup_timeout = _facade()._TASK_CANCEL_TIMEOUT_SECONDS + remaining_drain_timeout = shutdown_state.remaining_drain_timeout_seconds() + cleanup_timeout = ( + _WEBSOCKET_SCOPE_CLEANUP_TIMEOUT_SECONDS + if remaining_drain_timeout is None + else max(float(remaining_drain_timeout), 0.0) + ) + task_cleanup_timeout = ( + _facade()._TASK_CANCEL_TIMEOUT_SECONDS if remaining_drain_timeout is None else cleanup_timeout + ) async def finalize_websocket_scope() -> None: nonlocal replay_request_state @@ -2552,7 +2561,7 @@ async def finalize_websocket_scope() -> None: try: await _facade()._await_cancelled_task( retired_create_lease_release_task, - timeout_seconds=cleanup_timeout, + timeout_seconds=task_cleanup_timeout, label="proxy websocket retired create lease release", cancel=False, ) @@ -2566,7 +2575,7 @@ async def finalize_websocket_scope() -> None: try: await _facade()._await_cancelled_task( request_state_failure_task, - timeout_seconds=cleanup_timeout, + timeout_seconds=task_cleanup_timeout, label="proxy websocket unsent request finalization", cancel=False, ) diff --git a/openspec/changes/websocket-scope-cleanup-budget/.openspec.yaml b/openspec/changes/websocket-scope-cleanup-budget/.openspec.yaml new file mode 100644 index 0000000000..4af864176c --- /dev/null +++ b/openspec/changes/websocket-scope-cleanup-budget/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-08-14 diff --git a/openspec/changes/websocket-scope-cleanup-budget/design.md b/openspec/changes/websocket-scope-cleanup-budget/design.md new file mode 100644 index 0000000000..eb96d1d45b --- /dev/null +++ b/openspec/changes/websocket-scope-cleanup-budget/design.md @@ -0,0 +1,68 @@ +## Context + +The WebSocket handler publishes one tracked `finalize_websocket_scope()` task +from its `finally` block. The task must preserve cancellation while completing +the terminal request cleanup sequence. During process drain, +`shutdown_state.remaining_drain_timeout_seconds()` provides the shared absolute +deadline. Outside drain it returns `None`, so the current code falls back to +the generic one-second `_TASK_CANCEL_TIMEOUT_SECONDS` value. + +The generic timeout is used across HTTP bridge and proxy child-task cancellation +paths. Increasing it globally would slow unrelated cancellation and would +change the semantics of a helper that is intentionally a short cancellation +observation bound. The scope finalizer needs a different bounded allowance +because its work is a sequence of request-state and lease finalization steps. + +## Goals / Non-Goals + +**Goals:** + +- Give normal direct WebSocket scope cleanup enough bounded time for the + existing request finalization and lease-release sequence. +- Preserve the existing tracked-task ownership and cancellation behavior when + the bound is reached. +- Keep shutdown cleanup governed by the one shared remaining drain deadline. +- Prove the behavior through the real WebSocket route finalizer. + +**Non-Goals:** + +- Changing the `response.created` watchdog or any upstream request budget. +- Retrying, replaying, or moving an interrupted request to another account. +- Increasing the generic `_TASK_CANCEL_TIMEOUT_SECONDS` value. +- Adding an operator setting, environment variable, database state, or a new + background cleanup registry. + +## Decisions + +1. **Use one internal five-second scope budget.** The value is deliberately + fixed and bounded because this is a lifecycle safety allowance, not an + operator tuning surface. Five seconds is long enough to absorb ordinary + persistence/lease scheduling variance while still returning promptly when + teardown is stuck. + +2. **Prefer the active drain deadline.** When shutdown drain is active, the + finalizer continues to use the remaining shared deadline exactly as today. + The normal-operation budget is only the fallback for the no-drain case and + cannot extend process shutdown. + +3. **Keep child cancellation semantics separate.** Individual task waits keep + the one-second generic cancellation bound during normal operation. During + drain they remain capped by the shared remaining deadline. Only the outer + scope-finalization wait receives the five-second normal-operation allowance. + +4. **Retain tracked cleanup after the bound.** `asyncio.wait()` continues to + observe the finalizer without cancelling it at the scope budget. The + existing `_background_cleanup_tasks` registry and persistence drain remain + the owner of unfinished cleanup, so a timeout is honest and does not cause + lease or request finalization to be abandoned. + +## Verification Strategy + +- Run the focused WebSocket terminal-cancellation tests, including a regression + that lowers the generic task timeout and delays finalization beyond it while + allowing completion within the separate scope budget. +- Run Ruff check/format on changed Python files, the proxy architecture check, + and the applicable type/test targets. +- Validate the OpenSpec delta if the CLI is available; otherwise record the + unavailable local CLI as a handoff limitation and keep the artifacts in the + repository for CI validation. diff --git a/openspec/changes/websocket-scope-cleanup-budget/proposal.md b/openspec/changes/websocket-scope-cleanup-budget/proposal.md new file mode 100644 index 0000000000..6666fe1962 --- /dev/null +++ b/openspec/changes/websocket-scope-cleanup-budget/proposal.md @@ -0,0 +1,41 @@ +## Why + +Direct Responses WebSocket scope teardown currently reuses the generic +`_TASK_CANCEL_TIMEOUT_SECONDS` value as its entire normal-operation cleanup +budget. That value is intentionally one second for individual task +cancellation, but scope teardown can also have to finalize request logs, +release response-create ownership, and release the account connection lease. +Under ordinary load those operations can exceed one second, producing +`Websocket scope cleanup exceeded its remaining drain budget` even when the +server is not draining. The cleanup task remains tracked, but the warning and +unfinished teardown increase the chance of follow-up reconnect churn. + +## What Changes + +- Give normal-operation WebSocket scope teardown its own fixed five-second + bounded budget. +- Keep the existing one-second generic task-cancellation timeout for ordinary + child-task waits. +- Continue using the remaining shared shutdown deadline whenever process drain + is active; the new budget must not extend shutdown. +- Add a route-level cancellation regression proving that cleanup which takes + longer than the generic task timeout can still finish within the scope budget + and does not leave an orphaned cleanup task. + +## Capabilities + +### New Capabilities + +None. + +### Modified Capabilities + +- `responses-api-compat`: Direct Responses WebSocket scope cleanup has a + separate bounded normal-operation budget while preserving the shared + shutdown deadline and task ownership guarantees. + +## Impact + +The change is limited to the direct Responses WebSocket finalizer, its focused +unit coverage, and the OpenSpec contract. It adds no setting, dependency, +database migration, API shape, upstream watchdog change, or retry policy. diff --git a/openspec/changes/websocket-scope-cleanup-budget/specs/responses-api-compat/spec.md b/openspec/changes/websocket-scope-cleanup-budget/specs/responses-api-compat/spec.md new file mode 100644 index 0000000000..35fb23b2db --- /dev/null +++ b/openspec/changes/websocket-scope-cleanup-budget/specs/responses-api-compat/spec.md @@ -0,0 +1,39 @@ +# responses-api-compat Delta + +## ADDED Requirements + +### Requirement: Direct WebSocket scope cleanup has a bounded normal-operation budget + +When a direct Responses WebSocket scope exits while the process is not using an +active shutdown drain deadline, the proxy MUST allow its existing scope +finalization task a fixed five-second bounded observation budget, separate from +the one-second generic child-task cancellation timeout. The finalizer MUST +continue to own request finalization and lease cleanup through that budget. If +the budget expires, the proxy MUST preserve the existing cancellation result, +leave unfinished cleanup tracked by the existing cleanup-task registry, and +MUST NOT cancel or silently abandon that cleanup solely because the observation +budget expired. + +When an active shutdown drain deadline exists, the proxy MUST use the remaining +shared drain deadline instead of the normal-operation budget, so normal cleanup +allowance MUST NOT extend process shutdown. + +#### Scenario: normal scope cleanup outlives generic child cancellation + +- **GIVEN** a direct Responses WebSocket scope is cancelled while its existing + request finalization takes longer than the generic one-second child-task + cancellation timeout +- **AND** the finalization completes within the five-second normal-operation + scope budget +- **WHEN** scope cleanup runs +- **THEN** the finalizer completes and request/lease ownership is released +- **AND** the scope preserves its cancellation result +- **AND** no cleanup task remains orphaned after the finalizer completes + +#### Scenario: shutdown drain remains the upper bound + +- **GIVEN** a direct Responses WebSocket scope is cancelled while an active + shutdown drain deadline has less than five seconds remaining +- **WHEN** scope cleanup runs +- **THEN** the remaining shared drain deadline remains the upper bound +- **AND** the normal-operation five-second budget does not extend shutdown diff --git a/openspec/changes/websocket-scope-cleanup-budget/tasks.md b/openspec/changes/websocket-scope-cleanup-budget/tasks.md new file mode 100644 index 0000000000..b77807eb25 --- /dev/null +++ b/openspec/changes/websocket-scope-cleanup-budget/tasks.md @@ -0,0 +1,23 @@ +## 1. Regression Coverage + +- [x] 1.1 Add a real direct Responses WebSocket cancellation regression that + distinguishes the generic task timeout from the scope cleanup budget. +- [x] 1.2 Confirm the regression fails against the baseline implementation and + passes with the scoped budget. + +## 2. Scope Cleanup Budget + +- [x] 2.1 Add the fixed normal-operation WebSocket scope cleanup budget. +- [x] 2.2 Preserve the active shared shutdown deadline and one-second generic + child-task cancellation behavior. +- [x] 2.3 Keep unfinished cleanup tracked and prevent cancellation/lease + ownership regressions when the bound expires. + +## 3. Verification + +- [x] 3.1 Run focused WebSocket terminal-cancellation tests. +- [x] 3.2 Run changed-file Ruff check/format, proxy architecture checks, and + applicable type checks. +- [x] 3.3 Validate the OpenSpec delta and inspect the final diff/status. +- [ ] 3.4 Open a Draft PR targeting upstream `main` and add the live 1.23.0 + evidence to issue #1711. diff --git a/tests/unit/test_websocket_terminal_cancellation.py b/tests/unit/test_websocket_terminal_cancellation.py index 45122c622e..e830db164e 100644 --- a/tests/unit/test_websocket_terminal_cancellation.py +++ b/tests/unit/test_websocket_terminal_cancellation.py @@ -240,6 +240,89 @@ async def block_cleanup(*_args: object, **_kwargs: object) -> None: assert service._background_cleanup_tasks == set() +@pytest.mark.asyncio +async def test_normal_websocket_scope_cleanup_uses_separate_scope_budget( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + @asynccontextmanager + async def repo_factory() -> AsyncIterator[SimpleNamespace]: + yield SimpleNamespace(request_logs=_RequestLogsRecorder(), api_keys=object()) + + service = proxy_service.ProxyService(cast(proxy_service.ProxyRepoFactory, repo_factory)) + settings = SimpleNamespace( + prefer_earlier_reset_accounts=False, + sticky_threads_enabled=False, + openai_cache_affinity_max_age_seconds=0, + prohibit_fast_mode=False, + ) + + class _SettingsCache: + async def get(self) -> SimpleNamespace: + return settings + + receive_started = asyncio.Event() + cleanup_started = asyncio.Event() + release_cleanup = asyncio.Event() + + class _BlockingDownstreamWebSocket: + async def receive(self) -> dict[str, object]: + receive_started.set() + await asyncio.Event().wait() + raise AssertionError("unreachable") + + async def close(self, code: int = 1000, reason: str | None = None) -> None: + del code, reason + + async def block_cleanup(*_args: object, **_kwargs: object) -> None: + cleanup_started.set() + await release_cleanup.wait() + + async def release_cleanup_after_scope_budget_margin() -> None: + await asyncio.sleep(0.03) + release_cleanup.set() + + monkeypatch.setattr(proxy_service, "_TASK_CANCEL_TIMEOUT_SECONDS", 0.01) + monkeypatch.setattr(websocket_mixin, "_WEBSOCKET_SCOPE_CLEANUP_TIMEOUT_SECONDS", 0.08) + monkeypatch.setattr(proxy_service, "get_settings_cache", lambda: _SettingsCache()) + monkeypatch.setattr( + proxy_service, + "get_settings", + lambda: SimpleNamespace(proxy_downstream_websocket_idle_timeout_seconds=30.0), + ) + monkeypatch.setattr(proxy_service, "_routing_strategy", lambda _settings: "usage_weighted") + monkeypatch.setattr(service, "_websocket_continuity_state_for_request", lambda *_args, **_kwargs: None) + monkeypatch.setattr(service, "_fail_pending_websocket_requests", block_cleanup) + monkeypatch.setattr(service._load_balancer, "release_account_lease", AsyncMock()) + caplog.set_level(logging.WARNING) + + scope_task = asyncio.create_task( + service.proxy_responses_websocket( + cast(WebSocket, _BlockingDownstreamWebSocket()), + {}, + codex_session_affinity=False, + openai_cache_affinity=False, + api_key=None, + ) + ) + await asyncio.wait_for(receive_started.wait(), timeout=1) + + started_at = asyncio.get_running_loop().time() + scope_task.cancel() + await asyncio.wait_for(cleanup_started.wait(), timeout=1) + release_task = asyncio.create_task(release_cleanup_after_scope_budget_margin()) + + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(scope_task, timeout=0.5) + elapsed = asyncio.get_running_loop().time() - started_at + await release_task + await asyncio.sleep(0) + + assert elapsed >= 0.02 + assert "Websocket scope cleanup exceeded its remaining drain budget" not in caplog.messages + assert service._background_cleanup_tasks == set() + + @pytest.mark.asyncio @pytest.mark.parametrize( "failing_child", From 902ff02b60cbf77725c16c55aa995ce155f76b95 Mon Sep 17 00:00:00 2001 From: kevinsslin Date: Fri, 14 Aug 2026 00:28:44 +0800 Subject: [PATCH 2/4] test(proxy): harden cleanup budget regression --- tests/unit/test_websocket_terminal_cancellation.py | 12 +++--------- 1 file changed, 3 insertions(+), 9 deletions(-) diff --git a/tests/unit/test_websocket_terminal_cancellation.py b/tests/unit/test_websocket_terminal_cancellation.py index e830db164e..1036f11439 100644 --- a/tests/unit/test_websocket_terminal_cancellation.py +++ b/tests/unit/test_websocket_terminal_cancellation.py @@ -278,10 +278,6 @@ async def block_cleanup(*_args: object, **_kwargs: object) -> None: cleanup_started.set() await release_cleanup.wait() - async def release_cleanup_after_scope_budget_margin() -> None: - await asyncio.sleep(0.03) - release_cleanup.set() - monkeypatch.setattr(proxy_service, "_TASK_CANCEL_TIMEOUT_SECONDS", 0.01) monkeypatch.setattr(websocket_mixin, "_WEBSOCKET_SCOPE_CLEANUP_TIMEOUT_SECONDS", 0.08) monkeypatch.setattr(proxy_service, "get_settings_cache", lambda: _SettingsCache()) @@ -307,18 +303,16 @@ async def release_cleanup_after_scope_budget_margin() -> None: ) await asyncio.wait_for(receive_started.wait(), timeout=1) - started_at = asyncio.get_running_loop().time() scope_task.cancel() await asyncio.wait_for(cleanup_started.wait(), timeout=1) - release_task = asyncio.create_task(release_cleanup_after_scope_budget_margin()) + await asyncio.sleep(0.03) + assert scope_task.done() is False + release_cleanup.set() with pytest.raises(asyncio.CancelledError): await asyncio.wait_for(scope_task, timeout=0.5) - elapsed = asyncio.get_running_loop().time() - started_at - await release_task await asyncio.sleep(0) - assert elapsed >= 0.02 assert "Websocket scope cleanup exceeded its remaining drain budget" not in caplog.messages assert service._background_cleanup_tasks == set() From de260bc05396e25e0e738d9891a7dc23575912b0 Mon Sep 17 00:00:00 2001 From: kevinsslin Date: Fri, 14 Aug 2026 00:36:56 +0800 Subject: [PATCH 3/4] feat(proxy): report websocket cleanup phase --- app/modules/proxy/_service/websocket/mixin.py | 14 +++++++- .../proposal.md | 33 +++++++++++++++++++ .../specs/proxy-runtime-observability/spec.md | 27 +++++++++++++++ .../tasks.md | 12 +++++++ .../test_websocket_terminal_cancellation.py | 7 ++++ 5 files changed, 92 insertions(+), 1 deletion(-) create mode 100644 openspec/changes/attribute-websocket-scope-cleanup-phase/proposal.md create mode 100644 openspec/changes/attribute-websocket-scope-cleanup-phase/specs/proxy-runtime-observability/spec.md create mode 100644 openspec/changes/attribute-websocket-scope-cleanup-phase/tasks.md diff --git a/app/modules/proxy/_service/websocket/mixin.py b/app/modules/proxy/_service/websocket/mixin.py index ebf860febc..fcc353bc68 100644 --- a/app/modules/proxy/_service/websocket/mixin.py +++ b/app/modules/proxy/_service/websocket/mixin.py @@ -2522,8 +2522,10 @@ def take_reader_replay_request_state() -> _WebSocketRequestState | None: task_cleanup_timeout = ( _facade()._TASK_CANCEL_TIMEOUT_SECONDS if remaining_drain_timeout is None else cleanup_timeout ) + cleanup_phase = "not_started" async def finalize_websocket_scope() -> None: + nonlocal cleanup_phase nonlocal replay_request_state nonlocal request_state_failure_task nonlocal request_state_to_fail @@ -2536,6 +2538,7 @@ async def finalize_websocket_scope() -> None: # release that wait. reader_to_await.cancel() if upstream is not None: + cleanup_phase = "upstream_close" await _close_websocket_upstream_for_cleanup( proxy, upstream, @@ -2543,6 +2546,7 @@ async def finalize_websocket_scope() -> None: ) if reader_to_await is not None: try: + cleanup_phase = "upstream_reader" await _facade()._await_cancelled_task( reader_to_await, label="proxy websocket upstream reader", @@ -2559,6 +2563,7 @@ async def finalize_websocket_scope() -> None: upstream_reader = None if retired_create_lease_release_task is not None: try: + cleanup_phase = "retired_create_lease" await _facade()._await_cancelled_task( retired_create_lease_release_task, timeout_seconds=task_cleanup_timeout, @@ -2573,6 +2578,7 @@ async def finalize_websocket_scope() -> None: retired_create_lease_release_task = None if request_state_failure_task is not None: try: + cleanup_phase = "unsent_request" await _facade()._await_cancelled_task( request_state_failure_task, timeout_seconds=task_cleanup_timeout, @@ -2589,6 +2595,7 @@ async def finalize_websocket_scope() -> None: replay_request_state = upstream_control.replay_request_state upstream_control.replay_request_state = None if request_state_to_fail is not None: + cleanup_phase = "unsent_request" await proxy._fail_pending_websocket_requests( account=None, account_id_value=account.id if account is not None else upstream_account_id, @@ -2607,6 +2614,7 @@ async def finalize_websocket_scope() -> None: ) request_state_to_fail = None if replay_request_state is not None: + cleanup_phase = "replay_request" await proxy._fail_pending_websocket_requests( account=None, account_id_value=account.id if account is not None else upstream_account_id, @@ -2624,6 +2632,7 @@ async def finalize_websocket_scope() -> None: penalize_account=False, ) client_disconnected = downstream_activity.disconnected + cleanup_phase = "pending_requests" await proxy._fail_pending_websocket_requests( account=None if client_disconnected or scope_cancelled else account, account_id_value=account.id if account is not None else upstream_account_id, @@ -2646,6 +2655,7 @@ async def finalize_websocket_scope() -> None: penalize_account=not (client_disconnected or scope_cancelled), ) try: + cleanup_phase = "connection_lease" await release_current_account_lease() except Exception: # Connection-lease cleanup must never replace cancellation @@ -2654,6 +2664,7 @@ async def finalize_websocket_scope() -> None: "Failed to release websocket connection lease during scope cleanup", exc_info=True, ) + cleanup_phase = "complete" cleanup_task = asyncio.create_task( finalize_websocket_scope(), @@ -2679,8 +2690,9 @@ def log_scope_cleanup_failure(done_task: asyncio.Task[None]) -> None: if not done: _facade().logger.warning( "Websocket scope cleanup exceeded its remaining drain budget " - "timeout_seconds=%.3f background_cleanup_tasks=%d", + "timeout_seconds=%.3f cleanup_phase=%s background_cleanup_tasks=%d", max(float(cleanup_timeout), 0.0), + cleanup_phase, sum(1 for task in proxy._background_cleanup_tasks if not task.done()), ) diff --git a/openspec/changes/attribute-websocket-scope-cleanup-phase/proposal.md b/openspec/changes/attribute-websocket-scope-cleanup-phase/proposal.md new file mode 100644 index 0000000000..b30d747d74 --- /dev/null +++ b/openspec/changes/attribute-websocket-scope-cleanup-phase/proposal.md @@ -0,0 +1,33 @@ +## Why + +When WebSocket scope cleanup exceeds its drain budget, the warning reports the +timeout and total background cleanup task count but not the operation that is +still blocked. Operators cannot distinguish an upstream-close stall from +reader observation, request finalization, or lease release without reproducing +the incident under instrumentation. + +## What Changes + +- Track the current WebSocket scope cleanup phase locally while the existing + finalization sequence runs. +- Add that fixed, low-cardinality phase to the existing timeout warning. +- Keep cleanup ordering, timeout budgets, retries, and ownership unchanged. +- Do not log request ids, account ids, payloads, credentials, or exception + content in the phase field. + +## Capabilities + +### New Capabilities + +(none) + +### Modified Capabilities + +- `proxy-runtime-observability`: WebSocket scope cleanup timeout warnings MUST + identify the blocked cleanup phase with a fixed low-cardinality value. + +## Impact + +`app/modules/proxy/_service/websocket/mixin.py` and its route-level WebSocket +cleanup regression coverage. No API, schema, setting, timeout, or dashboard +change. diff --git a/openspec/changes/attribute-websocket-scope-cleanup-phase/specs/proxy-runtime-observability/spec.md b/openspec/changes/attribute-websocket-scope-cleanup-phase/specs/proxy-runtime-observability/spec.md new file mode 100644 index 0000000000..648f0bdf77 --- /dev/null +++ b/openspec/changes/attribute-websocket-scope-cleanup-phase/specs/proxy-runtime-observability/spec.md @@ -0,0 +1,27 @@ +# proxy-runtime-observability Delta + +## ADDED Requirements + +### Requirement: WebSocket scope cleanup timeout identifies its blocked phase + +When WebSocket scope finalization exceeds its cleanup budget, the proxy MUST +include the current cleanup phase in the existing warning. The phase MUST be a +fixed low-cardinality value that identifies the cleanup operation and MUST NOT +contain request ids, account ids, request payloads, credentials, or exception +content. This diagnostic MUST NOT change cleanup ordering, timeout budgets, +retry behavior, or task ownership. + +#### Scenario: Pending request finalization exceeds the cleanup budget + +- **GIVEN** a cancelled WebSocket scope whose pending request finalization does + not finish within the cleanup budget +- **WHEN** the proxy emits the cleanup-budget warning +- **THEN** the warning includes `cleanup_phase=pending_requests` +- **AND** the cleanup remains owned by the existing background drain + +#### Scenario: Diagnostic phase remains low-cardinality + +- **WHEN** any WebSocket scope cleanup phase exceeds the cleanup budget +- **THEN** the warning identifies only a fixed cleanup phase +- **AND** the phase contains no request id, account id, payload, credential, or + exception content diff --git a/openspec/changes/attribute-websocket-scope-cleanup-phase/tasks.md b/openspec/changes/attribute-websocket-scope-cleanup-phase/tasks.md new file mode 100644 index 0000000000..668b52e9e1 --- /dev/null +++ b/openspec/changes/attribute-websocket-scope-cleanup-phase/tasks.md @@ -0,0 +1,12 @@ +## 1. Implementation + +- [x] 1.1 Track the current fixed WebSocket scope cleanup phase. +- [x] 1.2 Include the phase in the existing cleanup-budget warning without + changing cleanup control flow or timeout behavior. + +## 2. Validation + +- [x] 2.1 Add a route-level regression proving a blocked request-finalization + cleanup is attributed to `pending_requests`. +- [x] 2.2 Run focused WebSocket tests, proxy integration tests, lint, type + checks, architecture checks, and strict OpenSpec validation. diff --git a/tests/unit/test_websocket_terminal_cancellation.py b/tests/unit/test_websocket_terminal_cancellation.py index 1036f11439..841736e752 100644 --- a/tests/unit/test_websocket_terminal_cancellation.py +++ b/tests/unit/test_websocket_terminal_cancellation.py @@ -152,6 +152,7 @@ async def test_transport_end_replay_requires_send_boundary_only_for_direct_webso @pytest.mark.asyncio async def test_cancelled_websocket_scope_cleanup_is_deadline_bounded_and_remains_drain_owned( monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, ) -> None: @asynccontextmanager async def repo_factory() -> AsyncIterator[SimpleNamespace]: @@ -213,6 +214,7 @@ async def block_cleanup(*_args: object, **_kwargs: object) -> None: ) await asyncio.wait_for(receive_started.wait(), timeout=1) + caplog.set_level(logging.WARNING) shutdown_state.commit_shutdown(timeout_seconds=0.1) started_at = asyncio.get_running_loop().time() scope_task.cancel() @@ -231,6 +233,11 @@ async def block_cleanup(*_args: object, **_kwargs: object) -> None: for task in service._background_cleanup_tasks if not task.done() ) + assert any( + "Websocket scope cleanup exceeded its remaining drain budget" in message + and "cleanup_phase=pending_requests" in message + for message in caplog.messages + ) persistence_drain = asyncio.create_task(service.drain_persistence_tasks(timeout_seconds=1)) await asyncio.sleep(0) From c3c5bf67f7c9d6a41046292bb37a3ca018fa9feb Mon Sep 17 00:00:00 2001 From: kevinsslin Date: Fri, 14 Aug 2026 12:49:30 +0800 Subject: [PATCH 4/4] fix(proxy): recover anchored bridge requests after missing response.created --- .../_service/http_bridge/request_submit.py | 66 ++++- .../_service/http_bridge/retry_circuit.py | 12 + .../_service/http_bridge/upstream_events.py | 21 +- app/modules/proxy/_service/support.py | 4 + .../.openspec.yaml | 2 + .../design.md | 32 +++ .../proposal.md | 28 ++ .../specs/responses-api-compat/spec.md | 41 +++ .../tasks.md | 9 + .../integration/test_http_responses_bridge.py | 241 ++++++++++++++++++ tests/unit/test_proxy_http_bridge.py | 58 ++++- 11 files changed, 501 insertions(+), 13 deletions(-) create mode 100644 openspec/changes/recover-precreated-anchored-bridge/.openspec.yaml create mode 100644 openspec/changes/recover-precreated-anchored-bridge/design.md create mode 100644 openspec/changes/recover-precreated-anchored-bridge/proposal.md create mode 100644 openspec/changes/recover-precreated-anchored-bridge/specs/responses-api-compat/spec.md create mode 100644 openspec/changes/recover-precreated-anchored-bridge/tasks.md diff --git a/app/modules/proxy/_service/http_bridge/request_submit.py b/app/modules/proxy/_service/http_bridge/request_submit.py index 23216a8c36..d75c5ff3ec 100644 --- a/app/modules/proxy/_service/http_bridge/request_submit.py +++ b/app/modules/proxy/_service/http_bridge/request_submit.py @@ -218,6 +218,7 @@ _REQUEST_TRANSPORT_HTTP = "http" _WEBSOCKET_AUTH_INVALIDATED_FAILURE_CODE = "account_auth_invalidated" +_HTTP_BRIDGE_SAME_ANCHOR_PRECREATED_MAX_REPLAYS = 1 _NO_SECURITY_WORK_AUTHORIZED_ACCOUNTS_CODE = "no_security_work_authorized_accounts" _SECURITY_WORK_NO_AUTHORIZED_ACCOUNTS_MESSAGE = ( "Upstream flagged this request as possible cybersecurity work, but no account is marked as authorized for " @@ -329,6 +330,23 @@ def _http_bridge_client_full_history_recovery_error() -> OpenAIErrorEnvelope: return payload +def _http_bridge_can_replay_same_anchor_before_created(request_state: _WebSocketRequestState) -> bool: + if not request_state.request_text: + return False + if request_state.missing_response_created_retry_count >= _HTTP_BRIDGE_SAME_ANCHOR_PRECREATED_MAX_REPLAYS: + return False + return ( + request_state.previous_response_id is not None + and request_state.response_id is None + and request_state.awaiting_response_created + and request_state.response_event_count == 0 + and request_state.last_downstream_sequence_number is None + and not request_state.downstream_visible + and not request_state.upstream_model_output_seen + and not request_state.file_required_preferred_account + ) + + async def _rollback_http_bridge_recovery_turn_state_registration( service: Any, receipt: DurableBridgeAliasRegistrationReceipt, @@ -2856,6 +2874,7 @@ async def _retry_http_bridge_precreated_request( *, request_state: _WebSocketRequestState | None = None, restart_reader: bool = False, + allow_same_anchor_before_created: bool = False, ) -> bool: clean_close_retry_max_count = self._http_bridge_clean_close_retry_max_count() account_neutral_recovery = is_http_bridge_account_neutral_replay( @@ -2863,9 +2882,17 @@ async def _retry_http_bridge_precreated_request( key=session.key.affinity_key, ) + hard_owner_bound = _http_bridge_key_strength(session.key) == "hard" + def request_is_retryable(request_state: _WebSocketRequestState) -> bool: if _websocket_request_can_replay_before_visible_output(request_state): return True + if ( + allow_same_anchor_before_created + and hard_owner_bound + and _http_bridge_can_replay_same_anchor_before_created(request_state) + ): + return True if ( clean_close_retry_max_count <= 0 or request_state.replay_count != 1 @@ -2886,6 +2913,7 @@ def request_is_retryable(request_state: _WebSocketRequestState) -> bool: fresh_hard_request_account_switch_candidate = False proof_gated_continuity_replay_candidate = False server_anchored_replay_candidate = False + eventless_same_anchor_replay_candidate = False if session.key.strength == "hard": async with session.pending_lock: retryable_candidates = [ @@ -2911,12 +2939,21 @@ def request_is_retryable(request_state: _WebSocketRequestState) -> bool: and candidate.replay_count == 0 ) server_anchored_replay_candidate = _http_bridge_server_anchored_replay_enabled(candidate) + eventless_same_anchor_replay_candidate = ( + allow_same_anchor_before_created + and hard_owner_bound + and _http_bridge_can_replay_same_anchor_before_created(candidate) + and not ( + candidate.fresh_upstream_request_is_retry_safe and candidate.fresh_upstream_request_text + ) + ) if not await self._http_bridge_precreated_retry_allowed( session, allow_fresh_hard_account_switch=fresh_hard_request_account_switch_candidate, allow_proof_gated_continuity_replay=( proof_gated_continuity_replay_candidate or server_anchored_replay_candidate ), + allow_eventless_same_anchor_replay=eventless_same_anchor_replay_candidate, ): return False @@ -2924,7 +2961,6 @@ def request_is_retryable(request_state: _WebSocketRequestState) -> bool: kind=session.key.affinity_kind, key=session.key.affinity_key, ) - hard_owner_bound = _http_bridge_key_strength(session.key) == "hard" async with session.pending_lock: if request_state is not None: if ( @@ -2945,8 +2981,24 @@ def request_is_retryable(request_state: _WebSocketRequestState) -> bool: return False request_state = retryable_requests[0] model_fallback_replay = request_state.precreated_replay_reason == _ACCOUNT_MODEL_UNSUPPORTED_ERROR_CODE - if request_state.previous_response_id is not None and not ( - request_state.fresh_upstream_request_is_retry_safe and request_state.fresh_upstream_request_text + eventless_same_anchor_replay = ( + allow_same_anchor_before_created + and hard_owner_bound + and _http_bridge_can_replay_same_anchor_before_created(request_state) + and not ( + request_state.fresh_upstream_request_is_retry_safe and request_state.fresh_upstream_request_text + ) + ) + if ( + request_state.previous_response_id is not None + and not ( + request_state.fresh_upstream_request_is_retry_safe and request_state.fresh_upstream_request_text + ) + and not ( + allow_same_anchor_before_created + and hard_owner_bound + and _http_bridge_can_replay_same_anchor_before_created(request_state) + ) ): # Once a continuation is pending upstream, reconnecting without # replay cannot complete the current request, while replaying it @@ -3046,10 +3098,16 @@ def request_is_retryable(request_state: _WebSocketRequestState) -> bool: request_state.clean_close_retry_close_generation = close_generation if additional_clean_close_retry: request_state.clean_close_replay_count += 1 + if eventless_same_anchor_replay: + request_state.missing_response_created_retry_count += 1 retry_jitter_seconds = ( self._http_bridge_clean_close_retry_jitter_seconds() if additional_clean_close_retry else 0.0 ) - retry_event = "retry_precreated_clean_close" if additional_clean_close_retry else "retry_precreated" + retry_event = ( + "retry_precreated_same_anchor" + if eventless_same_anchor_replay + else ("retry_precreated_clean_close" if additional_clean_close_retry else "retry_precreated") + ) _log_http_bridge_event( retry_event, session.key, diff --git a/app/modules/proxy/_service/http_bridge/retry_circuit.py b/app/modules/proxy/_service/http_bridge/retry_circuit.py index 891bf49dca..b9e30ff79d 100644 --- a/app/modules/proxy/_service/http_bridge/retry_circuit.py +++ b/app/modules/proxy/_service/http_bridge/retry_circuit.py @@ -262,6 +262,7 @@ async def _http_bridge_precreated_retry_allowed( allow_fresh_hard_account_switch: bool = False, allow_proof_gated_continuity_replay: bool = False, allow_operation_fenced_continuity_replay: bool = False, + allow_eventless_same_anchor_replay: bool = False, ) -> bool: """Avoid replaying a repeatedly failing hard-affinity request in a tight loop.""" if session.key.strength != "hard": @@ -278,6 +279,7 @@ async def _http_bridge_precreated_retry_allowed( and state.half_open_until > now and not allow_fresh_hard_account_switch and not allow_proof_gated_continuity_replay + and not allow_eventless_same_anchor_replay ): if PROMETHEUS_AVAILABLE and http_bridge_retry_circuit_total is not None: http_bridge_retry_circuit_total.labels(outcome="suppressed").inc() @@ -324,6 +326,16 @@ async def _http_bridge_precreated_retry_allowed( retry_after, ) return True + if allow_eventless_same_anchor_replay: + logger.info( + "http_bridge_retry_circuit event=bypass_eventless_same_anchor_replay bridge_kind=%s " + "bridge_key=%s failures=%s retry_after_seconds=%.1f", + session.key.affinity_kind, + _hash_identifier(session.key.affinity_key), + state.consecutive_failures, + retry_after, + ) + return True if PROMETHEUS_AVAILABLE and http_bridge_retry_circuit_total is not None: http_bridge_retry_circuit_total.labels(outcome="suppressed").inc() logger.info( diff --git a/app/modules/proxy/_service/http_bridge/upstream_events.py b/app/modules/proxy/_service/http_bridge/upstream_events.py index 431aaea01a..aaeeb25be0 100644 --- a/app/modules/proxy/_service/http_bridge/upstream_events.py +++ b/app/modules/proxy/_service/http_bridge/upstream_events.py @@ -1288,14 +1288,19 @@ async def _relay_http_bridge_upstream_messages( _extract_model_class(session.request_model) if session.request_model else None ), ) - # A fresh, self-contained hard request can use the - # same bounded pre-created recovery as the idle - # timeout path. Keep the session open until the - # recovery routine claims the handoff; otherwise - # its retry gate would reject the request as - # already retired. Continuity-bound requests still - # fail closed in _retry_http_bridge_precreated_request. - retried = await self._retry_http_bridge_precreated_request(session) + # A fresh, self-contained hard request, or the + # narrower eventless same-anchor continuation + # recovery, can use the same bounded pre-created + # path. Keep the session open until the recovery + # routine claims the handoff; otherwise its retry + # gate would reject the request as already + # retired. Continuity requests that do not satisfy + # the narrow proof still fail closed in + # _retry_http_bridge_precreated_request. + retried = await self._retry_http_bridge_precreated_request( + session, + allow_same_anchor_before_created=True, + ) if retried: continue session.closed = True diff --git a/app/modules/proxy/_service/support.py b/app/modules/proxy/_service/support.py index c92e2681cb..db775f697c 100644 --- a/app/modules/proxy/_service/support.py +++ b/app/modules/proxy/_service/support.py @@ -817,6 +817,10 @@ class _WebSocketRequestState: request_usage_budget: ApiKeyRequestUsageBudget | None = None request_text: str | None = None replay_count: int = 0 + # Counts the one watchdog-owned same-anchor recovery permitted after an + # eventless pre-response-created timeout. Keep this separate from + # ``replay_count``, which tracks client/security/fresh-replay attempts. + missing_response_created_retry_count: int = 0 # Counts only the one extra replay permitted after the initial recovery # replay when the replacement upstream socket also closes cleanly before # producing any response event. diff --git a/openspec/changes/recover-precreated-anchored-bridge/.openspec.yaml b/openspec/changes/recover-precreated-anchored-bridge/.openspec.yaml new file mode 100644 index 0000000000..4af864176c --- /dev/null +++ b/openspec/changes/recover-precreated-anchored-bridge/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-08-14 diff --git a/openspec/changes/recover-precreated-anchored-bridge/design.md b/openspec/changes/recover-precreated-anchored-bridge/design.md new file mode 100644 index 0000000000..8028a26d92 --- /dev/null +++ b/openspec/changes/recover-precreated-anchored-bridge/design.md @@ -0,0 +1,32 @@ +## Context + +The eventless watchdog already owns the pending request under the session +lifecycle lock, cancels the old receive task, and invokes +`_retry_http_bridge_precreated_request` before terminal retirement. The generic +replay predicate intentionally rejects an anchored continuation because a +second ambiguous submission can duplicate a turn. That predicate is too broad +for the narrower watchdog state: a hard owner has not observed any upstream +response event, and the request has no downstream-visible output. + +## Decision + +Add a call-site-only opt-in to the existing pre-created retry path. The opt-in +is accepted only when the session key is hard and the request satisfies a new +pure predicate for the zero-event, pre-`response.created` same-anchor case. The +retry remains on the established account and carries the existing anchor; it +does not clear continuity, select another account, or use a fresh unanchored +replay body. + +The recovery budget is one additional dispatch for this watchdog event. The +existing `replay_count` and clean-close controls remain authoritative, and the +predicate refuses a second attempt. If reconnect or resend fails, the current +fail-closed retirement path settles the request and releases its gate and +reservation exactly as before. + +## Explicit exclusions + +- No default-on indefinite recovery or multi-account replay. +- No retry after `response.created`, any response event, model output, or + downstream sequence/output. +- No changes to durable recovery journal semantics or operation settlement. +- No changes to direct WebSocket behavior. diff --git a/openspec/changes/recover-precreated-anchored-bridge/proposal.md b/openspec/changes/recover-precreated-anchored-bridge/proposal.md new file mode 100644 index 0000000000..4a6a311cc7 --- /dev/null +++ b/openspec/changes/recover-precreated-anchored-bridge/proposal.md @@ -0,0 +1,28 @@ +## Why + +An HTTP Responses bridge can send a hard-continuity `response.create` and +remain completely silent before upstream emits `response.created`. The current +eventless watchdog retires the bridge because the generic pre-created replay +guard treats an anchored continuation as unsafe, even when no response event, +model output, or downstream output exists. Codex then exhausts its own retry +budget and pauses the task instead of receiving a bounded server-side recovery. + +## What Changes + +- Permit one same-account, same-anchor pre-created recovery after the + eventless response-created watchdog fires. +- Keep the recovery proof narrow: hard bridge ownership, an existing + `previous_response_id`, no response id, zero response events, no downstream + sequence/output, and no model output. +- Keep durable operation fencing, account ownership, admission, reservation + settlement, and the existing terminal retirement path unchanged. +- Preserve fail-closed behavior for file-pinned requests, requests with any + response/model/downstream output, soft affinity, and subsequent retries. +- Add route-level regression coverage for a silent upstream followed by a + successful replacement socket, plus negative coverage for unsafe continuations. + +## Impact + +- HTTP Responses bridge pre-created timeout/reconnect behavior. +- No public API, database schema, environment variable, or WebSocket policy + change. diff --git a/openspec/changes/recover-precreated-anchored-bridge/specs/responses-api-compat/spec.md b/openspec/changes/recover-precreated-anchored-bridge/specs/responses-api-compat/spec.md new file mode 100644 index 0000000000..918fb91153 --- /dev/null +++ b/openspec/changes/recover-precreated-anchored-bridge/specs/responses-api-compat/spec.md @@ -0,0 +1,41 @@ +## ADDED Requirements + +### Requirement: Eventless hard continuations receive one bounded recovery + +When an HTTP Responses bridge request has a hard continuity owner and its +upstream socket reaches the missing-`response.created` watchdog before any +upstream response event, the service MUST allow at most one same-account, +same-`previous_response_id` reconnect and resend if all of the following hold: +the request has no assigned response id, no downstream sequence or visible +output, no upstream model output, and remains pending under the current bridge +session. The retry MUST preserve the existing durable operation fence, +admission leases, API-key reservation lifecycle, and account ownership. If the +bounded recovery does not succeed, the service MUST use the existing terminal +retirement path. + +#### Scenario: Silent hard continuation reconnects once + +- **GIVEN** a hard HTTP bridge request carries `previous_response_id` +- **AND** the first upstream socket receives `response.create` but emits no + response event before the client-safe watchdog deadline +- **WHEN** the watchdog handles the timeout +- **THEN** the bridge reconnects on the same account and resends the unchanged + anchored request at most once +- **AND** a replacement socket that emits a normal response completes the + original HTTP request without requiring a client retry + +#### Scenario: Eventless recovery remains fail-closed after one attempt + +- **GIVEN** the first replacement socket also fails before any response event +- **WHEN** the bounded recovery is exhausted +- **THEN** the request is settled through the existing terminal failure path +- **AND** the bridge session is retired rather than retried indefinitely + +#### Scenario: Unsafe continuation is not replayed + +- **GIVEN** an anchored request has a response id, any response event, model + output, downstream-visible output, downstream sequence, soft affinity, or a + file-pinned account requirement +- **WHEN** the upstream socket becomes eventless before completion +- **THEN** the bridge MUST NOT use the same-anchor eventless recovery +- **AND** it MUST preserve the existing fail-closed behavior diff --git a/openspec/changes/recover-precreated-anchored-bridge/tasks.md b/openspec/changes/recover-precreated-anchored-bridge/tasks.md new file mode 100644 index 0000000000..e67112cd43 --- /dev/null +++ b/openspec/changes/recover-precreated-anchored-bridge/tasks.md @@ -0,0 +1,9 @@ +- [x] 1. Add the narrow same-anchor pre-created recovery predicate and pass its + opt-in only from the missing-`response.created` watchdog. +- [x] 2. Keep account, durable-operation, admission, and reservation ownership + unchanged across the recovery; retire and settle on failure. +- [x] 3. Add route-level regression coverage for silent upstream recovery and + exhaustion, plus negative unsafe-continuation coverage. +- [x] 4. Run targeted pytest, Ruff, type/architecture checks, whitespace checks, + and strict OpenSpec validation via + `npx --yes @fission-ai/openspec@1.9.0`. diff --git a/tests/integration/test_http_responses_bridge.py b/tests/integration/test_http_responses_bridge.py index 6498e5b7d5..691b07442f 100644 --- a/tests/integration/test_http_responses_bridge.py +++ b/tests/integration/test_http_responses_bridge.py @@ -8914,6 +8914,247 @@ async def fake_connect_responses_websocket( assert connect_count == 2 +@pytest.mark.asyncio +async def test_v1_responses_http_bridge_retries_when_upstream_never_acknowledges_response_create( + async_client, + monkeypatch, +): + _install_bridge_settings_with_limits( + monkeypatch, + enabled=True, + ) + proxy_module.get_settings().http_responses_session_bridge_stuck_gate_retire_after_seconds = 0.01 + account_id = await _import_account( + async_client, + "acc_http_bridge_missing_created_retry", + "http-bridge-missing-created-retry@example.com", + ) + account = await _get_account(account_id) + silent_upstream = _SilentUpstreamWebSocket() + recovered_upstream = _FakeBridgeUpstreamWebSocket() + upstreams = [silent_upstream, recovered_upstream] + connect_count = 0 + + async def fake_select_account_with_budget( + self, + deadline, + *, + request_id, + kind, + request_stage="first_turn", + sticky_key, + sticky_kind, + reallocate_sticky, + sticky_max_age_seconds, + prefer_earlier_reset_accounts, + routing_strategy, + model, + exclude_account_ids=None, + additional_limit_name=None, + api_key=None, + preferred_account_id=None, + ): + del preferred_account_id + del ( + self, + deadline, + request_id, + kind, + request_stage, + sticky_key, + sticky_kind, + reallocate_sticky, + sticky_max_age_seconds, + prefer_earlier_reset_accounts, + routing_strategy, + model, + exclude_account_ids, + additional_limit_name, + api_key, + ) + return AccountSelection(account=account, error_message=None, error_code=None) + + async def fake_ensure_fresh_with_budget(self, target, *, force=False, timeout_seconds): + del self, force, timeout_seconds + return target + + async def fake_connect_responses_websocket( + headers, + access_token, + account_id_header, + *, + base_url=None, + session=None, + ): + del headers, access_token, account_id_header, base_url, session + nonlocal connect_count + upstream = upstreams[connect_count] + connect_count += 1 + return upstream + + monkeypatch.setattr(proxy_module.ProxyService, "_select_account_with_budget", fake_select_account_with_budget) + monkeypatch.setattr(proxy_module.ProxyService, "_ensure_fresh_with_budget", fake_ensure_fresh_with_budget) + monkeypatch.setattr(proxy_module, "connect_responses_websocket", fake_connect_responses_websocket) + + response = await asyncio.wait_for( + async_client.post( + "/v1/responses", + json={ + "model": "gpt-5.1", + "instructions": "Return exactly OK.", + "input": "retry missing response.created", + "prompt_cache_key": "missing-created-retry-key", + }, + ), + timeout=_TEST_SYNC_TIMEOUT_SECONDS, + ) + + assert response.status_code == 200 + assert connect_count == 2 + assert silent_upstream.closed is True + assert len(silent_upstream.sent_text) == 1 + assert len(recovered_upstream.sent_text) == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "replacement_recovers", + [ + pytest.param(True, id="replacement-completes"), + pytest.param(False, id="replacement-is-also-silent"), + ], +) +async def test_v1_responses_http_bridge_recovers_eventless_anchored_continuation( + async_client, + monkeypatch, + replacement_recovers, +): + """A silent hard continuation gets one bounded same-anchor recovery.""" + + class _FirstThenSilentUpstream(_FakeBridgeUpstreamWebSocket): + async def send_text(self, text: str) -> None: + if self.sent_text: + self.sent_text.append(text) + return + await super().send_text(text) + + _install_bridge_settings_with_limits( + monkeypatch, + enabled=True, + ) + proxy_module.get_settings().http_responses_session_bridge_stuck_gate_retire_after_seconds = 0.01 + account_id = await _import_account( + async_client, + "acc_http_bridge_anchored_missing_created_retry", + "http-bridge-anchored-missing-created-retry@example.com", + ) + account = await _get_account(account_id) + first_then_silent_upstream = _FirstThenSilentUpstream("resp_anchored_first") + recovered_upstream = ( + _FakeBridgeUpstreamWebSocket("resp_anchored_recovered") if replacement_recovers else _SilentUpstreamWebSocket() + ) + upstreams = [first_then_silent_upstream, recovered_upstream] + connect_count = 0 + + async def fake_select_account_with_budget( + self, + deadline, + *, + request_id, + kind, + request_stage="first_turn", + sticky_key, + sticky_kind, + reallocate_sticky, + sticky_max_age_seconds, + prefer_earlier_reset_accounts, + routing_strategy, + model, + exclude_account_ids=None, + additional_limit_name=None, + api_key=None, + preferred_account_id=None, + ): + del preferred_account_id + del ( + self, + deadline, + request_id, + kind, + request_stage, + sticky_key, + sticky_kind, + reallocate_sticky, + sticky_max_age_seconds, + prefer_earlier_reset_accounts, + routing_strategy, + model, + exclude_account_ids, + additional_limit_name, + api_key, + ) + return AccountSelection(account=account, error_message=None, error_code=None) + + async def fake_ensure_fresh_with_budget(self, target, *, force=False, timeout_seconds): + del self, force, timeout_seconds + return target + + async def fake_connect_responses_websocket( + headers, + access_token, + account_id_header, + *, + base_url=None, + session=None, + ): + del headers, access_token, account_id_header, base_url, session + nonlocal connect_count + upstream = upstreams[connect_count] + connect_count += 1 + return upstream + + monkeypatch.setattr(proxy_module.ProxyService, "_select_account_with_budget", fake_select_account_with_budget) + monkeypatch.setattr(proxy_module.ProxyService, "_ensure_fresh_with_budget", fake_ensure_fresh_with_budget) + monkeypatch.setattr(proxy_module, "connect_responses_websocket", fake_connect_responses_websocket) + + session_headers = {"session_id": "hard-eventless-anchored-continuation"} + first_response = await async_client.post( + "/v1/responses", + headers=session_headers, + json={ + "model": "gpt-5.1", + "instructions": "Return exactly OK.", + "input": "first turn", + "prompt_cache_key": "anchored-missing-created-key", + }, + ) + assert first_response.status_code == 200, first_response.text + previous_response_id = first_response.json()["id"] + + second_response = await asyncio.wait_for( + async_client.post( + "/v1/responses", + headers=session_headers, + json={ + "model": "gpt-5.1", + "instructions": "Return exactly OK.", + "input": "anchored continuation", + "previous_response_id": previous_response_id, + "prompt_cache_key": "anchored-missing-created-key", + }, + ), + timeout=_TEST_SYNC_TIMEOUT_SECONDS, + ) + + assert second_response.status_code == (200 if replacement_recovers else 502), second_response.text + assert connect_count == 2 + assert first_then_silent_upstream.closed is True + assert len(first_then_silent_upstream.sent_text) == 2 + assert len(recovered_upstream.sent_text) == 1 + assert json.loads(first_then_silent_upstream.sent_text[1])["previous_response_id"] == previous_response_id + assert json.loads(recovered_upstream.sent_text[0])["previous_response_id"] == previous_response_id + + @pytest.mark.asyncio async def test_backend_responses_http_bridge_retries_precreated_server_overload(async_client, monkeypatch): _install_bridge_settings(monkeypatch, enabled=True) diff --git a/tests/unit/test_proxy_http_bridge.py b/tests/unit/test_proxy_http_bridge.py index fea5f4e594..75cf91e4f3 100644 --- a/tests/unit/test_proxy_http_bridge.py +++ b/tests/unit/test_proxy_http_bridge.py @@ -828,6 +828,62 @@ def _make_eventless_http_bridge_owner( ) +@pytest.mark.parametrize( + ("field_name", "field_value"), + [ + ("previous_response_id", None), + ("response_id", "resp-created"), + ("response_event_count", 1), + ("downstream_visible", True), + ("last_downstream_sequence_number", 0), + ("upstream_model_output_seen", True), + ("awaiting_response_created", False), + ("file_required_preferred_account", True), + ("missing_response_created_retry_count", 1), + ], +) +def test_http_bridge_same_anchor_eventless_recovery_requires_unambiguous_owner_state( + field_name: str, + field_value: object, +) -> None: + request_state = _make_eventless_http_bridge_owner() + request_state.request_text = '{"type":"response.create","input":"continue"}' + request_state.previous_response_id = "resp-parent" + setattr(request_state, field_name, field_value) + + assert http_bridge_request_submit_module._http_bridge_can_replay_same_anchor_before_created(request_state) is False + + +@pytest.mark.asyncio +async def test_http_bridge_same_anchor_eventless_recovery_requires_hard_session_owner( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + request_state = _make_eventless_http_bridge_owner() + request_state.request_text = '{"type":"response.create","input":"continue"}' + request_state.previous_response_id = "resp-parent" + session = _make_bridge_session( + key=proxy_service._HTTPBridgeSessionKey("prompt_cache", "soft-anchor", None), + pending_requests=deque([request_state]), + queued_request_count=1, + ) + session.last_upstream_close_code = 1011 + session.upstream = cast(UpstreamWebSocket, SimpleNamespace(send_text=AsyncMock(), close=AsyncMock())) + reconnect = AsyncMock() + monkeypatch.setattr(proxy_service, "get_settings", lambda: _make_app_settings()) + monkeypatch.setattr(service, "_reconnect_http_bridge_session", reconnect) + + assert ( + await service._retry_http_bridge_precreated_request( + session, + allow_same_anchor_before_created=True, + ) + is False + ) + reconnect.assert_not_awaited() + assert request_state.missing_response_created_retry_count == 0 + + class _SilentEventlessUpstream: """Upstream double that never produces a response event, for eventless-timeout tests.""" @@ -22466,7 +22522,7 @@ async def close(self) -> None: assert owner.response_event_count == (1 if leading_telemetry else 0) if leading_telemetry: assert owner.latency_first_upstream_event_ms is not None - retry_precreated.assert_awaited_once_with(session) + retry_precreated.assert_awaited_once_with(session, allow_same_anchor_before_created=True) assert write_request_log.await_count == 2 assert {call.kwargs["error_code"] for call in write_request_log.await_args_list} == {"upstream_request_timeout"} fail_reader.assert_awaited_once()