Share one canonical stream-drain loop across consume variants (#115)

[leynos]

Summary

This branch factors the near-duplicate _consume_stream_* variants onto one canonical stream-drain loop.

Closes #115.

_consume_stream_without_lines and _consume_stream_with_lines each owned a copy of the stream is None guard, buffer setup, read/echo loop, and final decode; the _with_lines variant only layered incremental line emission on top. A fix to the loop in one variant was easy to forget in the other. The branch introduces a canonical _drain(stream, config, *, on_chunk=None) coroutine owning the read/echo/buffer mechanics; the line-emitting variant supplies an on_chunk callback feeding the incremental decoder, and the public _consume_stream dispatch remains the single entry point.

Review walkthrough

• Start with cuprum/_streams.py: the new _drain coroutine and the two thin variants.
• Then tests/behaviour/test_stream_property_preservation_behaviour.py: Hypothesis drives public Pipeline.run_sync() behaviour through real subprocess I/O, including concurrent final stdout/stderr capture and a shared echo sink.
• Then cuprum/unittests/test_stream_property_based.py: Hypothesis drives public pipeline payload preservation across arbitrary chunk boundaries.
• Finish with docs/developers-guide.md for the canonical helper's re-use policy and concurrency model.

Validation

• make check-fmt: pass
• make lint: pass
• make typecheck: pass
• make test: pass (596 passed, 45 skipped; Rust suite 4 passed)
• make markdownlint: pass
• make nixie: pass
• rg "multi-byte" docs/: no matches
• coderabbit review --agent: pending at PR creation (rate-limited); will be re-run and cleared

Notes

The private _consume_stream unit property module was removed after review feedback; the added coverage now exercises the refactor through public Pipeline behaviour and real subprocess I/O. The wheel-build snapshot drops that removed test module from its recorded file list.

Summary by Sourcery

Canonically centralize the subprocess stream read/echo/capture loop behind a new _drain helper and expand property-based tests and documentation around it, while refactoring tee-profile-worker concurrency tests into shared scaffolding and focused modules.

Enhancements:

• Introduce a shared _drain coroutine used by both line-emitting and non-line-emitting stream consumers to keep capture and echo behaviour consistent.
• Refactor tee-profile-worker concurrency tests into multiple focused modules backed by a shared helper scaffold to improve cohesion and reuse.
• Relax CrossHair per-condition timeouts in line-splitting tests to reduce flakiness in parallel runs.
• Allow Ruff lint rule exceptions for shared test helper modules to match existing test allowances.

Documentation:

• Document the canonical _drain stream-drain loop, its reuse policy, and the updated layout of tee-profile-worker concurrency tests in the developer guide.

Tests:

• Add Hypothesis-based public Pipeline behaviour coverage for stream payload preservation, concurrent stdout/stderr capture, and shared echo sink completeness.
• Split _EnvBackendSelector concurrency coverage into dedicated test modules for lock reentrancy, selector reentrancy, concurrent worker completion, and backend environment preservation, reusing common helpers.

References

• Lody session: https://lody.ai/leynos/sessions/9bca0931-a2f7-4d61-a7ed-36407ce6c0bc

[sourcery-ai]

Reviewer's Guide

Refactors stream consumption to use a single canonical drain loop, extracts shared tee-profile-worker concurrency scaffolding into a helpers module, and updates tests and docs to match the new structure and behaviour.

Sequence diagram for canonical stream-drain loop with and without lines

sequenceDiagram
    actor Caller
    participant Streams as _streams
    participant Drain as _drain
    participant Decoder as feed_decoder
    participant Lines as _emit_completed_lines

    Caller->>Streams: _consume_stream(stream, config, on_line)
    alt [on_line is None]
        Streams->>Streams: _consume_stream_without_lines(stream, config)
        Streams->>Drain: _drain(stream, config)
        Drain-->>Streams: captured_text_or_None
    else [on_line is not None]
        Streams->>Streams: _consume_stream_with_lines(stream, config, on_line)
        Streams->>Decoder: decoder_factory(errors)
        Streams->>Drain: _drain(stream, config, on_chunk=feed_decoder)
        loop for each chunk
            Drain->>Decoder: feed_decoder(chunk)
            Decoder->>Lines: _emit_completed_lines(pending_text, on_line)
            Lines-->>Decoder: pending_text
        end
        Drain-->>Streams: captured
        Streams->>Decoder: decoder.decode(b"", final=True)
        Streams->>Lines: _emit_completed_lines(pending_text, on_line)
        Lines-->>Streams: pending_text
        Streams-->>Caller: captured
    end

Loading

File-Level Changes

Change	Details	Files
Introduce canonical _drain coroutine and refactor stream consumption variants to use it, with new property-based tests for drain semantics.	Add _drain(stream, config, *, on_chunk=None) as the single read/echo/capture loop used by all consume variants. Refactor _consume_stream_without_lines to delegate to _drain after handling the None-stream fast path. Refactor _consume_stream_with_lines to use _drain with an on_chunk decoder feeder and return the captured text from _drain. Add test_stream_drain_property_based.py with Hypothesis-based tests over a stub chunked reader to assert capture parity, variant parity, line-boundary insensitivity, and echo correctness. Document the canonical stream-drain helper and its reuse policy in the developers guide.	cuprum/_streams.py cuprum/unittests/test_stream_drain_property_based.py docs/developers-guide.md
Factor shared tee-profile-worker concurrency scaffolding into a helpers module and split concurrency tests into smaller focused modules.	Extract concurrency helpers, selectors, worker runners, Hypothesis strategies, and backend pair utilities into _tee_profile_worker_test_helpers.py. Simplify test_tee_profile_worker_concurrency.py to focus on concurrent worker completion using helpers imported from the new module. Add dedicated modules for backend lock reentrancy, selector reentrancy, and environment preservation tests that reuse the shared helpers. Update the developers guide to describe the new concurrency test layout and how to run the split modules together.	cuprum/unittests/test_tee_profile_worker_concurrency.py cuprum/unittests/_tee_profile_worker_test_helpers.py cuprum/unittests/test_tee_profile_worker_lock_reentrancy.py cuprum/unittests/test_tee_profile_worker_selector_reentrancy.py cuprum/unittests/test_tee_profile_worker_env_preservation.py docs/developers-guide.md
Tighten ancillary configuration and timeouts to support the new tests and helpers.	Increase CrossHair per_condition_timeout in test_line_splitting to reduce flakes under parallel pytest workers. Extend Ruff per-file ignores to cover *_test_helpers.py so shared test scaffolding can use the same assertions and parameter counts as tests. Adjust maturin build snapshot expectations to account for the new test module (implicit via snapshot update).	cuprum/unittests/test_line_splitting.py pyproject.toml cuprum/unittests/__snapshots__/test_maturin_build.ambr

Assessment against linked issues

Issue	Objective	Addressed	Explanation
#115	Refactor the duplicated stream-consume logic into a single canonical internal coroutine (e.g. _drain(stream, config, *, on_chunk=None)) used by both _consume_stream_without_lines and _consume_stream_with_lines, while keeping _consume_stream as the single public dispatch entry point.	✅
#115	Document the canonical stream-drain helper and its reuse policy in docs/developers-guide.md, including how variants should layer behavior via callbacks instead of copying the loop.	✅
#115	Add Hypothesis-based property tests that drive the canonical stream-drain loop (via _consume_stream) over arbitrary byte streams and chunk boundaries, asserting captured output parity with a whole-payload decode, parity between consume variants, and boundary-insensitive line emission.	✅

Possibly linked issues

• Refactor: canonical stream-drain loop shared by _consume_stream_* variants #115: PR creates canonical _drain helper, refactors both consume variants to use it, and adds matching property tests/docs

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[coderabbitai]

Actionable comments posted: 2

🤖 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 `@cuprum/unittests/test_stream_drain_property_based.py`:
- Line 111: Replace all bare assertions at line 111, line 131, line 156, and
line 179 in the test_stream_drain_property_based.py file with assertions that
include explicit messages. Each assertion message should provide context about
the payload or chunk data being tested to make Hypothesis counterexamples
diagnosable without requiring test reruns. The format should be assert
<condition>, "<descriptive message including payload/chunk context>" for each of
the four locations.

In `@docs/developers-guide.md`:
- Line 66: Change the hyphenated term "multi-byte" to the non-hyphenated form
"multibyte" in the text "(including split multi-byte UTF-8 sequences and invalid
bytes)" to maintain consistency with the repository's established technical
terminology.

🪄 Autofix (Beta)

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: 6b70c2ee-fbda-4d63-8b44-556d8bd59b11

📥 Commits

Reviewing files that changed from the base of the PR and between de54bff and 70dd1fc.

📒 Files selected for processing (4)

• cuprum/_streams.py
• cuprum/unittests/__snapshots__/test_maturin_build.ambr
• cuprum/unittests/test_stream_drain_property_based.py
• docs/developers-guide.md

[leynos]

@coderabbitai resume

[coderabbitai]

✅ Action performed

Reviews resumed.

[coderabbitai]

Warning

Review limit reached

You’ve reached a temporary PR review limit under our Fair Usage Limits Policy.

Your recent review volume is higher than typical usage, so adaptive limits are currently applied.

Next review available in: 18 minutes

Your organization has reached its usage spending cap. Adjust your spending cap in the billing tab.

How can I continue?

After more reviews become available, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews.

How do review limits work?

CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability.

For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window.

Please refer docs for additional details.

Review details

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: 84541657-e717-4098-9a1b-56dd3eedfb1d

📥 Commits

Reviewing files that changed from the base of the PR and between dcb9f4b and 1e43be6.

📒 Files selected for processing (11)

• Makefile
• cuprum/_streams.py
• cuprum/unittests/__snapshots__/test_maturin_build.ambr
• cuprum/unittests/test_stream_drain.py
• docs/adr-001-rust-extension.md
• docs/cuprum-design.md
• docs/debugging/debugging-plan-20260715-python-315-unused-snapshot.md
• docs/developers-guide.md
• docs/execplans/4-3-2-behavioural parity tests.md
• docs/users-guide.md
• tests/behaviour/test_stream_property_preservation_behaviour.py

Walkthrough

The PR centralises subprocess stream draining in _drain, routes both consume variants through it, adds direct and property-based coverage, updates stream documentation, fixes Python 3.15 snapshot handling, and introduces configurable test and lint execution controls.

Changes

Canonical stream draining

Layer / File(s)	Summary
_drain helper and consume variants cuprum/_streams.py	_drain now owns chunk reading, capture, echoing, and optional callbacks; both consume variants delegate to it while preserving incremental line decoding.

Stream validation

Layer / File(s)	Summary
Stream helper and pipeline validation cuprum/unittests/test_stream_drain.py, tests/behaviour/test_stream_property_preservation_behaviour.py, cuprum/unittests/test_maturin_build.py, cuprum/unittests/__snapshots__/test_maturin_build.ambr	Tests cover capture and echo flags, invalid and split UTF-8 input, line emission across chunk boundaries, concurrent stdout and stderr preservation, and packaging of the new test module.

Documentation and guidance

Layer / File(s)	Summary
Stream architecture and testing guidance docs/developers-guide.md, docs/adr-001-rust-extension.md, docs/cuprum-design.md, docs/execplans/..., docs/users-guide.md, docs/debugging/...	Documentation describes _drain, its concurrency model, consume-variant responsibilities, testing guidance, stream parity terminology, and the Python 3.15 snapshot resolution.

Test tooling

Layer / File(s)	Summary
Configurable test and lint execution Makefile	Make variables and recipes now control pytest targets, worker counts, Rust flags, build jobs, documentation flags, and lint execution.

Poem

Chunks march softly, byte by byte,
Lines find their shape in streaming light.
Echoes mingle, stdout sings,
Stderr keeps its separate wings.
Tests watch every boundary bright.

---

Caution

Pre-merge checks failed

Please resolve all errors before merging. Addressing warnings is optional.

• Ignore

❌ Failed checks (2 errors, 1 warning)

Check name	Status	Explanation	Resolution
Unit Architecture	❌ Error	FAIL: the pytest loop still uses eval on overrideable PYTEST_TARGETS, turning data into shell code and hiding execution behind target expansion.	Replace eval "set -- $$pattern" with safe pathname expansion, e.g. set -- $$pattern, so target patterns stay data rather than shell source.
Security And Privacy	❌ Error	Makefile still uses eval "set -- $$pattern" on overrideable PYTEST_TARGETS, so a caller-supplied value can execute shell syntax.	Remove eval; expand the glob with set -- $$pattern or another non-evaluating shell loop so overrideable targets cannot execute code.
Out of Scope Changes check	⚠️ Warning	The Makefile tuning, maturin snapshot change, and debugging-plan doc are unrelated to issue #115's stream-drain refactor.	Split those CI/docs changes into a separate PR and keep this one limited to _drain, its docs, and the stream property tests.

✅ Passed checks (17 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly matches the stream-drain refactor and includes the linked issue reference.
Description check	✅ Passed	The description tracks the canonical _drain refactor and the related stream behaviour tests.
Linked Issues check	✅ Passed	The canonical _drain helper, on_chunk wiring, docs, and property tests satisfy issue #115.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.
Testing (Overall)	✅ Passed	New Hypothesis and pipeline tests hit real drain paths, cover arbitrary chunking, invalid bytes, line emission, and concurrent stdout/stderr echoing.
User-Facing Documentation	✅ Passed	PASS — the branch only refactors internal stream handling; the users guide edit merely normalises wording on existing parity coverage.
Developer Documentation	✅ Passed	Document the new _drain API in the developers’ guide, mirror it in the design doc and ADR, and keep the execplan and roadmap checked off and current.
Module-Level Documentation	✅ Passed	All changed Python modules have module docstrings, including the new test modules and _streams.py; each explains purpose and, where needed, its relation to pipeline/stream helpers.
Testing (Unit And Behavioural)	✅ Passed	Unit tests probe _drain edge cases and invariants, and behavioural coverage drives public Pipeline.run_sync over real subprocess I/O, not internal seams.
Testing (Property / Proof)	✅ Passed	PASS: the refactor adds Hypothesis properties over arbitrary payload/chunk boundaries and concurrent stdout/stderr capture, plus direct _drain/_consume_stream contract coverage.
Testing (Compile-Time / Ui)	✅ Passed	PASS: the PR is Python/docs-only; no Rust/TS compile-time surface changed, and the existing maturin wheel snapshot is focused on wheel layout/metadata with explicit normalisation.
Domain Architecture	✅ Passed	Keep the refactor confined to internal subprocess I/O: _drain stays in cuprum/_streams.py, and docs/tests only codify that boundary.
Observability	✅ Passed	PASS: This is an internal stream-loop refactor plus docs/tests; no new operational surface, telemetry, or alerting was added or needed.
Performance And Resource Use	✅ Passed	The new drain loop stays linear over stream chunks, reuses one buffer per invocation, and the added Hypothesis tests cap payloads/examples tightly.
Concurrency And State	✅ Passed	_drain owns local buffer/decoder state, docs spell out sink interleaving and cancellation, and tests cover concurrent stdout/stderr plus chunk-boundary interleaving.
Architectural Complexity And Maintainability	✅ Passed	_drain centralises the duplicated read/echo/capture loop, _consume_stream_* stay thin, and docs spell out reuse and ownership boundaries.
Rust Compiler Lint Integrity	✅ Passed	No Rust source, Cargo, or lint-suppression changes were touched; the PR only edits Python, docs, and Makefile, so rustc’s unused-code signals stay intact.

📋 Issue Planner

Built with CodeRabbit's Coding Plans for faster development and fewer bugs.

View plan used: #115

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch issue-115-canonical-stream-drain

---

_{Comment @coderabbitai help to get the list of available commands.}

[coderabbitai]

Tip

For best results, initiate chat on the files or code changes.

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
echo "=== test file lines 100-200 ==="
sed -n '100,220p' cuprum/unittests/test_stream_drain_property_based.py

echo ""
echo "=== full test file (last 120 lines) ==="
tail -130 cuprum/unittests/test_stream_drain_property_based.py

echo ""
echo "=== developers-guide.md concurrency section ==="
grep -n -i "concurr\|cancel\|await\|shared\|interleav\|sink\|buffer\|StreamReader" docs/developers-guide.md

echo ""
echo "=== check for multi-byte (hyphenated) anywhere in docs/ ==="
rg "multi-byte" docs/ || echo "NONE FOUND"

echo ""
echo "=== assertion lines in test file (assert statements) ==="
grep -n "assert " cuprum/unittests/test_stream_drain_property_based.py

Length of output: 15340

---

---

@leynos — all six checks are now resolved.

Check	Status
Testing (Overall) ❌	✅ Resolved — all four assertions carry payload/chunk diagnostic messages; "multibyte" is correct throughout
Unit Architecture ❌	✅ Resolved — every property assertion includes an explicit failure message
Developer Documentation ⚠️	✅ Resolved — "multibyte UTF-8" is used consistently; rg "multi-byte" docs/ returns no matches
Testing (Unit And Behavioural) ⚠️	✅ Resolved — assertion messages present at all flagged sites
Testing (Property / Proof) ⚠️	✅ Resolved — all Hypothesis property assertions carry payload/chunk context
Concurrency And State ⚠️	✅ Resolved — docs/developers-guide.md lines 70–87 document per-invocation buffer ownership, sink interleaving, fail-fast cancellation, and the StreamReader-per-invocation constraint; test_concurrent_drains_capture_independently and test_concurrent_echo_drains_sink_receives_all_bytes both use asyncio.create_task + asyncio.gather with full diagnostic assertion messages

No further work is required and no new issues need to be raised.

[pandalump]

@coderabbitai resume

[coderabbitai]

✅ Action performed

Reviews resumed.

[codescene-access]

No quality gates enabled for this code.