Fix Command-Query Separation violations (#118)

[leynos]

Summary

This branch fixes three Command-Query Separation violations flagged by the refactoring audit — helpers that read as a command or a query but did the opposite, or both.

Closes #118.

• _run_before_hooks(cmd) was documented as "collect hooks" (a query) but its real job was the ctx.check_allowed(...) side effect. It is split into _enforce_allowlist(cmd) (a command) and _collect_hooks(ctx) (a pure query).
• _emit_exec_event appended scheduled tasks to a caller-supplied pending_tasks out-parameter; it now returns the scheduled tasks and the caller extends its own collection.
• _write_to_stream_writer was a "write" command whose StreamWriter | None return was load-bearing control flow; it now returns a semantic _WriteOutcome (OPEN/CLOSED) and leaves writer ownership and closure to the caller.

Review walkthrough

• Start with cuprum/_pipeline_internals.py for the _enforce_allowlist / _collect_hooks split, then cuprum/sh.py and cuprum/_process_lifecycle.py for the updated call sites (enforce, then collect).
• cuprum/_observability.py returns the scheduled tasks; cuprum/_pipeline_types.py extends its own list.
• cuprum/_streams.py for the _WriteOutcome enum and the revised _pump_stream loop that keeps draining after downstream close.
• Finish with cuprum/unittests/test_cqrs_helpers.py for the behavioural coverage.

Validation

• make check-fmt: pass
• make lint: pass
• make typecheck: pass
• make test: pass (626 passed, 45 skipped; Rust suite 4 passed)
• coderabbit review --agent: 0 findings

Notes

The wheel-build snapshot gains the new test module. _collect_hooks reads current_context() once per call site rather than once per command; behaviour is unchanged because the context is constant within each synchronous build step.

Summary by Sourcery

Enforce clearer command-query separation in pipeline execution helpers and stream handling, and update observability, tests, and docs accordingly.

Enhancements:

• Split allowlist enforcement from hook collection into separate command and query helpers and update all call sites, including testing hooks exposure.
• Change observability event emission to return scheduled async tasks instead of mutating a caller-supplied collection, and adjust pipeline tracking to consume the new API.
• Refine stream pumping to distinguish write outcomes via a semantic enum and keep draining upstream when downstream closes, while leaving writer lifecycle to callers.
• Tighten concurrency test lock instrumentation docstrings and adjust CrossHair-related timeouts for more reliable analysis runs.

Documentation:

• Reflow and clarify developer guide documentation around environment overlay resolution, benchmark configuration, and selector model-checking invariants.

Tests:

• Add focused unit tests covering the new command/query helpers, observability event emission contract, and stream write outcome handling, and update existing concurrency tests.

References

• Lody session: https://lody.ai/leynos/sessions/cdd4039a-b710-408d-bbc9-b55f8097fa0c

[coderabbitai]

Note

Reviews paused

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

Use the following commands to manage reviews:

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

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

Walkthrough

The PR separates allowlist enforcement from hook collection, preserves asynchronous observe-hook tasks, improves stream shutdown after downstream closure, and adds locked maturin build diagnostics plus CI dependency preparation.

Changes

CQRS execution and stream refactor

Layer / File(s)	Summary
Separate allowlist enforcement from hook collection cuprum/_pipeline_internals.py, cuprum/sh.py, cuprum/_testing.py, cuprum/unittests/test_*	Split allowlist checking from context hook retrieval and enforce commands before input encoding or hook dispatch.
Return and preserve observe-hook tasks cuprum/_observability.py, cuprum/_pipeline_types.py, cuprum/_pipeline_internals.py, cuprum/unittests/test_cqrs_*, cuprum/unittests/test_observe.py, docs/*guide.md	Return scheduled observe tasks, retain tasks when later hooks fail, and drain them during pipeline finalisation.
Report downstream closure and drain upstream data cuprum/_streams.py, cuprum/unittests/test_cqrs_helpers.py, cuprum/unittests/test_stream_pump_runtime_behaviour.py	Use _WriteOutcome, perform bounded upstream draining after downstream closure, and explicitly manage writer teardown and diagnostics.
Document and validate execution contracts cuprum/unittests/_cqrs_fixtures.py, cuprum/unittests/__snapshots__/*, cuprum/unittests/test_stage_observation_builder.py, docs/*guide.md	Add shared fixtures, update snapshot entries, document execution contracts, and strengthen related assertions.

Native wheel build and coverage preparation

Layer / File(s)	Summary
Enrich maturin build failures tests/helpers/maturin.py, cuprum/unittests/test_maturin_build.py	Add MaturinBuildError, invoke maturin with locked dependencies and captured output, and test successful and failed subprocess paths.
Prepare locked Cargo dependencies in CI .github/workflows/ci.yml, docs/developers-guide.md, rust/cuprum-rust/src/lib_tests.rs	Configure Cargo retries, Rust setup, caching, credential handling, locked dependency prefetching, toolchain-pinned fixture regeneration, and Rust test syntax cleanup.

Sequence Diagram(s)

sequenceDiagram
  participant SafeCmd
  participant StageObservation
  participant ObserveHooks
  participant PendingTasks
  participant PipelineFinalisation
  SafeCmd->>StageObservation: emit execution event
  StageObservation->>ObserveHooks: invoke observe hooks
  ObserveHooks-->>StageObservation: return scheduled tasks or emission error
  StageObservation->>PendingTasks: retain scheduled tasks
  PipelineFinalisation->>PendingTasks: await and clear tasks

Loading

sequenceDiagram
  participant UpstreamReader
  participant PumpStream
  participant DownstreamWriter
  UpstreamReader->>PumpStream: provide chunks
  PumpStream->>DownstreamWriter: write and drain chunks
  DownstreamWriter-->>PumpStream: report CLOSED on pipe failure
  PumpStream->>UpstreamReader: bounded drain to EOF or timeout
  PumpStream->>DownstreamWriter: close writer once

Loading

Possibly related PRs

• leynos/cuprum#155: Touches the same stage-observation construction paths while focusing on canonical environment and tag helpers.

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

Poem

Hooks split cleanly, tasks take flight,
Streams drain through closing night.
Locked wheels build, logs retain,
Pipelines finish without a strain.
🛠️✨

---

Important

Pre-merge checks failed

Please resolve all errors before merging. Addressing warnings is optional.

❌ Failed checks (1 inconclusive)

Check name	Status	Explanation	Resolution
Module-Level Documentation	❓ Inconclusive	placeholder	pending

✅ Passed checks (19 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title matches the CQRS refactor and includes the linked issue reference required by the rules.
Description check	✅ Passed	The description clearly describes the same refactor and the three affected helpers.
Linked Issues check	✅ Passed	The changes satisfy #118 by splitting allowlist/query logic, returning scheduled tasks, and returning a semantic stream-writer outcome with tests.
Out of Scope Changes check	✅ Passed	The extra docs, CI, test, and snapshot updates support the refactor and do not appear unrelated to the linked objectives.
Docstring Coverage	✅ Passed	Docstring coverage is 99.35% which is sufficient. The required threshold is 80.00%.
Testing (Overall)	✅ Passed	New tests exercise real command, pipeline, stream, and build paths, with property-based and integration cases that would fail on plausible regressions.
User-Facing Documentation	✅ Passed	Confirm the users' guide covers awaitable observe hooks and early-drain pipeline behaviour in the user-facing pipeline/runtime sections.
Developer Documentation	✅ Passed	PASS: docs/developers-guide.md now records the new helper contracts and runtime boundaries; no roadmap/execplan sync issues were found.
Testing (Unit And Behavioural)	✅ Passed	Unit tests cover allowlist, emission, and cleanup edge cases; behavioural tests exercise real run_sync/pipeline and stream-pump runtime boundaries.
Testing (Property / Proof)	✅ Passed	Hypothesis property tests cover generated hook orderings and chunk sequences, matching the new scheduling and drain invariants; no substantive proof obligation appears.
Testing (Compile-Time / Ui)	✅ Passed	PASS: Rust has trybuild compile_time_ui pass/fail UI fixtures, and the wheel snapshot normalises metadata and entries into semantic assertions.
Unit Architecture	✅ Passed	Allowlist enforcement, hook collection, task scheduling, and stream outcomes are now explicit; tests exercise the boundaries.
Domain Architecture	✅ Passed	Keep the CQRS/stream refactor in runtime-adjacent helpers; no domain-model leakage or infrastructure coupling enters the model.
Observability	✅ Passed	Structured logs were added at hook-scheduling, hook-failure, early-close, timeout and cleanup boundaries, with tests asserting contextual fields.
Security And Privacy	✅ Passed	No secrets, auth bypasses, over-broad permissions, or sensitive-data exposures were introduced; the new logs and tests avoid raw credentials and the allowlist gate stays at the execution boundary.
Performance And Resource Use	✅ Passed	Accept the patch: all new loops are linear or bounded by the 0.25s drain timeout, and task buffers stay proportional to scheduled hooks.
Concurrency And State	✅ Passed	Single-loop ownership is explicit in docs, and tests cover cancellation, concurrent emits, cleanup grouping, and bounded stream draining.
Architectural Complexity And Maintainability	✅ Passed	PASS: The new helpers are small, local seams that remove duplication; docs spell out their contracts, and the touched modules have no runtime import cycle.
Rust Compiler Lint Integrity	✅ Passed	No new Rust lint suppressions or clone-based ownership work were added; the only Rust edit removes an explicit -> () in a test closure.

📋 Issue Planner

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

View plan used: #118

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch issue-118-cqrs-fixes

---

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

[sourcery-ai]

Reviewer's Guide

Refactors three helper functions to enforce clear Command-Query Separation, introduces a semantic write outcome enum for stream pumping, reshapes observability task scheduling to avoid out-parameters, and adds focused unit tests and minor documentation/test comment updates.

Sequence diagram for allowlist enforcement and hook collection

sequenceDiagram
    participant Run as run
    participant Enforce as _enforce_allowlist
    participant Collect as _collect_hooks
    participant Ctx as current_context

    Run->>Enforce: _enforce_allowlist(self)
    Enforce->>Ctx: current_context().check_allowed(program)
    Ctx-->>Enforce: None
    Enforce-->>Run: None

    Run->>Ctx: current_context()
    Ctx-->>Run: CuprumContext
    Run->>Collect: _collect_hooks(CuprumContext)
    Collect-->>Run: _ExecutionHooks

    Run->>Run: _ExecutionTracking(execution_hooks, pending_tasks)

Loading

Sequence diagram for exec event emission without out-parameters

sequenceDiagram
    participant Emit as emit
    participant EmitExec as _emit_exec_event
    participant Hook as ExecHook
    participant Wait as _wait_for_exec_hook_tasks

    Emit->>EmitExec: _emit_exec_event(hooks, event)
    loop for each ExecHook
        EmitExec->>Hook: hook(event)
        Hook-->>EmitExec: result
        alt [result is awaitable]
            EmitExec->>EmitExec: asyncio.create_task(_await_awaitable(result))
        end
    end
    EmitExec-->>Emit: list[asyncio.Task]
    Emit->>Emit: pending_tasks.extend(list[asyncio.Task])

    Emit->>Wait: _wait_for_exec_hook_tasks(pending_tasks)
    Wait-->>Emit: None

Loading

File-Level Changes

Change	Details	Files
Split allowlist enforcement from hook collection to separate commands from queries in pipeline execution.	Replaces _run_before_hooks with _enforce_allowlist (command) and _collect_hooks (pure query). Updates pipeline and process lifecycle builders to enforce allowlists per command, then collect hooks once from the current context. Adjusts sh.run tracking to call allowlist enforcement explicitly and pass collected hooks, and exports the new helpers via the testing shim.	cuprum/_pipeline_internals.py cuprum/_process_lifecycle.py cuprum/sh.py cuprum/_testing.py
Change observability hook dispatch to return scheduled tasks instead of mutating a caller-supplied list.	Refactors _emit_exec_event to build and return a list of asyncio tasks for async hooks while running sync hooks inline. Updates pipeline execution types to extend their own pending_tasks list from the helper’s return value. Clarifies pending task handling in _wait_for_exec_hook_tasks docstring.	cuprum/_observability.py cuprum/_pipeline_types.py
Make stream pumping treat writer liveness as a semantic outcome while always draining the reader.	Introduces _WriteOutcome enum representing OPEN/CLOSED downstream states. Refactors _write_to_stream_writer to return _WriteOutcome and stop closing the writer on BrokenPipe/ConnectionReset. Updates _pump_stream to track downstream_open separately so it keeps draining input even after downstream closure and closes the writer once at the end.	cuprum/_streams.py
Add targeted CQRS unit tests and tighten test documentation and configuration.	Adds test_cqrs_helpers.py to cover _enforce_allowlist/_collect_hooks, _emit_exec_event, and _write_to_stream_writer behaviour. Documents RLock-like test helpers with docstrings for acquire/release and context-manager methods. Raises CrossHair per_condition_timeout to reduce flakiness in line-splitting contracts and adds a new snapshot artefact for the test suite.	cuprum/unittests/test_cqrs_helpers.py cuprum/unittests/test_tee_profile_worker_concurrency.py cuprum/unittests/test_line_splitting.py cuprum/unittests/__snapshots__/test_maturin_build.ambr
Reflow and clarify developer documentation for environment overlays and benchmarks.	Rewraps and slightly clarifies developers-guide environment overlay section while preserving semantics. Improves wording and line wrapping in the pipeline throughput benchmark and model-checking sections.	docs/developers-guide.md

Assessment against linked issues

Issue	Objective	Addressed	Explanation
#118	Refactor _run_before_hooks into separate command/query helpers (_enforce_allowlist(cmd) for allowlist enforcement and _collect_hooks(ctx) for hook collection), update documentation, and ensure allowlist enforcement happens exactly once and independently of hook collection at all call sites.	✅
#118	Change _emit_exec_event so that it returns the scheduled async tasks instead of mutating a caller-supplied pending_tasks list, and update all callers to use the return value.	✅
#118	Change _write_to_stream_writer so that it reports downstream closure via a semantic result (e.g. _WriteOutcome.OPEN/CLOSED) instead of returning `StreamWriter	None`, leaving ownership and closure of the writer to the caller, and update the stream pump logic accordingly with tests.	✅

Possibly linked issues

• Refactor: fix Command-Query Separation violations (_run_before_hooks, _emit_exec_event, _write_to_stream_writer) #118: PR implements the issue’s proposed refactors for _run_before_hooks, _emit_exec_event, and _write_to_stream_writer plus tests.

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[sourcery-ai]

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

Please try again later or upgrade to continue using Sourcery

[pandalump]

@coderabbitai review

[coderabbitai]

✅ Action performed

Review finished.

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

[pandalump]

@coderabbitai resume

[coderabbitai]

✅ Action performed

Reviews resumed.

[pandalump]

@coderabbitai resume

[coderabbitai]

✅ Action performed

Reviews resumed.

[pandalump]

@coderabbitai review

[coderabbitai]

✅ Action performed

Review finished.

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

[coderabbitai]

Actionable comments posted: 1

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@cuprum/_pipeline_internals.py`:
- Around line 256-276: Update _drain_tasks_during_cleanup to use a single-line
docstring summarizing its cleanup behavior. Expand the existing # noqa: BLE001
on the BaseException handler with an inline rationale explaining that all task
failures, including non-Exception BaseException values, must be aggregated
without masking active_error.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: 346046b2-f048-42e3-99ff-45aff6de2c1a

📥 Commits

Reviewing files that changed from the base of the PR and between d9e3ed0 and 51c751e.

📒 Files selected for processing (2)

• cuprum/_pipeline_internals.py
• cuprum/unittests/test_cqrs_hook_behaviour.py

🔗 Linked repositories identified

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

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

[codescene-access]

No quality gates enabled for this code.