refactor(device): extract the live-sample async stream into a collaborator (part of #344)

[tylerkron]

Part of #344. Not merging — opened for review.

Problem

DaqifiStreamingDevice hosted the entire pull-based live-sample path itself — the bounded channel, the drop-oldest overflow callback, the per-enumeration subscribe/unsubscribe bookkeeping, and the device-wide drop counter field. None of that is device state. It is an adapter over the IChannel.SampleReceived events the decoder already raises, and it was sitting in the god class this issue exists to break up.

Fix

Moved it to Device/Internal/LiveSampleStream, reached through the existing IDeviceOperationHost.SnapshotChannels().

• No interface change, no new host member, no constructor-signature change. The only wiring is one more field built in InitializeStreamingDevice, alongside the six collaborators already there.
• No public API change. DefaultLiveSampleBufferCapacity, DroppedLiveSampleCount and StreamSamplesAsync keep their signatures and their semantics; docs/DEVICE_INTERFACES.md needed no edit.
• System.Runtime.CompilerServices and System.Threading.Channels were used only by this block, so both usings go with it.
• DaqifiStreamingDevice.cs: 1271 → 1239 lines.

The one subtle part

StreamSamplesAsync now hands back the collaborator's async iterator directly rather than wrapping it in one of its own. That is deliberate, and it is what preserves the two deferred behaviors a caller can actually observe:

1. WithCancellation(token) still reaches the iterator's own [EnumeratorCancellation] parameter. A wrapper without that attribute would drop the token silently and hang.
2. An invalid bufferCapacity still throws on the first MoveNextAsync, not at the call — async-iterator bodies are deferred.

Both are now pinned by tests, because neither was before.

Tests

+8 (2623 → 2631). The existing DaqifiStreamingDeviceLiveStreamTests are deliberately left alone — they are the evidence that the extraction changed nothing.

• 7 new LiveSampleStreamTests covering what only a direct test can see: every snapshot channel subscribed once at start and unsubscribed on every exit path including cancellation (a leaked handler would keep the decode path writing into a dead buffer for the device's lifetime); the channel set snapshotted exactly once per enumeration and later arrivals ignored; concurrent enumerations each getting their own buffer; the drop counter accumulating across enumerations rather than per-enumeration; and the deferred ArgumentOutOfRangeException. The fake host throws on every member outside this block's remit, so a future change that sends a command or takes the channels lock fails loudly.
• 1 new device-level test pinning the WithCancellation forwarding described above.

FULL suite green on net9 + net10 — 2631 passed / 2 skipped each (+23 Daqifi.Mcp.Tests on net9), 0 warnings.

Bench (real Nq1, fw 3.7.2, USB, non-destructive)

Two sequential enumerations on one connection, consuming StreamSamplesAsync end to end against live hardware:

	pass 1 (default buffer, fast consumer)	pass 2 (1-slot buffer, slowed consumer)
delivered	1586 (793 × ch0/ch1 over 5 s)	798
dropped	0	790
timestamps monotonic per channel	yes	yes
IsStreaming after enumeration ended	true	true

• Pass 2 delivered + dropped ≈ 1588, i.e. the same production rate as pass 1 — real proof that pass 1's handlers were unsubscribed rather than leaked, which no mocked channel can show.
• The drop counter grew across enumerations (0 → 790), confirming it stayed device-wide after the move.
• Cancelling enumeration ended it promptly and did not stop the device stream.
• 158.6 Hz effective against 200 Hz requested is the known fw 3.7.2 clock mismatch, not a regression.

Conflict note

Hunks are disjoint from the open #439 apart from the using block, where the two edits are separated by three unchanged lines; whichever lands second should merge cleanly, and a conflict there would be trivial.

🤖 Generated with Claude Code

[tylerkron]

/agentic_review

[qodo-code-review]

PR Summary by Qodo

Refactor: extract live-sample async stream into LiveSampleStream collaborator

✨ Enhancement 🧪 Tests 🕐 40+ Minutes

AI Description

• Extract pull-based live-sample buffering/drop-oldest logic out of DaqifiStreamingDevice into
 LiveSampleStream.
• Keep public APIs/semantics unchanged by directly returning the collaborator’s async iterator.
• Add unit tests that pin subscription lifecycle, snapshot semantics, concurrency, and deferred
 exceptions.

Diagram

graph TD
  C[Consumer] --> D["DaqifiStreamingDevice"] --> L["LiveSampleStream"] --> B[("Bounded buffer")]
  L --> H["IDeviceOperationHost"] --> S["Snapshot channels"]
  S --> E["IChannel.SampleReceived"] --> L

Loading

High-Level Assessment

The following are alternative approaches to this PR:

1. Keep a device-level async-iterator wrapper

• ➕ Device remains the single semantic choke point for streaming behavior/docs
• ➕ Can add device-specific instrumentation without touching the collaborator
• ➖ Easy to accidentally break observable deferred behaviors (WithCancellation -> [EnumeratorCancellation], and when exceptions surface)
• ➖ Adds an extra layer that complicates reasoning about cancellation/cleanup

2. Move buffering into the decoder (StreamFrameDecoder)

• ➕ Places buffering logic closer to the decode hot path
• ➕ Potentially fewer cross-object calls on event delivery
• ➖ Mixes decoding responsibilities with consumer-facing buffering/backpressure policy
• ➖ Harder to unit test subscription/cleanup semantics without involving decoding internals

3. Switch to a reactive API (IObservable) for live samples

• ➕ Natural model for push-based SampleReceived events
• ➕ Rich ecosystem for buffering/backpressure operators
• ➖ Public API change (explicitly ruled out here)
• ➖ Higher cognitive overhead and larger refactor surface

Recommendation: The PR’s approach is the best fit given the constraints (no public API change, preserve subtle async-iterator semantics). Returning the collaborator’s iterator directly is a good decision to keep WithCancellation and deferred-throw timing intact, while still extracting non-device-state responsibilities out of the device class.

Files changed (4) +431 / -49

Refactor (2) +114 / -49

DaqifiStreamingDevice.csDelegate StreamSamplesAsync and drop count to LiveSampleStream collaborator +17/-49

Delegate StreamSamplesAsync and drop count to LiveSampleStream collaborator

• Removes inline bounded-channel buffering and event subscription logic from DaqifiStreamingDevice, wiring in a new LiveSampleStream field created during initialization. StreamSamplesAsync now returns the collaborator’s IAsyncEnumerable directly, and DroppedLiveSampleCount proxies the collaborator’s counter.

src/Daqifi.Core/Device/DaqifiStreamingDevice.cs

LiveSampleStream.csIntroduce LiveSampleStream collaborator for pull-based live samples +97/-0

Introduce LiveSampleStream collaborator for pull-based live samples

• Adds an internal component that snapshots channels from IDeviceOperationHost, subscribes to SampleReceived, writes to a bounded drop-oldest channel, and exposes an async iterator with [EnumeratorCancellation]. Maintains a cumulative dropped-sample counter across enumerations.

src/Daqifi.Core/Device/Internal/LiveSampleStream.cs

Tests (2) +317 / -0

DaqifiStreamingDeviceLiveStreamTests.csPin WithCancellation behavior on device-level live sample stream +25/-0

Pin WithCancellation behavior on device-level live sample stream

• Adds a device-level test ensuring WithCancellation-driven cancellation ends enumeration promptly without stopping device streaming. This guards against regressions where forwarding/wrapping could drop the enumerator cancellation token.

src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceLiveStreamTests.cs

LiveSampleStreamTests.csAdd focused unit tests for LiveSampleStream lifecycle and semantics +292/-0

Add focused unit tests for LiveSampleStream lifecycle and semantics

• Introduces unit tests for per-enumeration subscription/unsubscription (including cancellation paths), snapshot-once channel semantics, concurrent enumerations, and device-wide drop-count accumulation. Also pins deferred ArgumentOutOfRangeException timing (throws on first MoveNextAsync, not at call).

src/Daqifi.Core.Tests/Device/Internal/LiveSampleStreamTests.cs

[qodo-code-review]

Code Review by Qodo

🐞 Bugs (0) 📘 Rule violations (0) 📎 Requirement gaps (0) 🎨 UX issues (0) 🔗 Cross-repo conflicts (0) 📜 Skill insights (0)

1. Tests can hang CI ✓ Resolved 🐞 Bug ☼ Reliability

Description

New LiveSampleStreamTests contain unbounded awaits (awaiting MoveNextAsync directly after
cancellation, and an await-foreach inside Assert.ThrowsAsync without any timeout). If
cancellation/validation regresses, these tests can block indefinitely and stall the whole test run
instead of failing quickly.

Code

src/Daqifi.Core.Tests/Device/Internal/LiveSampleStreamTests.cs[R86-90]

+        using var cts = new CancellationTokenSource();
+        var e = stream.StreamSamplesAsync(cts.Token).GetAsyncEnumerator();
+        var moveNext = e.MoveNextAsync();
+        cts.Cancel();
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(async () => await moveNext);

Relevance

●●● Strong

Repo recently accepted adding WaitAsync/timeouts to prevent regression tests hanging CI
indefinitely.

PR-#364
PR-#411

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The cancellation test awaits moveNext without a timeout, and the invalid-capacity test runs an
await foreach inside Assert.ThrowsAsync without any bound; either path can wait forever if the
expected exception is not produced. Prior fixes in this repo explicitly added WaitAsync(...)
bounds to prevent test hangs from regressions.

src/Daqifi.Core.Tests/Device/Internal/LiveSampleStreamTests.cs[80-92]
src/Daqifi.Core.Tests/Device/Internal/LiveSampleStreamTests.cs[162-177]
PR-#364
PR-#411

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
Some newly added tests can hang indefinitely if the behavior under test regresses (cancellation no longer stops `MoveNextAsync`, or invalid-capacity validation stops throwing promptly). This risks hanging CI rather than producing a clear failing assertion.

### Issue Context
`LiveSampleStream.StreamSamplesAsync(...)` is an async iterator that can wait forever in `ReadAllAsync(...)` unless cancellation/validation triggers. Tests should bound waits so regressions fail fast.

### Fix Focus Areas
- src/Daqifi.Core.Tests/Device/Internal/LiveSampleStreamTests.cs[80-95]
- src/Daqifi.Core.Tests/Device/Internal/LiveSampleStreamTests.cs[162-178]

### Suggested change
- In `Enumeration_EndedByCancellation_StillUnsubscribes`, wrap the awaited `moveNext` in a bounded wait, e.g.:
 - `await Assert.ThrowsAnyAsync<OperationCanceledException>(() => moveNext.AsTask().WaitAsync(TimeSpan.FromSeconds(5)));`
- In `InvalidBufferCapacity_ThrowsOnFirstMoveNext_NotAtTheCall`, ensure the `await foreach` is also time-bounded (e.g., run the enumeration in a Task and `WaitAsync(...)` it), so if the exception stops being thrown the test fails with a timeout instead of hanging.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

_{To customize comments, go to the Qodo configuration screen, or learn more in the docs.}

[tylerkron]

/agentic_review

[qodo-code-review]

Code review by qodo was updated up to the latest commit 2cf3bd1

[tylerkron]

Ready for review. Not merging — leaving the merge to you.

State at head 2cf3bd1:

• Qodo: clean. Bugs (0) / Rule violations (0), and 0 unresolved review threads — the one finding ("Tests can hang CI") is fixed and shows struck-through/✓ Resolved, so this isn't the "summary says 0 while findings hide inline" case; I checked the threads directly via GraphQL, not just the headline.
• CI build: success on this exact head.
• MERGEABLE, up to date with main — the branch had gone BEHIND when refactor(device): extract raw-command session interpretation into a collaborator (part of #344) #439 landed, and the merge back in was conflict-free despite refactor(device): extract raw-command session interpretation into a collaborator (part of #344) #439 touching the same DaqifiStreamingDevice.cs.
• Full suite green on net9.0 + net10.0: 2662 passed / 2 skipped on each, plus 23 Daqifi.Mcp.Tests (net9-only). 0 warnings in Release.

Since the last review pass, the only change is 2cf3bd1, which bounds the waits in LiveSampleStreamTests — test-harness timing only, no production code touched, so the hardware validation reported in the PR description still applies as-is.

One thing I deliberately left out of scope: DaqifiStreamingDeviceLiveStreamTests.cs has the same two unbounded shapes (lines ~66 and ~91-93). This PR keeps that file untouched on purpose — it's the "the extraction changed nothing" evidence — so bounding it belongs in a separate change.