Introduce streaming Pipeline API with Pipeline and PipelineResult - #14
Conversation
![terragon-labs[bot]](https://github.githubassets.com/images/gravatars/gravatar-user-420.png){kind=avatar}
- Introduce Pipeline and PipelineResult dataclasses for representing and capturing results of piped SafeCmd stages - Implement | operator overloads to compose SafeCmd pipelines - Support asynchronous and synchronous execution with streaming, backpressure, and capture of final stage output - Add extensive unit and behavioural tests - Update documentation and public API exports - Mark pipeline execution feature as complete in roadmap Co-authored-by: terragon-labs[bot] <terragon-labs[bot]@users.noreply.github.com>
|
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. WalkthroughAdd Pipeline and PipelineResult to compose SafeCmd stages with stdin/stdout streaming, async and sync execution paths, backpressure-aware stream pumps, minimum-stage validation, tests (unit and behavioural), and documentation updates exposing the new API. (≤50 words) Changes
Sequence Diagram(s)sequenceDiagram
actor Caller
participant Pipeline
participant Orchestrator as _run_pipeline
participant ProcMgr as Process Manager
participant StreamMgr as Stream Handlers
participant Stage1 as Stage 1
participant Stage2 as Stage 2
Caller->>Pipeline: run(capture?, echo?)
Pipeline->>Orchestrator: _run_pipeline(stages, config)
Orchestrator->>ProcMgr: _prepare_pipeline_config()
ProcMgr-->>Orchestrator: config
Orchestrator->>ProcMgr: _spawn_pipeline_processes()
ProcMgr->>Stage1: spawn
ProcMgr->>Stage2: spawn
ProcMgr-->>Orchestrator: processes
Orchestrator->>StreamMgr: _create_pipe_tasks()
StreamMgr->>StreamMgr: _pump_stream(Stage1.stdout -> Stage2.stdin)
StreamMgr->>StreamMgr: apply backpressure (drain)
par Concurrent
Stage1->>StreamMgr: emit stdout chunk
StreamMgr->>Stage2: write chunk to stdin
Stage2->>StreamMgr: produce stdout chunk
end
Orchestrator->>ProcMgr: _wait_for_pipeline()
ProcMgr-->>Orchestrator: exit codes, pids
Orchestrator->>Orchestrator: _collect_pipeline_streams()
Orchestrator->>Orchestrator: _build_pipeline_stage_results()
Orchestrator->>Pipeline: return PipelineResult(stages, final_stdout)
Pipeline-->>Caller: PipelineResult
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60–90 minutes
Poem
Pre-merge checks and finishing touches❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
Comment |
Reviewer's GuideImplements a new streaming Pipeline execution API that composes SafeCmd stages via the | operator, executes them with asyncio-based backpressure, and returns a structured PipelineResult exposing per-stage CommandResult metadata and the final stage’s captured stdout, with full wiring into hooks, public exports, tests, and documentation. Sequence diagram for async Pipeline.run executionsequenceDiagram
actor User
participant SC as SafeCmd
participant PL as Pipeline
participant RP as _run_pipeline
participant SPC as _spawn_pipeline_processes
participant CPT as _create_pipe_tasks
participant WFP as _wait_for_pipeline
participant CPS as _collect_pipeline_streams
participant BSR as _build_pipeline_stage_results
participant RAH as _run_pipeline_after_hooks
User->>SC: create SafeCmd instances
User->>PL: compose via __or__(SafeCmd, SafeCmd)
PL-->>User: Pipeline(parts)
User->>PL: run(capture, echo, context)
PL->>RP: _run_pipeline(parts, capture, echo, context)
RP->>RP: _prepare_pipeline_config(capture, echo, context)
RP->>RP: _run_before_hooks for each SafeCmd
RP->>SPC: _spawn_pipeline_processes(parts, config)
SPC-->>RP: processes, stderr_tasks, stdout_task
RP->>CPT: _create_pipe_tasks(processes)
CPT-->>RP: pipe_tasks
RP->>WFP: _wait_for_pipeline(processes, pipe_tasks, stream_tasks, cancel_grace)
WFP-->>RP: exit_codes
RP->>CPS: _collect_pipeline_streams(stderr_tasks, stdout_task)
CPS-->>RP: stderr_by_stage, final_stdout
RP->>BSR: _build_pipeline_stage_results(parts, _PipelineCompleted)
BSR-->>RP: stage_results
RP->>RAH: _run_pipeline_after_hooks(parts, after_hooks_by_stage, stage_results)
RAH-->>RP: hooks completed
RP-->>PL: PipelineResult(stages)
PL-->>User: PipelineResult
User->>PL: access result.stdout and result.stages
PL-->>User: final stdout and per-stage metadata
Class diagram for SafeCmd, Pipeline, and PipelineResultclassDiagram
class SafeCmd {
+Program program
+tuple argv
+tuple argv_with_program()
+__or__(other SafeCmd or Pipeline) Pipeline
+run(capture bool, echo bool, context ExecutionContext) CommandResult
+run_sync(capture bool, echo bool, context ExecutionContext) CommandResult
}
class Pipeline {
+tuple~SafeCmd~ parts
+__post_init__()
+__or__(other SafeCmd or Pipeline) Pipeline
+run(capture bool, echo bool, context ExecutionContext) PipelineResult
+run_sync(capture bool, echo bool, context ExecutionContext) PipelineResult
}
class PipelineResult {
+tuple~CommandResult~ stages
+final CommandResult
+ok bool
+stdout str or None
}
class ExecutionContext {
+float cancel_grace
+str encoding
+str errors
+IO stdout_sink
+IO stderr_sink
+dict env
+Path cwd
}
class CommandResult {
+Program program
+tuple argv
+int exit_code
+int pid
+str or None stdout
+str or None stderr
+ok bool
}
class _PipelineRunConfig {
+ExecutionContext ctx
+bool capture
+bool echo
+IO stdout_sink
+IO stderr_sink
+capture_or_echo bool
+stream_config _StreamConfig
}
class _PipelineCompleted {
+tuple~Process~ processes
+tuple~int~ exit_codes
+tuple~str or None~ stderr_by_stage
+str or None final_stdout
}
SafeCmd --> CommandResult
SafeCmd --> ExecutionContext
SafeCmd --> Pipeline
Pipeline --> SafeCmd
Pipeline --> PipelineResult
PipelineResult --> CommandResult
_PipelineRunConfig --> ExecutionContext
_PipelineRunConfig --> _StreamConfig
_PipelineCompleted --> CommandResult
_PipelineCompleted --> Process
class AfterHook {
+__call__(cmd SafeCmd, result CommandResult) None
}
SafeCmd --> AfterHook
Pipeline --> AfterHook
_PipelineCompleted --> PipelineResult
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![codescene-delta-analysis[bot]](https://avatars.githubusercontent.com/in/53921?s=60&v=4){kind=small}
Add streaming pipeline execution and per-stage results by introducing
|
Add streaming
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey there - I've reviewed your changes and they look great!
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location> `cuprum/unittests/test_pipeline.py:39-62` </location>
<code_context>
+ assert pipeline.parts == (first, second, third)
+
+
+def test_pipeline_run_streams_stdout_between_stages() -> None:
+ """Pipeline.run_sync streams stdout into the next stage stdin."""
+ catalogue, python_program = python_catalogue()
+ python = sh.make(python_program, catalogue=catalogue)
+ echo = sh.make(ECHO)
+
+ pipeline = echo("-n", "hello") | python(
+ "-c",
+ "import sys; sys.stdout.write(sys.stdin.read().upper())",
+ )
+
+ with scoped(allowlist=frozenset([ECHO, python_program])):
+ result = pipeline.run_sync()
+
+ assert isinstance(result, PipelineResult)
+ assert result.stdout == "HELLO"
+ assert len(result.stages) == 2
+ assert result.stages[0].stdout is None
+ assert result.stages[0].exit_code == 0
+ assert result.stages[1].exit_code == 0
+ assert result.stages[0].pid > 0
+ assert result.stages[1].pid > 0
+
+
</code_context>
<issue_to_address>
**suggestion (testing):** Add coverage for PipelineResult.ok/final behaviour and failure cases
This only verifies the success path and some per-stage fields. Please add at least one test where a stage exits non‑zero (e.g. a Python command that calls `sys.exit(1)`) and assert that:
- `PipelineResult.ok` is `False` when any stage fails
- `PipelineResult.final` is the last stage’s `CommandResult` and its `exit_code` matches the failure
That will lock in the error semantics of `PipelineResult`, not just the happy path.
```suggestion
def test_pipeline_run_streams_stdout_between_stages() -> None:
"""Pipeline.run_sync streams stdout into the next stage stdin."""
catalogue, python_program = python_catalogue()
python = sh.make(python_program, catalogue=catalogue)
echo = sh.make(ECHO)
pipeline = echo("-n", "hello") | python(
"-c",
"import sys; sys.stdout.write(sys.stdin.read().upper())",
)
with scoped(allowlist=frozenset([ECHO, python_program])):
result = pipeline.run_sync()
assert isinstance(result, PipelineResult)
# Pipeline-level expectations
assert result.ok is True
assert result.stdout == "HELLO"
assert len(result.stages) == 2
assert result.final is result.stages[-1]
assert result.final.exit_code == 0
# Stage-level expectations
assert result.stages[0].stdout is None
assert result.stages[0].exit_code == 0
assert result.stages[1].exit_code == 0
assert result.stages[0].pid > 0
assert result.stages[1].pid > 0
def test_pipeline_run_sync_failure_sets_ok_false_and_final_to_failed_stage() -> None:
"""Pipeline.run_sync marks failure when a stage exits non-zero and exposes it via final."""
catalogue, python_program = python_catalogue()
python = sh.make(python_program, catalogue=catalogue)
# Single-stage pipeline that exits with status 1
failing = python(
"-c",
"import sys; sys.exit(1)",
)
pipeline = failing
with scoped(allowlist=frozenset([python_program])):
result = pipeline.run_sync()
assert isinstance(result, PipelineResult)
assert len(result.stages) >= 1
# PipelineResult semantics on failure
assert result.ok is False
assert result.final is result.stages[-1]
assert result.final.exit_code == 1
# Sanity check: at least one stage failed
assert any(stage.exit_code != 0 for stage in result.stages)
```
</issue_to_address>
### Comment 2
<location> `cuprum/unittests/test_pipeline.py:15-24` </location>
<code_context>
+from tests.helpers.catalogue import python_catalogue
+
+
+def test_or_operator_composes_pipeline() -> None:
+ """The | operator composes SafeCmd stages into a Pipeline."""
+ echo = sh.make(ECHO)
+ first = echo("-n", "hello")
+ second = echo("-n", "world")
+
+ pipeline = first | second
+
+ assert isinstance(pipeline, Pipeline)
+ assert pipeline.parts == (first, second)
+
+
</code_context>
<issue_to_address>
**suggestion (testing):** Add explicit tests for SafeCmd | Pipeline and Pipeline | Pipeline composition
This currently exercises `SafeCmd | SafeCmd` only. Since the implementation also supports `SafeCmd | Pipeline` and `Pipeline | Pipeline`, please add targeted tests for those cases, e.g.:
- `pipeline = first | (second | third)` → parts `(first, second, third)`
- `(first | second) | (third | fourth)` → parts `(first, second, third, fourth)`
This will better cover the more complex composition paths.
```suggestion
def test_or_operator_composes_pipeline() -> None:
"""The | operator compposes SafeCmd stages into a Pipeline."""
echo = sh.make(ECHO)
first = echo("-n", "hello")
second = echo("-n", "world")
pipeline = first | second
assert isinstance(pipeline, Pipeline)
assert pipeline.parts == (first, second)
def test_or_operator_composes_safe_cmd_and_pipeline() -> None:
"""The | operator composes a SafeCmd and a Pipeline into a single Pipeline."""
echo = sh.make(ECHO)
first = echo("-n", "hello")
second = echo("-n", "beautiful")
third = echo("-n", "world")
right_pipeline = second | third
pipeline = first | right_pipeline
assert isinstance(right_pipeline, Pipeline)
assert isinstance(pipeline, Pipeline)
assert pipeline.parts == (first, second, third)
def test_or_operator_composes_pipeline_and_pipeline() -> None:
"""The | operator composes two Pipelines into a single Pipeline."""
echo = sh.make(ECHO)
first = echo("-n", "hello")
second = echo("-n", "beautiful")
third = echo("-n", "world")
fourth = echo("-n", "!")
left_pipeline = first | second
right_pipeline = third | fourth
pipeline = left_pipeline | right_pipeline
assert isinstance(left_pipeline, Pipeline)
assert isinstance(right_pipeline, Pipeline)
assert isinstance(pipeline, Pipeline)
assert pipeline.parts == (first, second, third, fourth)
```
</issue_to_address>
### Comment 3
<location> `cuprum/unittests/test_pipeline.py:63-72` </location>
<code_context>
+def test_pump_stream_drains_per_chunk() -> None:
</code_context>
<issue_to_address>
**suggestion (testing):** Add tests for _pump_stream behaviour when downstream stdin closes early
Given the explicit `BrokenPipeError` / `ConnectionResetError` handling in `_write_to_stream_writer` and `_close_stream_writer`, it’d be good to add a test that simulates the downstream pipe closing mid‑stream (e.g. a stub writer whose `drain` raises `BrokenPipeError` on the second call) and asserts that `_pump_stream` finishes without surfacing the exception and does not hang while draining remaining reader data. This would directly exercise the early‑termination path and its deadlock prevention behaviour.
Suggested implementation:
```python
def test_pump_stream_drains_per_chunk() -> None:
"""Streaming between stages awaits drain for backpressure."""
import asyncio
from cuprum.pipeline import _pump_stream
class StubReader:
def __init__(self, chunks: list[bytes]) -> None:
self._chunks = list(chunks)
self.read_calls = 0
async def read(self, n: int) -> bytes: # pragma: no cover - exercised via _pump_stream
self.read_calls += 1
if self._chunks:
return self._chunks.pop(0)
return b""
class StubWriter:
def __init__(self) -> None:
self.data = bytearray()
self.drain_calls = 0
self.write_calls = 0
self.closed = False
self.write_eof_calls = 0
def write(self, data: bytes) -> None:
self.write_calls += 1
self.data.extend(data)
async def drain(self) -> None: # pragma: no cover - exercised via _pump_stream
self.drain_calls += 1
def write_eof(self) -> None:
self.write_eof_calls += 1
def is_closing(self) -> bool:
return self.closed
def close(self) -> None:
self.closed = True
async def wait_closed(self) -> None: # pragma: no cover - exercised via _pump_stream
self.closed = True
async def _run() -> None:
reader = StubReader([b"hello", b"world"])
writer = StubWriter()
await _pump_stream(reader, writer, chunk_size=3)
# Ensure all data made it through and drain was used for backpressure
assert writer.data == b"helloworld"
assert writer.write_calls >= 2
assert writer.drain_calls >= writer.write_calls
asyncio.run(_run())
def test_pump_stream_downstream_stdin_closes_early() -> None:
"""_pump_stream should tolerate downstream stdin closing mid-stream.
Simulates a downstream StreamWriter whose drain starts raising BrokenPipeError
part-way through the stream. The pump must finish without surfacing the
exception and without hanging while the reader still has data.
"""
import asyncio
from cuprum.pipeline import _pump_stream
class StubReader:
def __init__(self, chunks: list[bytes]) -> None:
self._chunks = list(chunks)
self.read_calls = 0
async def read(self, n: int) -> bytes: # pragma: no cover - exercised via _pump_stream
self.read_calls += 1
if self._chunks:
return self._chunks.pop(0)
return b""
class EarlyCloseStubWriter:
def __init__(self) -> None:
self.data = bytearray()
self.drain_calls = 0
self.write_calls = 0
self.closed = False
self.write_eof_calls = 0
def write(self, data: bytes) -> None:
# We still accept writes up to the point where drain starts failing.
self.write_calls += 1
self.data.extend(data)
async def drain(self) -> None: # pragma: no cover - exercised via _pump_stream
self.drain_calls += 1
# Simulate downstream stdin being closed during the second drain
if self.drain_calls == 2:
raise BrokenPipeError()
def write_eof(self) -> None:
self.write_eof_calls += 1
def is_closing(self) -> bool:
return self.closed
def close(self) -> None:
self.closed = True
async def wait_closed(self) -> None: # pragma: no cover - exercised via _pump_stream
self.closed = True
async def _run() -> None:
# Multiple chunks so that we hit the early-termination path mid-stream.
reader = StubReader([b"aaaa", b"bbbb", b"cccc"])
writer = EarlyCloseStubWriter()
# _pump_stream is expected to swallow BrokenPipeError/ConnectionResetError
# raised by drain/close and complete cleanly without hanging.
await _pump_stream(reader, writer, chunk_size=2)
# We should have attempted at least two drains, with the second one failing.
assert writer.drain_calls >= 2
# The pump should complete even though not all chunks could be written.
assert reader.read_calls >= 1
# The writer should have been told to close or EOF at some point.
assert writer.write_eof_calls >= 0 # existence check; exact semantics are implementation-dependent
asyncio.run(_run())
```
The edits above assume:
1. `cuprum.pipeline._pump_stream` exists and has a signature compatible with `await _pump_stream(reader, writer, chunk_size=...)`.
2. Importing `BrokenPipeError` is not necessary as it is a built-in; however, if your codebase shadows it or requires explicit typing, you may want to import `BrokenPipeError` from `builtins` or `asyncio` as appropriate.
3. There are no conflicting local `StubReader`/`StubWriter` definitions earlier in the file. If such helpers already exist, you should:
- Reuse the shared helpers instead of redefining them inside the tests, or
- Move these inner classes to the shared helper definitions and adjust the tests accordingly.
4. If your test style prefers `pytest.mark.asyncio` over `asyncio.run`, you can convert these tests to async tests and mark them with `@pytest.mark.asyncio` instead of using `asyncio.run`, aligning with the rest of your test suite.
</issue_to_address>
### Comment 4
<location> `tests/behaviour/test_pipeline_execution.py:80-85` </location>
<code_context>
+ assert pipeline_result.stdout == "BEHAVIOUR"
+
+
+@then("the pipeline exposes per stage exit metadata")
+def then_pipeline_exposes_stage_metadata(pipeline_result: PipelineResult) -> None:
+ """Stage results include exit codes and process identifiers."""
+ assert len(pipeline_result.stages) == 2
+ assert all(stage.exit_code == 0 for stage in pipeline_result.stages)
+ assert all(stage.pid > 0 for stage in pipeline_result.stages)
</code_context>
<issue_to_address>
**suggestion (testing):** Consider adding behavioural coverage for non-zero exit codes in pipelines
Right now these BDDs only cover successful stages and basic metadata. Since the pipeline API is meant to improve observability, consider adding a scenario where one stage exits non‑zero and assert that:
- Individual stage `exit_code` values are exposed (e.g. first fails, second succeeds).
- `PipelineResult.ok` reflects the failure (in line with the suggested unit tests).
This will lock in the expected behaviour when part of the pipeline fails and guard against streaming/cleanup logic hiding errors.
Suggested implementation:
```python
@then("the pipeline exposes per stage exit metadata")
def then_pipeline_exposes_stage_metadata(pipeline_result: PipelineResult) -> None:
"""Stage results include exit codes and process identifiers."""
assert len(pipeline_result.stages) == 2
assert all(stage.exit_code == 0 for stage in pipeline_result.stages)
assert all(stage.pid > 0 for stage in pipeline_result.stages)
@then("the pipeline exposes per stage exit metadata when a stage fails")
def then_pipeline_exposes_stage_metadata_on_failure(
pipeline_result: PipelineResult,
) -> None:
"""Stage results expose mixed success/failure and pipeline ok reflects failure."""
assert len(pipeline_result.stages) == 2
first_stage, second_stage = pipeline_result.stages
# Individual stage exit codes are exposed (first fails, second succeeds)
assert first_stage.exit_code != 0
assert second_stage.exit_code == 0
# PIDs are still populated for all stages
assert first_stage.pid > 0
assert second_stage.pid > 0
# PipelineResult.ok reflects the failure of the pipeline
assert not pipeline_result.ok
```
To make this step runnable and meaningful you will also need to:
1. **Feature file**: Add a new scenario in the relevant `.feature` file, for example:
- A `Scenario: pipeline reports metadata when a stage fails` (or similar) that ends with the step text:
- `Then the pipeline exposes per stage exit metadata when a stage fails`
2. **Given/When wiring**: Ensure the `Given`/`When` steps in that scenario produce a `pipeline_result` where:
- The pipeline has exactly two stages.
- The first stage exits with a non‑zero exit code, the second with `0`.
- `pipeline_result.ok` is `False` when any stage fails (in line with your unit-test expectations).
3. **Fixture / pipeline construction**: If `pipeline_result` is produced by a fixture or helper, add a variant (or parameterisation) that runs or simulates a pipeline with a failing first stage and a succeeding second stage.
</issue_to_address>
### Comment 5
<location> `cuprum/sh.py:450` </location>
<code_context>
+ )
+
+
+async def _run_pipeline(
+ parts: tuple[SafeCmd, ...],
+ *,
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying the new pipeline orchestration code by inlining small helper structures/functions and centralizing composition logic so the control flow and dataflow are easier to follow in one place.
You can keep all current behavior but reduce orchestration complexity with a few targeted refactors.
---
### 1. Drop `_PipelineCompleted` and inline `_collect_pipeline_streams`
`_PipelineCompleted` is only a transient bundle for `_build_pipeline_stage_results`. You can pass the values directly and inline `_collect_pipeline_streams` in `_run_pipeline` to keep the dataflow in one place.
**Before (excerpt):**
```python
exit_codes = await _wait_for_pipeline(...)
stderr_by_stage, final_stdout = await _collect_pipeline_streams(
stderr_tasks,
stdout_task,
)
stage_results = _build_pipeline_stage_results(
parts,
_PipelineCompleted(
processes=tuple(processes),
exit_codes=tuple(exit_codes),
stderr_by_stage=tuple(stderr_by_stage),
final_stdout=final_stdout,
),
)
```
**After (keep `_wait_for_pipeline` as-is):**
```python
exit_codes = await _wait_for_pipeline(
processes,
pipe_tasks=pipe_tasks,
stream_tasks=_flatten_stream_tasks(stderr_tasks, stdout_task),
cancel_grace=config.ctx.cancel_grace,
)
stderr_by_stage = [
None if t is None else await t
for t in stderr_tasks
]
final_stdout = None if stdout_task is None else await stdout_task
stage_results = _build_pipeline_stage_results(
parts=parts,
processes=tuple(processes),
exit_codes=tuple(exit_codes),
stderr_by_stage=tuple(stderr_by_stage),
final_stdout=final_stdout,
)
```
Then simplify `_build_pipeline_stage_results` and delete `_PipelineCompleted` and `_collect_pipeline_streams`:
```python
def _build_pipeline_stage_results(
parts: tuple[SafeCmd, ...],
*,
processes: tuple[asyncio.subprocess.Process, ...],
exit_codes: tuple[int, ...],
stderr_by_stage: tuple[str | None, ...],
final_stdout: str | None,
) -> list[CommandResult]:
last_idx = len(parts) - 1
results: list[CommandResult] = []
for idx, cmd in enumerate(parts):
results.append(
CommandResult(
program=cmd.program,
argv=cmd.argv,
exit_code=exit_codes[idx],
pid=processes[idx].pid or -1,
stdout=final_stdout if idx == last_idx else None,
stderr=stderr_by_stage[idx],
)
)
return results
```
This keeps behavior identical while removing an extra indirection layer.
---
### 2. Inline `_flatten_stream_tasks` at the call-site
`_flatten_stream_tasks` is a thin list comprehension; inlining makes `_run_pipeline` easier to scan without losing clarity.
**Before:**
```python
exit_codes = await _wait_for_pipeline(
processes,
pipe_tasks=pipe_tasks,
stream_tasks=_flatten_stream_tasks(stderr_tasks, stdout_task),
cancel_grace=config.ctx.cancel_grace,
)
```
**After (delete `_flatten_stream_tasks`):**
```python
stream_tasks: list[asyncio.Task[str | None]] = [
t for t in stderr_tasks if t is not None
]
if stdout_task is not None:
stream_tasks.append(stdout_task)
exit_codes = await _wait_for_pipeline(
processes,
pipe_tasks=pipe_tasks,
stream_tasks=stream_tasks,
cancel_grace=config.ctx.cancel_grace,
)
```
---
### 3. Merge `_write_to_stream_writer` and `_close_stream_writer` into `_pump_stream`
You can keep the backpressure and “swallow broken pipe” logic but reduce three helpers to one, making the streaming behavior self-contained.
**Before:**
```python
async def _pump_stream(reader, writer):
if reader is None:
return
active_writer = writer
while True:
chunk = await reader.read(_READ_SIZE)
if not chunk:
break
active_writer = await _write_to_stream_writer(active_writer, chunk)
_close_stream_writer(active_writer)
```
**After (delete `_write_to_stream_writer` and `_close_stream_writer`):**
```python
async def _pump_stream(
reader: asyncio.StreamReader | None,
writer: asyncio.StreamWriter | None,
) -> None:
if reader is None:
return
active_writer = writer
try:
while True:
chunk = await reader.read(_READ_SIZE)
if not chunk:
break
if active_writer is None:
continue
try:
active_writer.write(chunk)
await active_writer.drain()
except (BrokenPipeError, ConnectionResetError):
active_writer = None
finally:
if active_writer is not None:
with contextlib.suppress(
AttributeError,
NotImplementedError,
BrokenPipeError,
ConnectionResetError,
):
active_writer.write_eof()
with contextlib.suppress(BrokenPipeError, ConnectionResetError):
active_writer.close()
```
Behavior is preserved, but the “read/write/drain/close” lifecycle is visible in one place.
---
### 4. Centralize pipeline composition semantics
To reduce cross‑type coupling between `SafeCmd.__or__` and `Pipeline.__or__`, you can make `Pipeline` the owner of composition semantics and have `SafeCmd.__or__` delegate without knowing about `Pipeline` internals.
**Before:**
```python
class SafeCmd:
def __or__(self, other: SafeCmd | Pipeline) -> Pipeline:
if isinstance(other, Pipeline):
return Pipeline((self, *other.parts))
return Pipeline((self, other))
@dc.dataclass(...)
class Pipeline:
parts: tuple[SafeCmd, ...]
def __or__(self, other: SafeCmd | Pipeline) -> Pipeline:
if isinstance(other, Pipeline):
return Pipeline((*self.parts, *other.parts))
return Pipeline((*self.parts, other))
```
**After (behavior unchanged, composition logic centralized):**
```python
@dc.dataclass(frozen=True, slots=True)
class Pipeline:
parts: tuple[SafeCmd, ...]
def __post_init__(self) -> None:
if len(self.parts) < _MIN_PIPELINE_STAGES:
raise ValueError("Pipeline must contain at least two stages")
def extend(self, *more: SafeCmd) -> Pipeline:
return Pipeline((*self.parts, *more))
@classmethod
def concat(cls, left: Pipeline, right: Pipeline) -> Pipeline:
return cls((*left.parts, *right.parts))
def __or__(self, other: SafeCmd | Pipeline) -> Pipeline:
if isinstance(other, Pipeline):
return Pipeline.concat(self, other)
return self.extend(other)
class SafeCmd:
def __or__(self, other: SafeCmd | Pipeline) -> Pipeline:
if isinstance(other, Pipeline):
return Pipeline.concat(Pipeline((self,)), other)
return Pipeline((self, other))
```
This keeps the “prepend vs append” behavior but consolidates the rules into `Pipeline`, which makes the `|` semantics easier to reason about.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
docs/roadmap.md (1)
47-51: Wrap the completed bullet to 80 columns.Keep the completion status, and wrap the bullet text to meet the documentation style constraints.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: ASSERTIVE
Plan: Pro
📒 Files selected for processing (9)
cuprum/__init__.py(2 hunks)cuprum/sh.py(7 hunks)cuprum/unittests/test_pipeline.py(1 hunks)cuprum/unittests/test_public_api.py(1 hunks)docs/cuprum-design.md(3 hunks)docs/roadmap.md(1 hunks)docs/users-guide.md(1 hunks)tests/behaviour/test_pipeline_execution.py(1 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/roadmap.mddocs/users-guide.mddocs/cuprum-design.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/roadmap.mddocs/users-guide.mddocs/cuprum-design.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/roadmap.mddocs/users-guide.mddocs/cuprum-design.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/roadmap.mddocs/users-guide.mddocs/cuprum-design.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/roadmap.mddocs/users-guide.mddocs/cuprum-design.md
docs/**/*.{md,mdx}
📄 CodeRabbit inference engine (docs/documentation-style-guide.md)
Follow markdownlint recommendations for Markdown formatting
Files:
docs/roadmap.mddocs/users-guide.mddocs/cuprum-design.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/__init__.pycuprum/unittests/test_public_api.pycuprum/unittests/test_pipeline.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/__init__.pycuprum/unittests/test_public_api.pycuprum/unittests/test_pipeline.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_public_api.pycuprum/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_public_api.pycuprum/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_public_api.pycuprum/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.
Files:
docs/users-guide.md
🧠 Learnings (1)
📚 Learning: 2025-12-13T01:01:34.740Z
Learnt from: CR
Repo: leynos/cuprum PR: 0
File: docs/roadmap.md:0-0
Timestamp: 2025-12-13T01:01:34.740Z
Learning: Implement `Pipeline` composition via the `|` operator with streaming between stages and backpressure handling; expose exit metadata per stage
Applied to files:
docs/roadmap.md
🧬 Code graph analysis (4)
cuprum/__init__.py (1)
cuprum/sh.py (5)
CommandResult(75-105)ExecutionContext(140-168)Pipeline(357-397)PipelineResult(109-136)SafeCmd(183-353)
cuprum/unittests/test_public_api.py (1)
cuprum/sh.py (2)
Pipeline(357-397)PipelineResult(109-136)
cuprum/sh.py (2)
tests/behaviour/test_pipeline_execution.py (1)
run(67-69)cuprum/unittests/test_pipeline.py (4)
write(74-76)drain(78-80)write_eof(82-83)close(85-86)
tests/behaviour/test_pipeline_execution.py (4)
cuprum/context.py (1)
scoped(230-262)tests/helpers/catalogue.py (1)
python_catalogue(17-26)cuprum/sh.py (8)
Pipeline(357-397)PipelineResult(109-136)make(643-659)run_sync(326-353)run_sync(389-397)run(207-324)run(374-387)stdout(134-136)cuprum/catalogue.py (1)
allowlist(67-69)
🪛 LanguageTool
docs/cuprum-design.md
[uncategorized] ~384-~384: Possible missing article found.
Context: ...simple: we do not attempt to encode full pipeline structure at the type level. ...
(AI_HYDRA_LEO_MISSING_THE)
[uncategorized] ~388-~388: Use a comma before ‘so’ if it connects two independent clauses (unless they are closely connected and short).
Context: ...t` Pipelines return a structured result so callers can inspect per-stage exit meta...
(COMMA_COMPOUND_SENTENCE_2)
🔍 Remote MCP
Let me try searching the web for cuprum project information:The search results don't contain information about the cuprum project specifically. Let me search more directly for the cuprum GitHub repository:The first result is a different Cuprum project (a Ruby command pattern library). Let me search for cuprum shell command execution more specifically:Let me try a different approach and check for general Python async/subprocess patterns for pipeline backpressure which the PR uses:Based on my research, I have gathered sufficient relevant context from industry standards and best practices for reviewing this PR on streaming pipeline execution with backpressure. Let me provide a comprehensive summary.
Summary of Relevant Context for PR Review
Key Technical Patterns and Best Practices Found
Async/Subprocess Streaming & Backpressure:
Python's asyncio documentation recommends not using write() on its own but to follow up with drain() to handle backpressure. This is critical to the PR's implementation, as the PR mentions using drain() semantics for backpressure handling.
To prevent memory inflation under bursts, writer.drain() should be awaited after writing chunks to cooperate with the transport's high-water marks. The PR's _pump_stream helper function should properly implement this pattern.
Subprocess Piping Architecture:
In asyncio, the Process.stdin attribute is a StreamWriter instance when PIPE is passed to stdin, and Process.stdout and Process.stderr are StreamReader instances when PIPE is passed to those arguments. The PR chains multiple subprocesses' stdin/stdout streams, which is the standard pattern.
Stream Reader Buffer Limits:
The limit argument in asyncio.create_subprocess_exec() and create_subprocess_shell() sets the buffer limit for StreamReader wrappers for stdout and stderr. The PR's implementation should consider whether explicit buffer limits are set to prevent unbounded memory growth.
Backpressure Best Practices:
Backpressure in asyncio should be enforced at three layers: bounded admission at the edge (queues, semaphores), flow control at transports/streams via drain(), and deadlines on work. The PR implements flow control via drain() but should verify no memory bloat occurs in intermediate buffers.
Key Points for PR Review Focus
-
Backpressure Implementation: Verify that
_pump_streamand_write_to_stream_writerproperly awaitdrain()after writes to prevent downstream pipe closure from causing unbounded buffering in upstream stages. -
Intermediate Stage Buffering: Confirm that intermediate pipeline stages (non-final stages) have their stdout set to None as documented, preventing accidental capture of large intermediate streams.
-
Stream Closure Handling: Verify proper closure semantics when a downstream pipe closes to avoid broken-pipe errors propagating unexpectedly.
-
Process Resource Management: Ensure all spawned subprocess processes are properly awaited and cleaned up to prevent zombie processes.
-
Minimum Pipeline Size Validation: The implementation enforces
_MIN_PIPELINE_STAGESviaPipeline.__post_init__, which should raise ValueError for pipelines with fewer than 2 stages—verify test coverage for this. -
Async/Sync Wrapper Correctness: The
run_syncwrapper usesasyncio.run(), which should properly isolate event loop context; verify it handles exception propagation correctly.
⏰ 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 (10)
cuprum/sh.py (1)
108-137: Ship the public Pipeline API as implemented.Keep
PipelineResult.stdoutand per-stagestdout=Nonesemantics; keep the minimum-stage invariant enforcement and re-exports aligned.Also applies to: 356-398, 201-206, 816-825
cuprum/unittests/test_public_api.py (1)
8-23: Assert new public exports.Keep the export checks for
PipelineandPipelineResultto lock the public surface.tests/features/pipeline_execution.feature (1)
1-14: Keep the feature coverage focused and aligned to steps.Maintain the two scenarios as written; they exercise sync and async execution paths cleanly.
cuprum/__init__.py (1)
45-52: Re-export Pipeline symbols via the top-level package.Keep
PipelineandPipelineResultin__all__alongside the existing execution surface.Also applies to: 58-93
cuprum/unittests/test_pipeline.py (4)
15-37: Lock pipeline composition order with unit tests.Keep the operator tests to prevent regressions in stage ordering.
39-61: Exercise real streaming through a transform stage.Keep the python-transform pipeline test; it validates stdout piping and per-stage metadata end-to-end.
63-107: Keep per-chunk drain assertions for backpressure.Retain the
write_calls == drain_callsinvariant to ensure each write participates in flow control.
109-115: Keep the minimum-stage invariant test.Retain the
ValueErrorassertion to lock thePipelineconstruction contract.tests/behaviour/test_pipeline_execution.py (1)
19-85: Keep behavioural coverage for sync and async pipelines.Maintain the scenarios and assertions; they validate the public contract (transformed stdout and per-stage exit metadata) through the feature file.
docs/cuprum-design.md (1)
386-401: Remove this comment — the rule is already documented.The
PipelineResultclass docstring already states the behaviour explicitly: "For stages whose stdout is streamed into the next stage,stdoutisNone. The final stage carries captured stdout when enabled." The implementation confirms this: intermediate stages always havestdout=None, and the final stage hasstdout=Nonewhencapture=False. No additional documentation is required.Likely an incorrect or invalid review comment.
- Introduce Pipeline.concat() class method to unify pipeline stage concatenation. - Simplify SafeCmd and Pipeline __or__ operator implementations to use Pipeline.concat. - Improve subprocess spawning in pipelines with enhanced error handling to terminate started processes on spawn failure. - Refactor pipeline output handling by building CommandResult per stage directly without intermediate dataclass. - Await stream closure properly to avoid resource leaks. - Update tests to cover pipeline composition and failure scenarios. - Enhance feature tests to validate pipeline metadata exposure on stage failures. 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.
There was a problem hiding this comment.
Actionable comments posted: 4
♻️ Duplicate comments (1)
docs/cuprum-design.md (1)
554-564: Restatecapturelinkage in “Design decisions” to prevent a split-brain contract.Rewrite Line 561-563 to explicitly bind capture to behaviour (e.g. “When
capture=True, only the final stage’s stdout is captured … Whencapture=False,result.stdoutisNone…”).- - Only the final stage's stdout is captured. Intermediate stage stdout is - streamed into the next stage and represented as `None` on its stage result. + - When `capture=True`, only the final stage's stdout is captured. Intermediate + stage stdout is streamed into the next stage and represented as `None` on its + stage result. + - When `capture=False`, no stage stdout is captured and `result.stdout` is + `None`.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: ASSERTIVE
Plan: Pro
📒 Files selected for processing (6)
cuprum/sh.py(7 hunks)cuprum/unittests/test_pipeline.py(1 hunks)docs/cuprum-design.md(3 hunks)docs/users-guide.md(1 hunks)tests/behaviour/test_pipeline_execution.py(1 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/users-guide.mddocs/cuprum-design.md
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.
Files:
docs/users-guide.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/users-guide.mddocs/cuprum-design.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/users-guide.mddocs/cuprum-design.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/users-guide.mddocs/cuprum-design.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/users-guide.mddocs/cuprum-design.md
docs/**/*.{md,mdx}
📄 CodeRabbit inference engine (docs/documentation-style-guide.md)
Follow markdownlint recommendations for Markdown formatting
Files:
docs/users-guide.mddocs/cuprum-design.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/sh.pycuprum/unittests/test_pipeline.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/sh.pycuprum/unittests/test_pipeline.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
🧬 Code graph analysis (1)
cuprum/sh.py (2)
tests/behaviour/test_pipeline_execution.py (1)
run(92-94)cuprum/unittests/test_pipeline.py (7)
wait(84-89)read(27-32)write(45-47)drain(49-55)write_eof(57-58)close(60-61)wait_closed(63-64)
🪛 LanguageTool
docs/cuprum-design.md
[uncategorized] ~384-~384: Possible missing article found.
Context: ...simple: we do not attempt to encode full pipeline structure at the type level. ...
(AI_HYDRA_LEO_MISSING_THE)
[typographical] ~405-~405: The word ‘When’ starts a question. Add a question mark (“?”) at the end of the sentence.
Context: ....stdoutcontains the final stage stdout. - Whencapture=False, PipelineResult...
(WRB_QUESTION_MARK)
[typographical] ~406-~406: The word ‘When’ starts a question. Add a question mark (“?”) at the end of the sentence.
Context: ...alse, PipelineResult.stdoutisNone. ### 6.2 cuprum.sh` – Safe Facade The...
(WRB_QUESTION_MARK)
🔍 Remote MCP Ref
Summary of findings:
-
No external repository/docs could be fetched because the Ref tool is not configured (error: "Ref is not correctly configured").
-
Given that external searches failed, I relied on the PR context you provided. From that context, key review-relevant facts are:
- New types: Pipeline and PipelineResult added and exported via cuprum.init.all.
- Pipeline enforces minimum 2 stages via post_init/_MIN_PIPELINE_STAGES; ValueError raised for smaller pipelines; unit tests cover this.
- SafeCmd.or and Pipeline.or implement composition returning Pipeline.
- Execution: Pipeline.run (async) and Pipeline.run_sync (sync via asyncio.run) added; only final stage stdout is captured; PipelineResult.stages holds per-stage CommandResult with intermediate stdout set to None.
- Streaming/backpressure: helpers _pump_stream, _write_to_stream_writer, and use of drain() are implemented; unit tests simulate backpressure with stub writers to validate chunked draining, EOF and closure behavior.
- Process management: _spawn_pipeline_processes and cleanup on failure are implemented and tested (tests ensure started stages are terminated on spawn failure).
- Public API change: cuprum.init re-exports Pipeline and PipelineResult; tests updated (test_public_api.py).
Sources used:
- PR context provided by you (AI summary, PR objectives, tests, docs). (external search tool unavailable; relied on provided PR context)
⏰ 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 (27)
docs/users-guide.md (3)
99-103: LGTM!The Pipeline execution section introduction is clear and accurately describes the streaming and backpressure semantics. Line wrapping adheres to the 80-column guideline.
105-108: LGTM!The
PipelineResultattributes are documented concisely and match the implementation incuprum/sh.py.
110-138: LGTM!The code example correctly demonstrates pipeline composition and execution with the updated
sys.executablepattern, addressing the previous review feedback.tests/features/pipeline_execution.feature (1)
1-18: LGTM!The feature file covers essential pipeline behaviour: streaming between stages, asynchronous execution, and failure metadata propagation. The scenarios align well with the step definitions in
tests/behaviour/test_pipeline_execution.py.tests/behaviour/test_pipeline_execution.py (3)
64-78: LGTM!The failing pipeline fixture correctly sets up a two-stage pipeline where the first stage exits non-zero. The allowlist is correctly scoped to only
python_program.
81-96: LGTM!Both sync and async execution steps correctly scope the allowlist using
scoped(). The async variant properly wraps the context manager inside the coroutine to ensureContextVarsemantics are preserved.
99-122: LGTM!The then-steps provide thorough assertions: transformed stdout, per-stage exit codes, PIDs, and failure propagation via
result.ok. Coverage aligns with the feature scenarios and past review feedback.cuprum/unittests/test_pipeline.py (8)
22-65: LGTM!The stub classes
_StubPumpReaderand_StubPumpWriterare well-designed for testing_pump_streambehaviour. They correctly simulate chunked reads, drain backpressure, and configurable failure scenarios. Thewait_closedtracking enables verification of proper cleanup.
67-90: LGTM!
_StubSpawnProcessprovides adequate simulation of subprocess lifecycle for testing_spawn_pipeline_processescleanup behaviour on spawn failure.
92-141: LGTM!The composition tests comprehensively cover
SafeCmd | SafeCmd,SafeCmd | Pipeline,Pipeline | Pipeline, and multi-stage appending. Assertions verify correct stage ordering inpipeline.parts.
143-165: LGTM!The streaming test validates end-to-end pipeline execution with stdout transformation, per-stage metadata, and intermediate stdout being
None. Good coverage of success path semantics.
167-187: LGTM!Failure semantics are correctly tested:
result.okisFalse,result.finalreferences the last stage, and individual stage exit codes are exposed. This addresses prior review feedback about failure coverage.
189-212: LGTM!The backpressure test verifies that
_pump_streamcallsdrain()per chunk and properly closes the writer with EOF signalling andwait_closed().
234-271: LGTM!The spawn failure cleanup test correctly verifies that already-started processes are terminated when a subsequent spawn fails. The monkeypatch approach is appropriate for isolating the subprocess creation behaviour.
273-279: LGTM!The validation test confirms that
PipelineraisesValueErrorwhen constructed with fewer than two stages, matching the_MIN_PIPELINE_STAGESconstraint.cuprum/sh.py (10)
108-137: LGTM!
PipelineResultprovides clean accessors forfinal,ok, andstdout. Theokproperty correctly aggregates success across all stages.
201-203: LGTM!Delegating to
Pipeline.concatcentralizes composition logic and handles bothSafeCmdandPipelineoperands uniformly.
354-401: LGTM!The
Pipelinedataclass with__post_init__validation,concatfactory method, andrun/run_syncentry points is well-structured. The composition logic inconcatcorrectly flattens nested pipelines.
403-443: LGTM!
_PipelineRunConfigand_prepare_pipeline_configprovide clean normalisation of runtime options, including sink defaults and computed properties.
486-563: LGTM!
_spawn_pipeline_processescorrectly handles cleanup on spawn failure: terminating already-started processes, cancelling stream tasks, and re-raising the original exception. This addresses prior review feedback about subprocess leaks.
591-629: LGTM!
_wait_for_pipelineproperly surfaces unexpected pipe task exceptions while suppressing expectedBrokenPipeError/ConnectionResetError. The finally block ensures pipe results are collected on both success and failure paths. This addresses prior review feedback.
716-738: LGTM!
_pump_streamcorrectly implements backpressure-aware streaming with graceful handling of downstream closure. The pattern of continuing to drain the reader after the writer fails prevents upstream deadlocks.
740-781: LGTM!
_write_to_stream_writerand_close_stream_writernow properly close the writer onBrokenPipeErrorand awaitwait_closed()when available, addressing prior review feedback about transport cleanup.
827-836: LGTM!
__all__correctly exportsPipelineandPipelineResultas part of the public API.
466-467: Clarify why stream tasks are intentionally not awaited in_wait_for_pipelinesuccess path.Stream tasks are deliberately excluded from the
_wait_for_pipelinesuccess path (line 602) and only awaited in the exception handler (line 612). The comment at line 620–621 explains pipe task handling but omits explanation for stream tasks. The design is correct—stream tasks are awaited immediately after_wait_for_pipelinereturns (lines 466–467)—but add a comment to_wait_for_pipelineclarifying that stream tasks are intentionally delegated to the caller on success, since process termination guarantees stream closure and task completion.docs/cuprum-design.md (2)
386-407: Keep this contract block as the single source of truth.Retain the two
capturebullets exactly as written and do not restate them elsewhere with different wording; use references back to this block if needed. (LanguageTool’s “question mark” hints here are false positives.)
484-487: Guardresult.stdoutcorrectly before use.Keep
assert text is not None(or equivalent) to avoid implyingstdoutis always present.
|
@coderabbitai Have the following now been resolved? |
|
@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 lines +591 to +628 async def _wait_for_pipeline(
processes: list[asyncio.subprocess.Process],
*,
pipe_tasks: list[asyncio.Task[None]],
stream_tasks: list[asyncio.Task[str | None]],
cancel_grace: float,
) -> list[int]:
"""Wait for pipeline completion, ensuring subprocess cleanup on cancellation."""
caught: BaseException | None = None
pipe_results: list[object] | None = None
try:
return list(await asyncio.gather(*(p.wait() for p in processes)))
except BaseException as exc:
caught = exc
await asyncio.gather(
*(_terminate_process(p, cancel_grace) for p in processes),
return_exceptions=True,
)
pipe_results = list(
await asyncio.gather(*pipe_tasks, return_exceptions=True),
)
await asyncio.gather(*stream_tasks, return_exceptions=True)
raise
finally:
if pipe_results is None:
pipe_results = list(
await asyncio.gather(*pipe_tasks, return_exceptions=True),
)
# Surface unexpected pipe failures. Broken pipes are handled inside the
# streaming helpers and should not cause pipeline execution to fail.
if caught is None:
for result in pipe_results:
if isinstance(result, Exception) and not isinstance(
result,
(BrokenPipeError, ConnectionResetError),
):
raise result❌ New issue: Complex Method |
|
@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 lines +486 to +562 async def _spawn_pipeline_processes(
parts: tuple[SafeCmd, ...],
config: _PipelineRunConfig,
) -> tuple[
list[asyncio.subprocess.Process],
list[asyncio.Task[str | None] | None],
asyncio.Task[str | None] | None,
]:
"""Start subprocesses for each stage and wire up capture tasks."""
processes: list[asyncio.subprocess.Process] = []
stderr_tasks: list[asyncio.Task[str | None] | None] = []
stdout_task: asyncio.Task[str | None] | None = None
last_idx = len(parts) - 1
try:
for idx, cmd in enumerate(parts):
stdin = asyncio.subprocess.DEVNULL if idx == 0 else asyncio.subprocess.PIPE
stdout = (
asyncio.subprocess.PIPE
if idx != last_idx or config.capture_or_echo
else asyncio.subprocess.DEVNULL
)
stderr = (
asyncio.subprocess.PIPE
if config.capture_or_echo
else asyncio.subprocess.DEVNULL
)
process = await asyncio.create_subprocess_exec(
*cmd.argv_with_program,
stdin=stdin,
stdout=stdout,
stderr=stderr,
env=_merge_env(config.ctx.env),
cwd=str(config.ctx.cwd) if config.ctx.cwd is not None else None,
)
processes.append(process)
stderr_tasks.append(
asyncio.create_task(
_consume_stream(
process.stderr,
dc.replace(config.stream_config, sink=config.stderr_sink),
),
)
if config.capture_or_echo
else None
)
if idx == last_idx and config.capture_or_echo:
stdout_task = asyncio.create_task(
_consume_stream(
process.stdout,
config.stream_config,
),
)
except BaseException:
# If any stage fails to spawn, terminate the already-started stages.
# Otherwise long-lived upstream processes can leak in the background.
await asyncio.gather(
*(_terminate_process(p, config.ctx.cancel_grace) for p in processes),
return_exceptions=True,
)
tasks: list[asyncio.Task[typ.Any]] = []
for task in stderr_tasks:
if task is None:
continue
task.cancel()
tasks.append(task)
if stdout_task is not None:
stdout_task.cancel()
tasks.append(stdout_task)
await asyncio.gather(*tasks, return_exceptions=True)
raise
return processes, stderr_tasks, stdout_task❌ New issue: Complex Method |
This comment was marked as resolved.
This comment was marked as resolved.
|
@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 lines +486 to +562 async def _spawn_pipeline_processes(
parts: tuple[SafeCmd, ...],
config: _PipelineRunConfig,
) -> tuple[
list[asyncio.subprocess.Process],
list[asyncio.Task[str | None] | None],
asyncio.Task[str | None] | None,
]:
"""Start subprocesses for each stage and wire up capture tasks."""
processes: list[asyncio.subprocess.Process] = []
stderr_tasks: list[asyncio.Task[str | None] | None] = []
stdout_task: asyncio.Task[str | None] | None = None
last_idx = len(parts) - 1
try:
for idx, cmd in enumerate(parts):
stdin = asyncio.subprocess.DEVNULL if idx == 0 else asyncio.subprocess.PIPE
stdout = (
asyncio.subprocess.PIPE
if idx != last_idx or config.capture_or_echo
else asyncio.subprocess.DEVNULL
)
stderr = (
asyncio.subprocess.PIPE
if config.capture_or_echo
else asyncio.subprocess.DEVNULL
)
process = await asyncio.create_subprocess_exec(
*cmd.argv_with_program,
stdin=stdin,
stdout=stdout,
stderr=stderr,
env=_merge_env(config.ctx.env),
cwd=str(config.ctx.cwd) if config.ctx.cwd is not None else None,
)
processes.append(process)
stderr_tasks.append(
asyncio.create_task(
_consume_stream(
process.stderr,
dc.replace(config.stream_config, sink=config.stderr_sink),
),
)
if config.capture_or_echo
else None
)
if idx == last_idx and config.capture_or_echo:
stdout_task = asyncio.create_task(
_consume_stream(
process.stdout,
config.stream_config,
),
)
except BaseException:
# If any stage fails to spawn, terminate the already-started stages.
# Otherwise long-lived upstream processes can leak in the background.
await asyncio.gather(
*(_terminate_process(p, config.ctx.cancel_grace) for p in processes),
return_exceptions=True,
)
tasks: list[asyncio.Task[typ.Any]] = []
for task in stderr_tasks:
if task is None:
continue
task.cancel()
tasks.append(task)
if stdout_task is not None:
stdout_task.cancel()
tasks.append(stdout_task)
await asyncio.gather(*tasks, return_exceptions=True)
raise
return processes, stderr_tasks, stdout_task❌ New issue: Bumpy Road Ahead |
|
@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix: tests/behaviour/test_pipeline_execution.py Comment on lines +50 to +61 def given_simple_pipeline() -> _ScenarioPipeline:
"""Create a two stage pipeline that uppercases its input."""
catalogue, python_program = python_catalogue()
python = sh.make(python_program, catalogue=catalogue)
echo = sh.make(ECHO)
pipeline = echo("-n", "behaviour") | python(
"-c",
"import sys; sys.stdout.write(sys.stdin.read().upper())",
)
allowlist = frozenset([ECHO, python_program])
return _ScenarioPipeline(pipeline=pipeline, allowlist=allowlist)❌ New issue: Code Duplication |
This comment was marked as resolved.
This comment was marked as resolved.
This comment was marked as resolved.
This comment was marked as resolved.
This comment was marked as resolved.
This comment was marked as resolved.
This comment was marked as resolved.
This comment was marked as resolved.
- Introduce _collect_pipe_results to gather pipe task results with exceptions - Add _surface_unexpected_pipe_failures to raise non-expected pipe errors - Extract _cleanup_pipeline_on_error to clean up and await tasks on errors - Use the above helpers in _wait_for_pipeline to improve clarity and error management - Preserve prior behavior ignoring BrokenPipeError and ConnectionResetError while surfacing other exceptions 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.
leynos
changed the title
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
cuprum/sh.py (1)
445-483: Pipeline orchestration is clear, but consider inlining _flatten_stream_tasks.The execution flow correctly spawns processes, waits for completion, and builds results. However, _flatten_stream_tasks at line 462 is a thin list comprehension that past reviews suggested inlining for clarity.
Apply this diff to inline the helper:
+ stream_tasks: list[asyncio.Task[str | None]] = [ + t for t in stderr_tasks if t is not None + ] + if stdout_task is not None: + stream_tasks.append(stdout_task) + exit_codes = await _wait_for_pipeline( processes, pipe_tasks=_create_pipe_tasks(processes), - stream_tasks=_flatten_stream_tasks(stderr_tasks, stdout_task), + stream_tasks=stream_tasks, cancel_grace=config.ctx.cancel_grace, )Then delete _flatten_stream_tasks (lines 580-588).
In cuprum/sh.py around lines 445-483, inline the call to _flatten_stream_tasks at line 462 by replacing it with a list comprehension that collects non-None tasks from stderr_tasks and appends stdout_task if not None, then delete the _flatten_stream_tasks function definition at lines 580-588 to reduce indirection as suggested in past reviews.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: ASSERTIVE
Plan: Pro
📒 Files selected for processing (1)
cuprum/sh.py(7 hunks)
🧰 Additional context used
📓 Path-based instructions (1)
**/*.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/sh.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/sh.py
🧬 Code graph analysis (1)
cuprum/sh.py (2)
tests/behaviour/test_pipeline_execution.py (1)
run(92-94)cuprum/unittests/test_pipeline.py (7)
wait(84-89)read(27-32)write(45-47)drain(49-55)write_eof(57-58)close(60-61)wait_closed(63-64)
🔍 Remote MCP Ref
Summary of additional review-relevant facts (concise, factual)
- Public API: Pipeline and PipelineResult are exported from cuprum via cuprum.init.all (ensures re-export). Tests updated to assert these exports.
- Types and validation: Pipeline is a dataclass enforcing a minimum of two stages (ValueError on <2 stages). SafeCmd.or and Pipeline.or return Pipeline to support chaining.
- Execution behavior:
- Pipeline.run is async and returns PipelineResult; Pipeline.run_sync wraps run via asyncio.run and returns PipelineResult.
- Only the final stage’s stdout is captured (PipelineResult.stdout); intermediate stages stream into downstream and have stdout == None.
- PipelineResult.stages is a tuple[CommandResult,...] with per-stage exit codes, pids, and metadata; .final references the final stage result.
- Streaming and backpressure:
- Implementation includes _pump_stream, _write_to_stream_writer, _close_stream_writer and uses asyncio drain() semantics to apply backpressure between stages.
- Tests simulate backpressure and downstream closure with stub readers/writers; tests also cover BrokenPipe-like behavior and ensure pumps complete and EOF/closure are handled.
- Process management and failure handling:
- _spawn_pipeline_processes spawns stage processes and wires stdio; on spawn failure started stages are terminated—unit tests verify cleanup behavior.
- After-hooks are invoked per-stage after results are produced.
- Tests and behavioural coverage:
- Unit tests added (cuprum/unittests/test_pipeline.py) covering composition, run/run_sync, streaming/backpressure, pump behavior, PID/exit metadata, and minimum-stage validation.
- Behavioural BDD tests and feature file added (tests/behaviour/test_pipeline_execution.py and tests/features/pipeline_execution.feature) covering sync/async runs and failure reporting.
- Docs: Design and user-guide docs updated to describe PipelineResult shape, run/run_sync signatures, stdout capture semantics, examples, and roadmap marked pipeline work completed.
Notes for reviewer / actionable checks
- Verify run/run_sync type-signatures and exported types align with public docs and init exports.
- Confirm PipelineResult semantics (stdout None for intermediate stages) are clear in API/docs and reflected in tests.
- Run full test suite (pytest) to validate streaming/backpressure, process cleanup on spawn failure, and behavioural scenarios.
Tooling attempt
- Attempted external documentation search via Ref (Ref_ref_search_documentation) but Ref is not configured (error). No external docs retrieved.
⏰ 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 (12)
cuprum/sh.py (12)
12-12: LGTM: Import and constant additions are appropriate.The contextlib import supports the suppress context manager in streaming helpers, and _MIN_PIPELINE_STAGES correctly enforces the two-stage minimum.
Also applies to: 39-39
108-137: LGTM: PipelineResult dataclass is well-structured.The type annotations, properties, and docstring are correct. The frozen and slots configuration follows best practices.
201-203: LGTM: Composition delegation is clean.SafeCmd.or correctly delegates to Pipeline.concat, centralising composition logic as recommended.
354-401: LGTM: Pipeline class structure is sound.The validation in post_init, composition via concat, and run/run_sync methods are correctly implemented. Type annotations and docstrings follow conventions.
403-424: LGTM: Configuration dataclass reduces parameter passing._PipelineRunConfig correctly bundles execution settings and provides convenient properties.
426-442: LGTM: Configuration preparation is consistent with SafeCmd.The normalisation of parameters and default sink selection mirror the single-command execution path.
565-577: LGTM: Pipe task creation is straightforward.The list comprehension correctly wires adjacent stages with _pump_stream tasks.
591-664: LGTM: Refactored wait logic correctly handles cleanup and pipe failures.The extraction of _collect_pipe_results, _surface_unexpected_pipe_failures, and _cleanup_pipeline_on_error addresses the cyclomatic complexity concern. The finally block correctly surfaces unexpected pipe failures when no exception was caught.
666-674: LGTM: After-hook invocation is correct.The use of zip with strict=True ensures all stages have matching hooks and results. LIFO execution matches the before-hook pattern.
751-816: Streaming helpers correctly handle backpressure and closure.The refactored helpers now close the writer on broken pipe (line 786) and await wait_closed (lines 806-815), addressing past review concerns. The separation into three functions is acceptable, though past reviews suggested merging into a single _pump_stream for self-contained lifecycle management.
Optionally, merge these three helpers into a single _pump_stream function as suggested in past reviews to make the read/write/drain/close lifecycle more self-contained. The current implementation is correct and well-tested.
865-866: LGTM: Public API exports are correct.Pipeline and PipelineResult are correctly added to all and re-exported from cuprum.init per external context.
1-871: Overall: Type safety, dataclass usage, and resource management follow guidelines.The implementation uses modern type annotations (PEP 604), frozen dataclasses with slots, context managers for suppression, and async resource cleanup. The refactored helpers reduce complexity and improve maintainability.
- Introduced _StageStreamConfig to encapsulate stream fds per pipeline stage. - Added _get_stage_stream_fds and _create_stage_capture_tasks helpers to simplify _spawn_pipeline_processes. - Added _cleanup_spawned_processes to manage process termination and task cancellation on spawn failure. - Refactored _spawn_pipeline_processes to use new helpers for clarity and resource safety. - Added _build_scenario_pipeline_from_commands to construct test pipelines declaratively. - Rewrote behavioural tests to use the new test pipeline builder for consistency and readability. 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: tests/behaviour/test_pipeline_execution.py Comment on lines +76 to +88 def given_simple_pipeline() -> _ScenarioPipeline:
"""Create a two stage pipeline that uppercases its input."""
catalogue, python_program = python_catalogue()
python = sh.make(python_program, catalogue=catalogue)
echo = sh.make(ECHO)
return _build_scenario_pipeline_from_commands(
[
(echo, ("-n", "behaviour")),
(python, ("-c", "import sys; sys.stdout.write(sys.stdin.read().upper())")),
],
allowlist=frozenset([ECHO, python_program]),
)❌ New issue: Code Duplication |
This comment was marked as resolved.
This comment was marked as resolved.
Co-authored-by: terragon-labs[bot] <terragon-labs[bot]@users.noreply.github.com>
![macroscopeapp[bot]](https://avatars.githubusercontent.com/in/900172?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
📜 Review details
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
📒 Files selected for processing (5)
cuprum/sh.py(7 hunks)cuprum/unittests/test_pipeline.py(1 hunks)docs/cuprum-design.md(3 hunks)docs/users-guide.md(1 hunks)tests/behaviour/test_pipeline_execution.py(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/users-guide.mddocs/cuprum-design.md
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.
Files:
docs/users-guide.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/users-guide.mddocs/cuprum-design.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/users-guide.mddocs/cuprum-design.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/users-guide.mddocs/cuprum-design.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/users-guide.mddocs/cuprum-design.md
docs/**/*.{md,mdx}
📄 CodeRabbit inference engine (docs/documentation-style-guide.md)
Follow markdownlint recommendations for Markdown formatting
Files:
docs/users-guide.mddocs/cuprum-design.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:
tests/behaviour/test_pipeline_execution.pycuprum/unittests/test_pipeline.pycuprum/sh.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:
tests/behaviour/test_pipeline_execution.pycuprum/unittests/test_pipeline.pycuprum/sh.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:
tests/behaviour/test_pipeline_execution.pycuprum/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:
tests/behaviour/test_pipeline_execution.pycuprum/unittests/test_pipeline.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
🧬 Code graph analysis (1)
tests/behaviour/test_pipeline_execution.py (3)
cuprum/context.py (1)
scoped(230-262)tests/helpers/catalogue.py (1)
python_catalogue(17-26)cuprum/sh.py (10)
Pipeline(355-400)PipelineResult(109-136)make(738-754)run_sync(324-351)run_sync(392-400)run(205-322)run(377-390)stdout(134-136)ok(103-105)ok(129-131)
🪛 LanguageTool
docs/users-guide.md
[uncategorized] ~144-~144: Possible missing comma found.
Context: ...Noneinresult.stages. - echo=True` echoes the final stage stdout and all stage st...
(AI_HYDRA_LEO_MISSING_COMMA)
docs/cuprum-design.md
[uncategorized] ~384-~384: Possible missing article found.
Context: ...simple: we do not attempt to encode full pipeline structure at the type level. ...
(AI_HYDRA_LEO_MISSING_THE)
[typographical] ~405-~405: The word ‘When’ starts a question. Add a question mark (“?”) at the end of the sentence.
Context: ....stdoutcontains the final stage stdout. - Whencapture=False, PipelineResult...
(WRB_QUESTION_MARK)
[typographical] ~406-~406: The word ‘When’ starts a question. Add a question mark (“?”) at the end of the sentence.
Context: ...alse, PipelineResult.stdoutisNone. ### 6.2 cuprum.sh` – Safe Facade The...
(WRB_QUESTION_MARK)
⏰ 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 (4)
docs/users-guide.md (1)
99-146: LGTM!The Pipeline execution documentation is clear, accurate, and well-structured. The example correctly uses
Program(str(Path(sys.executable))), the streaming semantics are properly explained, and the notes appropriately describe stdout capture and echo behaviour.cuprum/unittests/test_pipeline.py (1)
1-281: LGTM!The unit tests comprehensively cover Pipeline composition, execution semantics, streaming with backpressure, spawn failure cleanup, and validation constraints. The stub classes are well-designed for exercising the pipeline logic without real subprocesses.
cuprum/sh.py (1)
108-877: LGTM!The Pipeline implementation is well-structured and correct:
- PipelineResult properly aggregates per-stage metadata with convenient accessors.
- Pipeline class enforces the 2-stage minimum and provides clear composition semantics via
concat.- Runtime orchestration has been refactored into focused helpers, reducing cyclomatic complexity.
- Spawn failure cleanup properly terminates started processes and cancels tasks.
- Pipe task results are collected and non-benign exceptions are surfaced whilst suppressing expected BrokenPipeError/ConnectionResetError.
- Stream writer closing properly handles broken pipes, calls write_eof, and awaits wait_closed.
All previously flagged issues have been addressed.
docs/cuprum-design.md (1)
369-563: LGTM!The design documentation accurately describes the new PipelineResult type and its semantics:
- The type signature includes inline comments documenting the capture parameter behaviour.
- The PipelineResult section clearly describes the structure and contract for both capture=True and capture=False cases.
- The example code properly guards against None with an assertion.
- Design decisions explicitly document per-stage metadata exposure and streaming behaviour.
All previously flagged documentation issues have been addressed.
Added a comment to explain the use of default and custom catalogues for ECHO and Python programs in the test_pipeline_execution.py to improve test code readability. Co-authored-by: terragon-labs[bot] <terragon-labs[bot]@users.noreply.github.com>
leynos
changed the title
…e for tests Created a ProjectSettings instance that includes both ECHO and python_program and built a combined ProgramCatalogue from it. Updated command builder mappings to use this unified catalogue to improve test pipeline setup and reflect realistic program catalogue usage. Co-authored-by: terragon-labs[bot] <terragon-labs[bot]@users.noreply.github.com>
leynos
changed the title
1 participant
Summary
Changes
Core functionality
Public API
Tests
Documentation
Migration / compatibility
Rationale
📎 Task: https://www.terragonlabs.com/task/bbe4052f-aa89-4bb3-afcc-83a24c89fc12