Implement fail-fast pipeline with failure metadata and internals - #15
Conversation
![terragon-labs[bot]](https://github.githubassets.com/images/gravatars/gravatar-user-420.png){kind=avatar}
…lure - Pipeline stages now fail fast: when any stage exits non-zero, remaining stages are terminated. - `PipelineResult` includes `failure_index` and `failure` properties identifying the failing stage. - Internal `_wait_for_pipeline` captures failure index, terminates downstream stages, and returns detailed exit metadata. - Updated tests and behavior scenarios to cover fail-fast semantics and failure stage attribution. - Documentation updated to describe fail-fast pipeline behavior, termination policy, and failure metadata exposure. This improves pipeline robustness and diagnostic capability by aborting on the first error and surfacing the failing stage explicitly. Co-authored-by: terragon-labs[bot] <terragon-labs[bot]@users.noreply.github.com>
Reviewer's GuideRefactors pipeline execution internals by extracting stream and pipeline coordination logic into new internal modules, implements fail-fast semantics in _wait_for_pipeline that track the first failing stage and terminate remaining stages, and threads this information through to PipelineResult via new failure_index and failure accessors, with tests and docs updated accordingly. Sequence diagram for fail-fast pipeline execution and terminationsequenceDiagram
actor Client
participant sh as cuprum_sh
participant internals as _pipeline_internals
participant proc as asyncio_subprocess
participant streams as _streams
Client->>sh: pipeline.run(capture, echo, context)
sh->>internals: _run_pipeline(parts, capture, echo, context)
internals->>internals: _prepare_pipeline_config()
internals->>internals: _run_before_hooks() per SafeCmd
internals->>proc: _spawn_pipeline_processes(parts, config)
proc-->>internals: processes, stderr_tasks, stdout_task
internals->>internals: _create_pipe_tasks(processes)
internals->>internals: _flatten_stream_tasks(stderr_tasks, stdout_task)
internals->>internals: _wait_for_pipeline(processes, pipe_tasks, stream_tasks, cancel_grace)
internals->>internals: _PipelineWaitState.from_processes(processes)
loop wait for stages
internals->>proc: process.wait() as Task
proc-->>internals: exit_code
internals->>internals: _process_completed_task(task, state, processes, cancel_grace)
alt first non_zero exit
internals->>internals: set state.failure_index
internals->>internals: _terminate_pipeline_remaining_stages(processes, wait_tasks, failure_index, cancel_grace)
internals->>internals: _terminate_process_via_wait_task() for each remaining stage
internals->>proc: process.terminate()/kill()
proc-->>internals: wait_task completes
end
end
internals->>internals: _finalize_pipeline_wait(pipe_tasks, pipe_results, caught)
internals-->>internals: _PipelineWaitResult(exit_codes, failure_index)
internals->>streams: _consume_stream() for stderr/stdout
streams-->>internals: captured text
internals->>internals: build CommandResult per stage
internals->>internals: _run_pipeline_after_hooks()
internals-->>sh: PipelineResult(stages, failure_index)
sh-->>Client: PipelineResult
Client->>Client: inspect result.failure and result.failure_index
Class diagram for pipeline coordination and fail-fast typesclassDiagram
class CommandResult {
+str program
+tuple argv
+int exit_code
+int pid
+str stdout
+str stderr
}
class PipelineResult {
+tuple~CommandResult~ stages
+int failure_index
+CommandResult final()
+CommandResult failure()
+bool ok()
}
class ExecutionContext {
+float cancel_grace
+str encoding
+str errors
+IO stdout_sink
+IO stderr_sink
}
class SafeCmd {
+str program
+tuple argv
+tuple argv_with_program
}
class _StreamConfig {
+bool capture_output
+bool echo_output
+IO sink
+str encoding
+str errors
}
class _PipelineRunConfig {
+ExecutionContext ctx
+bool capture
+bool echo
+IO stdout_sink
+IO stderr_sink
+bool capture_or_echo
+_StreamConfig stream_config
}
class _PipelineWaitResult {
+list~int~ exit_codes
+int failure_index
}
class _PipelineWaitState {
+list~Task~ wait_tasks
+dict~Task,int~ task_to_index
+list~int~ exit_codes
+int failure_index
+from_processes(processes)
}
PipelineResult "*" --> "*" CommandResult : stages
_PipelineRunConfig --> ExecutionContext : ctx
_PipelineRunConfig --> _StreamConfig : creates
_PipelineWaitResult --> CommandResult : indexes into
_PipelineWaitState --> _PipelineWaitResult : builds
SafeCmd --> ExecutionContext : validated_by
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
|
Note Other AI code review bot(s) detectedCodeRabbit has detected other AI code review bot(s) in this pull request and will avoid duplicating their findings in the review comments. This may lead to a less comprehensive review. WalkthroughRefactor pipeline runtime into two new internal modules: Changes
Sequence Diagram(s)sequenceDiagram
participant User
participant Pipeline
participant Spawner as Spawner/Launcher
participant Stage1 as Stage 1
participant Stage2 as Stage 2
participant Stage3 as Stage 3
participant Monitor as Wait/Monitor
participant Cleanup as Cleaner
User->>Pipeline: run(parts)
Pipeline->>Spawner: _spawn_pipeline_processes(parts, config)
Spawner->>Stage1: start (stdin/stdout/stderr)
Spawner->>Stage2: start (pipe from Stage1)
Spawner->>Stage3: start (pipe from Stage2)
Spawner-->>Monitor: return processes & stream tasks
par execution
Stage1->>Monitor: exit 0
Stage2->>Monitor: exit non-zero
Stage3->>Monitor: running / blocked on pipe
end
Monitor->>Monitor: detect first non-zero (failure_index)
rect rgb(255,230,230)
Monitor->>Stage3: _terminate_pipeline_remaining_stages(failure_index)
Monitor->>Cleanup: _cleanup_spawned_processes()
end
Monitor->>Pipeline: assemble PipelineResult(failure_index, per-stage results)
Pipeline-->>User: return PipelineResult
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes
Poem
Pre-merge checks and finishing touches✅ Passed checks (3 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
Comment |
![codescene-delta-analysis[bot]](https://avatars.githubusercontent.com/in/53921?s=60&v=4){kind=small}
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
|
@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix: Comment on file Command results for each pipeline stage, in execution order. For stages
whose stdout is streamed into the next stage, ``stdout`` is ``None``.
The final stage carries captured stdout when enabled.
failure_index:❌ New issue: Lines of Code in a Single File |
|
@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix: Comment on file stream_tasks: list[asyncio.Task[str | None]],
cancel_grace: float,
) -> list[int]:
) -> _PipelineWaitResult:❌ New issue: Complex Method |
This comment was marked as resolved.
This comment was marked as resolved.
This comment was marked as resolved.
This comment was marked as resolved.
…dules - Extracted internal pipeline execution coordination and fail-fast semantics into `cuprum/_pipeline_internals.py`. - Added internal stream handling utilities for subprocess I/O in `cuprum/_streams.py`. - Cleaned up `cuprum/sh.py` by importing relevant internals and removing duplicated code. - Improved code modularity and separation of concerns for pipeline and streaming logic. Co-authored-by: terragon-labs[bot] <terragon-labs[bot]@users.noreply.github.com>
leynos
changed the title
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
📜 Review details
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
📒 Files selected for processing (9)
cuprum/_pipeline_internals.py(1 hunks)cuprum/_streams.py(1 hunks)cuprum/sh.py(2 hunks)cuprum/unittests/test_pipeline.py(3 hunks)docs/cuprum-design.md(4 hunks)docs/roadmap.md(1 hunks)docs/users-guide.md(1 hunks)tests/behaviour/test_pipeline_execution.py(3 hunks)tests/features/pipeline_execution.feature(1 hunks)
🧰 Additional context used
📓 Path-based instructions (10)
docs/**/*.md
📄 CodeRabbit inference engine (AGENTS.md)
docs/**/*.md: Use markdown files within thedocs/directory as a knowledge base and source of truth for project requirements, dependency choices, and architectural decisions.
When new decisions are made, requirements change, libraries are added/removed, or architectural patterns evolve, proactively update the relevant file(s) in thedocs/directory to reflect the latest state.
All documentation must adhere to the documentation style guide atdocs/documentation-style-guide.md.
Record any design decisions made in the relevant design document.
Files:
docs/cuprum-design.mddocs/users-guide.mddocs/roadmap.md
**/*.md
📄 CodeRabbit inference engine (AGENTS.md)
**/*.md: For Markdown files (.md only), ensure linting passes by runningmake markdownlint.
For Markdown files, validate Mermaid diagrams by runningmake nixie.
Markdown paragraphs and bullet points must be wrapped at 80 columns.
Code blocks in Markdown must be wrapped at 120 columns.
Markdown tables and headings must not be wrapped.
Use dashes (-) for list bullets in Markdown.
Use GitHub-flavoured Markdown footnotes ([^1]) for references and footnotes.
Validate Markdown files usingmake markdownlint.
Runmake fmtafter any documentation changes to format all Markdown files and fix table markup.
Validate Mermaid diagrams in Markdown files by runningmake nixie.
Files:
docs/cuprum-design.mddocs/users-guide.mddocs/roadmap.md
⚙️ CodeRabbit configuration file
**/*.md: * Avoid 2nd person or 1st person pronouns ("I", "you", "we")
- Use en-GB-oxendict (-ize / -yse / -our) spelling and grammar
- Headings must not be wrapped.
- Documents must start with a level 1 heading
- Headings must correctly increase or decrease by no more than one level at a time
- Use GitHub-flavoured Markdown style for footnotes and endnotes.
- Numbered footnotes must be numbered by order of appearance in the document.
Files:
docs/cuprum-design.mddocs/users-guide.mddocs/roadmap.md
docs/**/*.{md,mdx,rst,txt}
📄 CodeRabbit inference engine (docs/documentation-style-guide.md)
docs/**/*.{md,mdx,rst,txt}: Use British English based on Oxford English Dictionary (en-GB-oxendict) conventions: use -ize suffixes (realize, organization), -lyse suffixes (analyse, paralyse, catalyse), -our suffixes (colour, behaviour, neighbour), -re suffixes (calibre, centre, fibre), double 'l' (cancelled, counsellor, cruellest), maintain 'e' (likeable, liveable, rateable), -ogue suffixes (analogue, catalogue)
The word 'outwith' is acceptable in documentation
Use the Oxford comma in documentation: 'ships, planes, and hovercraft' where it aids comprehension
Treat company names as collective nouns in documentation, for example 'Lille Industries are expanding'
Write headings in sentence case in documentation
Use Markdown headings (#,##,###, and so on) in order without skipping levels
Always provide a language identifier for fenced code blocks in documentation; use 'plaintext' for non-code text
Use-as the first level bullet and renumber lists when items change in documentation
Prefer inline links using[text](url)or angle brackets around the URL in documentation
Ensure blank lines before and after bulleted lists and fenced blocks in documentation
Ensure tables have a delimiter line below the header row in documentation
Expand any uncommon acronym on first use in documentation, for example 'Continuous Integration (CI)'
Wrap paragraphs at 80 columns in documentation
Wrap code at 120 columns in documentation
Do not wrap tables in documentation
Use footnotes referenced with[^label]in documentation
Include Mermaid diagrams in documentation where they add clarity
When embedding figures in documentation, useand provide brief alt text describing the content
Add a short description before each Mermaid diagram in documentation so screen readers can understand it
Files:
docs/cuprum-design.mddocs/users-guide.mddocs/roadmap.md
docs/**/*.{md,mdx,rst,txt,rs}
📄 CodeRabbit inference engine (docs/documentation-style-guide.md)
Keep US spelling when used in API contexts, for example 'color'
Files:
docs/cuprum-design.mddocs/users-guide.mddocs/roadmap.md
docs/**/*.{md,mdx}
📄 CodeRabbit inference engine (docs/documentation-style-guide.md)
Follow markdownlint recommendations for Markdown formatting
Files:
docs/cuprum-design.mddocs/users-guide.mddocs/roadmap.md
**/*.py
📄 CodeRabbit inference engine (AGENTS.md)
**/*.py: For Python files, ensure testing passes all relevant unit and behavioral tests by runningmake test.
For Python files, ensure linting passes by runningmake lint.
For Python files, ensure formatting adheres to standards by runningmake check-fmtand applyingmake fmtif needed.
For Python files, ensure type checking passes by runningmake typecheck.
For Python development, refer to Python-specific guidelines in the.rules/directory, including Python Code Style Guidelines, Context Managers, Exceptions and Logging, Generators, Project Configuration, Return Patterns, and Typing.
**/*.py: Use snake_case for Python filenames (e.g., http_client.py, task_queue.py)
Use PascalCase for class names
Use snake_case for variable and function names
Use UPPER_SNAKE_CASE for module-level constants
Prefix private/internal functions and variables with a single underscore (_)
Enable full static type coverage using Pyright and maintain typing throughout the codebase
Use TypedDict or Dataclass for structured data, preferring @DataClass(slots=True) for internal-only usage
Avoid using Any type; use Unknown, generics, or cast() with documentation instead
Provide explicit return type annotations (e.g., -> None, -> str) for all public functions and class methods
Enforce strict mode in Pyright and treat all Pyright warnings as CI errors; use # pyright: ignore sparingly with explanation
Avoid side effects at import time; modules should not modify global state or perform actions on import
Use .env or settings modules for environment-specific configuration; never hardcode secrets
Use Ruff for formatting; let Ruff handle whitespace and formatting entirely
Use NumPy-format docstrings for public functions, classes, and modules
Use inline comments to explain tricky or non-obvious code logic and decisions
**/*.py: Use context managers (withcontextlib.contextmanageror class-based__enter__/__exit__) to encapsulate setup and teardown logic for resource management (f...
Files:
cuprum/unittests/test_pipeline.pycuprum/_streams.pycuprum/_pipeline_internals.pycuprum/sh.pytests/behaviour/test_pipeline_execution.py
⚙️ CodeRabbit configuration file
**/*.py: - Keep C90 / mccabe complexity ≤ 9
- Follow single responsibility and CQRS (command/query segregation)
- Prefer structural pattern matching to
- Prefer structural pattern matching over
isinstance()or imperative decomposition.- Docstrings must follow the
numpystyle guide. Use a single-line summary for private functions and methods, and full structured docs for all public interfaces.- Move conditionals with >2 branches to predicate/helper functions
- Avoid
eval,exec,pickle, monkey-patching,ctypes, unsafe shell- Every module must begin with a triple-quoted docstring explaining its purpose, utility, and usage, including example calls if appropriate.
- Comments and docs must follow en-GB-oxendict (-ize / -yse / -our) spelling and grammar
- Lint suppressions:
- Blanket
# noqa, file-level skips, and categories are forbidden- Only narrow in-line disables (
# noqa: XYZ) are permitted, and must be accompanied byFIXME:or a ticket link, and used only as a last resort.- Use
pytestfixtures for shared setup (conftest.pyorfixtures/)- Replace duplicate tests with
@pytest.mark.parametrize- Prefer
pytest-mockorunittest.mockfor stubs/mocks- Use
assert …, "message"over bare asserts- Reflect all API/behaviour changes in
docs/and update roadmap on completion- Files must not exceed 400 logical lines:
- Decompose large modules into subpackages
- Split large
match/caseor dispatch tables by domain and collocate with targets if appropriate- Move bulky data (fixtures, templates) to external files for parsing at runtime
- Mutable defaults and shadowed built-ins are forbidden
- All code must have clear type hints using modern style (
A | B,list[str],class Foo[A]:,type Bar = int, etc.), with ABC imports drawn from the correct stdlib module.- All path manipulation must be performed using pathlib for cross platform safety. Do not use string manipulation for extraction of path components or filename ele...
Files:
cuprum/unittests/test_pipeline.pycuprum/_streams.pycuprum/_pipeline_internals.pycuprum/sh.pytests/behaviour/test_pipeline_execution.py
**/unittests/test_*.py
📄 CodeRabbit inference engine (.rules/python-00.md)
Colocate unit tests with code using an unittests subdirectory with test_ prefix (e.g., user_auth/unittests/test_models.py)
Files:
cuprum/unittests/test_pipeline.py
**/test_*.py
📄 CodeRabbit inference engine (.rules/python-00.md)
**/test_*.py: Use pytest idioms: prefer fixtures over setup/teardown methods, parametrize broadly, avoid unnecessary mocks
Group related tests using class with method names prefixed by test_
Write tests from a user's perspective; test public behavior rather than internals
Avoid mocking too much; prefer test doubles only for external services or non-deterministic behaviors
Files:
cuprum/unittests/test_pipeline.pytests/behaviour/test_pipeline_execution.py
**/*test*.py
📄 CodeRabbit inference engine (.rules/python-exception-design-raising-handling-and-logging.md)
Use specific exception types and message constraints with
pytest.raises(SpecificError, match=r"pattern")in tests; avoid overly broad exception assertions (B017)
Files:
cuprum/unittests/test_pipeline.pytests/behaviour/test_pipeline_execution.py
docs/users-guide.md
📄 CodeRabbit inference engine (AGENTS.md)
docs/users-guide.md: Ensure new functionality is clearly documented in thedocs/users-guide.mdfile.
Ensure revised functionality is clearly documented in thedocs/users-guide.mdfile.
docs/users-guide.md: Document builder pattern indocs/users-guide.mdfor constructingSafeCmdinstances with typed argv handling
Document hook usage patterns, including basic logging hook that emits start/exit events compatible withlogging
Provide a scaffold and guidance for project-specific builders, including a template module and checklist indocs/users-guide.md
Files:
docs/users-guide.md
🧬 Code graph analysis (3)
cuprum/_streams.py (1)
cuprum/unittests/test_pipeline.py (6)
read(29-34)write(47-49)drain(51-57)write_eof(59-60)close(62-63)wait_closed(65-66)
cuprum/sh.py (2)
cuprum/_pipeline_internals.py (5)
_merge_env(44-52)_PipelineWaitResult(461-463)_prepare_pipeline_config(100-117)_terminate_process(55-74)_wait_for_pipeline(411-457)cuprum/_streams.py (5)
_close_stream_writer(94-118)_consume_stream(26-51)_pump_stream(54-75)_StreamConfig(16-23)_write_chunk(121-140)
tests/behaviour/test_pipeline_execution.py (2)
tests/helpers/catalogue.py (1)
python_catalogue(17-26)cuprum/sh.py (4)
PipelineResult(123-161)ok(117-119)ok(154-156)failure(147-151)
🪛 LanguageTool
docs/cuprum-design.md
[style] ~411-~411: Three successive sentences begin with the same word. Consider rewording the sentence or use a thesaurus to find a synonym.
Context: ..., PipelineResult.stdoutisNone`. - When a stage exits non-zero, Cuprum terminat...
(ENGLISH_WORD_REPEAT_BEGINNING_RULE)
🔍 Remote MCP
Let me try a web search for relevant best practices and patterns:Based on the search results, I have gathered relevant context for reviewing this PR. Let me compile a focused summary of the key architectural and technical considerations.
Summary of Relevant Context for PR Review
Asyncio Subprocess Termination Patterns & Graceful Shutdown
Calling terminate() starts the termination process but doesn't wait for it to complete; you should usually await process.wait() to actually wait for the subprocess to finish. The PR implements this pattern via the _terminate_process and _terminate_pipeline_remaining_stages helpers in _pipeline_internals.py.
The recommended pattern pairs terminate() with an awaited wait(), often within a try-except block or with a timeout; if termination times out, escalate to kill() to forcefully shutdown. This aligns with the PR's fail-fast termination design that uses both graceful termination (SIGTERM) with timeout escalation to forceful termination (SIGKILL).
Pipe and Stream I/O Deadlock Prevention
Use the communicate() method when using pipes to avoid deadlock conditions where child processes generate output that fills the OS pipe buffer. The new _streams.py module implements dedicated stream consumption (_consume_stream, _pump_stream) to prevent this via async StreamReader/StreamWriter handling, which is important for multi-stage pipelines with I/O capture.
Cyclomatic Complexity Reduction via Refactoring
The EXTRACT METHOD refactoring is the primary weapon to address "bumpy road" code complexity that contains multiple chunks of nested logic. The PR directly addresses this by:
- Extracting task-completion handling into helper methods
- Splitting
_wait_for_pipelineresponsibilities across_process_completed_taskand_finalize_pipeline_waithelpers - Separating pipeline internals into dedicated modules
Nested control structures increase cyclomatic complexity and make code harder to understand; refactor nested structures into simpler sequential logic or extract them into separate functions. The PR's modularization strategy aligns with this guidance.
File Size and Maintainability Considerations
Refactoring involves splitting large functions into smaller, more manageable functions, removing redundant code, and simplifying complex conditional statements. The target reduction from ~674-1059 lines to ~400-450 lines in sh.py follows industry guidance on function and module sizing for maintainability.
Key Review Considerations
- Graceful Termination with Escalation: Verify that
_terminate_processcorrectly implements SIGTERM→timeout→SIGKILL escalation for downstream stages. - Stream Safety: Confirm
_streams.pyhelpers correctly prevent deadlocks in multi-stage pipelines with I/O capture via proper async backpressure handling (drain()). - Complexity Reduction: Validate that extracted helpers (
_process_completed_task,_finalize_pipeline_wait) preserve all exit-code collection and failure-index logic originally in_wait_for_pipeline. - API Stability: Ensure
PipelineResult.failure_indexand.failureadditions are backward-compatible and properly integrate with existing pipeline semantics. - Test Coverage: Verify fail-fast behavior across early, middle, and final-stage failures; confirm per-stage exit codes and unexpected pipe failures are correctly surfaced.
⏰ Context from checks skipped due to timeout of 120000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)
- GitHub Check: Sourcery review
- GitHub Check: Sourcery review
🔇 Additional comments (36)
docs/users-guide.md (1)
146-148: LGTM!The fail-fast documentation is clear and aligns with the implementation. The
result.failure/result.failure_indexsurface is correctly referenced.cuprum/_streams.py (6)
1-13: LGTM!Module structure follows best practices: deferred annotations, TYPE_CHECKING guard for asyncio, and sensible chunk size constant.
15-24: LGTM!Clean configuration dataclass with appropriate frozen/slots settings.
26-51: LGTM!Stream consumption logic correctly handles capture/echo modes with proper chunked reading and final decoding.
54-75: LGTM!The deadlock prevention pattern (continuing to drain stdout when downstream closes) is correctly implemented. This aligns with best practices for subprocess pipe handling.
94-118: LGTM with a note.The defensive error suppression is appropriate for pipe cleanup where writer state may be indeterminate. The
getattrfallback forwait_closedhandles compatibility gracefully.
121-140: LGTM!Efficient sink writing that bypasses redundant encoding when the underlying buffer is accessible. The blocking write trade-off is well-documented.
docs/roadmap.md (1)
49-51: LGTM!The roadmap update correctly reflects the completed implementation. The PR delivers fail-fast termination, downstream stage cleanup, and failure surface via
failure_index/failure, with comprehensive test coverage for early, middle, and late stage failures.cuprum/unittests/test_pipeline.py (4)
11-20: LGTM!Importing internal symbols for unit testing internal behaviour is appropriate.
184-185: LGTM!The new assertions correctly validate the
failureproperty returns the expected stage andfailure_indexis set appropriately.
288-324: LGTM!Well-designed test double with explicit control over termination sequencing via the ready event. The signal codes (-15 for SIGTERM, -9 for SIGKILL) are accurate.
327-429: LGTM!Comprehensive test coverage for fail-fast behaviour across all failure positions:
- Early stage (0): downstream stages 1 and 2 terminated
- Middle stage (1): both upstream (0) and downstream (2) terminated
- Final stage (2): no termination needed as all stages already complete
The ready-event pattern provides deterministic control over completion order.
tests/features/pipeline_execution.feature (1)
15-28: LGTM!BDD scenarios comprehensively cover fail-fast behaviour for first, middle, and final stage failures. The step wording is clear and testable.
docs/cuprum-design.md (3)
392-413: LGTM!The
PipelineResultdocumentation correctly describes the new failure surface. The parallel "When..." structure in the contract notes is appropriate for clarity, despite the style lint.
571-573: LGTM!Design decision documentation aligns with implementation: fail-fast terminates remaining stages and surfaces the triggering stage.
974-978: LGTM!Error propagation policy is clearly documented, consistent with the implementation and other documentation sections.
cuprum/sh.py (2)
122-161: LGTM!The
failure_indexfield andfailureproperty are correctly implemented:
- Proper typing with
int | None- Sensible default of
Nonefor success cases- Clean property implementation that handles the None case
391-414: LGTM!Pipeline execution cleanly delegates to
_run_pipelinefrom the internal module, maintaining a clear separation between public API and implementation details.cuprum/_pipeline_internals.py (14)
1-18: LGTM!Clean module structure with proper use of
TYPE_CHECKINGguard for type-only imports and deferred annotation evaluation.
22-27: LGTM!Sound approach to avoid circular imports by deferring module access to runtime.
30-41: LGTM!Clean separation of hook invocation with clear docstring.
44-52: LGTM!Deferred
osimport avoids import-time side effects. Dict merge with|=is idiomatic.
55-74: LGTM!Correct implementation of graceful termination with SIGTERM→timeout→SIGKILL escalation. Exception handling appropriately guards against race conditions where the process exits between checks.
77-117: LGTM!Well-structured configuration dataclass with computed properties. Follows
@dataclass(frozen=True, slots=True)guideline.
120-162: Acceptable orchestration complexity.The
PLR0914suppression is justified here. This is a coordination function that necessarily tracks multiple concerns (processes, tasks, results). The logic flow remains readable.
165-222: LGTM!Clear separation between stream configuration and task creation. Docstrings effectively explain the FD selection logic.
225-302: LGTM!Robust cleanup implementation. Catching
BaseExceptionat line 293 is correct—it ensures cleanup onKeyboardInterruptandSystemExit, not just standard exceptions.
305-354: LGTM!Good decomposition of pipe orchestration concerns. Correctly treats
BrokenPipeErrorandConnectionResetErroras expected conditions when downstream stages terminate early.
411-457: LGTM!Clean decomposition of the wait loop into
_process_completed_taskand_finalize_pipeline_waithelpers. The fail-fast logic is clear: first non-zero exit triggers termination of remaining stages.
486-541: LGTM!Correct use of
asyncio.shieldto prevent wait task cancellation during graceful termination. Thestrict=Trueinzipat line 525 ensures process and task list lengths match.
544-551: LGTM!Clean after-hook invocation with
strict=Trueenforcement on zip.
19-19:_MIN_PIPELINE_STAGESis imported and used incuprum/sh.py(line 376). This is appropriate modularisation—defining the constant in the internals module and exporting it for external use is correct.Likely an incorrect or invalid review comment.
tests/behaviour/test_pipeline_execution.py (4)
44-57: LGTM!Clear scenario definitions with descriptive docstrings.
148-187: LGTM!Clear fixture separation for each failure scenario. The 2-second sleeps are necessary to validate fail-fast termination of long-running stages.
222-237: LGTM!Assertions correctly verify fail-fast semantics: the failure index must precede the final stage, and subsequent stages must have non-zero exit codes (from termination signals).
240-251: LGTM!Correct assertions for final-stage failure: preceding stages complete successfully, and
failurereferences the last stage.
- Extracted general _terminate_process_with_wait() to reuse termination logic with customizable completion check and waiter. - Simplified _terminate_process() and _terminate_process_via_wait_task() using the new helper. - Enhanced _wait_for_pipeline() and associated pipeline wait state management by consolidating wait task tracking and error cleanup. - Removed unnecessary stream_tasks parameter from internal wait and cleanup functions for clearer ownership and responsibilities. - Improved exception handling in _run_pipeline to properly cancel and await stream tasks on failure. - Added _PipelineWaitState data class to encapsulate state for pipeline process wait tasks. - Updated imports and test code to align with refactoring. This refactor improves code clarity, error handling, and reuse in pipeline subprocess termination and waiting mechanisms. Co-authored-by: terragon-labs[bot] <terragon-labs[bot]@users.noreply.github.com>
leynos
changed the title
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
|
@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix: cuprum/unittests/test_pipeline.py Comment on lines +183 to +184 assert result.failure is result.stages[-1]
assert result.failure_index == 1❌ New issue: Code Duplication |
This comment was marked as resolved.
This comment was marked as resolved.
|
@coderabbitai Have the following now been resolved? |
|
Tip For best results, initiate chat on the files or code changes. 🧩 Analysis chain🏁 Script executed: Length of output: 17876 🏁 Script executed: Length of output: 13841 🏁 Script executed: Length of output: 213 @leynos: Yes, all the review comments have been successfully resolved! Here's the status of each issue: ✅ Comment 1: stream_tasks Lifecycle & OwnershipResolved. The async def _wait_for_pipeline(
processes: list[asyncio.subprocess.Process],
*,
pipe_tasks: list[asyncio.Task[None]],
cancel_grace: float,
) -> _PipelineWaitResult:The caller ( stderr_by_stage = [
None if task is None else await task for task in stderr_tasks
]
final_stdout = None if stdout_task is None else await stdout_taskThis makes ownership explicit and prevents task leaks. ✅ Comment 2: Termination Logic ConsolidationResolved. The common termination logic has been extracted into
✅ Comment 3: stream_tasks Ownership SplitResolved. As noted above, ✅ Comment 4: Missing Success-Case TestResolved. The test assert result.ok is True
assert result.failure is None
assert result.failure_index is NoneThis ensures the "no failure" semantics are properly tested and won't regress. All structural issues have been addressed. The refactoring maintains backward compatibility whilst improving code clarity and lifecycle management. 🎉 |
…and termination - Introduced helper _run_test_pipeline to reduce duplication in pipeline stage exit code tests. - Refactored existing synchronous tests to use the helper for clarity. - Added async _exercise_wait_for_pipeline helper to test _wait_for_pipeline with stub processes. - Added assertions to verify pipeline failure indices and proper termination of downstream processes on failures. - Improved coverage for fail-fast behavior of pipeline stages in async scenarios. - Removed duplicated test code by consolidating common test setup logic. Co-authored-by: terragon-labs[bot] <terragon-labs[bot]@users.noreply.github.com>
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
|
@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix: cuprum/unittests/test_pipeline.py Comment on file with scoped(allowlist=frozenset([python_program])):
result = pipeline.run_sync()
result = _run_test_pipeline([0, 1])❌ New issue: Code Duplication |
This comment was marked as resolved.
This comment was marked as resolved.
Refactor pipeline run_sync and fail-fast asyncio tests to use pytest parametrize for multiple scenarios covering success and failure cases. - Combines success and failure cases into single parameterized test for run_sync failure semantics. - Parameterizes fail-fast pipeline tests for early, middle, and final stage failures, verifying termination and failure indexes. - Enhances test clarity and coverage by consolidating similar tests. Co-authored-by: terragon-labs[bot] <terragon-labs[bot]@users.noreply.github.com>
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
|
@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix: cuprum/unittests/test_pipeline.py Comment on lines +487 to +517 def test_wait_for_pipeline_fail_fast_scenarios(
scenario: str,
exit_codes: tuple[int, int, int],
ready_stages: frozenset[int],
*,
expected_failure_index: int,
expected_exit_codes: list[int],
terminated_stages: frozenset[int],
) -> None:
"""Validate fail-fast termination behaviour across different failure scenarios.
Tests that:
- Early stage failures terminate all downstream stages
- Middle stage failures terminate all downstream stages
- Final stage failures record failure index without terminating others
"""
p0, p1, p2, result = asyncio.run(
_exercise_wait_for_pipeline(
exit_codes=exit_codes,
ready_stages=ready_stages,
),
)
_assert_pipeline_failure(
result,
failure_index=expected_failure_index,
exit_codes=expected_exit_codes,
)
for idx, process in enumerate([p0, p1, p2]):
_assert_stage_terminated(process, should_terminate=(idx in terminated_stages))❌ New issue: Excess Number of Function Arguments |
This comment was marked as resolved.
This comment was marked as resolved.
Refactored the fail-fast pipeline test scenarios by introducing a frozen dataclass `_FailFastScenario` to encapsulate test parameters. Updated parametrized test cases to use instances of this dataclass instead of multiple separate parameters, enhancing readability and maintainability of the test code. Co-authored-by: terragon-labs[bot] <terragon-labs[bot]@users.noreply.github.com>
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
There was a problem hiding this comment.
Actionable comments posted: 2
📜 Review details
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
📒 Files selected for processing (4)
cuprum/_pipeline_internals.py(1 hunks)cuprum/_testing.py(1 hunks)cuprum/sh.py(2 hunks)cuprum/unittests/test_pipeline.py(4 hunks)
🧰 Additional context used
📓 Path-based instructions (4)
**/*.py
📄 CodeRabbit inference engine (AGENTS.md)
**/*.py: For Python files, ensure testing passes all relevant unit and behavioral tests by runningmake test.
For Python files, ensure linting passes by runningmake lint.
For Python files, ensure formatting adheres to standards by runningmake check-fmtand applyingmake fmtif needed.
For Python files, ensure type checking passes by runningmake typecheck.
For Python development, refer to Python-specific guidelines in the.rules/directory, including Python Code Style Guidelines, Context Managers, Exceptions and Logging, Generators, Project Configuration, Return Patterns, and Typing.
**/*.py: Use snake_case for Python filenames (e.g., http_client.py, task_queue.py)
Use PascalCase for class names
Use snake_case for variable and function names
Use UPPER_SNAKE_CASE for module-level constants
Prefix private/internal functions and variables with a single underscore (_)
Enable full static type coverage using Pyright and maintain typing throughout the codebase
Use TypedDict or Dataclass for structured data, preferring @DataClass(slots=True) for internal-only usage
Avoid using Any type; use Unknown, generics, or cast() with documentation instead
Provide explicit return type annotations (e.g., -> None, -> str) for all public functions and class methods
Enforce strict mode in Pyright and treat all Pyright warnings as CI errors; use # pyright: ignore sparingly with explanation
Avoid side effects at import time; modules should not modify global state or perform actions on import
Use .env or settings modules for environment-specific configuration; never hardcode secrets
Use Ruff for formatting; let Ruff handle whitespace and formatting entirely
Use NumPy-format docstrings for public functions, classes, and modules
Use inline comments to explain tricky or non-obvious code logic and decisions
**/*.py: Use context managers (withcontextlib.contextmanageror class-based__enter__/__exit__) to encapsulate setup and teardown logic for resource management (f...
Files:
cuprum/unittests/test_pipeline.pycuprum/_testing.pycuprum/sh.pycuprum/_pipeline_internals.py
⚙️ CodeRabbit configuration file
**/*.py: - Keep C90 / mccabe complexity ≤ 9
- Follow single responsibility and CQRS (command/query segregation)
- Prefer structural pattern matching to
- Prefer structural pattern matching over
isinstance()or imperative decomposition.- Docstrings must follow the
numpystyle guide. Use a single-line summary for private functions and methods, and full structured docs for all public interfaces.- Move conditionals with >2 branches to predicate/helper functions
- Avoid
eval,exec,pickle, monkey-patching,ctypes, unsafe shell- Every module must begin with a triple-quoted docstring explaining its purpose, utility, and usage, including example calls if appropriate.
- Comments and docs must follow en-GB-oxendict (-ize / -yse / -our) spelling and grammar
- Lint suppressions:
- Blanket
# noqa, file-level skips, and categories are forbidden- Only narrow in-line disables (
# noqa: XYZ) are permitted, and must be accompanied byFIXME:or a ticket link, and used only as a last resort.- Use
pytestfixtures for shared setup (conftest.pyorfixtures/)- Replace duplicate tests with
@pytest.mark.parametrize- Prefer
pytest-mockorunittest.mockfor stubs/mocks- Use
assert …, "message"over bare asserts- Reflect all API/behaviour changes in
docs/and update roadmap on completion- Files must not exceed 400 logical lines:
- Decompose large modules into subpackages
- Split large
match/caseor dispatch tables by domain and collocate with targets if appropriate- Move bulky data (fixtures, templates) to external files for parsing at runtime
- Mutable defaults and shadowed built-ins are forbidden
- All code must have clear type hints using modern style (
A | B,list[str],class Foo[A]:,type Bar = int, etc.), with ABC imports drawn from the correct stdlib module.- All path manipulation must be performed using pathlib for cross platform safety. Do not use string manipulation for extraction of path components or filename ele...
Files:
cuprum/unittests/test_pipeline.pycuprum/_testing.pycuprum/sh.pycuprum/_pipeline_internals.py
**/unittests/test_*.py
📄 CodeRabbit inference engine (.rules/python-00.md)
Colocate unit tests with code using an unittests subdirectory with test_ prefix (e.g., user_auth/unittests/test_models.py)
Files:
cuprum/unittests/test_pipeline.py
**/test_*.py
📄 CodeRabbit inference engine (.rules/python-00.md)
**/test_*.py: Use pytest idioms: prefer fixtures over setup/teardown methods, parametrize broadly, avoid unnecessary mocks
Group related tests using class with method names prefixed by test_
Write tests from a user's perspective; test public behavior rather than internals
Avoid mocking too much; prefer test doubles only for external services or non-deterministic behaviors
Files:
cuprum/unittests/test_pipeline.py
**/*test*.py
📄 CodeRabbit inference engine (.rules/python-exception-design-raising-handling-and-logging.md)
Use specific exception types and message constraints with
pytest.raises(SpecificError, match=r"pattern")in tests; avoid overly broad exception assertions (B017)
Files:
cuprum/unittests/test_pipeline.pycuprum/_testing.py
🧬 Code graph analysis (3)
cuprum/unittests/test_pipeline.py (5)
cuprum/context.py (1)
scoped(230-262)cuprum/_pipeline_internals.py (2)
_PipelineWaitResult(382-384)_wait_for_pipeline(460-504)cuprum/_streams.py (1)
_pump_stream(54-75)cuprum/sh.py (11)
Pipeline(361-406)PipelineResult(115-153)run_sync(330-357)run_sync(398-406)ok(109-111)ok(146-148)final(134-136)failure(139-143)stdout(151-153)run(211-328)run(383-396)cuprum/catalogue.py (1)
allowlist(67-69)
cuprum/sh.py (2)
cuprum/_pipeline_internals.py (4)
_merge_env(44-52)_run_before_hooks(30-41)_run_pipeline(136-186)_terminate_process(55-65)cuprum/_streams.py (2)
_consume_stream(26-51)_StreamConfig(16-23)
cuprum/_pipeline_internals.py (3)
cuprum/_streams.py (3)
_consume_stream(26-51)_pump_stream(54-75)_StreamConfig(16-23)cuprum/context.py (2)
current_context(176-178)check_allowed(68-79)cuprum/sh.py (4)
CommandResult(81-111)ExecutionContext(157-185)SafeCmd(189-357)stdout(151-153)
⏰ Context from checks skipped due to timeout of 120000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: Sourcery review
🔇 Additional comments (36)
cuprum/unittests/test_pipeline.py (12)
1-21: LGTM on imports and module structure.The import reorganization to use
cuprum._testingaddresses the prior review comment about brittle re-exports. The dedicated testing surface cleanly separates test-facing internals from the public API.
24-92: Well-designed test doubles.The stub classes properly simulate async subprocess behaviour with configurable failure scenarios and call tracking. The
_StubSpawnProcesscorrectly mimics SIGTERM exit code (-15) whenreturncodeis unset.
94-143: Composition tests correctly verify pipeline construction semantics.
176-198: Streaming test correctly validates inter-stage communication.
251-296: Pump stream tests thoroughly verify backpressure and error handling.
298-335: Spawn failure cleanup test correctly validates resource cleanup.
337-344: Validation test correctly enforces minimum stage constraint.
346-383: Stub correctly simulates signal-based process termination for fail-fast tests.
422-447: Assertion helpers improve test readability and reduce duplication.
449-458: Scenario dataclass cleanly encapsulates parametrized test configuration.
473-482: Clarify termination behaviour in scenario ID and docstring.The scenario ID says "middle-stage-failure-terminates-downstream" and the docstring states "Middle stage failures terminate all downstream stages", but the expected behaviour terminates both stage 0 (upstream) and stage 2 (downstream). Either update the naming/documentation to reflect that all other stages are terminated, or verify this is the intended fail-fast behaviour.
385-420: Use NumPy-format docstrings.Same as
_run_test_pipeline, use NumPy format for docstrings.Proposed fix
async def _exercise_wait_for_pipeline( exit_codes: tuple[int, int, int], ready_stages: frozenset[int], ) -> tuple[...]: """Execute _wait_for_pipeline with stub processes and custom exit scenarios. - Args: - exit_codes: Exit code for each of the three stages. - ready_stages: Set of stage indices that should be immediately ready. + Parameters + ---------- + exit_codes: + Exit code for each of the three stages. + ready_stages: + Set of stage indices that should be immediately ready. - Returns: - Tuple of (process0, process1, process2, wait_result). + Returns + ------- + tuple + Tuple of (process0, process1, process2, wait_result). """Likely an incorrect or invalid review comment.
cuprum/_testing.py (1)
1-52: Clean test-facing surface for internal helpers.This module properly addresses the prior review comment about brittle re-exports via
# noqa: F401. The_EXPORTSdictionary pattern for building__all__and subsequent deletion keeps the module namespace clean.cuprum/sh.py (2)
17-27: Clean import delegation to internal modules.The imports from
_pipeline_internalsand_streamsproperly centralise runtime logic while keeping the public API surface insh.py.
124-144: Fail-fast metadata correctly exposed viafailure_indexandfailureproperty.The
failureproperty safely handles theNonecase and correctly indexes intostages. The docstring clearly documents the semantics.cuprum/_pipeline_internals.py (21)
1-28: Module docstring and imports are appropriate.The TYPE_CHECKING guard correctly avoids runtime circular imports while enabling static analysis.
30-42: Hook coordination correctly validates allowlist and returns after hooks.
44-53: Environment merge correctly handles None case with lazyosimport.
55-91: Termination logic correctly handles race conditions and escalates to SIGKILL.The exception handling for
ProcessLookupErrorandOSErrorproperly handles cases where the process exits between check and signal. Thenoqa: UP041annotation is appropriate for maintaining explicit asyncio semantics.
93-134: Configuration dataclass cleanly encapsulates pipeline runtime options.
136-187: Pipeline orchestration correctly owns stream task lifecycle.The
_run_pipelinefunction properly awaits stream tasks on success (lines 159-162) and cancels/gathers them on error (lines 164-166), addressing the prior review comment about ownership. Thenoqa: PLR0914is justified for this coordination function.
189-218: Stage stream FD configuration correctly implements pipeline wiring semantics.
220-247: Capture task creation correctly handles per-stage stream requirements.
249-274: Spawn cleanup correctly terminates processes and cancels capture tasks.
276-327: Process spawning correctly implements cleanup on failure.
329-342: Pipe task creation correctly wires adjacent stage streams.
344-353: Stream task flattening correctly filters None tasks.
355-379: Pipe result handling correctly distinguishes expected from unexpected failures.
BrokenPipeErrorandConnectionResetErrorare correctly treated as expected when downstream processes terminate early (e.g.,head). Other exceptions are properly surfaced.
381-405: Wait state classes correctly encapsulate pipeline completion tracking.
407-424: Error cleanup correctly terminates processes and documents stream task ownership.
426-445: Completed task processing correctly triggers fail-fast termination on non-zero exit.
447-458: Pipeline wait finalization correctly avoids surfacing failures during exception handling.
460-505: Pipeline wait correctly coordinates task completion and fail-fast semantics.The
asyncio.waitwithFIRST_COMPLETEDenables reactive fail-fast behaviour. The fallback to-1forNoneexit codes (line 488) is a defensive measure that shouldn't trigger in normal execution paths.
507-519: Wait task-based termination correctly usesasyncio.shieldto preserve shared state.
521-554: Remaining stage termination correctly terminates all non-failed, still-running stages.The implementation terminates both upstream and downstream stages (skipping only the failed stage and already-completed stages). This explains the test scenario where a middle-stage failure terminates stages 0 and 2. Update the test scenario documentation to reflect this "terminate all remaining" behaviour rather than "terminate downstream".
556-564: After-hook invocation correctly iterates with strict zip validation.
Refactored docstrings in test_pipeline.py to use NumPy style sections for parameters and returns, improving readability and consistency. Co-authored-by: terragon-labs[bot] <terragon-labs[bot]@users.noreply.github.com>
leynos
changed the title
1 participant
Summary
cuprum._pipeline_internals.pyandcuprum/_streams.pyto improve maintainability.failure_index: int | Noneand a newfailureproperty to expose the failing stage._PipelineWaitResultand_PipelineWaitStateto map per-stage outcomes to the overall result._terminate_pipeline_remaining_stagesand_terminate_process_via_wait_taskfor graceful termination.Changes
failure_indexand a newfailureproperty._PipelineWaitResultand_PipelineWaitStateto coordinate waiting on processes and mapping results to stages. (Inherited from internal module; not defined here.)PipelineResult.failure_indexandPipelineResult.failure._terminate_pipeline_remaining_stagesand_terminate_process_via_wait_taskto perform graceful termination._MIN_PIPELINE_STAGES, internal pipeline run/config handling, and fail-fast termination helpers.result.failure,result.failure_index, and downstream termination signals.PipelineResult.PipelineResult.failureandfailure_index, and describe fail-fast policy.result.failure/result.failure_index.API Changes
failure_index: int | None– index of the stage that triggered fail-fast termination (or None if all stages succeeded).failure: CommandResult | None– the failing stage's result when applicable.Testing Plan
pytest -k pipeline.Migration / Compatibility
Generated by Terry
📎 Task: https://www.terragonlabs.com/task/229b58af-8c49-496f-b1a3-092aa8746018