feat(proxy): report websocket cleanup phase

.all-contributorsrc

@@ -1222,6 +1222,16 @@
       "contributions": [
         "code"
       ]
+    },
+    {
+      "login": "kevinsslin",
+      "name": "Kevin Lin",
+      "avatar_url": "https://avatars.githubusercontent.com/u/86810837?v=4",
+      "profile": "https://github.com/kevinsslin",
+      "contributions": [
+        "code",
+        "test"
+      ]
     }
   ],
   "contributorsPerLine": 7,

README.md

@@ -282,6 +282,10 @@ Thanks goes to these wonderful people ([emoji key](https://allcontributors.org/e
       <td align="center" valign="top" width="14.28%"><a href="https://github.com/rknightion"><img src="https://avatars.githubusercontent.com/u/12484127?v=4?s=100" width="100px;" alt="rknightion"/><br /><sub><b>rknightion</b></sub></a><br /><a href="https://github.com/Soju06/codex-lb/commits?author=rknightion" title="Code">💻</a> <a href="https://github.com/Soju06/codex-lb/commits?author=rknightion" title="Tests">⚠️</a></td>
       <td align="center" valign="top" width="14.28%"><a href="https://github.com/glopyglerky"><img src="https://avatars.githubusercontent.com/u/189872235?v=4?s=100" width="100px;" alt="glopyglerky"/><br /><sub><b>glopyglerky</b></sub></a><br /><a href="https://github.com/Soju06/codex-lb/commits?author=glopyglerky" title="Code">💻</a> <a href="https://github.com/Soju06/codex-lb/commits?author=glopyglerky" title="Tests">⚠️</a></td>
       <td align="center" valign="top" width="14.28%"><a href="https://github.com/iqbalmaulana03"><img src="https://avatars.githubusercontent.com/u/78488507?v=4?s=100" width="100px;" alt="Ahmad Maulana Iqbal"/><br /><sub><b>Ahmad Maulana Iqbal</b></sub></a><br /><a href="https://github.com/Soju06/codex-lb/commits?author=iqbalmaulana03" title="Code">💻</a> <a href="https://github.com/Soju06/codex-lb/commits?author=iqbalmaulana03" title="Tests">⚠️</a></td>
+      <td align="center" valign="top" width="14.28%"><a href="https://github.com/Dvredin"><img src="https://avatars.githubusercontent.com/u/32620766?v=4?s=100" width="100px;" alt="Dvredin"/><br /><sub><b>Dvredin</b></sub></a><br /><a href="https://github.com/Soju06/codex-lb/commits?author=Dvredin" title="Code">💻</a></td>
+    </tr>
+    <tr>
+      <td align="center" valign="top" width="14.28%"><a href="https://github.com/kevinsslin"><img src="https://avatars.githubusercontent.com/u/86810837?v=4?s=100" width="100px;" alt="Kevin Lin"/><br /><sub><b>Kevin Lin</b></sub></a><br /><a href="https://github.com/Soju06/codex-lb/commits?author=kevinsslin" title="Code">💻</a> <a href="https://github.com/Soju06/codex-lb/commits?author=kevinsslin" title="Tests">⚠️</a></td>
     </tr>
   </tbody>
 </table>

app/modules/proxy/_service/websocket/mixin.py

@@ -2513,8 +2513,10 @@ def take_reader_replay_request_state() -> _WebSocketRequestState | None:
             cleanup_timeout = shutdown_state.remaining_drain_timeout_seconds()
             if cleanup_timeout is None:
                 cleanup_timeout = _facade()._TASK_CANCEL_TIMEOUT_SECONDS
+            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
@@ -2527,13 +2529,15 @@ 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,
                         timeout_seconds=cleanup_timeout,
                     )
                 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",
@@ -2550,6 +2554,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=cleanup_timeout,
@@ -2564,6 +2569,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=cleanup_timeout,
@@ -2580,6 +2586,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,
@@ -2598,6 +2605,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,
@@ -2615,6 +2623,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,
@@ -2637,6 +2646,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
@@ -2645,6 +2655,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(),
@@ -2670,8 +2681,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()),
                 )
 

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.

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

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.

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)