Publish Rust-pump routing decisions on a dedicated observation channel

[leynos]

Problem

When an inter-stage pipe hop declines the Rust stream pump and falls back to
the Python one, or when a hop cancelled mid-transfer turns out to have had a
failing Rust worker, the decision is visible only as a DEBUG log record. An
operator whose deployment has quietly stopped taking the fast path — a kernel
that no longer yields raw descriptors, a transport that refuses to pause, a
blocking-mode switch that fails — sees no counter move and no aggregate
change. The pipeline still returns exactly what it would have returned, which
is precisely why the regression is invisible: routing is not a result, so
nothing in PipelineResult records it, and grepping every worker's DEBUG
stream for rust_pump_declined is not an answer a dashboard can give.

Cuprum already ships a MetricsCollector protocol and a MetricsHook that
projects ExecEvent values onto it. The obvious move — add a phase to
ExecPhase — is the one that must not be taken. ExecPhase is a closed set
that registered consumers match exhaustively, and Cuprum's own MetricsHook
raises _UnhandledMetricsPhaseError on an unrecognized phase by design. A new
member would start raising inside third-party consumers that were correct when
they were written, and, because Cuprum re-raises observe-hook failures, would
fail the runs those consumers were merely observing. Pump facts need a channel
of their own, entered by opting in.

Required work

Publish Rust-pump routing decisions on a dedicated observation channel that
cannot perturb the existing ExecEvent contract.

1. A pump event type, separate from ExecEvent

cuprum/pump_events.py carries the vocabulary: a frozen, slotted PumpEvent
dataclass; a closed PumpPhase literal of declined and
failed_after_cancel; and RustPumpDeclineReason, promoted from a private
enum to public API because operators aggregate on it. The reason set is a
StrEnum rather than free text so that a typo at a new call site is a type
error rather than a value an operator's filter silently misses.

PumpEvent carries the phase and, for a decline, the reason — and nothing
else. Descriptor numbers, argument vectors, exception types, and tracebacks are
either unbounded as metric labels or a disclosure risk; the DEBUG records
emitted alongside these events already carry that detail for the cases that
need it.

2. A hook registry on its own ContextVar

cuprum/pump_observation.py provides observe_pump(), a
PumpHookRegistration handle, current_pump_hooks(), and the internal
_emit_pump_event. Hooks live on their own ContextVar, deliberately not
on CuprumContext. That is the design, not a shortcut: ScopeConfig,
CuprumContext.narrow(), and every consumer that reads the execution context
keep exactly the shape they had, so adding this channel cannot alter how an
existing caller's commands execute. Registration follows the repository's
token-restoration discipline, so a pump registration nests and detaches like
every other one.

3. A metrics adapter reusing the existing collector protocol

cuprum/adapters/pump_metrics.py provides PumpMetricsHook (and a
pump_metrics_hook factory) consuming PumpEvent values and writing through
the existing cuprum.adapters.metrics_adapter.MetricsCollector protocol, so
one collector backs both hooks and no new telemetry dependency enters the
project.

4. Two counters with a bounded label set

• cuprum_rust_pump_declined_total, labelled by reason and nothing else,
  incremented once per hop that fell back to the Python pump.
• cuprum_rust_pump_failed_after_cancel_total, unlabelled, incremented once
  per Rust-pump worker failure consumed during cancellation unwinding.

The reason domain is bounded by RustPumpDeclineReason plus the public
UNKNOWN_DECLINE_REASON constant — four values, no more. UNKNOWN_DECLINE_REASON
is public because it is operator-visible: a dashboard filtering on the enum
alone would silently miss it. It is not an enum member because no seam reports
it; it exists only so that a malformed event cannot introduce an unbounded
label.

5. An ADR recording the decision

docs/adr-008-rust-pump-observation-channel.md, indexed from
docs/contents.md, recording the options weighed — an ExecPhase member, a
module-global counter registry, a dedicated channel — the drivers, the
hook-failure policy, and the non-goals. The users guide, developers guide, and
design document describe registration, the counters, and the label domain.

Acceptance criteria

• cuprum/pump_events.py defines PumpEvent, a closed PumpPhase literal
  (declined, failed_after_cancel), and a public RustPumpDeclineReason
StrEnum with raw_fd_unavailable, reader_pause_failed, and
  blocking_mode_unavailable. PumpEvent, PumpHook,
  PumpHookRegistration, RustPumpDeclineReason, and observe_pump are
  re-exported from the cuprum package.
• No pump phase reaches the ExecEvent contract. ExecPhase gains no
  member, so a registration of the existing MetricsHook — which matches
  ExecPhase exhaustively and raises on anything else — cannot begin
  raising because this channel exists. A test pins ExecPhase's membership
  and asserts the two phase sets are disjoint, and a second test drives a
  real decline past an already-registered MetricsHook and asserts it
  never sees it.
• Hooks are held on a ContextVar private to cuprum.pump_observation,
  with CuprumContext, ScopeConfig, and CuprumContext.narrow()
  unchanged. observe_pump() returns a handle that is both a context
  manager and detachable; registration order is preserved on delivery;
  detach restores the preceding tuple, and detaching twice is harmless.
• _emit_pump_event logs a raising hook at WARNING and continues.
  It does not re-raise, and the remaining hooks still run. This diverges
  deliberately from _emit_exec_event: both pump emission sites lie on
  paths contracted to complete — the fall-back that keeps a hop working
  when the fast path is unavailable, and cancellation unwinding, where a
  raise would displace the CancelledError the caller is owed. Registering
  a metrics backend must not turn a pipeline that would have succeeded into
  one that fails. The failure is recorded, not swallowed: a
  pump_observer_failed record names the hook's error type and the event
  that provoked it.
• A BaseException that is not an Exception — SystemExit,
  KeyboardInterrupt, asyncio.CancelledError — propagates from a hook
  untouched, so a shutdown signal arriving here still travels.
• A raising collector leaves the fall-back signal intact: the hop still
  reports that it declined and still runs on the Python pump.
• Pump hooks are synchronous by contract; a hook returning an awaitable is
  reported at WARNING and the coroutine is closed, rather than left to
  surface later as an unrelated "never awaited" warning.
• With no hooks registered, _emit_pump_event returns before touching the
  event, so a caller that has not opted in pays nothing and behaves exactly
  as before.
• PumpMetricsHook increments cuprum_rust_pump_declined_total once per
  decline, labelled by reason alone, and
  cuprum_rust_pump_failed_after_cancel_total once, unlabelled, per
  failure recovered after cancellation. A successful hand-off counts
  nothing, and a cancelled hop whose pump succeeded counts nothing.
• The reason label is bounded by the closed enum. Each of the three
  real decline seams is driven through the actual fall-back path — not a
  synthesized event — and the resulting labels are asserted to be exactly
  {"reason"} with a value drawn from a bounded set derived from
  RustPumpDeclineReason itself plus UNKNOWN_DECLINE_REASON, rather than
  from a hard-coded copy that could drift from the enum. The assertion loop
  is guarded so that it cannot pass vacuously over an empty collection.
• An unrecognized phase is ignored rather than raised on, applying to this
  adapter the lesson that motivated the separate channel.
• docs/adr-008-rust-pump-observation-channel.md records the decision, the
  rejected options, the hook-failure policy, and the non-goals, and is
  indexed from docs/contents.md. The users guide documents registration,
  both counter names, and the four-value reason domain; the developers
  guide maps each counter to the seam that emits it.
• Covered by cuprum/unittests/test_pump_observation.py,
  cuprum/unittests/test_pump_metrics_adapter.py, and
  cuprum/unittests/test_pipeline_streams_observability.py, the last of
  which also asserts that registering an observer does not displace the
  existing DEBUG log record.

Notes

This work was specified for, and is delivered by, PR #244 alongside the
Hypothesis fault-injection work for #74, at the maintainer's written direction
that it be treated as required for that PR. The issue records the requirement
that PR satisfies.

Companion to #285, which records the equivalent requirement for the pipeline
fail-fast decision. The two channels differ deliberately: the fail-fast
decision is a command lifecycle fact and rides ExecPhase, accepting the
breaking change; pump routing is not, and gets a channel of its own so that no
existing consumer changes behaviour.

[coderabbitai]

🔗 Related PRs

#136 - Apply assorted consistency and dead-code cleanups (#129) [merged]
#144 - Fix Command-Query Separation violations (#118) [merged]
#154 - De-duplicate adapter event projection and collector base (#114) [merged]
#242 - Rust: pure pump state machine with proptest + Kani (#84) [merged]
#247 - Hypothesis stateful tests for the metrics event reducer (#78) [merged]

---

📝 Issue Planner

_{Check the box below or use the @coderabbitai plan command to generate an implementation plan and prompts that you can use with your favorite coding assistant.}

• Create Plan

---

🧪 Issue enrichment is currently in open beta.

You can configure auto-planning by selecting labels in the issue_enrichment configuration.

To disable automatic issue enrichment, add the following to your .coderabbit.yaml:

issue_enrichment:
  auto_enrich:
    enabled: false

💬 Have feedback or questions? Drop into our discord!