Property-based tests for _handle_subprocess_timeout and _terminate_pipeline_remaining_stages

[coderabbitai]

Overview

Add Hypothesis tests (and CrossHair on extracted transition data) for _handle_subprocess_timeout (cuprum/_subprocess_execution.py, line 263) and _terminate_pipeline_remaining_stages (cuprum/_process_lifecycle.py, line 235).

Justification

Timeout and cleanup paths are temporal and branch-heavy, with risk of inconsistent payloads, non-idempotent cleanup, and double-termination scheduling.

Benefit

• Proves timeout payload consistency
• Proves idempotent cleanup
• Proves no double-termination scheduling

Cost Trade-Off

Medium-high. Requires fakes around process state.

Verifiability Improvements

Represent the process lifecycle as explicit enum/state records, then verify reducers independently.

Recommended Order

Phase 3 — state-machine work after refactors isolate pure transitions.

---

Raised from PR #62 by @leynos.

[coderabbitai]

🔗 Related PRs

#55 - Guard tee backend selector reentry (#53) [merged]
leynos/falcon-correlate#33 - Validate optional Celery integration (4.2.4) [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.}

---

🧪 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!

[leynos]

@CodeRabbit plan

[coderabbitai]

Coding Plan

Summary

• Extract pure state transition logic from impure async/process code into deterministic helper functions, enabling direct property testing without subprocess mocking
• Follow established Hypothesis patterns: synchronous test functions, asyncio.run() for async exercise, derandomize=True for CI reproducibility
• Use CrossHair's PEP-316 contracts with tight bounds for symbolic edge-case verification, complementing Hypothesis's randomised exploration
• Leverage existing stub classes (_StubSpawnProcess, _StubPipelineWaitProcess) for integration tests rather than inventing new faking infrastructure

Implementation Steps

Phase 1: Extract Pure State Transition Functions

Isolate the decision logic from impure side effects (process termination, event emission, exception raising) to create independently testable pure functions. This follows the ticket's guidance to "represent the process lifecycle as explicit enum/state records, then verify reducers independently."

Task 1: Extract timeout payload resolution logic

Create a pure function that determines which source (exception vs context) provides each field of the timeout payload, separating this branching logic from the side effects in _handle_subprocess_timeout.

• Add a frozen dataclass _ResolvedTimeoutPayload in cuprum/_subprocess_execution.py with fields: timeout: float, exited_at: float, stdout: str | None, stderr: str | None
• Create _resolve_timeout_payload(ctx: _SubprocessTimeoutContext, exc: TimeoutError | _SubprocessTimeoutError, fallback_exited_at: float) -> _ResolvedTimeoutPayload that encapsulates the isinstance branching
• Accept fallback_exited_at as a parameter rather than calling time.perf_counter() internally, making the function fully deterministic
• Raise RuntimeError when a bare TimeoutError is received but ctx.execution.timeout is None
• Update _handle_subprocess_timeout to call _resolve_timeout_payload with time.perf_counter() as the fallback, preserving existing behaviour

Task 2: Extract termination target selection logic

Create a pure function that computes which pipeline stage indices require termination, separating this from the async process termination in _terminate_pipeline_remaining_stages.

• Create _compute_termination_targets(done_statuses: cabc.Sequence[bool], failure_index: int) -> frozenset[int] in cuprum/_process_lifecycle.py
• Return the set of indices where idx != failure_index and not done_statuses[idx]
• Update _terminate_pipeline_remaining_stages to use this helper, replacing the inline loop logic

Task 3: Export extracted functions for testing

Make the pure functions accessible to unit tests via the established testing module pattern.

• Add _resolve_timeout_payload, _ResolvedTimeoutPayload, and _compute_termination_targets to the exports in cuprum/_testing.py
• Follow the existing pattern of re-exporting private symbols for test access

🤖 Prompt for AI agents

Extract pure state transition logic from impure async/process code in
`cuprum/_subprocess_execution.py` and `cuprum/_process_lifecycle.py`, then
expose the new helpers via `cuprum/_testing.py`.

**`cuprum/_subprocess_execution.py`**
- Add a frozen dataclass `_ResolvedTimeoutPayload` with fields: `timeout:
float`, `exited_at: float`, `stdout: str | None`, `stderr: str | None`
- Add function `_resolve_timeout_payload(ctx: _SubprocessTimeoutContext, exc:
TimeoutError | _SubprocessTimeoutError, fallback_exited_at: float) ->
_ResolvedTimeoutPayload`:
- Encapsulate the `isinstance` branching that determines which source (exception
vs context) provides each payload field
- Accept `fallback_exited_at` as a parameter — do NOT call `time.perf_counter()`
internally
- Raise `RuntimeError` when a bare `TimeoutError` is received but
`ctx.execution.timeout` is `None`
- Update `_handle_subprocess_timeout` to delegate to `_resolve_timeout_payload`,
passing `time.perf_counter()` as `fallback_exited_at`, preserving existing
behaviour

**`cuprum/_process_lifecycle.py`**
- Add function `_compute_termination_targets(done_statuses: cabc.Sequence[bool],
failure_index: int) -> frozenset[int]`:
- Return the set of indices where `idx != failure_index` and `not
done_statuses[idx]`
- Update `_terminate_pipeline_remaining_stages` to call
`_compute_termination_targets`, replacing the inline loop logic

**`cuprum/_testing.py`**
- Re-export `_resolve_timeout_payload`, `_ResolvedTimeoutPayload`, and
`_compute_termination_targets` following the existing pattern of re-exporting
private symbols for test access

Phase 2: Hypothesis Property Tests

Add property-based tests exercising the extracted pure functions and integration tests using fake process stubs, following established patterns from test_pipeline.py and test_stdin_property_based.py.

Task 1: Create test file with shared strategies and settings

Establish the test module structure with reusable Hypothesis strategies and settings profile.

• Create cuprum/unittests/test_lifecycle_property_based.py
• Define settings constant following existing pattern: deadline=None, derandomize=True, max_examples=100 for pure tests
• Create @st.composite strategy _timeout_exception_inputs() generating: a _SubprocessTimeoutContext with varied timeout values (including None), and either a _SubprocessTimeoutError or bare TimeoutError
• Create strategy _termination_inputs() generating: done_statuses as st.lists(st.booleans(), min_size=1, max_size=8) and failure_index via st.integers(min_value=0).filter(lambda i: i < len(statuses))

Task 2: Property tests for timeout payload resolution

Test that _resolve_timeout_payload always produces consistent, complete payloads.

• Property: when exception is _SubprocessTimeoutError, all payload fields match exception attributes exactly
• Property: when exception is bare TimeoutError and ctx.execution.timeout is not None, payload uses context fields and the provided fallback timestamp
• Property: when exception is bare TimeoutError and ctx.execution.timeout is None, function raises RuntimeError
• Property: returned payload has all fields non-None except stdout and stderr (which may legitimately be None)

Task 3: Property tests for termination target selection

Test that _compute_termination_targets never schedules double-termination and produces deterministic results.

• Property: failure_index is never in the returned set (no termination of failed stage)
• Property: no index i where done_statuses[i] is True appears in the returned set (no termination of completed stages)
• Property: function is idempotent — calling twice with identical inputs yields identical results
• Property: returned indices are all valid (within bounds of done_statuses)
• Property: len(result) == len(done_statuses) - 1 - sum(done_statuses) when failure_index < len(done_statuses) and is not itself done

Task 4: Integration tests with fake process stubs

Test the full _handle_subprocess_timeout and _terminate_pipeline_remaining_stages functions using existing stub patterns.

• Reuse _StubSpawnProcess pattern for fake asyncio.subprocess.Process with .returncode attribute
• Create fake _StageObservation with empty observe_hooks (making emit() a no-op)
• Test _handle_subprocess_timeout: verify it raises TimeoutExpired with correct fields for both exception types
• Test _terminate_pipeline_remaining_stages with _StubPipelineWaitProcess stubs: verify terminate_calls counts match expectations and failure_index stage is untouched
• Use asyncio.run() pattern for async tests, following existing conventions

🤖 Prompt for AI agents

Create `cuprum/unittests/test_lifecycle_property_based.py` containing Hypothesis
property tests for the pure functions extracted in Phase 1, plus integration
tests using existing stub patterns.

**Module-level settings and strategies**
- Define a settings constant with `deadline=None`, `derandomize=True`,
`max_examples=100`
- Define `@st.composite` strategy `_timeout_exception_inputs()` that generates a
`_SubprocessTimeoutContext` with varied `timeout` values (including `None`) and
either a `_SubprocessTimeoutError` or bare `TimeoutError`
- Define strategy `_termination_inputs()` generating `done_statuses` as
`st.lists(st.booleans(), min_size=1, max_size=8)` and `failure_index` via
`st.integers(min_value=0).filter(lambda i: i < len(statuses))`

**Property tests for `_resolve_timeout_payload`**
- Assert that when `exc` is `_SubprocessTimeoutError`, all payload fields match
exception attributes exactly
- Assert that when `exc` is bare `TimeoutError` and `ctx.execution.timeout` is
not `None`, the payload uses context fields and the provided fallback timestamp
- Assert that when `exc` is bare `TimeoutError` and `ctx.execution.timeout` is
`None`, the function raises `RuntimeError`
- Assert that returned payload has `timeout` and `exited_at` always non-`None`;
`stdout` and `stderr` may legitimately be `None`

**Property tests for `_compute_termination_targets`**
- Assert `failure_index` is never in the returned set
- Assert no index `i` where `done_statuses[i]` is `True` appears in the returned
set
- Assert the function is idempotent — calling twice with identical inputs yields
identical results
- Assert all returned indices are valid (within bounds of `done_statuses`)
- Assert `len(result) == len(done_statuses) - 1 - sum(done_statuses)` when
`failure_index < len(done_statuses)` and is not itself done

**Integration tests**
- Reuse `_StubSpawnProcess` for a fake `asyncio.subprocess.Process` with
`.returncode`
- Create a fake `_StageObservation` with empty `observe_hooks` so `emit()` is a
no-op
- Test `_handle_subprocess_timeout`: verify it raises `TimeoutExpired` with
correct fields for both exception types
- Test `_terminate_pipeline_remaining_stages` using `_StubPipelineWaitProcess`
stubs: verify `terminate_calls` counts match expectations and the
`failure_index` stage is untouched
- Wrap all async exercise in `asyncio.run()` following existing conventions

Phase 3: CrossHair Symbolic Contracts

Add PEP-316 docstring contracts and symbolic verification for the pure transition functions, following the pattern in test_line_splitting.py.

Task 1: Add CrossHair contracts for timeout payload resolution

Define symbolic contracts verifying the field-sourcing logic for all input combinations.

• Create contract function _resolve_timeout_payload_contract with PEP-316 docstring specifying:
  • Pre-condition: input context and exception are well-formed
  • Post-condition: when exc is _SubprocessTimeoutError, result fields equal exception fields
  • Post-condition: when exc is TimeoutError, result timeout equals context timeout
• Use tight bounds on string lengths (≤4 chars) for tractable symbolic analysis
• Add test invoking check_states with AnalysisKind.PEP316 and per_condition_timeout=10
• Skip test if CrossHair unavailable or Python version incompatible (following existing pattern)

Task 2: Add CrossHair contracts for termination target selection

Define symbolic contracts verifying the no-double-termination invariants.

• Create contract function _compute_termination_targets_contract with PEP-316 docstring specifying:
  • Pre-condition: 0 <= failure_index < len(done_statuses) and len(done_statuses) <= 5 (tight bound)
  • Post-condition: failure_index not in result
  • Post-condition: all(not done_statuses[i] for i in result)
• Add test invoking check_states with appropriate analysis options
• Include skip conditions matching existing CrossHair test patterns

🤖 Prompt for AI agents

Add CrossHair symbolic contracts for the two pure functions, following the
PEP-316 docstring pattern established in `test_line_splitting.py`.

**Contract for `_resolve_timeout_payload`**
- Create a contract function `_resolve_timeout_payload_contract` with a PEP-316
docstring containing:
  - Pre-condition: input context and exception are well-formed
- Post-condition: when `exc` is `_SubprocessTimeoutError`, result fields equal
exception fields
- Post-condition: when `exc` is `TimeoutError`, result timeout equals context
timeout
- Use tight bounds on string lengths (≤4 chars) to keep symbolic analysis
tractable
- Add a test that invokes `check_states` with `AnalysisKind.PEP316` and
`per_condition_timeout=10`
- Add skip conditions: skip if CrossHair is unavailable or the Python version is
incompatible, matching existing patterns

**Contract for `_compute_termination_targets`**
- Create a contract function `_compute_termination_targets_contract` with a
PEP-316 docstring containing:
- Pre-condition: `0 <= failure_index < len(done_statuses)` and
`len(done_statuses) <= 5`
  - Post-condition: `failure_index not in result`
  - Post-condition: `all(not done_statuses[i] for i in result)`
- Add a test invoking `check_states` with appropriate analysis options
- Include skip conditions matching existing CrossHair test patterns

Research

The cuprum package implements subprocess execution and pipeline orchestration with observability hooks. Timeout handling lives in cuprum/_subprocess_execution.py (branching on exception type to resolve payload fields), whilst pipeline termination cleanup lives in cuprum/_process_lifecycle.py (selecting which stages to terminate based on failure index and completion status). Existing Hypothesis tests follow a consistent pattern: synchronous test functions using asyncio.run() for async exercise, @settings(deadline=None, derandomize=True, suppress_health_check=[HealthCheck.function_scoped_fixture]), and @st.composite strategies. CrossHair usage exists in test_line_splitting.py with PEP-316 docstring contracts. Stub classes _StubSpawnProcess and _StubPipelineWaitProcess provide fake process behaviour for unit tests.

---

🚀 Next Steps

🤖 All AI agent prompts combined

Task: 1

Extract pure state transition logic from impure async/process code in
`cuprum/_subprocess_execution.py` and `cuprum/_process_lifecycle.py`, then
expose the new helpers via `cuprum/_testing.py`.

**`cuprum/_subprocess_execution.py`**
- Add a frozen dataclass `_ResolvedTimeoutPayload` with fields: `timeout:
float`, `exited_at: float`, `stdout: str | None`, `stderr: str | None`
- Add function `_resolve_timeout_payload(ctx: _SubprocessTimeoutContext, exc:
TimeoutError | _SubprocessTimeoutError, fallback_exited_at: float) ->
_ResolvedTimeoutPayload`:
- Encapsulate the `isinstance` branching that determines which source (exception
vs context) provides each payload field
- Accept `fallback_exited_at` as a parameter — do NOT call `time.perf_counter()`
internally
- Raise `RuntimeError` when a bare `TimeoutError` is received but
`ctx.execution.timeout` is `None`
- Update `_handle_subprocess_timeout` to delegate to `_resolve_timeout_payload`,
passing `time.perf_counter()` as `fallback_exited_at`, preserving existing
behaviour

**`cuprum/_process_lifecycle.py`**
- Add function `_compute_termination_targets(done_statuses: cabc.Sequence[bool],
failure_index: int) -> frozenset[int]`:
- Return the set of indices where `idx != failure_index` and `not
done_statuses[idx]`
- Update `_terminate_pipeline_remaining_stages` to call
`_compute_termination_targets`, replacing the inline loop logic

**`cuprum/_testing.py`**
- Re-export `_resolve_timeout_payload`, `_ResolvedTimeoutPayload`, and
`_compute_termination_targets` following the existing pattern of re-exporting
private symbols for test access
===============================================================================

Task: 2

Create `cuprum/unittests/test_lifecycle_property_based.py` containing Hypothesis
property tests for the pure functions extracted in Phase 1, plus integration
tests using existing stub patterns.

**Module-level settings and strategies**
- Define a settings constant with `deadline=None`, `derandomize=True`,
`max_examples=100`
- Define `@st.composite` strategy `_timeout_exception_inputs()` that generates a
`_SubprocessTimeoutContext` with varied `timeout` values (including `None`) and
either a `_SubprocessTimeoutError` or bare `TimeoutError`
- Define strategy `_termination_inputs()` generating `done_statuses` as
`st.lists(st.booleans(), min_size=1, max_size=8)` and `failure_index` via
`st.integers(min_value=0).filter(lambda i: i < len(statuses))`

**Property tests for `_resolve_timeout_payload`**
- Assert that when `exc` is `_SubprocessTimeoutError`, all payload fields match
exception attributes exactly
- Assert that when `exc` is bare `TimeoutError` and `ctx.execution.timeout` is
not `None`, the payload uses context fields and the provided fallback timestamp
- Assert that when `exc` is bare `TimeoutError` and `ctx.execution.timeout` is
`None`, the function raises `RuntimeError`
- Assert that returned payload has `timeout` and `exited_at` always non-`None`;
`stdout` and `stderr` may legitimately be `None`

**Property tests for `_compute_termination_targets`**
- Assert `failure_index` is never in the returned set
- Assert no index `i` where `done_statuses[i]` is `True` appears in the returned
set
- Assert the function is idempotent — calling twice with identical inputs yields
identical results
- Assert all returned indices are valid (within bounds of `done_statuses`)
- Assert `len(result) == len(done_statuses) - 1 - sum(done_statuses)` when
`failure_index < len(done_statuses)` and is not itself done

**Integration tests**
- Reuse `_StubSpawnProcess` for a fake `asyncio.subprocess.Process` with
`.returncode`
- Create a fake `_StageObservation` with empty `observe_hooks` so `emit()` is a
no-op
- Test `_handle_subprocess_timeout`: verify it raises `TimeoutExpired` with
correct fields for both exception types
- Test `_terminate_pipeline_remaining_stages` using `_StubPipelineWaitProcess`
stubs: verify `terminate_calls` counts match expectations and the
`failure_index` stage is untouched
- Wrap all async exercise in `asyncio.run()` following existing conventions
===============================================================================

Task: 3

Add CrossHair symbolic contracts for the two pure functions, following the
PEP-316 docstring pattern established in `test_line_splitting.py`.

**Contract for `_resolve_timeout_payload`**
- Create a contract function `_resolve_timeout_payload_contract` with a PEP-316
docstring containing:
  - Pre-condition: input context and exception are well-formed
- Post-condition: when `exc` is `_SubprocessTimeoutError`, result fields equal
exception fields
- Post-condition: when `exc` is `TimeoutError`, result timeout equals context
timeout
- Use tight bounds on string lengths (≤4 chars) to keep symbolic analysis
tractable
- Add a test that invokes `check_states` with `AnalysisKind.PEP316` and
`per_condition_timeout=10`
- Add skip conditions: skip if CrossHair is unavailable or the Python version is
incompatible, matching existing patterns

**Contract for `_compute_termination_targets`**
- Create a contract function `_compute_termination_targets_contract` with a
PEP-316 docstring containing:
- Pre-condition: `0 <= failure_index < len(done_statuses)` and
`len(done_statuses) <= 5`
  - Post-condition: `failure_index not in result`
  - Post-condition: `all(not done_statuses[i] for i in result)`
- Add a test invoking `check_states` with appropriate analysis options
- Include skip conditions matching existing CrossHair test patterns

💡 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!