Implement structured pipeline events and telemetry

[leynos]

Summary

Introduce structured pipeline events and telemetry hooks to Cuprum. This enables emitting per-command and per-pipeline lifecycle events, line-based stdout/stderr observations, timing data, and merged metadata through a new observe API. Observers may be sync or async and are awaited to avoid leaking background tasks. Public API includes ExecEvent, ExecHook, and observe().

Changes

• New structured event model
  • Added cuprum/events.py with ExecEvent, ExecHook, and ExecPhase (plan, start, stdout, stderr, exit).
  • ExecEvent carries program, argv, cwd, env, pid, timestamp, line, exit_code, duration_s, and tags.
• Telemetry wiring and helpers
  • Added cuprum/_observability.py to centralize event emission helpers and async observer handling.
• Pipeline internals and observability
  • Refactored cuprum/_pipeline_internals.py to support structured events:
    • Introduced _StageObservation and _EventDetails for per-stage event emission.
    • Emit plan, start, and exit events with tagging (pipeline_stage_index, pipeline_stages).
    • Merge tags from ExecutionContext and per-stage metadata.
    • Track timing via started_at and ended_at for accurate duration_s on exit.
    • Ensure observe hooks can be empty; emit only when hooks exist.
• Line-based stdout/stderr emission
  • Updated cuprum/_streams.py to support per-line emission via on_line callbacks and incremental decoding with proper line handling.
• Async observe hooks and safety
  • Observe hooks may be sync or async; scheduled as background tasks and awaited before returning results.
• Command/Stage observation integration
  • Extended cuprum/sh.py to wire observation into SafeCmd execution and pipeline stages, including plan/start/exit events and line emission.
• Context and API surface
  • Extended cuprum/context.py with observe_hooks (ExecHook) and helper methods to attach/detach observe hooks.
  • HookRegistration updated to support observe hooks via observe(), alongside before/after hooks.
  • Exported ExecEvent, ExecHook, and observe() through cuprum/init.py.
• Public API surface
  • init now exposes ExecEvent, ExecHook, and observe; users can register observe hooks to receive ExecEvent data.
• Documentation and design updates
  • Updated docs/cuprum-design.md with a new section 8.1.3 describing structured execution events, timing, tagging, and async hooks.
  • Added user guide snippet demonstrating observe() usage and expected event fields.
  • Roadmap updated to reflect completed step for structured events.

Tests

• New tests for structured events and observe hooks:
  • cuprum/unittests/test_observe.py: unit tests for observe registration/detachment and event emission
  • tests/behaviour/test_structured_events.py: behaviour tests around observe hook capturing stdout/stderr and timing
  • tests/features/structured_events.feature: BDD feature covering observe hook behavior
• Test coverage ensures stdout/stderr line emission, timing metadata, and tag propagation in both single commands and pipelines.

Migration and Usage

• To enable structured telemetry, register an observe hook:
  • from cuprum import sh, ExecEvent
  • def hook(ev: ExecEvent) -> None: ...
  • with scoped(allowlist=my_allowlist), sh.observe(hook):
• Events emitted: plan, start, stdout, stderr, exit. stdout/stderr are emitted line-by-line; exit includes exit_code and duration_s. Tags merge in: project, runtime tags from ExecutionContext, and per-event stage metadata.
• Observe hooks may be sync or async. Async hooks run in the background and are awaited before run_sync returns.

Example

• Demonstrates capturing stdout lines and ensuring exit timing metadata is present:
  • See tests and docs users-guide for a complete example integrating with observability stacks.

Notes

• This feature is opt-in via observe() hooks and does not affect existing behavior unless hooks are registered.
• Performance considerations: observe hooks are non-blocking; async hooks are awaited to avoid leaking tasks.
• Public API additions include ExecEvent, ExecHook, and observe() for telemetry integrations.

🌿 Generated by Terry

---

ℹ️ Tag @terragon-labs to ask questions and address PR feedback

📎 Task: https://www.terragonlabs.com/task/85dd05a3-edc9-45f4-b76f-b12e7372e98c

Summary by Sourcery

Add structured execution event hooks and telemetry to command and pipeline execution, including per-phase events, line-based output observation, and tag-rich metadata, exposed via a new observe() API.

New Features:

• Introduce ExecEvent and ExecHook types and a new observe() registration API for structured execution events.
• Emit structured plan, start, stdout, stderr, and exit events for both single commands and pipeline stages, including timing and tag metadata.
• Support per-line stdout/stderr observation with incremental decoding and delivery to observe hooks.

Enhancements:

• Extend ExecutionContext and CuprumContext to carry arbitrary tag metadata and observe hooks, and propagate them into emitted events.
• Refactor subprocess and pipeline orchestration to centralize observability concerns, including timing capture and safe awaiting of async observer hooks.

Documentation:

• Document structured execution events, observe hooks, and tagging in the user guide and design docs, including usage examples and design rationale.
• Update the roadmap to mark structured events and telemetry wiring as completed.

Tests:

• Add unit tests for observe hook registration, detachment, and event emission for commands and pipelines.
• Add behavioural and BDD tests to validate stdout/stderr line emission, timing metadata, and tag propagation in structured events.

[sourcery-ai]

Reviewer's Guide

Adds a structured execution event system (ExecEvent/ExecHook) and observe hooks to Cuprum, wiring them through command and pipeline execution, with per-line stdout/stderr emission, timing, tagging, async-safe observer handling, and updated context/docs/tests/public API surface.

Sequence diagram for observe hooks during single command execution

sequenceDiagram
    actor User
    participant Context as CuprumContext
    participant CtxAPI as context_module
    participant Sh as sh_module
    participant Cmd as SafeCmd
    participant Obs as _CommandObservation
    participant ObsHelpers as _observability_module
    participant Hook as ExecHook

    User->>CtxAPI: scoped(..., observe_hooks=(hook,))
    activate CtxAPI
    CtxAPI->>Context: narrow(observe_hooks)
    CtxAPI-->>User: _ScopedContext
    deactivate CtxAPI

    User->>Sh: observe(hook)
    activate Sh
    Sh->>CtxAPI: observe(hook)
    activate CtxAPI
    CtxAPI->>Context: with_observe_hook(hook)
    CtxAPI-->>Sh: HookRegistration
    deactivate CtxAPI
    Sh-->>User: HookRegistration
    deactivate Sh

    User->>Cmd: run(capture, echo, context)
    activate Cmd
    Cmd->>Context: check_allowed(program)
    Cmd->>CtxAPI: current_context()
    CtxAPI-->>Cmd: CuprumContext

    Cmd->>Cmd: _run_before_hooks()
    Cmd-->>Cmd: _ExecutionHooks(before, after, observe)

    Cmd->>Obs: create _CommandObservation
    Note over Cmd,Obs: Merge ExecutionContext.tags with default tags

    Cmd->>Obs: emit("plan", _EventDetails(pid=None))
    activate Obs
    Obs->>ObsHelpers: _emit_exec_event(hooks, ExecEvent(plan,...), pending_tasks)
    activate ObsHelpers
    loop for each ExecHook
        ObsHelpers->>Hook: hook(ExecEvent(plan,...))
        alt async hook
            Hook-->>ObsHelpers: awaitable
            ObsHelpers->>ObsHelpers: create_task(awaitable)
        else sync hook
            Hook-->>ObsHelpers: None
        end
    end
    deactivate ObsHelpers
    deactivate Obs

    Cmd->>Cmd: spawn subprocess
    Cmd->>Obs: emit("start", _EventDetails(pid))
    activate Obs
    Obs->>ObsHelpers: _emit_exec_event(..., ExecEvent(start,...))
    deactivate Obs

    par stdout stream
        Cmd->>Cmd: _consume_stream(..., on_line=stdout_handler)
        loop each stdout line
            Cmd->>Obs: emit("stdout", _EventDetails(pid, line))
            Obs->>ObsHelpers: _emit_exec_event(..., ExecEvent(stdout,...))
        end
    and stderr stream
        Cmd->>Cmd: _consume_stream(..., on_line=stderr_handler)
        loop each stderr line
            Cmd->>Obs: emit("stderr", _EventDetails(pid, line))
            Obs->>ObsHelpers: _emit_exec_event(..., ExecEvent(stderr,...))
        end
    end

    Cmd->>Cmd: wait_for_exit_code()
    Cmd-->>Cmd: exit_code, duration_s

    Cmd->>Obs: emit("exit", _EventDetails(pid, exit_code, duration_s))
    Obs->>ObsHelpers: _emit_exec_event(..., ExecEvent(exit,...))

    Cmd-->>User: CommandResult

    Cmd->>CtxAPI: _wait_for_exec_hook_tasks(pending_tasks)
    activate CtxAPI
    CtxAPI->>ObsHelpers: _wait_for_exec_hook_tasks(pending_tasks)
    ObsHelpers-->>CtxAPI: all tasks done
    deactivate CtxAPI

    deactivate Cmd

Loading

Sequence diagram for observe hooks during pipeline execution

sequenceDiagram
    participant User
    participant Pipeline as _run_pipeline
    participant Context as CuprumContext
    participant ObsStage as _StageObservation
    participant Spawn as _spawn_pipeline_processes
    participant Wait as _wait_for_pipeline
    participant ObsHelpers as _observability_module
    participant Hook as ExecHook

    User->>Pipeline: _run_pipeline(parts, capture, echo, context)
    activate Pipeline
    Pipeline->>Context: current_context()

    loop per stage
        Pipeline->>Pipeline: _run_before_hooks(cmd)
        Pipeline-->>Pipeline: _ExecutionHooks(before, after, observe)
    end

    Pipeline->>Pipeline: create _StageObservation for each cmd
    Note over Pipeline,ObsStage: Merge ExecutionContext.tags with project and pipeline tags

    loop for each _StageObservation
        Pipeline->>ObsStage: emit("plan", _EventDetails(pid=None))
        ObsStage->>ObsHelpers: _emit_exec_event(..., ExecEvent(plan,...))
        loop before_hooks
            Pipeline->>Hook: before_hook(cmd)
        end
    end

    Pipeline->>Spawn: _spawn_pipeline_processes(..., observations)
    activate Spawn
    loop per stage index
        Spawn->>Spawn: create_subprocess_exec(argv_with_program,...)
        Spawn-->>Spawn: process, pid
        Spawn->>ObsStage: emit("start", _EventDetails(pid))
        ObsStage->>ObsHelpers: _emit_exec_event(..., ExecEvent(start,...))
        Spawn->>Spawn: _create_stage_capture_tasks(process, is_last_stage, observation)
        note over Spawn,ObsStage: _consume_stream uses on_line callbacks
        par stderr lines
            loop stderr lines
                Spawn->>ObsStage: emit("stderr", _EventDetails(pid, line))
                ObsStage->>ObsHelpers: _emit_exec_event(..., ExecEvent(stderr,...))
            end
        and last stage stdout lines
            loop stdout lines
                Spawn->>ObsStage: emit("stdout", _EventDetails(pid, line))
                ObsStage->>ObsHelpers: _emit_exec_event(..., ExecEvent(stdout,...))
            end
        end
    end
    Spawn-->>Pipeline: processes, stderr_tasks, stdout_task, started_at
    deactivate Spawn

    Pipeline->>Wait: _wait_for_pipeline(processes, pipe_tasks, cancel_grace, started_at)
    activate Wait
    Wait->>Wait: _PipelineWaitState.from_processes(processes, started_at)
    loop each completed process
        Wait->>Wait: record exit_code and ended_at
    end
    Wait-->>Pipeline: _PipelineWaitResult(exit_codes, failure_index, started_at, ended_at)
    deactivate Wait

    loop per stage index
        Pipeline->>ObsStage: emit("exit", _EventDetails(pid, exit_code, duration_s))
        ObsStage->>ObsHelpers: _emit_exec_event(..., ExecEvent(exit,...))
    end

    Pipeline->>Pipeline: build CommandResult per stage
    Pipeline->>Pipeline: _run_pipeline_after_hooks(parts, hooks_by_stage, results)

    Pipeline->>ObsHelpers: _wait_for_exec_hook_tasks(pending_tasks)
    ObsHelpers-->>Pipeline: all observe hooks finished

    Pipeline-->>User: PipelineResult
    deactivate Pipeline

Loading

Class diagram for structured execution events and observability internals

classDiagram
    direction LR

    class ExecEvent {
        <<dataclass>>
        +ExecPhase phase
        +Program program
        +tuple~str~ argv
        +Path cwd
        +Mapping~str,str~ env
        +int pid
        +float timestamp
        +str line
        +int exit_code
        +float duration_s
        +Mapping~str,object~ tags
    }

    class ExecHook {
        <<callable>>
        +__call__(ExecEvent) Awaitable~None~ | None
    }

    class CuprumContext {
        <<dataclass>>
        +frozenset~Program~ allowlist
        +tuple~BeforeHook~ before_hooks
        +tuple~AfterHook~ after_hooks
        +tuple~ExecHook~ observe_hooks
        +bool is_allowed(program)
        +check_allowed(program)
        +narrow(allowlist,before_hooks,after_hooks,observe_hooks) CuprumContext
        +with_before_hook(hook) CuprumContext
        +with_after_hook(hook) CuprumContext
        +with_observe_hook(hook) CuprumContext
        +without_before_hook(hook) CuprumContext
        +without_after_hook(hook) CuprumContext
        +without_observe_hook(hook) CuprumContext
    }

    class HookRegistration {
        +HookRegistration(hook, hook_type)
        +detach()
        +__enter__() HookRegistration
        +__exit__(exc_type, exc, tb) bool
        -object _hook
        -str _hook_type
        -Token~CuprumContext~ _token
    }

    class ExecutionContext {
        <<dataclass>>
        +Mapping~Program,bool~ allowlist
        +Mapping~str,str~ env
        +Path cwd
        +IO~str~ stdout_sink
        +IO~str~ stderr_sink
        +str encoding
        +str errors
        +Mapping~str,object~ tags
        +float cancel_grace
    }

    class SafeCmd {
        +Program program
        +Project project
        +tuple~str~ argv
        +tuple~str~ argv_with_program
        +run(capture, echo, context) CommandResult
        +run_sync(capture, echo, context) CommandResult
    }

    class CommandResult {
        <<dataclass>>
        +Program program
        +tuple~str~ argv
        +int exit_code
        +int pid
        +str stdout
        +str stderr
    }

    class _ExecutionHooks {
        <<dataclass>>
        +tuple~BeforeHook~ before_hooks
        +tuple~AfterHook~ after_hooks
        +tuple~ExecHook~ observe_hooks
    }

    class _CommandObservation {
        <<dataclass>>
        +SafeCmd cmd
        +tuple~ExecHook~ observe_hooks
        +Path cwd
        +Mapping~str,str~ env_overlay
        +Mapping~str,object~ tags
        +list~Task~ pending_tasks
        +emit(phase, details) void
    }

    class _StageObservation {
        <<dataclass>>
        +SafeCmd cmd
        +_ExecutionHooks hooks
        +Mapping~str,object~ tags
        +Path cwd
        +Mapping~str,str~ env_overlay
        +list~Task~ pending_tasks
        +emit(phase, details) void
    }

    class _EventDetails {
        <<dataclass>>
        +int pid
        +str line
        +int exit_code
        +float duration_s
    }

    class _SubprocessExecution {
        <<dataclass>>
        +SafeCmd cmd
        +ExecutionContext ctx
        +bool capture
        +bool echo
        +_CommandObservation observation
    }

    class _PipelineRunConfig {
        <<dataclass>>
        +ExecutionContext ctx
        +bool capture
        +bool echo
        +bool capture_or_echo
        +_StreamConfig stream_config
        +IO~str~ stderr_sink
    }

    class _PipelineWaitState {
        <<dataclass>>
        +list~Task~ wait_tasks
        +dict~Task,int~ task_to_index
        +list~int~ exit_codes
        +list~float~ started_at
        +list~float~ ended_at
        +int failure_index
        +from_processes(processes, started_at) _PipelineWaitState
    }

    class _PipelineWaitResult {
        <<dataclass>>
        +list~int~ exit_codes
        +int failure_index
        +list~float~ started_at
        +list~float~ ended_at
    }

    class _StreamConfig {
        <<dataclass>>
        +bool capture_output
        +bool echo_output
        +IO~str~ sink
        +str encoding
        +str errors
    }

    class _observability_module {
        <<module>>
        +_freeze_str_mapping(mapping) Mapping~str,str~
        +_merge_tags(*tags) Mapping~str,object~
        +_emit_exec_event(hooks, event, pending_tasks) void
        +_wait_for_exec_hook_tasks(pending_tasks) void
    }

    class context_module {
        <<module>>
        +current_context() CuprumContext
        +get_context() CuprumContext
        +scoped(allowlist,before_hooks,after_hooks,observe_hooks) _ScopedContext
        +before(hook) HookRegistration
        +after(hook) HookRegistration
        +observe(hook) HookRegistration
    }

    class sh_module {
        <<module>>
        +observe(hook) HookRegistration
    }

    ExecHook --> ExecEvent : parameter
    CuprumContext o--> ExecHook : observe_hooks
    CuprumContext o--> BeforeHook
    CuprumContext o--> AfterHook

    HookRegistration --> CuprumContext : updates
    context_module --> HookRegistration : creates

    ExecutionContext --> _StreamConfig : configures
    ExecutionContext --> _CommandObservation : provides tags, env, cwd

    SafeCmd --> _ExecutionHooks : uses via _run_before_hooks
    SafeCmd --> _CommandObservation : constructs
    SafeCmd --> _SubprocessExecution : wraps

    _ExecutionHooks o--> BeforeHook
    _ExecutionHooks o--> AfterHook
    _ExecutionHooks o--> ExecHook

    _CommandObservation --> ExecHook : notifies
    _StageObservation --> ExecHook : notifies
    _CommandObservation --> _EventDetails : uses
    _StageObservation --> _EventDetails : uses

    _SubprocessExecution --> _CommandObservation

    _PipelineRunConfig --> ExecutionContext
    _PipelineRunConfig --> _StreamConfig

    _PipelineWaitState --> _PipelineWaitResult

    _observability_module --> ExecHook : calls
    _observability_module --> ExecEvent : emits
    _observability_module --> _CommandObservation : used by
    _observability_module --> _StageObservation : used by

    context_module --> CuprumContext
    context_module --> ExecHook

    sh_module --> ExecEvent
    sh_module --> ExecHook
    sh_module --> context_module

Loading

File-Level Changes

Change	Details	Files
Introduce structured ExecEvent/ExecHook model and observability helpers for emitting execution telemetry.	Added cuprum/events.py defining ExecPhase, ExecEvent dataclass (plan/start/stdout/stderr/exit phases) and ExecHook type for sync/async observers. Added cuprum/_observability.py with helpers to freeze env mappings, merge tag mappings, emit events to hooks while collecting async tasks, and await pending hook tasks safely.	cuprum/events.py cuprum/_observability.py
Wire observe hooks through single-command execution with per-phase events, line-based output emission, timing, and tag metadata.	Extended ExecutionContext with optional tags metadata to attach to structured events. Refactored command execution in cuprum/sh.py into helper dataclasses (_CommandObservation, _SubprocessExecution, _EventDetails) and coroutine helpers (_wait_for_exit_code, _spawn_subprocess, _run_subprocess_with_streams, _execute_subprocess) that emit plan/start/stdout/stderr/exit ExecEvents via _emit_exec_event. Hook collection now returns an _ExecutionHooks object with before/after/observe hooks; observe hooks are invoked only when present, and async observe hooks are scheduled into a pending_tasks list and awaited via _wait_for_exec_hook_tasks after execution or cancellation. Updated run() to construct observation/tag state (project/capture/echo plus context tags), emit a plan event before before_hooks, and ensure pending observe tasks are awaited on both success and cancellation paths. Updated stream consumption for commands to pass on_line callbacks into _consume_stream so stdout/stderr lines generate per-line ExecEvents including pid and tags.	cuprum/sh.py
Extend pipeline internals to emit structured events per stage with stage-aware tags, line events, and timing for each subprocess.	Introduced _ExecutionHooks, _StageObservation, and _EventDetails in pipeline internals to encapsulate hooks and observation metadata (project, capture/echo, pipeline_stage_index, pipeline_stages, context tags). Updated _run_before_hooks to return _ExecutionHooks including observe hooks, and refactored _run_pipeline to build _StageObservation instances, emit plan events and before_hooks, and wire them into _spawn_pipeline_processes. Changed _spawn_pipeline_processes to accept precomputed observations, emit start events per stage, track started_at timestamps, and return started_at list alongside processes and stream tasks; also added a guard builder _build_spawn_observations for cases without explicit observations. Modified _create_stage_capture_tasks to always use _consume_stream with optional on_line callbacks that emit stdout/stderr events tagged with the stage index, restricting stdout capture to the last stage while still emitting stderr events for all stages. Extended pipeline waiting logic (_PipelineWaitState, _PipelineWaitResult, _wait_for_pipeline, _process_completed_task) to track per-stage start/end times and durations used when emitting exit events in _run_pipeline. Adjusted _run_pipeline_after_hooks to work with _ExecutionHooks and ensured observe hook tasks are awaited after pipelines finish or are cancelled. Updated existing unit test helper _exercise_wait_for_pipeline to pass dummy started_at data to _wait_for_pipeline.	cuprum/_pipeline_internals.py cuprum/unittests/test_pipeline.py
Enhance stream consumption to support incremental decoding and per-line callbacks while preserving existing capture/echo semantics.	Refactored _consume_stream into a dispatcher that either consumes without line awareness (_consume_stream_without_lines) or with line callbacks (_consume_stream_with_lines). Implemented incremental decoding path that maintains a pending_text buffer, uses a codecs incremental decoder, and emits completed lines through on_line callbacks, including handling of final partial lines without trailing newline. Added helpers to emit complete lines and manage line endings (_emit_completed_lines, _ends_with_line_ending, _strip_line_ending) while still returning captured text identical to previous behaviour for callers.	cuprum/_streams.py
Extend context and hook registration API to support observe hooks and expose them via the top-level package and sh facade.	Extended CuprumContext to track observe_hooks, added with_observe_hook/without_observe_hook, and threaded observe_hooks through CuprumContext.narrow and scoped(). Generalized HookRegistration to handle before/after/observe hook types and added a new observe() registration function that returns a HookRegistration usable as a context manager or for manual detach. Exported ExecHook and observe from context all, and updated cuprum/init.py to re-export ExecEvent, ExecHook, and observe alongside existing sh and context symbols. Imported observe and ExecEvent into cuprum/sh.py, added observe to sh.all, and provided a public sh.observe() entry point for registering observe hooks in the current context.	cuprum/context.py cuprum/__init__.py cuprum/sh.py
Document structured events and mark roadmap progress for telemetry, plus add tests covering observe hooks and structured event behaviour.	Added a "Structured execution events" section to the user guide describing ExecEvent phases, sh.observe usage, ExecutionContext.tags, and the tag-merging behaviour with a runnable example. Extended the design document with section 8.1.3 detailing design decisions around phases, per-line emission, timing (time.time vs time.perf_counter), tag semantics, and async observer handling. Marked the roadmap item for structured events and sh.observe as completed. Introduced unit tests for observe registration/detachment and event content (stdout/stderr, timing, tags, pipeline stage metadata) in cuprum/unittests/test_observe.py. Added behaviour tests and a BDD feature (structured_events.feature plus test_structured_events.py) to validate end-to-end observe hook behaviour for stdout/stderr lines, timing, and tags.	docs/users-guide.md docs/cuprum-design.md docs/roadmap.md cuprum/unittests/test_observe.py tests/behaviour/test_structured_events.py tests/features/structured_events.feature

---

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.

[coderabbitai]

Note

Other AI code review bot(s) detected

CodeRabbit 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.

Walkthrough

This pull request introduces a structured execution events system to Cuprum. It adds a new ExecEvent type and observe hook mechanism that emit telemetry events (plan, start, stdout, stderr, exit phases) during command and pipeline execution. The implementation includes new internal modules for observability, process lifecycle management, pipeline wait logic, and stream handling, alongside restructuring of pipeline internals to support hook orchestration.

Changes

Cohort / File(s)	Change summary
Public API exports cuprum/__init__.py	Expose ExecEvent, ExecHook, and observe as public symbols via updated __all__.
Structured events model cuprum/events.py	Define ExecPhase (literal for plan/start/stdout/stderr/exit), ExecEvent (frozen dataclass with event metadata), and ExecHook (callable type for async observers).
Context observability integration cuprum/context.py	Add observe_hooks field to CuprumContext; introduce observe() registration function; extend narrow(), scoped(), and HookRegistration to manage observe hooks alongside existing before/after hooks.
Internal event emission cuprum/_observability.py	Provide helpers: _emit_exec_event() (invoke hooks with event), _freeze_str_mapping(), _merge_tags(), _wait_for_exec_hook_tasks() (await pending hook tasks).
Command execution observability cuprum/sh.py	Integrate event emission into SafeCmd.run(); add ExecutionContext.tags field; introduce internal observation, event emission, and async hook orchestration during command lifecycle.
Pipeline process lifecycle cuprum/_process_lifecycle.py	Implement subprocess creation, termination with grace periods, environment merging, and spawn observation; provide cleanup paths for errors and failures; support fail-fast termination of remaining stages.
Pipeline wait orchestration cuprum/_pipeline_wait.py	Introduce _PipelineWaitResult and _PipelineWaitState for tracking process completion; implement _wait_for_pipeline() with fail-fast semantics and pipe task finalisation.
Pipeline stream coordination cuprum/_pipeline_streams.py	Define _PipelineRunConfig for execution context; provide _prepare_pipeline_config() and _get_stage_stream_fds() for per-stage stream setup; implement capture/echo task orchestration and inter-stage pipe pumping.
Pipeline spawn wrapper cuprum/_pipeline_spawn.py	Re-export spawn helpers from cuprum._process_lifecycle to avoid import cycles.
Stream consumption with line emission cuprum/_streams.py	Extend _consume_stream() to support optional on_line callback; introduce line-aware streaming via _consume_stream_with_lines() and line-emission helpers (_emit_completed_lines(), _ends_with_line_ending(), _strip_line_ending()).
Pipeline internals restructuring cuprum/_pipeline_internals.py	Introduce _ExecutionHooks, _StageObservation, _EventDetails, _PipelineStageResultInputs dataclasses; restructure pipeline execution with new helpers (_emit_plan_events_and_run_before_hooks(), _build_pipeline_observations(), _build_pipeline_stage_results(), _finalize_pipeline_execution()) to coordinate observations, event emission, and hook execution.
Testing infrastructure cuprum/_testing.py	Relocate imports: _merge_env and _prepare_pipeline_config from new modules; _PipelineWaitResult and _wait_for_pipeline from cuprum._pipeline_wait.
Pipeline test updates cuprum/unittests/test_pipeline.py	Add started_at parameter to _wait_for_pipeline() call in test fixture.
Observe hook tests cuprum/unittests/test_observe.py	Add unit tests verifying observe registration, event emission (phases, lines, tags, timing), and pipeline stage observation.
Feature file collection tests/conftest.py	Introduce FeatureFile and FeatureFileItem pytest collector classes to enable .feature file selection from CLI.
Behavioural tests tests/behaviour/test_structured_events.py	Add behaviour-driven test for structured events and telemetry hook integration.
BDD feature specification tests/features/structured_events.feature	Define feature scenario for observe hook receiving stdout/stderr lines, timing, and metadata.
Documentation docs/cuprum-design.md, docs/roadmap.md, docs/users-guide.md	Document structured events design, mark roadmap task as complete, provide usage guidance and examples.

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~50 minutes

• Dense logic across process lifecycle and pipeline orchestration: Termination with grace periods, fail-fast semantics, and pipe task coordination demand careful reasoning around timing and exception handling.
• Restructured pipeline internals: New _ExecutionHooks, _StageObservation, and event emission pathways introduce multiple interdependent dataclasses and helpers requiring verification of ordering, state passing, and side effects.
• Async event emission and hook scheduling: The _emit_exec_event() path with background task collection and subsequent _wait_for_exec_hook_tasks() must be verified for cancellation safety and exception propagation.
• Stream handling changes: Line-based emission with incremental decoding adds complexity to buffer management and line-ending normalisation.
• Heterogeneous file spread: Changes span eight internal modules plus public API, context, command execution, and tests—each requiring separate reasoning.

Areas requiring extra attention:

• Verification of fail-fast termination logic in _terminate_pipeline_remaining_stages() and exception handling in _wait_for_pipeline().
• Correctness of event emission sequencing (plan → start → stdout/stderr → exit) and phase ordering within _StageObservation.emit().
• Proper cleanup and cancellation semantics in _cleanup_spawned_processes() and error paths.
• Line-based stream emission completeness and handling of partial lines in _consume_stream_with_lines().
• Integration of ExecutionContext.tags merging into event tags across both command and pipeline execution.

Poem

🎭 Events now flow through the pipeline's veins,

Hooks capture secrets of stdout and stderr's domains,

Timing, tags, and telemetry dance,

Structured observability's grand advance! 📊✨

Pre-merge checks and finishing touches

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 66.27% which is insufficient. The required threshold is 80.00%.	You can run @coderabbitai generate docstrings to improve docstring coverage.

✅ Passed checks (2 passed)

Check name	Status	Explanation
Title check	✅ Passed	Title accurately summarises the main feature: structured pipeline events and telemetry hooks are the primary focus of this substantial changeset.
Description check	✅ Passed	Description comprehensively details the feature set, implementation, tests, and usage patterns; directly aligned with the changeset scope.

✨ Finishing touches

• 📝 Generate docstrings

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch terragon/implement-structured-events-telemetry-deqvhb

---

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

[leynos]

@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix:

cuprum/_pipeline_internals.py

Comment on lines +203 to +225

    hooks_by_stage = tuple(_run_before_hooks(cmd) for cmd in parts)
    pending_tasks: list[asyncio.Task[None]] = []
    cwd = None if config.ctx.cwd is None else Path(config.ctx.cwd)
    env_overlay = _freeze_str_mapping(config.ctx.env)
    observations = tuple(
        _StageObservation(
            cmd=cmd,
            hooks=hooks,
            tags=_merge_tags(
                {
                    "project": cmd.project.name,
                    "capture": capture,
                    "echo": echo,
                    "pipeline_stage_index": idx,
                    "pipeline_stages": len(parts),
                },
                config.ctx.tags,
            ),
            cwd=cwd,
            env_overlay=env_overlay,
            pending_tasks=pending_tasks,
        )
        for idx, (cmd, hooks) in enumerate(zip(parts, hooks_by_stage, strict=True))

❌ New issue: Complex Method
_run_pipeline has a cyclomatic complexity of 13, threshold = 9

[leynos]

@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix:

cuprum/_pipeline_internals.py

Comment on file

import asyncio
import dataclasses as dc
import sys
import time

❌ New issue: Lines of Code in a Single File
This module has 638 lines of code, improve code health by reducing it to 400

[sourcery-ai]

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

Please try again later or upgrade to continue using Sourcery

[coderabbitai]

Actionable comments posted: 7

📜 Review details

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between b76557d and 82acef9.

📒 Files selected for processing (20)

• cuprum/__init__.py (3 hunks)
• cuprum/_observability.py (1 hunks)
• cuprum/_pipeline_internals.py (2 hunks)
• cuprum/_pipeline_spawn.py (1 hunks)
• cuprum/_pipeline_streams.py (1 hunks)
• cuprum/_pipeline_wait.py (1 hunks)
• cuprum/_process_lifecycle.py (1 hunks)
• cuprum/_streams.py (4 hunks)
• cuprum/_testing.py (1 hunks)
• cuprum/context.py (13 hunks)
• cuprum/events.py (1 hunks)
• cuprum/sh.py (6 hunks)
• cuprum/unittests/test_observe.py (1 hunks)
• cuprum/unittests/test_pipeline.py (1 hunks)
• docs/cuprum-design.md (1 hunks)
• docs/roadmap.md (1 hunks)
• docs/users-guide.md (1 hunks)
• tests/behaviour/test_structured_events.py (1 hunks)
• tests/conftest.py (1 hunks)
• tests/features/structured_events.feature (1 hunks)

🧰 Additional context used

📓 Path-based instructions (10)

**/*.py

📄 CodeRabbit inference engine (AGENTS.md)

**/*.py: For Python files, ensure testing passes all relevant unit and behavioral tests by running make test.
For Python files, ensure linting passes by running make lint.
For Python files, ensure formatting adheres to standards by running make check-fmt and applying make fmt if needed.
For Python files, ensure type checking passes by running make 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 (with contextlib.contextmanager or class-based __enter__/__exit__) to encapsulate setup and teardown logic for resource management (f...

Files:

• tests/behaviour/test_structured_events.py
• tests/conftest.py
• cuprum/events.py
• cuprum/sh.py
• cuprum/_pipeline_streams.py
• cuprum/unittests/test_pipeline.py
• cuprum/context.py
• cuprum/unittests/test_observe.py
• cuprum/_pipeline_spawn.py
• cuprum/_streams.py
• cuprum/_observability.py
• cuprum/__init__.py
• cuprum/_process_lifecycle.py
• cuprum/_pipeline_wait.py
• cuprum/_pipeline_internals.py
• cuprum/_testing.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 numpy style 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 by FIXME: or a ticket link, and used only as a last resort.
• Use pytest fixtures for shared setup (conftest.py or fixtures/)
• Replace duplicate tests with @pytest.mark.parametrize
• Prefer pytest-mock or unittest.mock for 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/case or 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_structured_events.py
• tests/conftest.py
• cuprum/events.py
• cuprum/sh.py
• cuprum/_pipeline_streams.py
• cuprum/unittests/test_pipeline.py
• cuprum/context.py
• cuprum/unittests/test_observe.py
• cuprum/_pipeline_spawn.py
• cuprum/_streams.py
• cuprum/_observability.py
• cuprum/__init__.py
• cuprum/_process_lifecycle.py
• cuprum/_pipeline_wait.py
• cuprum/_pipeline_internals.py
• cuprum/_testing.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_structured_events.py
• cuprum/unittests/test_pipeline.py
• cuprum/unittests/test_observe.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_structured_events.py
• tests/conftest.py
• cuprum/unittests/test_pipeline.py
• cuprum/unittests/test_observe.py
• cuprum/_testing.py

docs/**/*.md

📄 CodeRabbit inference engine (AGENTS.md)

docs/**/*.md: Use markdown files within the docs/ 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 the docs/ directory to reflect the latest state.
All documentation must adhere to the documentation style guide at docs/documentation-style-guide.md.
Record any design decisions made in the relevant design document.

Files:

• docs/cuprum-design.md
• docs/roadmap.md
• docs/users-guide.md

**/*.md

📄 CodeRabbit inference engine (AGENTS.md)

**/*.md: For Markdown files (.md only), ensure linting passes by running make markdownlint.
For Markdown files, validate Mermaid diagrams by running make 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 using make markdownlint.
Run make fmt after any documentation changes to format all Markdown files and fix table markup.
Validate Mermaid diagrams in Markdown files by running make nixie.

Files:

• docs/cuprum-design.md
• docs/roadmap.md
• docs/users-guide.md

---

⚙️ CodeRabbit configuration file

**/*.md: * Avoid 2nd person or 1st person pronouns ("I", "you", "we")

• Use en-GB-oxendict (-ize / -yse / -our) spelling and grammar
• Headings must not be wrapped.
• Documents must start with a level 1 heading
• Headings must correctly increase or decrease by no more than one level at a time
• Use GitHub-flavoured Markdown style for footnotes and endnotes.
• Numbered footnotes must be numbered by order of appearance in the document.

Files:

• docs/cuprum-design.md
• docs/roadmap.md
• docs/users-guide.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, use ![alt text](path/to/image) and provide brief alt text describing the content
Add a short description before each Mermaid diagram in documentation so screen readers can understand it

Files:

• docs/cuprum-design.md
• docs/roadmap.md
• docs/users-guide.md

docs/**/*.{md,mdx,rst,txt,rs}

📄 CodeRabbit inference engine (docs/documentation-style-guide.md)

Keep US spelling when used in API contexts, for example 'color'

Files:

• docs/cuprum-design.md
• docs/roadmap.md
• docs/users-guide.md

docs/**/*.{md,mdx}

📄 CodeRabbit inference engine (docs/documentation-style-guide.md)

Follow markdownlint recommendations for Markdown formatting

Files:

• docs/cuprum-design.md
• docs/roadmap.md
• docs/users-guide.md

**/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
• cuprum/unittests/test_observe.py

docs/users-guide.md

📄 CodeRabbit inference engine (AGENTS.md)

docs/users-guide.md: Ensure new functionality is clearly documented in the docs/users-guide.md file.
Ensure revised functionality is clearly documented in the docs/users-guide.md file.

docs/users-guide.md: Add sh.make to construct SafeCmd instances with typed argv handling and minimal builder examples; document the expected builder pattern in docs/users-guide.md
Document hook usage patterns in docs/users-guide.md
Provide a scaffold and guidance for project-specific builders, including a template module and checklist in docs/users-guide.md

Files:

• docs/users-guide.md

🧬 Code graph analysis (8)

cuprum/sh.py (6)

cuprum/_observability.py (4)

• _emit_exec_event (33-43)
• _freeze_str_mapping (16-21)
• _merge_tags (24-30)
• _wait_for_exec_hook_tasks (50-57)

cuprum/_pipeline_internals.py (4)

• _run_before_hooks (54-62)
• _run_pipeline (204-273)
• emit (74-97)
• _EventDetails (101-105)

cuprum/_process_lifecycle.py (1)

• _merge_env (103-111)

cuprum/context.py (1)

• observe (476-491)

cuprum/events.py (1)

• ExecEvent (23-66)

cuprum/_streams.py (2)

• _StreamConfig (17-24)
• _consume_stream (27-36)

cuprum/context.py (3)

cuprum/catalogue.py (1)

• allowlist (67-69)

cuprum/unittests/test_observe.py (2)

• hook (41-42)
• hook (55-56)

tests/behaviour/test_structured_events.py (1)

• hook (63-64)

cuprum/_pipeline_spawn.py (1)

cuprum/_process_lifecycle.py (3)

• _build_spawn_observations (114-146)
• _cleanup_spawned_processes (57-81)
• _spawn_pipeline_processes (149-215)

cuprum/_observability.py (3)

cuprum/events.py (1)

• ExecEvent (23-66)

cuprum/unittests/test_observe.py (2)

• hook (41-42)
• hook (55-56)

tests/behaviour/test_structured_events.py (1)

• hook (63-64)

cuprum/__init__.py (2)

cuprum/context.py (7)

• HookRegistration (366-423)
• after (451-473)
• allow (344-363)
• before (426-448)
• current_context (195-197)
• observe (476-491)
• scoped (251-287)

cuprum/events.py (1)

• ExecEvent (23-66)

cuprum/_process_lifecycle.py (4)

cuprum/_observability.py (2)

• _freeze_str_mapping (16-21)
• _merge_tags (24-30)

cuprum/_pipeline_streams.py (4)

• _collect_pipe_results (181-189)
• _PipelineRunConfig (18-37)
• _create_stage_capture_tasks (93-152)
• capture_or_echo (26-27)

cuprum/_pipeline_internals.py (3)

• _StageObservation (66-97)
• _EventDetails (101-105)
• emit (74-97)

cuprum/sh.py (5)

• SafeCmd (399-517)
• _EventDetails (236-240)
• argv_with_program (413-415)
• stdout (160-162)
• emit (209-232)

cuprum/_pipeline_wait.py (2)

cuprum/_pipeline_streams.py (2)

• _collect_pipe_results (181-189)
• _surface_unexpected_pipe_failures (192-204)

cuprum/_process_lifecycle.py (2)

• _cleanup_pipeline_on_error (84-100)
• _terminate_pipeline_remaining_stages (232-264)

cuprum/_testing.py (3)

cuprum/_pipeline_streams.py (1)

• _prepare_pipeline_config (40-59)

cuprum/_pipeline_wait.py (2)

• _PipelineWaitResult (21-25)
• _wait_for_pipeline (89-136)

cuprum/_process_lifecycle.py (1)

• _merge_env (103-111)

🪛 LanguageTool

docs/cuprum-design.md

[uncategorized] ~932-~932: Possible missing comma found.
Context: ...xecution and awaited before returning results so run_sync() does not leak pending t...

(AI_HYDRA_LEO_MISSING_COMMA)

docs/users-guide.md

[grammar] ~373-~373: The verb ‘observe’ does not usually follow articles like ‘an’. Check that ‘observe’ is spelled correctly; using ‘observe’ as a noun may be non-standard.
Context: ...nts For richer observability, register an observe hook with sh.observe(). Observe hooks...

(A_INFINITIVE)

🔍 Remote MCP

Let me try searching the web for information about Cuprum and its architecture:The web search results are returning information about stock trading patterns (cup and handle) and React hooks, which are not relevant to the Cuprum library. Let me refine my search:Based on my research, I've gathered relevant context about structured events and telemetry patterns in Python. Let me compile the findings for the PR review.

Summary of Additional Context Found

Structured Events and Telemetry Patterns

The PR implements structured execution events following industry best practices found in telemetry libraries:

Event Design Pattern: In OpenTelemetry, events are discrete, time-stamped occurrences in an application's life cycle that provide specific information about what happened at a particular time. These can represent user actions, system state changes, or significant milestones in application processing. Each event typically includes attributes, such as timestamps, severity levels, and additional metadata that describe the event.

Async Hook Handling: If you are using async support, then hooks registered must be async functions, rather than plain functions. The PR's approach to handling both synchronous and asynchronous observers via _emit_exec_event and _await_awaitable aligns with this pattern.

Telemetry-Agnostic Design: Similar to Jupyter's Telemetry approach, telemetry systems enable applications to record events—actions by application users—and transmit them to remote (or local) destinations as structured data. It works with Python's standard logging library to handle the transmission of events allowing users to send events to local files, over the web, etc. Cuprum's design avoids tight coupling to specific telemetry stacks.

Implementation Review Points

1. Event Phases and Metadata: The PR's event phases (plan, start, stdout, stderr, exit) and metadata fields (pid, duration_s, exit_code, tags) align with standard telemetry event models.

2. Per-Line Emission: Structured event data such as the user ID, login status, and authentication method capture information that makes events easy to analyze and correlate with other telemetry data. The implementation's per-line stdout/stderr emission enables granular observability.

3. Complexity Management: The PR acknowledges code complexity through the proposed refactoring of _run_pipeline (cyclomatic complexity 13) and extraction of pipeline-related logic into dedicated modules (_pipeline_wait.py, _process_lifecycle.py, _pipeline_streams.py), addressing file size and maintainability concerns identified in code reviews.

4. Async Task Leak Prevention: The PR's careful handling of pending tasks in _wait_for_exec_hook_tasks prevents background task leaks, which is critical for long-running async operations.

[::Context7::]

🔇 Additional comments (63)

tests/features/structured_events.feature (1)

1-10: LGTM! Clear BDD feature definition.

The feature file clearly describes the structured execution events behavior. The scenario covers the key aspects: output events (stdout/stderr lines) and timing/tag metadata.

docs/roadmap.md (1)

55-56: LGTM! Roadmap accurately reflects completed work.

The task description now correctly captures the implemented scope: stdout/stderr line emissions, timings, tag metadata, and sh.observe() registration.

cuprum/unittests/test_pipeline.py (1)

423-423: LGTM! Test correctly updated for new timing parameter.

The started_at parameter addition aligns with the refactored pipeline wait logic that now tracks per-stage start times for timing metadata.

tests/behaviour/test_structured_events.py (5)

15-17: LGTM! Proper use of TYPE_CHECKING guard.

The ExecEvent import is correctly guarded with TYPE_CHECKING to avoid runtime import costs whilst maintaining type hint support.

---

37-50: LGTM! Test fixture clearly constructs observed command.

The fixture builds a Python command that writes to both stdout and stderr, providing comprehensive coverage for observe hook testing.

---

52-70: LGTM! Observe hook collection pattern is clear.

The test correctly demonstrates the observe hook API: registering a hook with sh.observe(hook) within a scoped allowlist context, and passing custom tags via ExecutionContext.

---

72-81: LGTM! Output event validation is comprehensive.

The test correctly validates that stdout and stderr line events are emitted and contain the expected output lines.

---

83-98: LGTM! Timing and tag metadata validation is thorough.

The test validates all key metadata: process ID, exit code, duration, custom tags (run_id), and default tags (project).

docs/cuprum-design.md (1)

913-933: LGTM! Comprehensive documentation of observe hooks.

The section clearly documents the structured event stream design decisions:

• Event phases and pipeline stage tagging
• Per-line emission semantics
• Timing measurement approaches (wall-clock vs monotonic)
• Tag precedence rules
• Async observer task lifecycle management

The documentation aligns with the implementation and provides clear guidance for users.

docs/users-guide.md (1)

371-408: LGTM! Clear user guide section for structured events.

The documentation provides a practical example demonstrating:

• Registering observe hooks with sh.observe()
• Collecting ExecEvent values
• Filtering events by phase (stdout, stderr, exit)
• Passing custom tags via ExecutionContext.tags
• Accessing event metadata (line, tags, etc.)

The example aligns with the behavioral test in tests/behaviour/test_structured_events.py, ensuring consistency between documentation and implementation.

cuprum/__init__.py (2)

34-45: LGTM! Public API exports are correctly added.

The new observability types and functions are properly imported and exposed:

• ExecHook from cuprum.context
• observe from cuprum.context
• ExecEvent from cuprum.events

These align with the documented API in the users guide and design documentation.

---

61-99: LGTM! all list correctly updated.

The new exports (ExecEvent, ExecHook, observe) are correctly added to __all__, maintaining the existing organisational structure of the list.

cuprum/_pipeline_spawn.py (1)

1-20: Wrapper module is justified and currently imported by other modules.

This module is imported by cuprum/_pipeline_internals.py, which depends on _spawn_pipeline_processes from this wrapper. The original suggestion to remove this module is invalid because callers already reference it. Removing it would break the import in _pipeline_internals.py.

The circular import claim in the docstring should be documented further if there is an actual cycle involving _process_lifecycle and _pipeline_internals. Verify whether the cycle exists and add a brief explanation to the docstring if it does (e.g., which modules would create the cycle and why the wrapper resolves it).

Likely an incorrect or invalid review comment.

cuprum/_testing.py (1)

17-24: LGTM—import provenance updated to reflect refactored modules.

The import reorganisation correctly reflects the extraction of pipeline internals into dedicated modules (_pipeline_streams, _pipeline_wait, _process_lifecycle). The exported test surface remains unchanged.

tests/conftest.py (1)

31-36: LGTM—minimal validation aligns with pytest-bdd semantics.

The runtest implementation performs a simple string check for "Feature:" to confirm the file resembles a Gherkin feature. Whilst this won't catch malformed features, it serves the stated purpose of enabling CLI selection without altering pytest-bdd behaviour.

cuprum/events.py (1)

19-72: LGTM—well-structured event model for telemetry integration.

The event design aligns with industry best practices for structured telemetry. The frozen dataclass with slots ensures immutability and memory efficiency. The ExecHook type alias correctly supports both synchronous and asynchronous observers, enabling flexible integration patterns.

cuprum/unittests/test_observe.py (3)

49-73: LGTM—hook registration lifecycle correctly validated.

The test confirms that observe hooks attach/detach cleanly and that detaching halts event collection. The assertions appropriately check hook count before/after registration and event capture behaviour.

---

76-123: LGTM—comprehensive event payload validation.

The test verifies all key event attributes: phase sequence (plan→start→exit), stdout/stderr line emission, timing data (duration_s), process metadata (pid, cwd, env), and tag propagation (project, run_id). The assertions are thorough and appropriate.

---

126-159: LGTM—pipeline stage tagging validated.

The test confirms that per-stage events include pipeline_stage_index tags and that the final stdout reflects combined stage outputs. The assertions correctly verify exit event count and stage-specific stdout emission.

cuprum/_observability.py (1)

33-43: LGTM—async task leak prevention correctly implemented.

The emission logic correctly schedules async hooks as background tasks and appends them to pending_tasks for later awaiting. This prevents task leaks and ensures hooks complete before run_sync returns.

cuprum/_pipeline_streams.py (3)

40-59: LGTM—dynamic import avoids circular dependency.

The _sh_module() dynamic import pattern is appropriate for internal helpers to avoid circular imports whilst still accessing ExecutionContext. The fallback to default context when context is None ensures robustness.

---

93-152: LGTM—conditional observation callbacks correctly implemented.

The creation of on_line callbacks only when observation.hooks.observe_hooks is non-empty avoids overhead when no observers are registered. The inline function definitions correctly capture observation, process.pid, and _EventDetails for event emission.

---

192-204: LGTM—expected pipe failures correctly filtered.

The filtering of BrokenPipeError and ConnectionResetError as normal termination (e.g., when head closes stdin early) is appropriate. Other exceptions are correctly surfaced to indicate genuine failures.

cuprum/_streams.py (2)

67-109: LGTM—incremental decoding correctly implemented.

The use of codecs.getincrementaldecoder ensures multibyte characters split across chunk boundaries are decoded correctly. The decoder.decode(b"", final=True) call flushes any remaining buffered bytes, and the final partial line handling (line 105) ensures all output is emitted.

---

201-230: LGTM—line emission logic correctly handles line endings.

The _emit_completed_lines function correctly uses splitlines(keepends=True) to preserve line endings, checks the last line for a trailing ending, and pops it as a remainder if incomplete. The _strip_line_ending helper normalises both \r\n and \n/\r endings. Edge cases (empty text, no newlines) are handled by returning the input as remainder.

cuprum/context.py (5)

54-62: LGTM—observe_hooks field correctly added to CuprumContext.

The addition of observe_hooks with FIFO ordering is documented in the docstring (line 55) and correctly initialised as an empty tuple. The design choice of FIFO (vs LIFO for after_hooks) aligns with typical event stream processing patterns.

---

86-140: LGTM—narrow() correctly propagates observe_hooks.

The narrow() method correctly extends observe_hooks with FIFO ordering (line 133) and passes them to the new context (line 139). The docstring (lines 105-106) documents the FIFO semantics.

---

168-175: LGTM—observe hook mutation methods correctly implemented.

The with_observe_hook method appends the hook to maintain FIFO order, and without_observe_hook correctly filters out the specific hook instance. The implementations mirror the before/after hook patterns.

---

384-401: LGTM—HookRegistration correctly supports observe hooks.

The updated signature accepts ExecHook in the union type (line 386), the hook_type Literal includes "observe" (line 387), and the dispatch logic (lines 399-400) correctly routes observe hooks through with_observe_hook. The token-based restoration semantics remain consistent.

---

476-491: LGTM—observe() public API correctly implemented.

The function signature, docstring, and return type align with the existing before() and after() patterns. The integration with HookRegistration ensures consistent detach() semantics and context manager support.

cuprum/sh.py (9)

18-45: LGTM!

Imports are well-organised: runtime helpers from _observability, type-only imports guarded under TYPE_CHECKING, and explicit re-export of observe.

---

185-198: LGTM!

The tags field follows the established pattern for optional context parameters and uses an immutable Mapping type appropriately.

---

243-259: LGTM!

The cancellation handling correctly terminates the process, awaits consumers to prevent resource leaks, and re-raises CancelledError. Using time.perf_counter() for duration measurement is the right choice for monotonic high-resolution timing.

---

262-268: LGTM!

Clean parameter object pattern that bundles execution context and observation together.

---

271-288: LGTM!

Subprocess creation correctly configures pipes based on capture/echo flags and safely handles the optional cwd parameter.

---

291-354: LGTM!

Stream consumption and line-by-line observation emission are correctly wired. The use of dc.replace for stderr configuration is clean, and consumer tasks are properly awaited after process exit.

---

357-395: LGTM!

The subprocess lifecycle is well-orchestrated: spawn → start event → stream consumption → exit event with timing. The defensive max(0.0, ...) for duration is a sensible guard.

---

450-488: LGTM!

The execution flow correctly sequences plan → before_hooks → execute → after_hooks → await pending tasks. Cancellation handling ensures pending async hooks are awaited before propagation.

---

597-598: LGTM!

Public API correctly exports observe for user access.

cuprum/_pipeline_wait.py (4)

37-51: LGTM!

The classmethod correctly initialises wait tasks and index mappings for efficient task-to-stage resolution.

---

54-73: LGTM!

Fail-fast logic correctly records the first failure, updates timing, and terminates remaining stages (excluding the final stage which needs no downstream termination).

---

76-86: LGTM!

Correctly defers pipe failure surfacing when an exception is already being propagated, preventing exception masking.

---

89-136: LGTM with minor observation.

The wait loop correctly implements fail-fast semantics using FIRST_COMPLETED. The finally block's assignment to pipe_results is unused after the function exits, but this is acceptable since _finalize_pipeline_wait is called for its side effects (surfacing unexpected failures).

cuprum/_process_lifecycle.py (10)

1-17: LGTM!

Clean import structure with TYPE_CHECKING guards preventing circular imports.

---

19-29: LGTM!

Clean delegation to the generic termination helper with appropriate callbacks.

---

32-54: LGTM!

Robust termination logic with proper exception handling for race conditions where the process may exit between checks. The noqa: UP041 annotation correctly documents the need for asyncio.TimeoutError.

---

57-81: LGTM!

Cleanup correctly terminates processes before cancelling capture tasks, with exception suppression to ensure complete cleanup.

---

84-100: LGTM!

Clean separation of concerns: this helper handles process termination and pipe result collection, leaving stream task cleanup to the caller.

---

103-111: LGTM!

Returns a mutable dict as required by subprocess APIs. The |= merge operator is clean and idiomatic.

---

143-146: LGTM!

Defensive check correctly ensures that callers provide explicit observations when observe hooks are registered, preventing accidental loss of observability wiring.

---

149-215: LGTM!

Pipeline spawning correctly wires stdin/stdout between stages, records per-stage timing, emits start events, and handles cleanup on spawn failures.

---

218-229: LGTM!

The use of asyncio.shield correctly protects the shared wait task from cancellation during the grace period timeout.

---

232-263: LGTM!

Fail-fast termination correctly skips the failed stage and already-completed stages, terminating only the still-running processes.

cuprum/_pipeline_internals.py (10)

1-35: LGTM!

Imports are well-organised with TYPE_CHECKING guards for type-only dependencies.

---

47-51: LGTM!

Clean immutable data structure for hook collection with appropriate tuple types.

---

54-62: LGTM!

Correctly collects hooks after enforcing the allowlist check. The function name reflects its role in the pre-execution phase.

---

65-97: LGTM!

The emit method correctly constructs ExecEvent with all required fields. Note: duplication with _CommandObservation in sh.py was flagged earlier.

---

100-105: LGTM!

Simple data carrier for event emission details.

---

115-143: LGTM!

Observations are correctly built with merged tags including context tags and pipeline metadata.

---

146-152: LGTM!

Clean helper that emits plan events and executes before hooks in the correct order.

---

155-190: LGTM!

Stage results are correctly constructed with duration calculation and exit event emission. The max(0.0, ...) guard is appropriately defensive.

---

193-201: LGTM!

Finalisation correctly runs after hooks and awaits pending async observe tasks.

---

276-284: LGTM!

Clean iteration with strict=True ensuring length consistency between parts, hooks, and results.

[chatgpt-codex-connector]

💡 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".