Split _subprocess_execution.py along its stated seams (#117)

[leynos]

Summary

This branch splits the 510-line cuprum/_subprocess_execution.py along the
seams its own TODO named, removing the too-many-lines suppression pragma
instead of carrying it.

Closes #117.

• cuprum/_subprocess_stdin.py — _emit_stdin_error, _write_stdin, _spawn_stdin_writer; the cuprum.stdin logger now lives in the module whose name matches it.
• cuprum/_subprocess_timeout.py — timeout data/errors, timeout translation and shared exit-event helpers. The review follow-up uses structural match/case binding while retaining the TimeoutError fallback.
• cuprum/_subprocess_execution.py keeps the runner (_execute_subprocess, _run_subprocess_with_streams, _spawn_subprocess, stream-consumer wiring).

The redundant _resolve_timeout re-export is dropped from __all__;
cuprum/sh.py imports it from cuprum._subprocess_context.

Review walkthrough

• The code moves verbatim apart from import plumbing, _require_timeout
  consolidation, and the requested structural timeout match.
• cuprum/unittests/test_observe.py patches _write_stdin at its new home.
• The private module boundaries are documented in the developers guide, design,
  and ADR-005; the unrelated duplicate ExecPlan artefact section was removed.

Validation

• make check-fmt: pass
• make lint: pass
• make typecheck: pass
• make test: pass (723 passed, 50 skipped; Rust nextest 4/4 passed)
• make markdownlint: pass
• make nixie: pass

Notes

Related: #75 tracks property
tests for _handle_subprocess_timeout, which now has a single, separately
importable home. The wheel-build snapshot reflects the new file list. The
observed Hypothesis health-check failure was replayed successfully with its
supplied seed, so no suppression or speculative test change was made.

References

• https://lody.ai/leynos/sessions/66292eb4-a28f-4468-b9d5-23d8763bfc0c

[sourcery-ai]

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

Please try again later or upgrade to continue using Sourcery

[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: 8 minutes

Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available.
You're only billed for reviews past your plan's rate limits ($0.25/file).

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: e06a373f-8d4e-4960-bff7-93469d49f47b

📥 Commits

Reviewing files that changed from the base of the PR and between 901ae4e and 8fbe035.

📒 Files selected for processing (24)

• .github/workflows/ci.yml
• cuprum/_pipeline_types.py
• cuprum/_subprocess_execution.py
• cuprum/_subprocess_stdin.py
• cuprum/_subprocess_timeout.py
• cuprum/adapters/tracing_adapter.py
• cuprum/events.py
• cuprum/sh.py
• cuprum/unittests/__snapshots__/test_maturin_build.ambr
• cuprum/unittests/_adapter_test_support.py
• cuprum/unittests/test_observe.py
• cuprum/unittests/test_safe_cmd_context.py
• cuprum/unittests/test_safe_cmd_run.py
• cuprum/unittests/test_safe_cmd_stdin.py
• cuprum/unittests/test_safe_cmd_streams.py
• cuprum/unittests/test_subprocess_timeout.py
• cuprum/unittests/test_tracing_adapter.py
• docs/adr-007-subprocess-execution-module-boundaries.md
• docs/contents.md
• docs/cuprum-design.md
• docs/debugging/debugging-plan-2026-07-18-hypothesis-too-slow.md
• docs/developers-guide.md
• rust/rust-toolchain.toml
• tests/helpers/execution.py

Walkthrough

Split subprocess execution into orchestration, stdin lifecycle, timeout handling, and observability modules. Update cancellation and timeout cleanup, add regression coverage, document module boundaries, package the new modules, and pin the Rust toolchain to 1.85.0.

Changes

Subprocess execution refactor

Layer / File(s)	Summary
Centralise stdin handling cuprum/_subprocess_stdin.py, cuprum/_pipeline_types.py, cuprum/events.py	Implement stdin writing, closure, diagnostics, cancellation, and structured stdin error observations.
Centralise timeout handling cuprum/_subprocess_timeout.py	Add timeout contexts, exit-event emission, timeout translation, and stream cleanup helpers.
Coordinate subprocess execution cuprum/_subprocess_execution.py, cuprum/sh.py	Wire process spawning, stream consumers, stdin tasks, exit waiting, cancellation cleanup, and timeout imports.
Record structured tracing events cuprum/adapters/tracing_adapter.py, cuprum/unittests/_adapter_test_support.py	Record stdin errors as span events with operation, error type, and note attributes.
Validate execution and packaging cuprum/unittests/*, cuprum/unittests/__snapshots__/*, tests/helpers/execution.py	Add stdin, timeout, cancellation, context, consumer-cleanup, tracing, and packaging coverage.
Record module boundaries docs/adr-007-subprocess-execution-module-boundaries.md, docs/contents.md, docs/cuprum-design.md, docs/developers-guide.md	Document ownership and compatibility boundaries for subprocess execution, stdin, and timeout handling.

Ancillary maintenance records

Layer / File(s)	Summary
Align Rust toolchain configuration rust/rust-toolchain.toml, .github/workflows/ci.yml	Pin local and CI Rust setup to version 1.85.0 with minimal tooling components.
Document Hypothesis investigation docs/debugging/debugging-plan-2026-07-18-hypothesis-too-slow.md	Record reproduction evidence, falsifiable hypotheses, execution order, and termination criteria for a slow health check.

Sequence Diagram(s)

sequenceDiagram
  participant SafeCmd
  participant _execute_subprocess
  participant _spawn_subprocess
  participant _spawn_stdin_writer
  participant _wait_for_exit_code
  participant _cancel_pending_consumers
  SafeCmd->>_execute_subprocess: run command
  _execute_subprocess->>_spawn_subprocess: spawn process and configure streams
  _execute_subprocess->>_spawn_stdin_writer: start writer when stdin data exists
  _execute_subprocess->>_wait_for_exit_code: await process exit
  _wait_for_exit_code->>_cancel_pending_consumers: cancel and drain consumers
  _execute_subprocess-->>SafeCmd: return result or propagate exception

Loading

Possibly related PRs

• leynos/cuprum#60: Modifies subprocess stdin writer creation and cleanup behaviour.

Suggested labels: Issue

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

Poem

Pipes close softly, tasks take flight,
Timeouts yield before the night.
Streams are gathered, errors shown,
Three neat modules find their home.
Rust tools march in steady line.

🚥 Pre-merge checks | ✅ 4 | ❌ 16

❌ Failed checks (1 warning, 15 inconclusive)

Check name	Status	Explanation	Resolution
Out of Scope Changes check	⚠️ Warning	The CI Rust toolchain downgrade, rust/rust-toolchain.toml, and the debugging-plan document fall outside #117.	Split the CI/toolchain tweaks and unrelated docs into separate PRs unless they are needed to validate the refactor.
Testing (Overall)	❓ Inconclusive	Repository clone failed, so this custom check could not run with code access.	Retry the review run. If this persists, inspect pre-merge custom-check logs for infrastructure or agent runtime failures.
User-Facing Documentation	❓ Inconclusive	Repository clone failed, so this custom check could not run with code access.	Retry the review run. If this persists, inspect pre-merge custom-check logs for infrastructure or agent runtime failures.
Developer Documentation	❓ Inconclusive	Repository clone failed, so this custom check could not run with code access.	Retry the review run. If this persists, inspect pre-merge custom-check logs for infrastructure or agent runtime failures.
Module-Level Documentation	❓ Inconclusive	Repository clone failed, so this custom check could not run with code access.	Retry the review run. If this persists, inspect pre-merge custom-check logs for infrastructure or agent runtime failures.
Testing (Unit And Behavioural)	❓ Inconclusive	Repository clone failed, so this custom check could not run with code access.	Retry the review run. If this persists, inspect pre-merge custom-check logs for infrastructure or agent runtime failures.
Testing (Property / Proof)	❓ Inconclusive	Repository clone failed, so this custom check could not run with code access.	Retry the review run. If this persists, inspect pre-merge custom-check logs for infrastructure or agent runtime failures.
Testing (Compile-Time / Ui)	❓ Inconclusive	Repository clone failed, so this custom check could not run with code access.	Retry the review run. If this persists, inspect pre-merge custom-check logs for infrastructure or agent runtime failures.
Unit Architecture	❓ Inconclusive	Repository clone failed, so this custom check could not run with code access.	Retry the review run. If this persists, inspect pre-merge custom-check logs for infrastructure or agent runtime failures.
Domain Architecture	❓ Inconclusive	Repository clone failed, so this custom check could not run with code access.	Retry the review run. If this persists, inspect pre-merge custom-check logs for infrastructure or agent runtime failures.
Observability	❓ Inconclusive	Repository clone failed, so this custom check could not run with code access.	Retry the review run. If this persists, inspect pre-merge custom-check logs for infrastructure or agent runtime failures.
Security And Privacy	❓ Inconclusive	Repository clone failed, so this custom check could not run with code access.	Retry the review run. If this persists, inspect pre-merge custom-check logs for infrastructure or agent runtime failures.
Performance And Resource Use	❓ Inconclusive	Repository clone failed, so this custom check could not run with code access.	Retry the review run. If this persists, inspect pre-merge custom-check logs for infrastructure or agent runtime failures.
Concurrency And State	❓ Inconclusive	Repository clone failed, so this custom check could not run with code access.	Retry the review run. If this persists, inspect pre-merge custom-check logs for infrastructure or agent runtime failures.
Architectural Complexity And Maintainability	❓ Inconclusive	Repository clone failed, so this custom check could not run with code access.	Retry the review run. If this persists, inspect pre-merge custom-check logs for infrastructure or agent runtime failures.
Rust Compiler Lint Integrity	❓ Inconclusive	Repository clone failed, so this custom check could not run with code access.	Retry the review run. If this persists, inspect pre-merge custom-check logs for infrastructure or agent runtime failures.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title names the refactor, matches the changed module, and includes the linked issue number.
Description check	✅ Passed	The description directly explains the subprocess split and related cleanup work.
Linked Issues check	✅ Passed	The refactor follows #117: it splits stdin and timeout helpers out, keeps runner wiring, removes the suppression, and drops the redundant export.
Docstring Coverage	✅ Passed	Docstring coverage is 98.97% which is sufficient. The required threshold is 80.00%.

📋 Issue Planner

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

View plan used: #117

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch issue-117-split-subprocess-execution

---

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

[leynos]

@coderabbitai Have the following failed checks now been resolved?

If further work is required, please provide an AI agent prompt for the remaining work to be done to address these failures.

Do not treat warnings as optional or aspirational. Where a change is out of scope for this PR, propose a GitHub issue unless one exists already. (Treat documentation and validation coverage as in scope).

Developer Documentation ⚠️ Warning The branch changes only code plus an execplan; no developer-guide or design-doc edits record the new _subprocess_stdin/_subprocess_timeout split. Document the new internal module boundaries in docs/developers-guide.md and docs/cuprum-design.md, and add an ADR or design note for the split.

[coderabbitai]

Tip

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

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo '== PR changed files =='
gh pr view 158 --repo leynos/cuprum --json files --jq '.files[].path'

echo
echo '== Candidate documentation files =='
fd -a -t f -i 'developers-guide.md' -i 'cuprum-design.md' -i '*adr*' -i '*design*' docs .github 2>/dev/null || true

echo
echo '== Documentation references to subprocess module boundaries =='
rg -n -i -C 3 \
  'subprocess_execution|subprocess_stdin|subprocess_timeout|stdin.*timeout|timeout.*stdin|architecture decision|ADR' \
  docs 2>/dev/null || true

Length of output: 53513

---

@leynos Mark the Developer Documentation warning as resolved.

PR #158 now includes all required documentation:

• docs/developers-guide.md defines the _subprocess_execution, _subprocess_stdin, and _subprocess_timeout ownership boundaries and lifecycle rules.
• docs/cuprum-design.md records the architectural split and links to the decision record.
• docs/adr-005-subprocess-execution-module-boundaries.md records the accepted decision, rationale, alternatives, compatibility constraints, and module responsibilities.
• docs/contents.md indexes ADR-005.

No further work, AI-agent prompt, or follow-up GitHub issue is required for this check.

[pandalump]

@coderabbitai resume

[coderabbitai]

✅ Action performed

Reviews resumed.

[leynos]

@coderabbitai resume

[coderabbitai]

✅ Action performed

Reviews resumed.

[codescene-access]

No quality gates enabled for this code.

[coderabbitai]

Actionable comments posted: 4

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)

175-198: 🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift

Keep stream consumers owned by the timeout path. _wait_for_exit_code(..., consumers=consumers) drains the consumer tasks on TimeoutError, then _handle_stream_timeout() gathers those same tasks again and maps the cancelled results to None, which drops captured stdout/stderr from TimeoutExpired. Move consumer cancellation/drain into the timeout branch here and let _wait_for_exit_code() wait for the exit code only.

🤖 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 175 - 198, Move stream-consumer
cancellation and draining out of _wait_for_exit_code and into the TimeoutError
branch of _run_subprocess_with_streams, ensuring consumers are drained exactly
once before _handle_stream_timeout processes them. Update _wait_for_exit_code to
wait only for the process exit code, and preserve captured stdout/stderr when
constructing TimeoutExpired.

cuprum/unittests/test_observe.py (1)

333-376: 🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift

Pipe-capacity race still unresolved.

This is the outstanding finding already recorded in the PR summary: the test still sizes the payload from _drain_blocking_payload_size() but never synchronizes the child's sys.stdin.close() with the writer's drain(). Correctness depends entirely on the child closing stdin before the parent's buffered write exhausts the probed capacity — nothing here guarantees that ordering. Deterministic synchronization (a control channel/marker the child signals after closing stdin, or a test seam that lets the harness confirm the writer has actually blocked before the child proceeds) is still needed rather than relying on size alone.

🤖 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_observe.py` around lines 333 - 376, Make
test_observe_emits_stdin_error_event_when_process_closes_stdin_early
deterministic by synchronizing the child’s sys.stdin.close() with the parent
writer reaching its blocked drain state. Add a control channel or explicit
marker/seam so the child signals after closing stdin and the harness confirms
the writer is blocked before allowing the child to proceed; do not rely solely
on _drain_blocking_payload_size().

🤖 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 314-330: Move _drain_blocking_payload_size from
cuprum/unittests/test_observe.py lines 314-330 into the shared test helper
module, preserving its dynamic pipe-capacity probing. In
cuprum/unittests/test_safe_cmd_stdin.py lines 261-289 and 292-347, import and
use this helper for the stdin payload in
test_direct_timeout_with_blocked_stdin_writer_does_not_hang and
test_stdin_input_cancellation_cleans_up_task, replacing both hardcoded 1 MiB
allocations.
- Around line 6-17: Update test_observe.py so the pipe-capacity logic using
fcntl and F_GETPIPE_SZ is only imported and executed on supported platforms,
allowing test collection elsewhere. Replace the payload-size-based early-close
trigger with an explicit synchronization seam that deterministically coordinates
the close path, while preserving the existing test behavior and assertions.

In `@cuprum/unittests/test_safe_cmd_context.py`:
- Around line 191-229: Update
test_run_does_not_invoke_after_hooks_on_cancellation to replace the fixed
asyncio.sleep readiness guess with the established sh.observe start-phase hook
and an asyncio.Event. Await the event before calling task.cancel(), ensuring
cancellation occurs only after the subprocess has actually started while
preserving the existing cancellation and after-hook assertions.

In `@docs/debugging/debugging-plan-2026-07-18-hypothesis-too-slow.md`:
- Around line 39-41: Insert a comma after “0.098 seconds” in the sentence
beginning “Sampling 100 `_TAGS` examples” to separate the two independent
clauses, while leaving the surrounding investigation results unchanged.

---

Outside diff comments:
In `@cuprum/_subprocess_execution.py`:
- Around line 175-198: Move stream-consumer cancellation and draining out of
_wait_for_exit_code and into the TimeoutError branch of
_run_subprocess_with_streams, ensuring consumers are drained exactly once before
_handle_stream_timeout processes them. Update _wait_for_exit_code to wait only
for the process exit code, and preserve captured stdout/stderr when constructing
TimeoutExpired.

In `@cuprum/unittests/test_observe.py`:
- Around line 333-376: Make
test_observe_emits_stdin_error_event_when_process_closes_stdin_early
deterministic by synchronizing the child’s sys.stdin.close() with the parent
writer reaching its blocked drain state. Add a control channel or explicit
marker/seam so the child signals after closing stdin and the harness confirms
the writer is blocked before allowing the child to proceed; do not rely solely
on _drain_blocking_payload_size().

🪄 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: 3885c852-f415-4619-805c-2de8f9cc1c88

📥 Commits

Reviewing files that changed from the base of the PR and between faf9741 and 901ae4e.

📒 Files selected for processing (24)

• .github/workflows/ci.yml
• cuprum/_pipeline_types.py
• cuprum/_subprocess_execution.py
• cuprum/_subprocess_stdin.py
• cuprum/_subprocess_timeout.py
• cuprum/adapters/tracing_adapter.py
• cuprum/events.py
• cuprum/sh.py
• cuprum/unittests/__snapshots__/test_maturin_build.ambr
• cuprum/unittests/_adapter_test_support.py
• cuprum/unittests/test_observe.py
• cuprum/unittests/test_safe_cmd_context.py
• cuprum/unittests/test_safe_cmd_run.py
• cuprum/unittests/test_safe_cmd_stdin.py
• cuprum/unittests/test_safe_cmd_streams.py
• cuprum/unittests/test_subprocess_timeout.py
• cuprum/unittests/test_tracing_adapter.py
• docs/adr-007-subprocess-execution-module-boundaries.md
• docs/contents.md
• docs/cuprum-design.md
• docs/debugging/debugging-plan-2026-07-18-hypothesis-too-slow.md
• docs/developers-guide.md
• rust/rust-toolchain.toml
• tests/helpers/execution.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)

[coderabbitai]

🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

printf '\n== test_observe outline ==\n'
ast-grep outline cuprum/unittests/test_observe.py --view expanded || true

printf '\n== relevant lines ==\n'
sed -n '1,120p' cuprum/unittests/test_observe.py
printf '\n---\n'
sed -n '300,470p' cuprum/unittests/test_observe.py

printf '\n== sibling platform skip patterns ==\n'
rg -n "win32|skip\\(" cuprum/unittests/test_safe_cmd_run.py cuprum/unittests -g 'test_*.py' || true

Repository: leynos/cuprum

Length of output: 11230

---

🏁 Script executed (no clone):

#!/bin/bash
python3 - <<'PY'
import sys
print(sys.platform)
PY

Length of output: 131

---

🏁 Script executed:

#!/bin/bash
set -euo pipefail

printf 'FILE EXISTS?\n'
test -f cuprum/unittests/test_observe.py && echo yes

printf '\nLINES AROUND IMPORTS\n'
nl -ba cuprum/unittests/test_observe.py | sed -n '1,40p'

printf '\nLINES AROUND HELPER AND TEST\n'
nl -ba cuprum/unittests/test_observe.py | sed -n '300,390p'

printf '\nPLATFORM SKIPS IN UNIT TESTS\n'
rg -n "win32|sys\.platform|pytest\.skip" cuprum/unittests/test_*.py || true

Repository: leynos/cuprum

Length of output: 230

---

Guard the pipe-capacity probe behind a platform check.

• Move fcntl behind a lazy import or platform guard; the module-level import and F_GETPIPE_SZ call make this test Linux-only and can break collection on other platforms.
• Replace the payload-size heuristic with an explicit synchronisation seam for the early-close path.

🤖 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_observe.py` around lines 6 - 17, Update test_observe.py
so the pipe-capacity logic using fcntl and F_GETPIPE_SZ is only imported and
executed on supported platforms, allowing test collection elsewhere. Replace the
payload-size-based early-close trigger with an explicit synchronization seam
that deterministically coordinates the close path, while preserving the existing
test behavior and assertions.

[coderabbitai]

📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Extract the pipe-blocking payload sizing into one shared helper. test_observe.py computes a reliable "drain-blocking" payload size by probing the real OS pipe capacity, while test_safe_cmd_stdin.py duplicates the same intent twice with a hardcoded 1 MiB literal and a comment assuming a "~64 KiB pipe buffer". The dynamic probe is strictly more correct; reuse it everywhere instead of letting the assumption drift.

• cuprum/unittests/test_observe.py#L314-L330: move _drain_blocking_payload_size() into a shared test helper module (e.g. tests/helpers/execution.py, which already hosts _RunKwargs/ExecuteFn).
• cuprum/unittests/test_safe_cmd_stdin.py#L261-L289: replace the b"x" * (1024 * 1024) literal in test_direct_timeout_with_blocked_stdin_writer_does_not_hang with the shared helper's computed size.
• cuprum/unittests/test_safe_cmd_stdin.py#L292-L347: replace the b"x" * (1024 * 1024) literal in test_stdin_input_cancellation_cleans_up_task with the shared helper's computed size.

📍 Affects 2 files

• cuprum/unittests/test_observe.py#L314-L330 (this comment)
• cuprum/unittests/test_safe_cmd_stdin.py#L261-L289
• cuprum/unittests/test_safe_cmd_stdin.py#L292-L347

🤖 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_observe.py` around lines 314 - 330, Move
_drain_blocking_payload_size from cuprum/unittests/test_observe.py lines 314-330
into the shared test helper module, preserving its dynamic pipe-capacity
probing. In cuprum/unittests/test_safe_cmd_stdin.py lines 261-289 and 292-347,
import and use this helper for the stdin payload in
test_direct_timeout_with_blocked_stdin_writer_does_not_hang and
test_stdin_input_cancellation_cleans_up_task, replacing both hardcoded 1 MiB
allocations.

[coderabbitai]

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

Replace the fixed sleep with a readiness signal before cancelling.

await asyncio.sleep(0.1) # Let the process start is a scheduler guess, not proof the subprocess has actually started before task.cancel() fires. Under CI load this can cancel before the process is even spawned, silently defeating the regression this test targets. Sibling files in this same PR (test_safe_cmd_streams.py) already fix the identical pattern with an sh.observe hook that sets an asyncio.Event on the "start" phase.

🔧 Proposed fix using the established readiness-signal pattern

+    from cuprum.events import ExecEvent
+
     command = python_builder("-c", "import time; time.sleep(10)")
+    started = asyncio.Event()
+
+    def on_start(ev: ExecEvent) -> None:
+        """Signal once the subprocess has actually started."""
+        if ev.phase == "start":
+            started.set()

     async def orchestrate() -> None:
         """Start the command under a scope, then cancel it."""
-        with scoped(
-            ScopeConfig(
-                allowlist=frozenset([command.program]), after_hooks=(after_hook,)
-            )
-        ):
+        with (
+            scoped(
+                ScopeConfig(
+                    allowlist=frozenset([command.program]), after_hooks=(after_hook,)
+                )
+            ),
+            sh.observe(on_start),
+        ):
             task = asyncio.create_task(
                 command.run(
                     output=RunOutputOptions(capture=False),
                     context=ExecutionContext(cancel_grace=0.1),
                 ),
             )
-            await asyncio.sleep(0.1)  # Let the process start
+            await asyncio.wait_for(started.wait(), timeout=2.0)
             task.cancel()
             with pytest.raises(asyncio.CancelledError):
                 await task

As per coding guidelines, "Changes involving shared state, asynchronous execution, ordering, cancellation, locks, transactions, task lifetimes, or parallelism must make the concurrency model explicit and test failure-prone interleavings."

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	def test_run_does_not_invoke_after_hooks_on_cancellation(
	python_builder: cabc.Callable[..., SafeCmd],
	) -> None:
	"""run() does not invoke after hooks when task is cancelled."""
	from cuprum.context import ScopeConfig, scoped
	after_called = False
	def after_hook(cmd: SafeCmd, result: CommandResult) -> None:
	"""Record that the after hook was invoked."""
	nonlocal after_called
	_, _ = cmd, result
	after_called = True
	# Use a long-running command that we can cancel
	command = python_builder("-c", "import time; time.sleep(10)")
	async def orchestrate() -> None:
	"""Start the command under a scope, then cancel it."""
	with scoped(
	ScopeConfig(
	allowlist=frozenset([command.program]), after_hooks=(after_hook,)
	)
	):
	task = asyncio.create_task(
	command.run(
	output=RunOutputOptions(capture=False),
	context=ExecutionContext(cancel_grace=0.1),
	),
	)
	await asyncio.sleep(0.1) # Let the process start
	task.cancel()
	with pytest.raises(asyncio.CancelledError):
	await task
	asyncio.run(orchestrate())
	assert after_called is False, (
	"after hooks must not run when the command is cancelled"
	)
	def test_run_does_not_invoke_after_hooks_on_cancellation(
	python_builder: cabc.Callable[..., SafeCmd],
	) -> None:
	"""run() does not invoke after hooks when task is cancelled."""
	from cuprum.context import ScopeConfig, scoped
	from cuprum.events import ExecEvent
	after_called = False
	started = asyncio.Event()
	def after_hook(cmd: SafeCmd, result: CommandResult) -> None:
	"""Record that the after hook was invoked."""
	nonlocal after_called
	_, _ = cmd, result
	after_called = True
	def on_start(ev: ExecEvent) -> None:
	"""Signal once the subprocess has actually started."""
	if ev.phase == "start":
	started.set()
	command = python_builder("-c", "import time; time.sleep(10)")
	async def orchestrate() -> None:
	"""Start the command under a scope, then cancel it."""
	with (
	scoped(
	ScopeConfig(
	allowlist=frozenset([command.program]), after_hooks=(after_hook,)
	)
	),
	sh.observe(on_start),
	):
	task = asyncio.create_task(
	command.run(
	output=RunOutputOptions(capture=False),
	context=ExecutionContext(cancel_grace=0.1),
	),
	)
	await asyncio.wait_for(started.wait(), timeout=2.0)
	task.cancel()
	with pytest.raises(asyncio.CancelledError):
	await task
	asyncio.run(orchestrate())
	assert after_called is False, (
	"after hooks must not run when the command is cancelled"
	)

🤖 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_safe_cmd_context.py` around lines 191 - 229, Update
test_run_does_not_invoke_after_hooks_on_cancellation to replace the fixed
asyncio.sleep readiness guess with the established sh.observe start-phase hook
and an asyncio.Event. Await the event before calling task.cancel(), ensuring
cancellation occurs only after the subprocess has actually started while
preserving the existing cancellation and after-hook assertions.

Source: Coding guidelines