Refactor: canonical stream-drain loop shared by _consume_stream_* variants

[lodyai]

Summary

_consume_stream_without_lines (cuprum/_streams.py:40-65) and
_consume_stream_with_lines (cuprum/_streams.py:68-110) are near-duplicates.
They share the stream is None guard, the buffer setup, the read/echo loop
(read(_READ_SIZE) -> extend -> _write_chunk on echo), and the final
buffer.decode(...). The _with_lines variant only layers incremental-decoder
line emission on top.

Problem

A fix to the read/echo loop in one variant is easy to forget in the other,
silently diverging the capture path from the line-emitting path.

Proposed resolution — single canonical implementation

• Factor the shared mechanics into one canonical internal coroutine, for
  example _drain(stream, config, *, on_chunk=None), that owns the read/echo/
  buffer loop. _consume_stream_with_lines supplies an on_chunk callback that
  feeds the incremental decoder; the no-lines variant supplies none.
• Keep the public _consume_stream dispatch (_streams.py:28-37) as the single
  entry point.
• Document the canonical drain helper and its re-use policy in
  docs/developers-guide.md (indexed via docs/contents.md).

Verification

• Property tests (hypothesis): over arbitrary byte streams chunked at
  arbitrary boundaries (including split multi-byte UTF-8 sequences and invalid
  bytes), assert the canonical drain produces identical captured output to the
  pre-refactor functions and that line emission is boundary-insensitive. Align
  with the existing stream property-test suite.
• Model checking / snapshot assertions: n/a.

References

• Cuprum refactoring audit — consistency / near-duplicate consume functions.

[coderabbitai]

🔗 Related PRs

#60 - Add direct stdin support to SafeCmd (#30) [merged]

---

📝 Issue Planner

_{Check the box below or use the @coderabbitai plan command to generate an implementation plan and prompts that you can use with your favorite coding assistant.}

✅ Successfully generated implementation plan.

💡 Use @coderabbitai <feedback> to replan

---

🧪 Issue enrichment is currently in open beta.

You can configure auto-planning by selecting labels in the issue_enrichment configuration.

To disable automatic issue enrichment, add the following to your .coderabbit.yaml:

issue_enrichment:
  auto_enrich:
    enabled: false

💬 Have feedback or questions? Drop into our discord!

[coderabbitai]

Coding Plan

Summary

• Extract the shared stream is None guard, buffer setup, read/echo loop, and final decode into a single _drain coroutine with an optional on_chunk callback.
• Rewrite both consumer functions as thin wrappers: the no-lines variant delegates directly; the with-lines variant supplies a closure that feeds the incremental decoder.
• Verify correctness via Hypothesis property tests over arbitrary byte payloads and chunk boundaries, following existing test patterns.
• Document the helper and its re-use policy in docs/developers-guide.md.

Implementation Steps

Phase 1: Extract canonical drain helper and refactor consumers

Consolidate the duplicated stream-consumption mechanics into a single internal coroutine. Existing tests must continue to pass after this refactor.

Task 1: Create the `_drain` coroutine

Introduce the canonical drain helper that owns all shared mechanics.

• Add _drain(stream, config, *, on_chunk=None) coroutine in cuprum/_streams.py, positioned immediately before _consume_stream_without_lines
• Accept an optional on_chunk: cabc.Callable[[bytes], None] callback invoked with each raw chunk as it arrives
• Own the stream is None guard: return "" if capture_output else None
• Own buffer allocation: bytearray() when capture_output is true, otherwise None
• Own the read loop: await stream.read(_READ_SIZE) until EOF; extend buffer if capturing; call _write_chunk if echo_output; invoke on_chunk(chunk) when the callback is supplied
• Own final decoding: return buffer.decode(config.encoding, config.errors) or None

Task 2: Refactor `_consume_stream_without_lines` to delegate to `_drain`

Eliminate the duplicated loop from the no-callback variant.

• Replace the body of _consume_stream_without_lines with a single return await _drain(stream, config) call
• Remove the now-unused local buffer setup and read loop
• Retain the function as a thin wrapper for API symmetry and explicitness in the dispatch

Task 3: Refactor `_consume_stream_with_lines` to delegate to `_drain`

Eliminate the duplicated loop from the line-emitting variant; keep decoder and flush logic local.

• Set up the incremental decoder and pending_text accumulator before calling _drain
• Define a closure on_chunk(chunk) that decodes the chunk via the incremental decoder and calls _emit_completed_lines, updating pending_text via nonlocal
• Call await _drain(stream, config, on_chunk=on_chunk) to perform the loop
• After _drain returns, flush the decoder with decode(b"", final=True), emit any remaining complete lines, and call on_line for any trailing partial line
• Remove the duplicated read loop and buffer logic from this function

🤖 Prompt for AI agents

Extract the shared stream-consumption mechanics from
`_consume_stream_without_lines` and `_consume_stream_with_lines` in
`cuprum/_streams.py` into a single canonical `_drain` coroutine, then rewrite
both consumer functions as thin wrappers that delegate to it.

**Task 1.1 — Add `_drain` in `cuprum/_streams.py`:**
- Position the new `_drain(stream, config, *, on_chunk=None)` coroutine
immediately before `_consume_stream_without_lines`
- Declare `on_chunk` as `cabc.Callable[[bytes], None] | None`
- Implement the `stream is None` guard: return `""` if `config.capture_output`
is true, else return `None`
- Allocate `bytearray()` when `config.capture_output` is true, otherwise set the
buffer to `None`
- Run the read loop: `await stream.read(_READ_SIZE)` until EOF
  - Extend the buffer if capturing
  - Call `_write_chunk` if `config.echo_output`
  - Invoke `on_chunk(chunk)` when the callback is not `None`
- Return `buffer.decode(config.encoding, config.errors)` if capturing, else
return `None`

**Task 1.2 — Refactor `_consume_stream_without_lines`:**
- Replace the entire body with `return await _drain(stream, config)`
- Remove all local buffer setup and the read loop
- Keep the function definition in place as a thin wrapper

**Task 1.3 — Refactor `_consume_stream_with_lines`:**
- Before calling `_drain`, set up the incremental decoder and `pending_text`
accumulator
- Define a `on_chunk(chunk)` closure that:
  - Decodes `chunk` using the incremental decoder
  - Calls `_emit_completed_lines`, updating `pending_text` via `nonlocal`
- Call `await _drain(stream, config, on_chunk=on_chunk)`
- After `_drain` returns, flush the decoder with `decode(b"", final=True)`, emit
any remaining complete lines, and call `on_line` for any trailing partial line
- Remove the duplicated read loop and buffer logic

Phase 2: Verification and documentation

Add property tests that exercise the canonical drain across edge-case byte streams, then document the helper for maintainers.

Task 1: Add property tests for canonical drain equivalence and line-emission stability

Verify the refactored implementation matches pre-refactor behaviour and that line callbacks are boundary-insensitive.

• Create new test functions in cuprum/unittests/test_stream_property_based.py
• Use Hypothesis @given with derandomize=True and deadline=None; parametrise via stream_backend fixture
• Generate arbitrary byte payloads (including multi-byte UTF-8 sequences and invalid bytes) with arbitrary chunk boundaries using patterns from existing _payload_and_chunk_sizes strategy
• Assert that captured output from the refactored path equals the original full payload (byte-perfect via hex comparison)
• Assert that line callbacks receive the same sequence of lines regardless of where chunk boundaries fall
• Include edge cases: empty stream, single-byte chunks, chunk boundary inside a multi-byte character, payload ending without newline

Task 2: Document the canonical drain helper in the developers' guide

Record the helper's purpose, design rationale, and re-use policy for future maintainers.

• Add a new ## Canonical stream-drain helper (cuprum/_streams.py) section in docs/developers-guide.md
• Explain that _drain owns the read/echo/buffer loop to prevent duplication between the no-lines and with-lines variants
• State the re-use policy: always invoke via _consume_stream; never call _drain directly from outside _streams.py
• Include the pytest command to run the associated property tests
• Cross-reference the test file path

🤖 Prompt for AI agents

Add Hypothesis property tests in
`cuprum/unittests/test_stream_property_based.py` and a documentation section in
`docs/developers-guide.md` to verify and record the canonical `_drain` helper.

**Task 2.1 — Add property tests in
`cuprum/unittests/test_stream_property_based.py`:**
- Follow the existing pattern: decorate test functions with `@given` using
`derandomize=True` and `deadline=None`
- Parametrise tests via the `stream_backend` fixture
- Use the existing `_payload_and_chunk_sizes` strategy (or the same pattern) to
generate:
- Arbitrary byte payloads including multi-byte UTF-8 sequences and invalid bytes
  - Arbitrary chunk boundary splits
- Write a test asserting byte-perfect equivalence: captured output from the
refactored path must equal the original full payload (compare via hex)
- Write a test asserting line-emission stability: line callbacks must receive
the same sequence of lines regardless of where chunk boundaries fall
- Cover the following edge cases explicitly: empty stream, single-byte chunks,
chunk boundary inside a multi-byte character, payload ending without a newline

**Task 2.2 — Document the helper in `docs/developers-guide.md`:**
- Add a `## Canonical stream-drain helper (cuprum/_streams.py)` section
- Explain that `_drain` owns the read/echo/buffer loop to prevent duplication
between the no-lines and with-lines code paths
- State the re-use policy: always invoke via `_consume_stream`; never call
`_drain` directly from outside `_streams.py`
- Include the pytest command to run the associated property tests
- Cross-reference `cuprum/unittests/test_stream_property_based.py`

Research

The stream consumption logic lives in cuprum/_streams.py, which defines _StreamConfig (a frozen dataclass bundling capture/echo settings), a dispatch function _consume_stream, and two implementation functions that share substantial logic: buffer setup, the read(_READ_SIZE) loop, echo via _write_chunk, and final decoding. The _with_lines variant layers an incremental decoder and line emission on top. Property tests using Hypothesis exist in cuprum/unittests/test_stream_property_based.py, following a pattern of @given with derandomize=True and parametric stream_backend coverage. Internal helpers are documented in docs/developers-guide.md using ## headings with prose, rationale, and test commands.

---

🚀 Next Steps

🤖 All AI agent prompts combined

Task: 1

Extract the shared stream-consumption mechanics from
`_consume_stream_without_lines` and `_consume_stream_with_lines` in
`cuprum/_streams.py` into a single canonical `_drain` coroutine, then rewrite
both consumer functions as thin wrappers that delegate to it.

**Task 1.1 — Add `_drain` in `cuprum/_streams.py`:**
- Position the new `_drain(stream, config, *, on_chunk=None)` coroutine
immediately before `_consume_stream_without_lines`
- Declare `on_chunk` as `cabc.Callable[[bytes], None] | None`
- Implement the `stream is None` guard: return `""` if `config.capture_output`
is true, else return `None`
- Allocate `bytearray()` when `config.capture_output` is true, otherwise set the
buffer to `None`
- Run the read loop: `await stream.read(_READ_SIZE)` until EOF
  - Extend the buffer if capturing
  - Call `_write_chunk` if `config.echo_output`
  - Invoke `on_chunk(chunk)` when the callback is not `None`
- Return `buffer.decode(config.encoding, config.errors)` if capturing, else
return `None`

**Task 1.2 — Refactor `_consume_stream_without_lines`:**
- Replace the entire body with `return await _drain(stream, config)`
- Remove all local buffer setup and the read loop
- Keep the function definition in place as a thin wrapper

**Task 1.3 — Refactor `_consume_stream_with_lines`:**
- Before calling `_drain`, set up the incremental decoder and `pending_text`
accumulator
- Define a `on_chunk(chunk)` closure that:
  - Decodes `chunk` using the incremental decoder
  - Calls `_emit_completed_lines`, updating `pending_text` via `nonlocal`
- Call `await _drain(stream, config, on_chunk=on_chunk)`
- After `_drain` returns, flush the decoder with `decode(b"", final=True)`, emit
any remaining complete lines, and call `on_line` for any trailing partial line
- Remove the duplicated read loop and buffer logic
===============================================================================

Task: 2

Add Hypothesis property tests in
`cuprum/unittests/test_stream_property_based.py` and a documentation section in
`docs/developers-guide.md` to verify and record the canonical `_drain` helper.

**Task 2.1 — Add property tests in
`cuprum/unittests/test_stream_property_based.py`:**
- Follow the existing pattern: decorate test functions with `@given` using
`derandomize=True` and `deadline=None`
- Parametrise tests via the `stream_backend` fixture
- Use the existing `_payload_and_chunk_sizes` strategy (or the same pattern) to
generate:
- Arbitrary byte payloads including multi-byte UTF-8 sequences and invalid bytes
  - Arbitrary chunk boundary splits
- Write a test asserting byte-perfect equivalence: captured output from the
refactored path must equal the original full payload (compare via hex)
- Write a test asserting line-emission stability: line callbacks must receive
the same sequence of lines regardless of where chunk boundaries fall
- Cover the following edge cases explicitly: empty stream, single-byte chunks,
chunk boundary inside a multi-byte character, payload ending without a newline

**Task 2.2 — Document the helper in `docs/developers-guide.md`:**
- Add a `## Canonical stream-drain helper (cuprum/_streams.py)` section
- Explain that `_drain` owns the read/echo/buffer loop to prevent duplication
between the no-lines and with-lines code paths
- State the re-use policy: always invoke via `_consume_stream`; never call
`_drain` directly from outside `_streams.py`
- Include the pytest command to run the associated property tests
- Cross-reference `cuprum/unittests/test_stream_property_based.py`

💡 Iterate on the plan with: `@coderabbitai` <feedback>

Example Feedback
- `@coderabbitai` You can skip phase 3. Add a simple unit test case for phase 2.
- `@coderabbitai` For design choice 1 go ahead with option 3 and replan.

---

💬 Have feedback or questions? Drop into our discord!