feat(asr): implement transport recovery mechanism

[diyuyi-agora]

• Added DEFAULT_TRANSPORT_RECONNECT_GRACE_SEC constant for default reconnection grace period.
• Updated AzureASRConfig to include transport_reconnect_grace_sec parameter.
• Enhanced AzureASRExtension to manage transport recovery after disconnection, including handling reconnection logic and session epoch validation.
• Introduced unit tests for transport recovery lifecycle and configuration parameter handling.

[github-actions]

Review: feat(asr): implement transport recovery mechanism

Well-scoped change, and the epoch guards on the Azure callbacks are the right instinct: the SDK delivers events from its own threads, and stale callbacks arriving after a recognizer swap are a real source of double-reconnect bugs. A few notes below, one of which I think prevents the feature from firing in the scenario it targets.

Verification note: I read extension.py (before/after), reconnect_manager.py, config.py, const.py, and the existing azure tests. I could not read ten_ai_base (its interface directory is not populated in this checkout), so claims about base-class behaviour are inferred from this extension own state transitions. Please sanity-check point 1 against the base class.

1. self.connected is likely the wrong gate, so recovery may never run (high impact)

_transport_recovery_after_grace bails out when self.connected is true:

if (
    self.stopped
    or self.connected
    or disconnect_epoch != self._transport_disconnect_epoch
):
    return

Within this file self.connected tracks the recognizer session, not the transport:

• set True in _azure_event_handler_on_session_started
• set False in _azure_event_handler_on_session_stopped, stop_connection, _handle_finalize_disconnect

_azure_event_handler_on_disconnected never clears it. So in the exact case this PR targets, where the websocket drops but the recognizer session is still alive and the SDK fails to reconnect, self.connected is still True when the grace timer expires, and the task returns without reconnecting.

The new test passes only because make_extension() leaves connected at its __init__ default of False; it never exercises the state the extension is actually in at that moment.

Suggested fix: track transport state separately, e.g. add self._transport_connected: bool = False in __init__, set it True in _azure_event_handler_on_connected and False in _azure_event_handler_on_disconnected, and gate on that instead. The _cancel_transport_recovery() call already present in the connected handler is the primary "SDK recovered" signal, so this check is really a backstop against a lost cancel, but as written it is a backstop that always trips.

Worth adding a test that sets extension.connected = True before the disconnect event and still expects a reconnect after the grace period. That test fails against the current implementation, which is the point.

2. Exceptions in the recovery task are swallowed, and recovery never retries

self._transport_recovery_task = None
self._transport_recovery_in_flight = True
try:
    await self.stop_connection()
    await self._handle_reconnect()
finally:
    self._transport_recovery_in_flight = False

Nothing holds a reference to the task and nothing awaits it, so if stop_connection() raises, the exception surfaces only as the asyncio "Task exception was never retrieved" warning: no log_error, no send_asr_error, and no further recovery attempt, since the disconnect epoch is not re-armed. stop_connection() calls self.client.stop_continuous_recognition(), a blocking SDK call that can throw; the existing code already measures its duration because it is slow enough to matter.

Suggest wrapping the body in except Exception with a log_error at minimum, so a failed recovery is visible rather than silent.

Related: that blocking stop_continuous_recognition() now runs on the event loop from a timer task, stalling audio pumping for its duration. Pre-existing rather than introduced here, but this change adds a new path that hits it. asyncio.to_thread would be the fix if you want to address it.

3. A late session_stopped can still double-reconnect

The in-flight flag is reset in finally, but the session_stopped coroutine is created on the SDK thread and scheduled via call_soon_threadsafe, so it can run after the flag clears. Normally the epoch check saves you, because start_connection bumped _recognizer_epoch. But if _handle_reconnect fails before reaching that bump (say SpeechConfig construction throws), the epoch is unchanged and in-flight is False, so the late session_stopped starts a second independent reconnect chain. Narrow window and low severity; noting it because the mitigation depends on where the epoch bump sits inside start_connection.

4. The self-cancel guard in _cancel_transport_recovery() is order-dependent

The task is asyncio.current_task() early return is what keeps stop_connection() from cancelling the very task that called it. But the recovery path also sets self._transport_recovery_task = None before awaiting stop_connection(), so in practice the task is None branch short-circuits first. Both mechanisms look load-bearing, only one is actually exercised, and moving the nulling below the awaits would silently reintroduce self-cancellation. A one-line comment explaining why the nulling happens where it does would save the next reader, or drop the nulling and rely on the guard.

5. No validation on transport_reconnect_grace_sec

The field accepts any float, including 0 and negatives, and update() copies straight from params with no bounds check. asyncio.sleep(-1) returns immediately, turning every transport blip into an instant teardown-and-reconnect. Consider Field(default=DEFAULT_TRANSPORT_RECONNECT_GRACE_SEC, gt=0).

Minor: DEFAULT_TRANSPORT_RECONNECT_GRACE_SEC = 10 is an int backing a float field. Write it as 10.0 for clarity.

6. Tests

• Timing flakiness. grace=0.05 with sleep(0.1), and grace=0.2 with sleep(0.3), give roughly 50-100ms of slack. On a loaded CI runner these will flake. Prefer driving the wait deterministically: monkeypatch asyncio.sleep, or have the recovery path set an asyncio.Event that the test awaits with a generous timeout.
• Style consistency. These use asyncio.run(run_test()) wrappers, while other async unit tests in the repo, including this extension test_metrics.py, use the pytest.mark.asyncio marker. Worth matching.
• License header. conftest.py and test_unlimited_reconnect.py carry the "This file is part of TEN Framework" block. It is inconsistent across existing files so this is low priority, but new files may as well have it.
• Coverage gaps beyond the connected=True case in point 1: a second disconnect superseding a pending recovery (the _transport_disconnect_epoch path), and stop_connection() cancelling a pending recovery. Both are cheap to add, and both cover logic this PR introduces.

7. Naming

_transport_reconnect_grace_sec() (method) and config.transport_reconnect_grace_sec (field) differ only by a leading underscore, and both appear in the same function. Something like _grace_sec(), or inlining the two-line lookup, would read better.

CI reminder

Per docs/ai/L1/04_conventions.md, task lint is strict and any pylint warning fails the build. Worth running before merge:

sudo docker exec ten_agent_dev bash -c "cd /app && task format && task check && task lint"

Summary: point 1 is the one I would want resolved before merge, since as written I believe the grace timer returns early in the common transport-drop case. Points 2 and 5 are small hardening items; the rest is polish.

[diyuyi-agora]

Code review (manual — Codex job failed: stream disconnected / builder error)

The automated Codex review did not finish (provider reconnect exhausted). This comment follows docs/ai/L1/L2/asr_plugin_design_review.md and the PR diff for azure_asr_python transport recovery.

Summary

Solid direction: bounded grace before extension-level reconnect when Azure transport drops, recognizer_epoch guards on transport/session callbacks, cancellation on reconnect/stop/deinit, and focused unit tests. This aligns with ASR guide §5.1 (SDK transport recovery vs extension reconnect).

Strengths

• Epoch guards on on_connected / on_disconnected / on_session_stopped reduce stale-callback damage when start_connection() replaces the recognizer (§5.2).
• Single deferred recovery task with disconnect_epoch coalesces repeated disconnected events.
• _transport_recovery_in_flight avoids on_session_stopped starting a second reconnect while stop + ReconnectManager runs.
• Configurable transport_reconnect_grace_sec with a documented default constant.
• Tests cover grace expiry, SDK reconnect cancelling grace, stale session_stopped, and params update.

Issues / questions

1. Grace vs on_session_stopped (behavioral — please confirm in PR description)
   on_disconnected schedules grace, but on_session_stopped calls _cancel_transport_recovery() and still await self._handle_reconnect() immediately when not in _transport_recovery_in_flight. If Azure often emits session_stopped soon after transport disconnected, the grace window may rarely apply. If that is intentional (grace only when transport drops without session stop), document the SDK event ordering; otherwise consider deferring extension reconnect from session_stopped through the same grace owner (§9.1 one reconnect entry point).

2. Duplicate on_disconnected reporting (minor)
   A failure path may call base on_disconnected from both transport disconnected and session_stopped. Worth confirming connection-status consumers tolerate duplicate disconnected transitions or gating the second report.

3. Stop during grace (SHOULD)
   Grace task checks self.stopped after sleep; _cancel_transport_recovery() runs in stop_connection / on_deinit. Consider mirroring smallest_asr_python by cancelling the recovery task early in on_stop (if overridden) so a sleeping task cannot proceed if stop_connection ordering differs from base on_stop.

4. Stale canceled callback (nit)
   Epoch is not passed to _azure_event_handler_on_canceled. If an old recognizer can still emit cancellation after swap, it could still send errors/disconnect on the new session. Low risk if stop_continuous_recognition nulls handlers, but epoch guard would match the rest of the change.

5. Docs / config surface
   New knob is only via params.update; consider a line in package README or property.json comment so operators know about transport_reconnect_grace_sec.

6. Test gap
   Add a test that on_session_stopped cancels a pending grace task (disconnect scheduled, then session_stopped → immediate reconnect, no double reconnect). Existing tests do not cover disconnect + session_stopped ordering.

Security / performance

• No new secrets; logging uses existing vendor categories.
• Default 10s grace adds latency only when transport stays down and SDK does not recover — acceptable tradeoff; make tunable for latency-sensitive graphs.

Test coverage

• New test_reconnect_lifecycle.py is helpful but isolated (mocked stop_connection / _handle_reconnect). Existing integration-style test_reconnect.py unchanged — confirm CI still passes for azure extension (task test-extension / guarder if required for this plugin).

---

ASR design review checklist

Item	Status
Lifecycle / super() chain	Pass (no regressions seen in diff)
Connection state	Pass with note on possible duplicate disconnected
Buffering	N/A (unchanged ASRBufferConfigModeKeep)
Finalize	N/A
Reconnect	Pass with question on grace vs immediate session_stopped reconnect
Result shape	N/A
Metrics	N/A
Tests	Partial — add disconnect+session_stopped ordering; guarder not evidenced in PR

Merge recommendation: Approve after clarifying Azure event ordering (item 1) and ideally adding the ordering test; not blocking on docs alone.

[github-actions]

Review: transport recovery mechanism

Reviewed against docs/ai/L1/L2/asr_plugin_design_review.md, since this touches ASR lifecycle, connection state, and reconnect.

Good direction overall. It implements what section 5.1 asks for (bounded grace before the extension takes over from SDK-internal recovery), and the recognizer_epoch threading through session_stopped / connected / disconnected adds the callback-generation guard from section 5.2 that this extension previously lacked. That closes a real section-14 merge blocker (stale callback from an old client mutating new-client state).

Blockers

1. Recovery task is never awaited during stop/deinit (section 9.2 MUST)

_cancel_transport_recovery() calls task.cancel() then immediately discards the reference, so nothing ever awaits it. cancel() only schedules a CancelledError. There is a window where the task has already passed its asyncio.sleep() and is inside await self.stop_connection() / await self._handle_reconnect() — that region is not covered by the try/except CancelledError around the sleep, and _handle_reconnect() -> ReconnectManager.handle_reconnect() -> await asyncio.sleep(delay) -> await connection_func() will call start_connection() after on_stop. That is section 14: a background task can reconnect after stop. Dropping the last reference to a pending task also risks GC before completion, and exceptions surface as bare "Task exception was never retrieved".

The method is def, not async def, so it structurally cannot await. Suggested shape:

async def _cancel_transport_recovery(self) -> None:
    task = self._transport_recovery_task
    self._transport_recovery_task = None
    if task is None or task is asyncio.current_task() or task.done():
        return
    task.cancel()
    with contextlib.suppress(asyncio.CancelledError):
        await task

Then re-check the stop latch after the grace sleep and again before _handle_reconnect(). Both call sites (on_deinit, stop_connection) need await. Note on_deinit currently cancels before await super().on_deinit(...); section 12.3 wants the stop latch set first, then tasks cancelled and awaited.

2. stop_connection() / recovery-task reentrancy can drop the reconnect entirely

The recovery task awaits stop_connection(), which calls _cancel_transport_recovery(). The task is asyncio.current_task() check correctly avoids self-cancellation but returns early, before self._transport_recovery_task = None, so the stale finished task stays referenced.

More importantly, _azure_event_handler_on_session_stopped suppresses reconnect while _transport_recovery_in_flight is set. If the recovery task is cancelled during stop_connection(), the finally resets the flag while CancelledError propagates — but the session_stopped that stop_continuous_recognition() triggers was already suppressed. Result: no transport recovery, no session-stopped reconnect, no reconnect owner. Audio then buffers into the 10 MB ASRBufferConfigModeKeep until overflow. Please scope the suppression flag strictly to the stop_connection() call, or add an explicit post-recovery verification/re-arm.

Should fix

3. Suppressed session_stopped sets connected = False but skips on_disconnected(). The normal path calls it. is_connected() reads self.connected, so audio routing stays correct, but base connection_status can desync and produce a missing transition in exactly the scenario this PR targets. _azure_event_handler_on_disconnected already emitted it before scheduling, so a comment saying so would suffice — currently a reader cannot tell whether the omission is deliberate.

4. FINALIZE_MODE_DISCONNECT schedules recovery for an expected close. _handle_finalize_disconnect() closes the stream and calls stop_continuous_recognition() without going through stop_connection(), so recovery is never cancelled. The resulting disconnected event has stopped == False and schedules a recovery task for a deliberate close. Section 5.3 requires expected and unexpected closes be distinguished; section 14 lists treating an expected close as failure as merge-blocking. The next start_connection() bumps the epoch and the self.connected check may save it in practice, but that is accident rather than design. Cancel recovery there or mark the close expected, and cover it with a test — mute_pkg is the default, so this path is likely under-exercised.

5. No validation on transport_reconnect_grace_sec. Section 3.2 requires explicit validation. Negative or 0 makes asyncio.sleep() return immediately, so extension reconnect races SDK recovery — the concurrent-recovery case 5.1 warns about. Add Field(ge=0) at minimum. The field is also missing from manifest.json under api.property.properties.params; it works via config.update(self.config.params), but the manifest no longer describes the accepted surface.

6. Epoch is not re-checked inside the recovery task. _azure_event_handler_on_disconnected guards the epoch at entry, then awaits on_disconnected(...), then schedules. A concurrent start_connection() can advance _recognizer_epoch across that await. _transport_disconnect_epoch only guards against a newer disconnect, not a newer recognizer. Re-checking recognizer_epoch in _transport_recovery_after_grace closes it.

7. DEFAULT_TRANSPORT_RECONNECT_GRACE_SEC = 10 is an int on a float field. Pydantic coerces, so no runtime bug, but declaring 10.0 makes the defensive float() wrapping in _transport_reconnect_grace_sec() unnecessary.

Test coverage

The five tests are focused, and the SDK-reconnects-cancels-recovery case is the right core scenario. Gaps:

• No stop/deinit-during-recovery test. Highest-risk path in the PR (findings 1 and 2), and section 13.1 explicitly requires "no reconnect during stop" for test_reconnect_lifecycle. Add: schedule recovery, set stopped = True mid-grace, assert _handle_reconnect never awaited. Also cover cancellation landing inside stop_connection().
• Timing sleeps will flake. asyncio.sleep(0.1) against a 0.05 grace leaves a 50 ms margin that can invert under CI load. Drive time deterministically or widen margins.
• The cancellation test never sets/asserts self.connected, which is the real-world trigger — it passes only because _azure_event_handler_on_connected runs immediately and cancels the task.
• make_extension() does not set self.stopped, relying on the base default; set it explicitly so the tests do not silently change meaning.
• Diverges from suite conventions. Every other test here (test_reconnect.py, test_unlimited_reconnect.py) drives the extension through the TEN runtime with patch_azure_ws from tests/mock.py. These use asyncio.run() + MagicMock and poke private attrs, bypassing the session-scoped FakeApp fixture in conftest.py. Unit-level is defensible for bookkeeping logic, but it does not verify the mechanism against the real base class. Please confirm task test-extension and the ASR guarder pass, especially guarder item 6 (continuous audio during reconnect without a crash).

ASR checklist

Item	Status	Note
Lifecycle	Fail	on_deinit cancels before setting stop latch; cancel not awaited (12.3, 9.2)
Connection state	Partial	Suppressed session_stopped skips on_disconnected() (finding 3)
Buffering	Pass	Unchanged, bounded 10 MB ModeKeep — but finding 2 can silently fill it
Finalize	Fail	FINALIZE_MODE_DISCONNECT schedules recovery for an expected close (5.3, 14)
Reconnect	Fail	Recovery task can call start_connection() after stop (9.2, 14)
Result shape	N/A	No result-path changes
Metrics	Pass	send_connect_delay_metrics unchanged; epoch guard correctly precedes it
Tests	Partial	Good core cases; missing stop-during-recovery; timing-sensitive

Epoch guards (5.2) are a clear pass and a genuine improvement.

Conventions

Commit message follows conventional commits correctly. Please confirm task format, task check, and task lint pass, and consider copying the section-15 checklist into the PR description as the guide asks.

Happy to re-review once findings 1, 2, and 4 are addressed — the core design is sound.

[github-actions]

Review: changes requested

The transport-recovery implementation has merge-blocking ASR lifecycle issues. These violate the repository ASR design-review MUST rules; they need correction before merge.

1. [Blocking] The recovery path can never recover a connection-only disconnect. session_started sets self.connected = True, but the new Connection.disconnected handler does not clear it. After the grace period, the recovery task returns because self.connected is still true. is_connected() therefore continues to report a live vendor connection and audio is handed to the old stream instead of using the configured bounded buffer. Clear the authoritative vendor-live state before scheduling recovery, and add a test that starts from connected=True; the current fixture leaves it false, so it cannot expose this failure.

2. [Blocking] Reconnect ownership and the SDK grace period are still split across callbacks. A disconnect schedules the delayed task, while session_stopped cancels that task and immediately awaits _handle_reconnect(). When the SDK emits both events, one ordering bypasses the intended grace period; the reverse ordering can schedule the delayed task while the immediate reconnect is in progress. Route both callbacks through one serialized recovery owner and explicitly classify finalize/stop closes as expected. The recovery also invokes the existing ReconnectManager, which deliberately retries forever and reports failed retries as FATAL_ERROR; the ASR baseline requires a bounded retry ceiling with a terminal fatal transition, not an unlimited fatal-error loop.

3. [Blocking] The epoch guard is incomplete. The new registration passes an epoch to session_stopped and connection callbacks, but the canceled callback is still registered without one. A fatal cancellation from a retired recognizer can consequently set self.stopped = True and report a disconnect for the replacement connection (handler). Guard every old-client callback that can mutate state or emit results/errors, and test a stale fatal cancellation after a new recognizer has started.

4. [Blocking] The new background task is cancelled but never drained at shutdown. _cancel_transport_recovery() only calls cancel(), including from deinit and stop_connection; no lifecycle path awaits or otherwise observes task completion. This leaves a stop/recovery race and unobserved exceptions, contrary to the required cancel-and-await cleanup rule. Make teardown cancellation async and await the task with gather(..., return_exceptions=True) (or otherwise safely drain it), then test stopping after the grace check but before reconnect work begins.

5. [Blocking] The new grace-period parameter is not safely validated when supplied through params. AzureASRConfig.update() assigns untyped params values directly, bypassing Pydantic validation. A negative, non-finite, or non-numeric transport_reconnect_grace_sec reaches float(...) in a background task and can disable recovery with an unobserved exception. Validate the value as a finite non-negative duration before connection attempts; invalid configuration must emit one fatal error and prevent recovery.

The new unit tests cover only the happy path with a default-false connected flag. Please add coverage for: a real connected-to-disconnected transition and buffered audio, both callback orders with exactly one reconnect, expected finalize close, stale fatal cancellation, shutdown/task draining, and invalid duration values. There is also no PR evidence yet for the required standalone extension test and sequential ASR guarder run.

ASR design review

• Lifecycle: FAIL - stale callbacks and task cleanup are not safe.
• Connection state: FAIL - transport disconnect can remain vendor-live.
• Buffering: FAIL - the stale live state bypasses the bounded buffer.
• Finalize: FAIL - expected finalize close is not classified or covered.
• Reconnect: FAIL - callback ownership is split and retry policy is unbounded.
• Result shape: N/A - this PR does not change result protocol.
• Metrics: N/A - this PR does not change metric shape.
• Tests: FAIL - required recovery race, stop, invalid-config, and guarder evidence are missing.