Drain subprocess stream consumers once and harden test synchronisation

[leynos]

Summary

This branch actions a batch of code-review findings on the subprocess
execution machinery and its tests. Each finding was first verified against the
current code; all six proved still valid, and every one has been fixed with
minimal changes. The substantive change is that stream-consumer draining on
timeout now happens exactly once rather than twice, with captured output
preserved on TimeoutExpired; the remainder replace timing-based test
coordination with deterministic synchronisation seams and move a shared helper
into the test-helper package.

No issue or roadmap task governs this branch, and there is no execplan; the
work originates from inline and out-of-diff review comments.

Review walkthrough

• Start with cuprum/_subprocess_execution.py — _wait_for_exit_code now waits only for the exit code (still terminating the process on timeout or cancellation), and a new _drain_stream_consumers helper cancels, drains, and decodes the stdout/stderr consumers exactly once in the timeout and cancellation branches of _run_subprocess_with_streams.
• Then review cuprum/_subprocess_timeout.py — _handle_stream_timeout is now synchronous and simply raises _SubprocessTimeoutError from the caller-supplied, already-decoded stdout/stderr, so captured output survives onto the public TimeoutExpired.
• Review the moved helper in tests/helpers/stream_pipes.py — drain_blocking_payload_size probes the real pipe capacity and imports fcntl lazily so off-POSIX collection still succeeds.
• Review the reworked observe tests in cuprum/unittests/test_observe.py — the early-close stdin test replaces payload-size timing with a deterministic go/closed marker handshake: the child closes stdin only once the writer is confirmed wedged in drain(), then signals back.
• Review cuprum/unittests/test_safe_cmd_context.py — the cancellation test replaces a fixed asyncio.sleep readiness guess with an sh.observe start-phase hook and an asyncio.Event.
• Review cuprum/unittests/test_safe_cmd_stdin.py — the two blocked-stdin tests now use the shared drain_blocking_payload_size helper instead of hardcoded 1 MiB payloads, and a capture-mode blocked-stdin timeout regression test exercises the stream path.
• Finish with cuprum/unittests/test_subprocess_timeout.py — the unit tests are re-pointed at the new decomposition: _drain_stream_consumers decoding, the synchronous _handle_stream_timeout, and _wait_for_exit_code's process teardown.
• The documentation fix is a single clause-separating comma in docs/debugging/debugging-plan-2026-07-18-hypothesis-too-slow.md.

Validation

• uv run ruff format --check: pass (all files formatted)
• make lint: pass (ruff clean, typos gate clean)
• make typecheck: pass (ty check, all checks passed)
• make markdownlint: pass (exit 0)
• make nixie: pass (all diagrams validated)
• uv run pytest -n0 test_observe.py test_safe_cmd_context.py test_safe_cmd_stdin.py test_subprocess_timeout.py: 57 passed

Notes

• The consumer-drain relocation preserves existing behaviour: the previous code
  drained the consumers twice (discarding the first pass, keeping the second),
  so collapsing to a single drain does not change observable output. A consumer
  still blocked on read() at drain time is cancelled and decodes to None, as
  before.
• The full make test suite, including the Rust nextest portion, was not run
  to completion locally; the changes are Python-only and do not touch the Rust
  crate. The known trybuild snapshot drift against the CI-pinned toolchain is
  unrelated to this branch.

[sourcery-ai]

Sorry @leynos, you have reached your weekly rate limit of 500000 diff characters.

Please try again later or upgrade to continue using Sourcery

[coderabbitai]

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: f9fe5acb-41f0-4c35-9577-3cacd7bff834

📥 Commits

Reviewing files that changed from the base of the PR and between f906b35 and ac68971.

📒 Files selected for processing (1)

• cuprum/unittests/test_observe_stdin_early_close.py

🔗 Linked repositories identified

CodeRabbit considers these linked repositories for cross-repo context during reviews:

• leynos/shared-actions (auto-detected)
• leynos/pylint-pypy-shim (auto-detected)
• leynos/whitaker (auto-detected)

---

Summary

• Updated subprocess timeout and cancellation sequencing so stream consumers are cancelled/drained exactly once, with decoded partial stdout/stderr preserved on TimeoutExpired.
• Simplified _wait_for_exit_code to focus on process termination only; introduced _drain_stream_consumers to centralise consumer teardown/decoding behaviour.
• Refactored _handle_stream_timeout to synchronously raise using caller-supplied, pre-drained stdout/stderr output (and updated exported signatures accordingly).
• Reworked tests to remove timing-based coordination in favour of deterministic event/marker handshakes, including an updated embedded child script that now uses pathlib for marker paths.
• Moved the portable pipe-capacity/blocked-stdin payload sizing helper to tests/helpers/stream_pipes.py and reused it across blocked-stdin tests.
• Consolidated blocked-stdin timeout coverage into a single parametrised direct/capture test (while eliminating duplicate test cases), and tightened stdin-close/early-error observation coverage with deterministic marker-based signalling.
• Updated developer and user documentation to reflect the new timeout/cancellation output retention semantics and deterministic pipe sizing guidance.

Testing

• Formatting, lint, type checking, Markdown, diagram checks, and 57 focused tests passed.
• The full test suite was not completed locally.

Walkthrough

Refactor subprocess cleanup so process termination, stream draining, and timeout translation occur in separate stages. Update timeout, cancellation, blocked-stdin, and early-stdin-close tests with deterministic coordination.

Changes

Subprocess cleanup refactor

Layer / File(s)	Summary
Execution teardown and stream draining cuprum/_subprocess_execution.py	_wait_for_exit_code terminates only the process, while _drain_stream_consumers cancels, drains, decodes, and exports stream-consumer results.
Pre-drained timeout translation cuprum/_subprocess_timeout.py	_handle_stream_timeout synchronously builds timeout details from supplied output and validates the configured timeout.
Cleanup behaviour validation cuprum/unittests/test_subprocess_timeout.py	Tests cover consumer-draining outcomes, timeout translation, and process termination during timeout and cancellation.
Deterministic subprocess test coordination tests/helpers/stream_pipes.py, cuprum/unittests/test_safe_cmd_stdin.py, cuprum/unittests/test_safe_cmd_context.py, cuprum/unittests/test_observe.py, cuprum/unittests/test_observe_stdin_early_close.py, docs/developers-guide.md, docs/users-guide.md, docs/debugging/..., cuprum/unittests/__snapshots__/*	Tests and documentation use OS pipe capacity, execution events, marker files, captured timeout output, and updated wheel snapshots to coordinate subprocess cleanup scenarios.

Sequence Diagram(s)

sequenceDiagram
  participant SubprocessRunner
  participant ProcessWaiter
  participant StreamConsumers
  participant TimeoutHandler
  SubprocessRunner->>ProcessWaiter: wait for process exit
  ProcessWaiter->>ProcessWaiter: terminate process on timeout or cancellation
  SubprocessRunner->>StreamConsumers: cancel and drain stdout/stderr
  StreamConsumers-->>SubprocessRunner: return decoded output
  SubprocessRunner->>TimeoutHandler: translate timeout with output

Loading

Possibly related PRs

• leynos/cuprum#158: Modifies the same subprocess teardown, timeout, and stream-consumer handling paths.

Suggested labels: Issue

Suggested reviewers: codescene-delta-analysis, codescene-access

Poem

Processes pause, then streams unwind,
Exit codes leave cleanup behind.
Pipes may wedge, but tests now know
When drains finish and errors flow.
Timeout details arrive in line.

🚥 Pre-merge checks | ✅ 20

✅ Passed checks (20 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title matches the main scope: single-pass stream-consumer draining and deterministic test synchronisation.
Description check	✅ Passed	The description is clearly about the same subprocess timeout and test refactor changeset.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Testing (Overall)	✅ Passed	Added integration and property-based tests hit real subprocess timeouts, blocked stdin drains, cancellation, and observe events, with concrete assertions that fail on plausible regressions.
User-Facing Documentation	✅ Passed	Document the timeout behaviour in docs/users-guide.md: it now states preserved stdout/stderr on TimeoutExpired and the capture=False case, matching the code.
Developer Documentation	✅ Passed	PASS: docs/developers-guide.md now documents the new subprocess timeout/drain sequencing and shared pipe helper; no roadmap item or execplan needed updating.
Module-Level Documentation	✅ Passed	Every touched Python module has a top-level docstring; the new helper/test modules also state their role and relationship to subprocess tests.
Testing (Unit And Behavioural)	✅ Passed	PASS: the new stdin-close, timeout, and cancellation tests hit real subprocess/observe boundaries, while the helper tests cover concrete drain and timeout invariants.
Testing (Property / Proof)	✅ Passed	Approve it: the new ordering-sensitive drain invariant is covered by Hypothesis, and the timeout/cancellation cases are unit-tested without any proof obligation.
Testing (Compile-Time / Ui)	✅ Passed	No Rust/TS compile-time surface changed; the only UI-adjacent output change is covered by a focused wheel snapshot test with stable semantics.
Unit Architecture	✅ Passed	PASS: keep wait, drain, and timeout concerns split into explicit helpers; tests use injected seams and deterministic markers.
Domain Architecture	✅ Passed	Pass: the diff only touches subprocess execution, timeout handling, tests, helpers, and docs; no domain model or business rules are altered.
Observability	✅ Passed	Keep the existing ExecEvent/logging/tracing pipeline intact: start/exit/stdin_error still expose pid, exit_code, duration_s, and note, and timeouts preserve stdout/stderr.
Security And Privacy	✅ Passed	Changes only adjust subprocess timeout cleanup, tests, and docs; no secrets, auth bypasses, unsafe deserialisation, or sensitive data exposure appear.
Performance And Resource Use	✅ Passed	New drains are bounded to two tasks, waits/polls have finite deadlines, and the only added loops are small test-only coordinations; no avoidable perf regressions found.
Concurrency And State	✅ Passed	PASS: stream/waiter cancellation is made explicit, task ownership is local, and new tests cover timeout, cancellation, and interleaving paths.
Architectural Complexity And Maintainability	✅ Passed	PASS: _drain_stream_consumers and drain_blocking_payload_size replace duplicated teardown/payload logic with concrete, reused seams; no generic layer or hidden lifecycle was added.
Rust Compiler Lint Integrity	✅ Passed	Only Python and docs files changed; no Rust sources, lint suppressions, or clone workarounds were added.

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch fix/review-findings-test-sync

---

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

[leynos]

@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix. Include the file and symbol names indicated in the issue at the head of your response. Ensure that this is validated against the current version of the codegraph.

If further refinement to address this finding would be deleterious, please supply a clear explanatory one to two paragraph markdown message I can paste into the CodeScene web ui's diagnostic suppression function so this diagnostic can be silenced.

cuprum/unittests/test_safe_cmd_stdin.py

Comment on lines +275 to +277

    # Probe the real pipe capacity so the writer is guaranteed to stall in
    # drain() rather than assuming a ~64 KiB buffer.
    payload = b"x" * drain_blocking_payload_size()

❌ New issue: Code Duplication
The module contains 2 functions with similar structure: test_capture_timeout_with_blocked_stdin_writer_does_not_hang,test_direct_timeout_with_blocked_stdin_writer_does_not_hang

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 12a9296a89

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[pandalump]

@coderabbitai review

[coderabbitai]

✅ Action performed

Review finished.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[coderabbitai]

Actionable comments posted: 2

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (2)

cuprum/_subprocess_execution.py (1)

229-239: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Reuse _drain_stream_consumers in the stdin-failure path.

The extraction leaves this branch as the only place that still hand-rolls cancel-then-gather. Route it through the new helper so all three failure paths share one drain.

♻️ Proposed de-duplication

             # An unexpected stdin-writer failure (or a cancellation landing on
             # this await) must still reconcile the stdout/stderr consumers,
             # mirroring the timeout and cancellation paths above, so those tasks
             # are cancelled and drained before the error propagates.
-            _cancel_pending_consumers(consumers)
-            await asyncio.gather(*consumers, return_exceptions=True)
+            await _drain_stream_consumers(consumers)
             raise

🤖 Prompt for 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.

In `@cuprum/_subprocess_execution.py` around lines 229 - 239, Replace the manual
_cancel_pending_consumers(consumers) and asyncio.gather sequence in the
stdin_task exception handler with the existing _drain_stream_consumers helper.
Preserve the current BaseException handling and re-raise behavior so stdin
failures and cancellations still propagate after consumers are drained.

cuprum/unittests/test_subprocess_timeout.py (1)

97-124: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

The generated delays buy nothing once every consumer is pre-gathered.

Line 106 drives all consumers to completion before the drain, so stdout_delay/stderr_delay cannot influence any asserted outcome and the task.done() check at lines 110-112 is vacuous. The property currently reduces to "two independent outcomes decode independently" and would still pass against an implementation that never cancels.

Use the delays to leave a consumer genuinely pending — for example, gather with a zero timeout or skip the pre-gather for a generated subset — so the drain's cancel-then-decode path is actually explored. Otherwise drop the delay fields and shrink the example budget.

♻️ Sketch: let a generated consumer stay pending

-        # Let every consumer reach its outcome before draining, so the decode
-        # path (text preserved, failure -> None) is exercised rather than the
-        # cancellation of a reader that never got to run.
-        await asyncio.gather(*consumers, return_exceptions=True)
+        # Yield a bounded number of times so consumers with a smaller delay
+        # finish while longer-delayed ones stay pending, exercising the decode
+        # path and the cancellation path in the same property.
+        for _ in range(case.settle_steps):
+            await asyncio.sleep(0)
 
         stdout_text, stderr_text = await _drain_stream_consumers(consumers)

Expected values then become None whenever the consumer's delay exceeded
settle_steps, which is exactly the invariant worth pinning down.

🤖 Prompt for 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.

In `@cuprum/unittests/test_subprocess_timeout.py` around lines 97 - 124, Update
run_case and its generated test setup so stdout_delay and stderr_delay can leave
consumers pending when _drain_stream_consumers runs, instead of always awaiting
every consumer to completion first. Preserve completed consumer outcomes, but
expect None for consumers whose delay exceeds the settling window, and keep
assertions that verify pending tasks are cancelled and all tasks are done after
draining; remove delay fields only if the test no longer exercises timing.

🤖 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_observe.py`:
- Around line 317-327: Update the _STDIN_EARLY_CLOSE_CHILD script so its wait
for the go marker has a finite deadline; when the deadline expires, exit with a
non-zero status instead of continuing to sleep. Preserve the existing marker,
stdin-close, and completion behavior when the marker appears.

In `@tests/helpers/stream_pipes.py`:
- Around line 50-58: Update the pipe-capacity calculation around
fcntl.F_GETPIPE_SZ to check whether the constant is available before calling
fcntl.fcntl. On hosts without F_GETPIPE_SZ, return a conservative default
capacity while preserving the existing Linux calculation and descriptor cleanup
behavior.

---

Outside diff comments:
In `@cuprum/_subprocess_execution.py`:
- Around line 229-239: Replace the manual _cancel_pending_consumers(consumers)
and asyncio.gather sequence in the stdin_task exception handler with the
existing _drain_stream_consumers helper. Preserve the current BaseException
handling and re-raise behavior so stdin failures and cancellations still
propagate after consumers are drained.

In `@cuprum/unittests/test_subprocess_timeout.py`:
- Around line 97-124: Update run_case and its generated test setup so
stdout_delay and stderr_delay can leave consumers pending when
_drain_stream_consumers runs, instead of always awaiting every consumer to
completion first. Preserve completed consumer outcomes, but expect None for
consumers whose delay exceeds the settling window, and keep assertions that
verify pending tasks are cancelled and all tasks are done after draining; remove
delay fields only if the test no longer exercises timing.

🪄 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: 0161eabe-d10e-45fd-85ba-f94dd274eaac

📥 Commits

Reviewing files that changed from the base of the PR and between aa1fd32 and 3295b26.

📒 Files selected for processing (8)

• cuprum/_subprocess_execution.py
• cuprum/_subprocess_timeout.py
• cuprum/unittests/test_observe.py
• cuprum/unittests/test_safe_cmd_context.py
• cuprum/unittests/test_safe_cmd_stdin.py
• cuprum/unittests/test_subprocess_timeout.py
• docs/debugging/debugging-plan-2026-07-18-hypothesis-too-slow.md
• tests/helpers/stream_pipes.py

🔗 Linked repositories identified

CodeRabbit considers these linked repositories for cross-repo context during reviews:

• leynos/shared-actions (auto-detected)
• leynos/pylint-pypy-shim (auto-detected)
• leynos/whitaker (auto-detected)

[leynos]

@coderabbitai resume

[coderabbitai]

✅ Action performed

Reviews resumed.

[coderabbitai]

Actionable comments posted: 3

🤖 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_observe_stdin_early_close.py`:
- Around line 142-143: Update the test around _orchestrate to accept pytest’s
tmp_path fixture and construct the go and closed marker paths from it, replacing
tempfile.TemporaryDirectory. Remove the now-unused tempfile import, while
retaining the Path import if it is still needed at runtime.

In `@cuprum/unittests/test_safe_cmd_stdin.py`:
- Around line 285-304: Preserve the execution-strategy label from
execution_strategy in the test setup instead of discarding it, and use that
label in the timeout assertion message. Keep mode available for the capture
variant, while ensuring failures identify whether run() or run_sync() hung.

In `@docs/developers-guide.md`:
- Around line 1571-1572: Update the non-streaming `_execute_subprocess`
documentation step to include that `_cancel_stdin_writer` cancels and drains the
writer when `_wait_for_exit_code` raises `TimeoutError` or
`asyncio.CancelledError`, while retaining the existing happy-path description.

🪄 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: b7a42a7a-ac7d-40aa-bd60-d4fd70b127a2

📥 Commits

Reviewing files that changed from the base of the PR and between aa1fd32 and e6377a3.

📒 Files selected for processing (11)

• cuprum/_subprocess_execution.py
• cuprum/_subprocess_timeout.py
• cuprum/unittests/__snapshots__/test_maturin_build.ambr
• cuprum/unittests/test_observe.py
• cuprum/unittests/test_observe_stdin_early_close.py
• cuprum/unittests/test_safe_cmd_context.py
• cuprum/unittests/test_safe_cmd_stdin.py
• cuprum/unittests/test_subprocess_timeout.py
• docs/debugging/debugging-plan-2026-07-18-hypothesis-too-slow.md
• docs/developers-guide.md
• tests/helpers/stream_pipes.py

🔗 Linked repositories identified

CodeRabbit considers these linked repositories for cross-repo context during reviews:

• leynos/shared-actions (auto-detected)
• leynos/pylint-pypy-shim (auto-detected)
• leynos/whitaker (auto-detected)

[leynos]

@coderabbitai resume

[coderabbitai]

✅ Action performed

Reviews resumed.

[buzzybee-df12]

@coderabbitai review

[coderabbitai]

✅ Action performed

Review finished.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[coderabbitai]

Actionable comments posted: 1

🤖 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_observe_stdin_early_close.py`:
- Around line 36-47: Update the embedded _STDIN_EARLY_CLOSE_CHILD script to
import pathlib.Path, convert the CUPRUM_STDIN_CLOSE_GO and CUPRUM_STDIN_CLOSED
values to Path objects, and replace os.path.exists() with Path.exists() and
open()/write with Path.write_text().

🪄 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: e2df1dcb-3d45-48b9-897d-cf6a3bb0ac6d

📥 Commits

Reviewing files that changed from the base of the PR and between e6377a3 and f906b35.

📒 Files selected for processing (12)

• cuprum/_subprocess_execution.py
• cuprum/_subprocess_timeout.py
• cuprum/unittests/__snapshots__/test_maturin_build.ambr
• cuprum/unittests/test_observe.py
• cuprum/unittests/test_observe_stdin_early_close.py
• cuprum/unittests/test_safe_cmd_context.py
• cuprum/unittests/test_safe_cmd_stdin.py
• cuprum/unittests/test_subprocess_timeout.py
• docs/debugging/debugging-plan-2026-07-18-hypothesis-too-slow.md
• docs/developers-guide.md
• docs/users-guide.md
• tests/helpers/stream_pipes.py

🔗 Linked repositories identified

CodeRabbit considers these linked repositories for cross-repo context during reviews:

• leynos/shared-actions (auto-detected)
• leynos/pylint-pypy-shim (auto-detected)
• leynos/whitaker (auto-detected)

[codescene-access]

No quality gates enabled for this code.