reliability: fix audited lifecycle and state edge cases - #962
Open
yisding wants to merge 4 commits into
Open
Conversation
📝 WalkthroughWalkthroughThis PR hardens configuration validation, framework-state serialization, journal ownership, session and STT teardown, WebSocket cleanup, telephony transitions, transport recovery, and latency data validation. It adds retryable cleanup state and broad regression coverage. ChangesConfiguration and data safety
Lifecycle and transport handling
Validation and supporting coverage
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant Session
participant STTCommitter
participant RuntimeSupervisor
participant Provider
Session->>STTCommitter: cancel or end stream
STTCommitter->>Provider: execute bounded provider operation
STTCommitter->>RuntimeSupervisor: retain unfinished task
RuntimeSupervisor->>STTCommitter: retry retained cleanup
STTCommitter-->>Session: report cleanup status
Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
yisding
marked this pull request as ready for review
August 7, 2026 14:07
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 2b0659c7f3
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
Comment on lines
+329
to
+332
| if type(telephony) is not TelephonyConfig: | ||
| raise EasyConfigError("telephony must be a TelephonyConfig instance or None.") | ||
| outbound = telephony.outbound | ||
| if outbound is not None and type(outbound) is not OutboundCallConfig: |
There was a problem hiding this comment.
When a caller supplies a subclass of TelephonyConfig or OutboundCallConfig, create_session() now rejects it before build validation, even though EasyConfig.validate_for_build() explicitly accepts subclasses with isinstance() and the previous copy path handled them. This breaks otherwise valid extensions that add telephony policy fields; use subclass-aware checks for both guards or validate before copying.
Useful? React with 👍 / 👎.
Resolve ratchet manifest conflicts by recounting merged entries. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…n errors Address PR 962 review comments: the runtime-copy guard now matches validate_for_build's isinstance semantics, and the WebRTC SDP failure response no longer echoes exception details to the client. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
- bound the rejected-start STT cleanup join after caller cancellation so an unresponsive provider cannot make stop(force=True) uncancellable - report cooperatively-cancelled listener waits as force-timeout failures and retain the listener for retry instead of discarding it - validate bridge event text only for text-bearing kinds; duck-typed tool events may carry text=None - restore substring credential-name filtering for the generic workflow __dict__ fallback snapshot (authtoken-style names leaked) - let observer-spawned tasks transition after the owning call-state transition settles; only an active transition rejects them - classify the new teardown budgets and refresh source ratchets Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 20
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/easycat/debugger/server.py`:
- Around line 705-708: Remove the duplicated “an” in the comment near the
state-changing request handling, so it reads “can pass an explicit same-origin
value.”
In `@src/easycat/integrations/agents/_agent_runner.py`:
- Around line 208-213: Update the validation branches in AgentRunnerConfig for
preemptive_generation and preemptive_max_retries so every line stays within
Ruff’s 99-character limit. Preserve the existing validation behavior and noqa
suppression while splitting the error statements or extracting their messages
into local constants.
- Around line 65-67: Update the text extraction in the event validation flow
around _require_agent_text() to use None or a sentinel instead of an
empty-string default when the event lacks text, ensuring text-bearing events
without a text attribute are rejected before forwarding.
In `@src/easycat/integrations/agents/_state_serialization.py`:
- Around line 74-77: Update _remove_secret_shaped_keys to canonicalize unordered
state before serialization: recursively scrub set items, sort them using a safe
canonical JSON key, and serialize scrubbed mappings with sort_keys=True so
equivalent states produce identical bytes. Preserve sequence handling, and add a
regression test covering a multi-item set with varying iteration order.
In `@src/easycat/runtime/scope.py`:
- Around line 584-591: Update prune_empty_child around
_close_admission_recursive so its contract reflects that a refused prune may
still close admission and leave the child subtree in CLOSING, or move the
emptiness check ahead of admission closure to preserve a no-side-effect False
result. Ensure the docstring and return behavior consistently describe whichever
contract is implemented.
In `@src/easycat/server/voice_server.py`:
- Around line 640-646: Guard the completed-task result in the sweep-success path
before calling _record_incomplete_hard_sweep: catch exceptions re-raised by
sweep_task.result(), append the exception to cleanup_errors, and pass None as
report when retrieval fails. Preserve the existing successful result behavior
and ensure _record_incomplete_hard_sweep still executes so
_finalize_stop_cleanup remains retry-safe.
In `@src/easycat/session/_stt_committer.py`:
- Line 244: Promote the repeated "stt_segment_commit" identifier to a class
constant alongside FINAL_CLOSE_TASK_NAME, PROVIDER_END_TASK_NAME, and
PROVIDER_CLOSE_TASK_NAME. Replace every occurrence in lifecycle task lookups,
cancellation targets, and journal payloads with that constant, including the
references in the relevant session methods.
- Around line 1016-1049: Bound the retry loop in
_finish_transferred_provider_close with an explicit attempt cap or wall-clock
deadline. When the bound is reached, stop retrying while preserving
_provider_close_error, allow the owned task to settle, and leave
_provider_close_pending consistent so retry_transferred_provider_close can
perform any later attempt.
In `@src/easycat/teardown_budgets.py`:
- Around line 39-44: Adjust the rejected-start STT cleanup timing constants used
by _cleanup_rejected_stt_start() so their combined join and cancel-grace
duration fits within SESSION_FORCE_START_LOCK_TIMEOUT_S when invoked from the
force-cancellation path, or consistently increase that force-path cap; preserve
bounded cleanup before the remaining stop path proceeds.
In `@src/easycat/telephony/call_state.py`:
- Around line 426-432: Move _transition_context out of
OutboundCallStateMachine.__init__ and define one module-level ContextVar shared
by all instances. Store both the owning state machine and task in its value,
updating all transition-context reads and writes to use that shared pair, while
retaining _active_transition_owner as the current task.
In `@src/easycat/telephony/retry.py`:
- Around line 128-135: Update get_delay() to handle a zero base_delay_s when
state.attempts is still zero before evaluating the exponentiation, returning the
defined zero delay (or otherwise applying the established pre-attempt contract).
Preserve the existing overflow and non-finite delay clamping for recorded
attempts.
In `@tests/debugger/test_server_replay_export.py`:
- Around line 211-215: Use the canonical _SAFE_HEADERS from
tests/debugger/_server_helpers.py across both modules: in
tests/debugger/test_server_replay_export.py, import it and replace the local
mappings at lines 193-197, 211-215, 246-250, 268-272, 301-305, 321-325, 352-356,
386-390, and 404-408; in tests/debugger/test_aec_diagnostics.py, remove the
module-level definition at lines 276-280 and import _SAFE_HEADERS instead.
In `@tests/debugger/test_server_rest_api.py`:
- Around line 523-547: Add a test for
test_api_audio_waveform_rejects_pcm_over_memory_limit where the raw audio blobs
remain within _WAVEFORM_MAX_PCM_BYTES but _coerce_frames_to_format expands the
coerced PCM beyond that limit. Assert the post-coercion request still returns
HTTP 413 with the existing WAVEFORM_AUDIO_TOO_LARGE payload, covering the guard
after format conversion rather than the raw accumulation guard.
In `@tests/server/test_capacity_gate_drain.py`:
- Around line 247-268: Strengthen
test_timed_safe_await_observes_cancelled_gather_result by directly verifying
that _observe_future_result was attached to the gathered future after
_safe_await returns. Inspect the future’s registered callbacks before deleting
gathered, while preserving the existing exception-handler assertion.
In `@tests/server/test_shutdown_draining.py`:
- Around line 411-415: Update the shutdown test around server.stop() to expect
and assert the intended cleanup exception for the cooperative cancellation path.
Remove the assertion that server._ws_server is None, while preserving assertions
that the hanging listener is closed and cleanup task tracking is cleared.
In `@tests/server/test_websocket_runtime_drain.py`:
- Around line 294-299: Update the assertions in the test around the retry flow
to inspect runtime._connection_cleanup_retry directly instead of using
runtime._connections as a proxy. Keep the existing setup and drain calls
unchanged, and assert the ledger’s expected state before and after the retry
cleanup.
In `@tests/session/test_stt_committer.py`:
- Around line 737-748: Update the test surrounding _HangingSTT and
committer.end_stream to drain the runtime/provider task scope after the timeout,
matching the cleanup pattern used by the sibling tests. Ensure the cleanup runs
in a finally block so the parked hanging task is released even if an assertion
fails, while preserving the existing assertions.
In `@tests/stages/test_stages.py`:
- Around line 1407-1427: Update
test_turn_stage_materializes_generator_input_once_for_detection to pass a
journal and artifact store using the existing test pattern, then assert the
stage_start artifact contains exactly chunk.data. Keep the existing detect-input
assertion and result checks so the test verifies both capture and detection
receive identical audio.
In `@tests/telephony/test_retry_strategy.py`:
- Around line 117-130: Add the return annotation -> None to both test function
definitions in this test module, including
test_backoff_overflow_caps_at_max_delay and the additional test around the
referenced second location; leave their bodies and behavior unchanged.
In `@tests/turns/test_turn_manager.py`:
- Around line 237-243: Set detector.release before calling manager.shutdown() in
the test cleanup flow, so the cancelled detect task can finish while shutdown
awaits replaced. Preserve the existing assertions and retain the finally block’s
cleanup and exception-gathering behavior.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 72ddc6af-7e4b-449a-87d4-6a5250e8d67a
📒 Files selected for processing (87)
src/easycat/_concurrency.pysrc/easycat/config/_factory.pysrc/easycat/config/easy.pysrc/easycat/debugger/server.pysrc/easycat/integrations/agents/_agent_runner.pysrc/easycat/integrations/agents/_state_serialization.pysrc/easycat/integrations/agents/generic_workflow.pysrc/easycat/integrations/agents/langchain.pysrc/easycat/integrations/agents/langgraph.pysrc/easycat/integrations/agents/llama_agents.pysrc/easycat/integrations/agents/openai_agents.pysrc/easycat/integrations/agents/pydantic_ai.pysrc/easycat/integrations/agents/responses_api.pysrc/easycat/integrations/agents/template.pysrc/easycat/runtime/journal_sql.pysrc/easycat/runtime/scope.pysrc/easycat/server/transports.pysrc/easycat/server/voice_server.pysrc/easycat/session/_audio_router.pysrc/easycat/session/_debug_backends.pysrc/easycat/session/_journal_sink.pysrc/easycat/session/_session.pysrc/easycat/session/_stt_committer.pysrc/easycat/session/_turn_runner.pysrc/easycat/session_manager.pysrc/easycat/stages/turn.pysrc/easycat/teardown_budgets.pysrc/easycat/telephony/call_state.pysrc/easycat/telephony/retry.pysrc/easycat/telephony/voicemail.pysrc/easycat/timeouts.pysrc/easycat/transports/_webrtc_stats.pysrc/easycat/transports/local.pysrc/easycat/transports/webrtc.pysrc/easycat/turn_manager.pysrc/easycat/validation/_latency_artifacts.pysrc/easycat/validation/_latency_baseline.pysrc/easycat/validation/_latency_budgets.pysrc/easycat/validation/_latency_models.pysrc/easycat/validation/redaction.pytests/cli/test_doctor.pytests/config/test_session_creation.pytests/core/test_timeouts.pytests/debugger/_server_helpers.pytests/debugger/test_aec_diagnostics.pytests/debugger/test_server_replay_export.pytests/debugger/test_server_rest_api.pytests/debugger/test_server_route_controller.pytests/debugger/test_server_security_origin.pytests/integration/test_session_pipeline.pytests/integrations/agents/test_agent_runner.pytests/integrations/agents/test_bridge_template.pytests/integrations/agents/test_generic_workflow_bridge.pytests/integrations/agents/test_pydantic_ai_v2.pytests/ratchets/pause-generation-manifest.jsontests/ratchets/source-baseline.jsontests/ratchets/teardown-budget-manifest.jsontests/ratchets/turn-commit-manifest.jsontests/ratchets/turn-lifecycle-manifest.jsontests/runtime/test_scope.pytests/runtime/test_sqlite_journal.pytests/server/test_capacity_gate_drain.pytests/server/test_shutdown_draining.pytests/server/test_voice_server_lifecycle.pytests/server/test_websocket_runtime_drain.pytests/session/test_audio_router.pytests/session/test_journal_sink.pytests/session/test_session_journal_accounting.pytests/session/test_session_lifecycle_teardown.pytests/session/test_session_manager.pytests/session/test_session_stop_ordering.pytests/session/test_session_streaming_behavior.pytests/session/test_stt_committer.pytests/session/test_turn_runner.pytests/stages/test_stages.pytests/teaching/test_chapter_15_doctor_contract.pytests/telephony/test_call_state_basics.pytests/telephony/test_call_state_gate.pytests/telephony/test_outbound_config.pytests/telephony/test_retry_strategy.pytests/telephony/test_voicemail.pytests/transports/test_local_transport.pytests/transports/test_webrtc_lifecycle_server.pytests/transports/test_webrtc_stats_artifacts.pytests/turns/test_turn_manager.pytests/validation/test_latency_boundaries.pytests/validation/test_latency_percentiles.py
Comment on lines
705
to
+708
| # On state-changing requests, a missing Origin from a | ||
| # browser is suspicious — refuse rather than trust the | ||
| # caller blindly. Server-to-server clients can pass an | ||
| # explicit ``Origin: http://localhost`` or use | ||
| # ``allow_remote``. | ||
| # an explicit same-origin value or use ``allow_remote``. |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Remove the duplicated article in the comment.
Line 708 reads "can pass an / an explicit same-origin value". The word an is repeated across the line break.
📝 Proposed fix
# On state-changing requests, a missing Origin from a
# browser is suspicious — refuse rather than trust the
# caller blindly. Server-to-server clients can pass an
- # an explicit same-origin value or use ``allow_remote``.
+ # explicit same-origin value or use ``allow_remote``.📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| # On state-changing requests, a missing Origin from a | |
| # browser is suspicious — refuse rather than trust the | |
| # caller blindly. Server-to-server clients can pass an | |
| # explicit ``Origin: http://localhost`` or use | |
| # ``allow_remote``. | |
| # an explicit same-origin value or use ``allow_remote``. | |
| # On state-changing requests, a missing Origin from a | |
| # browser is suspicious — refuse rather than trust the | |
| # caller blindly. Server-to-server clients can pass an | |
| # explicit same-origin value or use ``allow_remote``. |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/easycat/debugger/server.py` around lines 705 - 708, Remove the duplicated
“an” in the comment near the state-changing request handling, so it reads “can
pass an explicit same-origin value.”
Comment on lines
+65
to
+67
| getattr(event, "text", ""), | ||
| source=f"agent bridge {kind} event text", | ||
| ) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Reject text-bearing events that omit text.
Line 65 converts a missing text attribute into "". The malformed event then passes validation and is forwarded without a text attribute. Downstream consumers can fail when they read event.text.
Use None or a sentinel as the default so _require_agent_text() rejects the event.
Proposed fix
- getattr(event, "text", ""),
+ getattr(event, "text", None),📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| getattr(event, "text", ""), | |
| source=f"agent bridge {kind} event text", | |
| ) | |
| getattr(event, "text", None), | |
| source=f"agent bridge {kind} event text", | |
| ) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/easycat/integrations/agents/_agent_runner.py` around lines 65 - 67,
Update the text extraction in the event validation flow around
_require_agent_text() to use None or a sentinel instead of an empty-string
default when the event lacks text, ensuring text-bearing events without a text
attribute are rejected before forwarding.
Comment on lines
+208
to
213
| if not isinstance(self.preemptive_generation, bool): | ||
| raise ValueError("AgentRunnerConfig.preemptive_generation must be a boolean") # noqa: TRY004 domain-specific validation error | ||
| if isinstance(self.preemptive_max_retries, bool) or not isinstance( | ||
| self.preemptive_max_retries, int | ||
| ): | ||
| raise ValueError("AgentRunnerConfig.preemptive_max_retries must be an integer") # noqa: TRY004 domain-specific validation error |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Keep validation lines within the configured limit.
Lines 209 and 213 exceed 99 characters because of the inline noqa comments. Split the error messages or extract them into local constants.
As per coding guidelines, **/*.py must “keep lines within Ruff's configured 99-character limit.”
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/easycat/integrations/agents/_agent_runner.py` around lines 208 - 213,
Update the validation branches in AgentRunnerConfig for preemptive_generation
and preemptive_max_retries so every line stays within Ruff’s 99-character limit.
Preserve the existing validation behavior and noqa suppression while splitting
the error statements or extracting their messages into local constants.
Source: Coding guidelines
Comment on lines
+74
to
+77
| if isinstance(value, AbstractSet): | ||
| return [_remove_secret_shaped_keys(item) for item in value] | ||
| if isinstance(value, Sequence) and not isinstance(value, str | bytes | bytearray): | ||
| return [_remove_secret_shaped_keys(item) for item in value] |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
Canonicalize unordered framework state before encoding.
AbstractSet iteration order is unspecified. The direct list conversion can produce different artifact bytes for the same logical state. Mapping insertion order can also vary by state producer.
Sort normalized set items with a safe canonical JSON key. Encode scrubbed mappings with sort_keys=True. Add a regression test with a multi-item set.
Proposed fix
+def _canonical_json(value: Any) -> str:
+ return json.dumps(
+ value,
+ default=_redacted_string,
+ ensure_ascii=False,
+ allow_nan=False,
+ sort_keys=True,
+ )
+
def _remove_secret_shaped_keys(value: Any) -> Any:
...
if isinstance(value, AbstractSet):
- return [_remove_secret_shaped_keys(item) for item in value]
+ items = [_remove_secret_shaped_keys(item) for item in value]
+ return sorted(items, key=_canonical_json)
...
- return json.dumps(
- scrubbed,
- default=_redacted_string,
- ensure_ascii=False,
- allow_nan=False,
- ).encode("utf-8")
+ return _canonical_json(scrubbed).encode("utf-8")Also applies to: 108-113
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/easycat/integrations/agents/_state_serialization.py` around lines 74 -
77, Update _remove_secret_shaped_keys to canonicalize unordered state before
serialization: recursively scrub set items, sort them using a safe canonical
JSON key, and serialize scrubbed mappings with sort_keys=True so equivalent
states produce identical bytes. Preserve sequence handling, and add a regression
test covering a multi-item set with varying iteration order.
Comment on lines
+584
to
+591
| child._close_admission_recursive() | ||
| if ( | ||
| child.tasks() | ||
| or child.children() | ||
| or child._pending_finalizer_names() | ||
| or child.terminal_results() | ||
| ): | ||
| return False |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Document that a refused prune still closes the child's admission.
_close_admission_recursive() runs before the emptiness check. When the check fails and the method returns False, the child subtree stays in CLOSING and survivor_registry.close_owner() has already run for every scope in it. The child can no longer admit tasks, finalizers, or sub-children after that point.
The current callers tolerate this: STTCommitter._provider_operation_scope only prunes scopes it has already marked retired, and _prune_provider_operation_scope guards with scope.empty first. The docstring and the False return still read as "no change was made", so a future caller can use prune_empty_child as a probe and silently disable a live child scope.
Either state the side effect in the docstring or move the emptiness check before the admission closure.
♻️ Option: check emptiness before closing admission
def prune_empty_child(self, child: RuntimeScope) -> bool:
- """Retire and unlink one settled direct child without reopening its owner."""
+ """Retire and unlink one settled direct child without reopening its owner.
+
+ A child that is not settled is left registered, but its admission is
+ already closed by this call.
+ """
if child.parent is not self:
raise ValueError("RuntimeScope child does not belong to this parent")
child._close_admission_recursive()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/easycat/runtime/scope.py` around lines 584 - 591, Update
prune_empty_child around _close_admission_recursive so its contract reflects
that a refused prune may still close admission and leave the child subtree in
CLOSING, or move the emptiness check ahead of admission closure to preserve a
no-side-effect False result. Ensure the docstring and return behavior
consistently describe whichever contract is implemented.
Comment on lines
+294
to
+299
| assert runtime._connections == {key: connection} | ||
|
|
||
| connection.fail_close = False | ||
| await runtime.drain(server, drain_timeout_s=0.0, force_timeout_s=1.0) | ||
|
|
||
| assert runtime._connections == {} |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Assert the retry ledger directly.
Line 294 and line 299 check runtime._connections, which is an indirect proxy for the retry state this test targets. test_force_timeout_is_shared_across_all_runtime_cleanup_steps already asserts runtime._connection_cleanup_retry directly at lines 577 and 590. Use the same assertion here so a regression that leaves the ledger populated is caught.
♻️ Proposed refactor
assert "WebSocket connection close task" in caplog.text
assert "retryable connection failure" in caplog.text
assert runtime._connections == {key: connection}
+ assert runtime._connection_cleanup_retry == {key: connection}
connection.fail_close = False
await runtime.drain(server, drain_timeout_s=0.0, force_timeout_s=1.0)
assert runtime._connections == {}
+ assert runtime._connection_cleanup_retry == {}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| assert runtime._connections == {key: connection} | |
| connection.fail_close = False | |
| await runtime.drain(server, drain_timeout_s=0.0, force_timeout_s=1.0) | |
| assert runtime._connections == {} | |
| assert runtime._connections == {key: connection} | |
| assert runtime._connection_cleanup_retry == {key: connection} | |
| connection.fail_close = False | |
| await runtime.drain(server, drain_timeout_s=0.0, force_timeout_s=1.0) | |
| assert runtime._connections == {} | |
| assert runtime._connection_cleanup_retry == {} |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/server/test_websocket_runtime_drain.py` around lines 294 - 299, Update
the assertions in the test around the retry flow to inspect
runtime._connection_cleanup_retry directly instead of using runtime._connections
as a proxy. Keep the existing setup and drain calls unchanged, and assert the
ledger’s expected state before and after the retry cleanup.
Comment on lines
737
to
+748
| stt = _HangingSTT() | ||
| no_turn = TurnContext("no-turn", CancelToken()) | ||
| tm = TurnManager(bus, config=TurnManagerConfig()) | ||
| committer = STTCommitter( | ||
| wiring=make_wiring(stt=lambda: stt, emit=_emit), | ||
| event_bus=bus, | ||
| journal_sink=sink, | ||
| runtime_scope=RuntimeScope(), | ||
| committer, _stt, emitted, _no_turn, _tm = _make_committer( | ||
| stt=stt, | ||
| timeout_config=TimeoutConfig(stt_timeout=0.01), | ||
| segment_silence_ms=0, | ||
| no_turn=no_turn, | ||
| turn_manager=tm, | ||
| ) | ||
| turn = _new_turn() | ||
|
|
||
| await committer.end_stream(turn) | ||
|
|
||
| assert stt.end_stream_calls == 1 | ||
| assert turn.pending_stt_segment_futures == [] | ||
| stt_errors = [e for e in errors if e.stage == ErrorStage.STT] | ||
| stt_errors = [e for e in emitted if isinstance(e, Error) and e.stage == ErrorStage.STT] |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Release the hanging provider task at the end of this test.
_HangingSTT.end_stream awaits asyncio.Event().wait() and never returns. end_stream times out and parks that owned task. The test then finishes without draining it.
The task stays pending until the event loop closes. That produces a "Task was destroyed but it is pending" warning and holds the survivor reservation for the rest of the loop's life. The sibling tests added in this file already handle this: Line 640 drains the runtime scope and Line 430 drains the provider-error scope in a finally.
Add the same cleanup so the test does not leak a pending task into later tests on the same loop.
💚 Proposed cleanup
turn = _new_turn()
- await committer.end_stream(turn)
-
- assert stt.end_stream_calls == 1
- assert turn.pending_stt_segment_futures == []
- stt_errors = [e for e in emitted if isinstance(e, Error) and e.stage == ErrorStage.STT]
- assert stt_errors
- assert stt_errors[0].provider == "openai"
+ try:
+ await committer.end_stream(turn)
+
+ assert stt.end_stream_calls == 1
+ assert turn.pending_stt_segment_futures == []
+ stt_errors = [e for e in emitted if isinstance(e, Error) and e.stage == ErrorStage.STT]
+ assert stt_errors
+ assert stt_errors[0].provider == "openai"
+ finally:
+ await committer._runtime_scope.cancel_and_drain()📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| stt = _HangingSTT() | |
| no_turn = TurnContext("no-turn", CancelToken()) | |
| tm = TurnManager(bus, config=TurnManagerConfig()) | |
| committer = STTCommitter( | |
| wiring=make_wiring(stt=lambda: stt, emit=_emit), | |
| event_bus=bus, | |
| journal_sink=sink, | |
| runtime_scope=RuntimeScope(), | |
| committer, _stt, emitted, _no_turn, _tm = _make_committer( | |
| stt=stt, | |
| timeout_config=TimeoutConfig(stt_timeout=0.01), | |
| segment_silence_ms=0, | |
| no_turn=no_turn, | |
| turn_manager=tm, | |
| ) | |
| turn = _new_turn() | |
| await committer.end_stream(turn) | |
| assert stt.end_stream_calls == 1 | |
| assert turn.pending_stt_segment_futures == [] | |
| stt_errors = [e for e in errors if e.stage == ErrorStage.STT] | |
| stt_errors = [e for e in emitted if isinstance(e, Error) and e.stage == ErrorStage.STT] | |
| stt = _HangingSTT() | |
| committer, _stt, emitted, _no_turn, _tm = _make_committer( | |
| stt=stt, | |
| timeout_config=TimeoutConfig(stt_timeout=0.01), | |
| ) | |
| turn = _new_turn() | |
| try: | |
| await committer.end_stream(turn) | |
| assert stt.end_stream_calls == 1 | |
| assert turn.pending_stt_segment_futures == [] | |
| stt_errors = [e for e in emitted if isinstance(e, Error) and e.stage == ErrorStage.STT] | |
| assert stt_errors | |
| assert stt_errors[0].provider == "openai" | |
| finally: | |
| await committer._runtime_scope.cancel_and_drain() |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/session/test_stt_committer.py` around lines 737 - 748, Update the test
surrounding _HangingSTT and committer.end_stream to drain the runtime/provider
task scope after the timeout, matching the cleanup pattern used by the sibling
tests. Ensure the cleanup runs in a finally block so the parked hanging task is
released even if an assertion fails, while preserving the existing assertions.
Comment on lines
+1407
to
+1427
| async def test_turn_stage_materializes_generator_input_once_for_detection(self): | ||
| chunk = AudioChunk(data=b"\x01\x02", format=PCM16_MONO_16K) | ||
| seen: list[AudioChunk] = [] | ||
|
|
||
| class _CapturingSmartTurn: | ||
| async def detect(self, audio_chunks): | ||
| seen.extend(audio_chunks) | ||
| return {"prediction": 1, "probability": 0.95} | ||
|
|
||
| def audio_window(): | ||
| yield chunk | ||
|
|
||
| result = await TurnStage(_CapturingSmartTurn()).execute( | ||
| audio_window(), | ||
| _make_ctx(), | ||
| _make_turn(), | ||
| ) | ||
|
|
||
| assert result["prediction"] == 1 | ||
| assert seen == [chunk] | ||
|
|
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Also assert the captured artifact bytes.
The test pins one half of the fix: detect receives the materialized chunk. The docstring of _materialize_one_shot_iterable states the full contract — "capture and detection see identical audio". The capture half is not asserted here, so a regression that materializes for detect but drains the generator before _concat_chunks would still pass.
Pass a journal and an artifact store, then assert the stage_start artifact equals chunk.data. This file already uses that pattern at Line 1259-1261.
💚 Proposed additional assertion
+ journal = InMemoryRingBuffer(capacity=100)
+ artifacts = InMemoryArtifactStore()
result = await TurnStage(_CapturingSmartTurn()).execute(
audio_window(),
- _make_ctx(),
+ _make_ctx(journal=journal, artifact_store=artifacts),
_make_turn(),
)
assert result["prediction"] == 1
assert seen == [chunk]
+ start = next(record for record in journal.read() if record.name == "stage_start")
+ assert start.input_ref is not None
+ assert artifacts.get(start.input_ref) == chunk.data📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| async def test_turn_stage_materializes_generator_input_once_for_detection(self): | |
| chunk = AudioChunk(data=b"\x01\x02", format=PCM16_MONO_16K) | |
| seen: list[AudioChunk] = [] | |
| class _CapturingSmartTurn: | |
| async def detect(self, audio_chunks): | |
| seen.extend(audio_chunks) | |
| return {"prediction": 1, "probability": 0.95} | |
| def audio_window(): | |
| yield chunk | |
| result = await TurnStage(_CapturingSmartTurn()).execute( | |
| audio_window(), | |
| _make_ctx(), | |
| _make_turn(), | |
| ) | |
| assert result["prediction"] == 1 | |
| assert seen == [chunk] | |
| async def test_turn_stage_materializes_generator_input_once_for_detection(self): | |
| chunk = AudioChunk(data=b"\x01\x02", format=PCM16_MONO_16K) | |
| seen: list[AudioChunk] = [] | |
| class _CapturingSmartTurn: | |
| async def detect(self, audio_chunks): | |
| seen.extend(audio_chunks) | |
| return {"prediction": 1, "probability": 0.95} | |
| def audio_window(): | |
| yield chunk | |
| journal = InMemoryRingBuffer(capacity=100) | |
| artifacts = InMemoryArtifactStore() | |
| result = await TurnStage(_CapturingSmartTurn()).execute( | |
| audio_window(), | |
| _make_ctx(journal=journal, artifact_store=artifacts), | |
| _make_turn(), | |
| ) | |
| assert result["prediction"] == 1 | |
| assert seen == [chunk] | |
| start = next(record for record in journal.read() if record.name == "stage_start") | |
| assert start.input_ref is not None | |
| assert artifacts.get(start.input_ref) == chunk.data |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/stages/test_stages.py` around lines 1407 - 1427, Update
test_turn_stage_materializes_generator_input_once_for_detection to pass a
journal and artifact store using the existing test pattern, then assert the
stage_start artifact contains exactly chunk.data. Keep the existing detect-input
assertion and result checks so the test verifies both capture and detection
receive identical audio.
Comment on lines
+117
to
+130
| def test_backoff_overflow_caps_at_max_delay(self) -> None: | ||
| config = RetryStrategyConfig( | ||
| max_retries=5, | ||
| sms_fallback_after=5, | ||
| base_delay_s=1.0, | ||
| backoff_factor=1e308, | ||
| max_delay_s=10.0, | ||
| jitter_fraction=0.0, | ||
| ) | ||
| strategy = RetryStrategy(config) | ||
| for _ in range(3): | ||
| strategy.record_attempt("+1555", "other") | ||
|
|
||
| assert strategy.get_delay("+1555") == 10.0 |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
Add return annotations to the test functions.
Add -> None to both test definitions. As per coding guidelines, "**/*.py: Use Python 3.11 or later with typing-first code."
Also applies to: 160-165
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/telephony/test_retry_strategy.py` around lines 117 - 130, Add the
return annotation -> None to both test function definitions in this test module,
including test_backoff_overflow_caps_at_max_delay and the additional test around
the referenced second location; leave their bodies and behavior unchanged.
Source: Coding guidelines
Comment on lines
+237
to
+243
| try: | ||
| await manager.shutdown() | ||
| assert replaced.done() | ||
| assert manager._silence_timer_tasks == set() | ||
| finally: | ||
| detector.release.set() | ||
| await asyncio.gather(replaced, return_exceptions=True) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Release the detector before awaiting shutdown.
manager.shutdown() waits for replaced to finish. detect() waits for detector.release after cancellation. The test sets that event only after shutdown() returns. The test therefore hangs.
Proposed fix
try:
- await manager.shutdown()
+ shutdown = asyncio.create_task(manager.shutdown())
+ await asyncio.sleep(0)
+ assert not shutdown.done()
+ detector.release.set()
+ await asyncio.wait_for(shutdown, timeout=1)
assert replaced.done()
assert manager._silence_timer_tasks == set()
finally:📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| try: | |
| await manager.shutdown() | |
| assert replaced.done() | |
| assert manager._silence_timer_tasks == set() | |
| finally: | |
| detector.release.set() | |
| await asyncio.gather(replaced, return_exceptions=True) | |
| try: | |
| shutdown = asyncio.create_task(manager.shutdown()) | |
| await asyncio.sleep(0) | |
| assert not shutdown.done() | |
| detector.release.set() | |
| await asyncio.wait_for(shutdown, timeout=1) | |
| assert replaced.done() | |
| assert manager._silence_timer_tasks == set() | |
| finally: | |
| detector.release.set() | |
| await asyncio.gather(replaced, return_exceptions=True) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/turns/test_turn_manager.py` around lines 237 - 243, Set
detector.release before calling manager.shutdown() in the test cleanup flow, so
the cancelled detect task can finish while shutdown awaits replaced. Preserve
the existing assertions and retain the finally block’s cleanup and
exception-gathering behavior.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
A broad reliability audit found lifecycle races, dropped task failures, inconsistent state serialization, and persistence/validation edge cases that could surface under cancellation, shutdown, parallel execution, or malformed inputs.
Changes
Impact
Shutdown remains bounded while retaining failed work for an explicit retry, background failures are observed instead of leaked, and persisted/validated state is deterministic across concurrent and degraded paths.
Validation
Summary by CodeRabbit
Bug Fixes
Security
Validation
New Features