fix(ws): hold SYNCING through the post-connect status replay

[shbatm]

Problem

The controller replays every node's current status (a burst of ST/DON/DOF frames) over the WebSocket on every connect. WebSocketEventStream notified EventStreamStatus.CONNECTED the instant the socket opened — before that burst arrives in the read loop. Consumers that fire on events (most notably Home Assistant event entities in hacs-udi-iox) therefore treated the replayed status as live changes, producing spurious automation triggers on every HA restart, config-entry reload, and WS reconnect.

Empirically reproduced against the IoX-Testing controller: device-triggers, plain state-triggers, and attribute state-triggers on a udi_iox event entity all fired with zero user action on both a config-entry reload and a full HA restart (on+off each).

Fix

Add EventStreamStatus.SYNCING — socket is open but the initial status replay is still draining. The reader holds SYNCING and only flips to CONNECTED once the burst goes quiet for _SYNC_QUIET_SECONDS (1.0 s with no frame), with a _SYNC_MAX_SECONDS (10 s) hard cap so a perpetually chatty controller can't stall the stream. A small sampled watcher task (_promote_when_quiet) does the debounce so the hot read loop stays a plain async for; it's cancelled in _connect_and_read's finally, so a socket that drops before settling never reports CONNECTED.

• Records still update during SYNCING (the dispatcher keeps feeding, so node/property state stays correct) — only the "stream is live" signal is withheld.
• Re-armed on every reconnect, so the WS-reconnect case is fixed too, not just restart/reload.
• connected / == CONNECTED consumers are unchanged structurally; they simply see connected go true ~1 s later (after the replay), which is the intended behavior.

Consumer follow-up (separate, in hacs-udi-iox)

event.py::_on_control will gate emission on the synced flag (the existing subscribe_ws_status → connected: bool plumbing already maps == CONNECTED). Not in this PR — pyisyox capability lands first; consumer policy follows.

Tests

• test_ws.py: FakeWebSocket gained an additive keep_open (socket stays open with no frames so the quiet timer can be exercised); _SYNC_* are module-level for monkeypatching.
• Rewrote the lifecycle assertions to INITIALIZING → SYNCING → CONNECTED; the reconnect-path tests now reach a live connection via keep_open instead of an instant-close.
• New test_ws_reader_holds_syncing_through_replay_then_connects: a replayed frame dispatches into the record while the stream is still SYNCING and connected is False, and CONNECTED is only emitted after the quiet window.
• Full suite: 836 passed; ruff / ruff-format / mypy / pylint (10.00) / codespell / pre-commit all green.

🤖 Generated with Claude Code

[github-actions]

Claude finished @shbatm's task in 2m 17s —— View job

---

Code Review

• Read changed files and gather context
• Analyze the SYNCING state implementation
• Review test coverage
• Post review findings

---

This is a well-designed fix for a real problem. The approach — holding SYNCING via a sampled watcher task rather than weaving timers into the hot read loop — keeps the architecture clean. The logic is sound, and the PR description accurately describes what the code does. Below are my findings, in descending priority.

---

Issues

1. Silent-controller edge case: first sample window always promotes immediately when _frame_count == seen == 0 (ws.py:317)

seen = self._frame_count   # 0, just reset above
while not self._stop_requested:
    await asyncio.sleep(_SYNC_QUIET_SECONDS)
    ...
    quiet = self._frame_count == seen   # True if 0 frames ever arrived

For a controller with no nodes (or one whose replay finishes in < 1 µs before any frame is seen), quiet is trivially True on the very first sample. That's the correct outcome — a silent controller should promote after one quiet window. But it also means the sampling is asymmetric: a burst of frames that arrives and completes before the first sample will look exactly like zero frames. The distinction matters if the replay is very fast (sub-_SYNC_QUIET_SECONDS).

In practice this is fine — a completed replay is quiet — but it's worth documenting the invariant in the docstring: "a window in which _frame_count did not change from the previous sample is treated as quiet, including the initial window when no frames have yet arrived."

2. FakeWebSocket.__anext__ has a StopAsyncIteration after the await inside keep_open path (test_ws.py:73–75)

if self._keep_open:
    await self._closed_evt.wait()
raise StopAsyncIteration

When close() fires _closed_evt, the await returns and falls through to raise StopAsyncIteration. That works, but self.closed is already True at this point (set in close()). A second call to __anext__ after that (e.g. if the async-for loop hasn't exited yet) would hit the if self.closed: guard at the top — correct. No bug; just worth knowing the path. Fix this →

---

Observations

3. No test for the _SYNC_MAX_SECONDS hard-cap path (continuous traffic)

The existing tests cover:

• Zero frames → quiet after one window → CONNECTED ✓
• One frame then silence → quiet after the next window → CONNECTED ✓

But there is no test for: frames keep arriving without pause until the deadline fires. A test like test_ws_reader_max_cap_promotes_even_under_constant_traffic would round out coverage and make the deadline logic regression-proof. Fix this →

4. SYNCING enum value has stream_ prefix; CONNECTED does not

CONNECTED  = "connected"
SYNCING    = "stream_syncing"   # vs "syncing"

INITIALIZING = "stream_initializing" follows the same pattern, so SYNCING is internally consistent with the newer constants. Just noting that consumers who compare against the raw string values (rather than the enum member) may be surprised — nothing to fix here unless you want to audit downstream consumers, but it's a documentation point.

5. _promote_when_quiet: stopping local copy comment is subtly misleading (ws.py:322–325)

# Local copy — mypy narrows ``self._stop_requested`` to its
# loop-entry value across the await; the read loop / stop()
# may flip it during the sleep (same idiom as ``_run``).
stopping: bool = self._stop_requested
if stopping:
    return

The comment says "mypy narrows self._stop_requested to its loop-entry value across the await" — that's the pattern in _run, where the variable is captured before a long await. Here, asyncio is single-threaded so self._stop_requested can only change at a yield point, and there is a yield point inside asyncio.sleep. The local copy is still correct, but the mypy narrowing rationale doesn't apply to instance attributes (mypy doesn't narrow self.x across awaits). The real reason is just clarity / single read of a mutable attribute. Minor, but the comment may confuse readers familiar with mypy's narrowing rules.

6. Potential timing fragility in test_ws_reader_holds_syncing_through_replay_then_connects

monkeypatch.setattr(ws_module, "_SYNC_QUIET_SECONDS", 0.2)
...
for _ in range(200):
    if nodes["3D 7D 87 1"].properties["ST"].formatted == "On":
        break
    await asyncio.sleep(0.005)   # 200 × 5 ms = 1 s budget

Then the test asserts stream.status == EventStreamStatus.SYNCING immediately after the property flipped. With _SYNC_QUIET_SECONDS=0.2, the watcher won't promote until 200 ms after the last frame. The 1 s budget to detect the property update is generous, but there's a narrow race: the property could flip and the quiet timer could expire within the same 5 ms tick, causing the assertion to catch CONNECTED rather than SYNCING. In practice this window is tiny, but on a heavily loaded CI host it could flap. Inserting one await asyncio.sleep(0) after detecting the property change (to flush any pending tasks) before asserting SYNCING would make this iron-clad. Fix this →

---

Summary

The core implementation is correct and clean. The _promote_when_quiet design isolates the debounce logic from the hot path, the finally-block cleanup is properly ordered, and the keep_open test infrastructure is well-thought-out. The main actionable items are: a hard-cap coverage test (#3) and a minor test-race hardening (#6). Everything else is observational.

[github-actions]

Claude finished @shbatm's task in 3m 53s —— View job

---

Code Review

• Read changed files and gather context
• Analyze the SYNCING state implementation
• Review test coverage
• Post review findings

---

This is a clean, well-thought-out fix. The second commit (a31c41c) that adds debug-level status-transition logging rounds out observability nicely. The _promote_when_quiet task design — sampled watcher rather than injecting timer logic into the hot read loop — is the right call, and the finally-block ordering ensures a socket that drops before settling never spuriously emits CONNECTED.

Below I cover the still-open items from the last review pass and a few new observations on the logging commit.

---

New commit (a31c41c — log lifecycle transitions at DEBUG)

Implementation looks good. The if status != self._status: guard in _notify correctly limits log output to actual state transitions (avoids duplicates when _run re-notifies INITIALIZING on each reconnect attempt at a cycle boundary). The %s → %s format uses the StrEnum values directly, so the log line reads "WS stream status: not_started -> stream_initializing" — clear and greppable.

test_ws_reader_logs_status_transitions covers the happy path. One minor point: the test asserts "stream_syncing" and "connected" appear somewhere in the concatenated log output, which is correct given the StrEnum formatting. Asserting INITIALIZING's "stream_initializing" as well would make the test a complete three-step lifecycle check, but it's not critical.

---

Still-open from previous review

1. No test for the _SYNC_MAX_SECONDS hard-cap path (previous item #3 — actionable)

The existing tests cover quiet-window promotion and the basic lifecycle. A test for the deadline path would exercise ws.py:328 (loop.time() >= deadline). A workable recipe:

monkeypatch.setattr(ws_module, "_SYNC_QUIET_SECONDS", 0.1)   # sample every 100 ms
monkeypatch.setattr(ws_module, "_SYNC_MAX_SECONDS", 0.05)    # deadline at 50 ms
# queue enough frames that the first sample sees activity (quiet=False)
# and keep_open=True so the socket stays alive past the deadline

With _SYNC_MAX_SECONDS < _SYNC_QUIET_SECONDS: the first sample wakes at ~100 ms, deadline (50 ms) has already elapsed, loop.time() >= deadline is True → promotes via the cap path even though quiet may be True. To force quiet=False at the first sample, pre-queue one frame — the reader consumes it before the first sleep ends, so _frame_count != seen. Fix this →

2. Timing race in test_ws_reader_holds_syncing_through_replay_then_connects (test_ws.py:261-265 — actionable)

# Wait for the replayed frame to dispatch into the record.
for _ in range(200):
    if nodes["3D 7D 87 1"].properties["ST"].formatted == "On":
        break
    await asyncio.sleep(0.005)

# Replay dispatched (state synced) but we're still SYNCING, not live.
assert stream.status == EventStreamStatus.SYNCING  # ← race here

With _SYNC_QUIET_SECONDS=0.2, the quiet timer is 200 ms, but the property flip is detected with a 5 ms poll. In theory the assertion could fire in the same 5 ms tick as the quiet timer's internal sleep wakeup, flipping to CONNECTED before this line runs. In practice the window is tiny, but on a loaded CI host it can flap. Inserting one or two await asyncio.sleep(0) after the detection loop flushes any concurrently-scheduled tasks without adding wall-clock delay. Fix this →

---

Observations (no change needed)

3. _promote_when_quiet comment: mypy narrowing rationale doesn't quite fit

# Local copy — mypy narrows ``self._stop_requested`` to its
# loop-entry value across the await; the read loop / stop()
# may flip it during the sleep (same idiom as ``_run``).
stopping: bool = self._stop_requested

Mypy doesn't narrow instance attributes across awaits, so "mypy narrows self._stop_requested" isn't the actual reason. The real motivation is just: single read of a mutable attribute to avoid a second attribute access between the check and the return. The same comment in _run refers to the same pattern but is more accurate there because the narrowing concern is about the stopping variable itself being re-evaluated. A quick fix: drop "mypy narrows self._stop_requested to its loop-entry value across the await" and just say "avoid a second attribute read after the await point."

4. SYNCING = "stream_syncing" prefix is internally consistent

stream_* is the prefix for internal stream-lifecycle states (INITIALIZING, LOADED, DISCONNECTING), while CONNECTED, DISCONNECTED, LOST_CONNECTION etc. are connection-facing. SYNCING follows the stream_* convention correctly. No change needed; worth knowing if consumers compare raw string values.

---

Summary

Core implementation is solid. The _promote_when_quiet task lifecycle is correct under both normal and teardown paths; the _stop_requested guard at the exit correctly prevents spurious CONNECTED on stop-while-syncing. The debug logging commit is a clean addition. Two actionable items remain: the hard-cap coverage test (#1 above) and the CI-race fix in the replay test (#2 above) — both small additions to the test file.

[github-actions]

Claude finished @shbatm's task in 2m 41s —— View job

---

Code Review

• Read changed files and gather context
• Analyze what was addressed in round 1 vs still-open items
• Review new commit (7b334e4 — "address claude-review round 1")
• Post review findings

---

All six items from the previous two review passes are addressed. This is clean and ready; one minor nit below.

---

Prior-review checklist

#	Item	Status
1	_promote_when_quiet docstring — quiet-window invariant + cap-boundary note	✅ Added
2	FakeWebSocket.__anext__ keep-open path comment	✅ Added
3	Hard-cap coverage test	✅ Added (test_ws_reader_max_cap_promotes_under_constant_traffic)
4	SYNCING = "stream_syncing" prefix intentional	✅ Acknowledged, no change
5	_stop_requested local-copy comment reworded	✅ Matches _run phrasing now
6	Two asyncio.sleep(0) flushes in replay test	✅ Added

---

New test: test_ws_reader_max_cap_promotes_under_constant_traffic (test_ws.py:686)

The design is correct: repeat= mode keeps _frame_count incrementing every tick so quiet is always False; the only escape is loop.time() >= deadline. The assert frame_count > 5 proves traffic was real, not stalled. The assert reached_connected is captured before stop() — the comment explains why; good.

One nit — the elapsed >= 0.12 lower bound works but the number isn't obviously derived from the parameters in the test (_SYNC_QUIET_SECONDS=0.05, _SYNC_MAX_SECONDS=0.2). The intent is "must have waited more than one quiet window, proving it didn't slip through via an accidental quiet sample". An inline comment tying the bound to the configured values would make it self-documenting for future readers. Not blocking. Fix this →

---

Observations (nothing to fix)

test_ws_reader_logs_status_transitions still only asserts stream_syncing and connected; stream_initializing is not checked. This was flagged in the previous pass as "not critical" and remains so — two out of three lifecycle steps verified is fine.

_promote_when_quiet double guard (not self._stop_requested and self._status == EventStreamStatus.SYNCING) at ws.py:338 is correct defensive layering: the finally-block cancellation is the primary guard, this is the belt-and-suspenders fallback for any cancellation timing edge. No change needed, just worth knowing the intent.

---

Summary

The round-1 response is thorough. All actionable items were addressed correctly, and the new hard-cap test is well-structured. The single nit (inline comment on the elapsed lower bound) is minor and not blocking. Core implementation remains sound.