Skip to content

feat(adr-001): streaming event iterator over SubLinear orchestrator - #45

Merged
ruvnet merged 1 commit into
mainfrom
adr/event-stream
May 19, 2026
Merged

ruvnet merged 1 commit into
mainfrom
adr/event-stream

Conversation

@ruvnet

@ruvnet ruvnet commented May 19, 2026

Copy link
Copy Markdown
Owner

Summary

The single-event call pattern (auto-tuned contrastive orchestrator + coherence gate + `PlanBudget`) was complete after PR #40. Real callers — RuView agents, Cognitum reflex loops — process an iterator of events, not a single one. This PR lifts the single-event call into a stdlib `Iterator` adapter.

API

```rust
pub fn event_stream_iter(
matrix: &dyn Matrix,
prev_solution: &[Precision],
events: I,
config: &EventStreamConfig,
budget: Option<&mut PlanBudget>,
) -> impl Iterator<Item = ProcessedEvent>
where I: Iterator<Item = (SparseDelta, Vec)>;

pub struct ProcessedEvent {
pub event_idx: usize,
pub anomalies: Vec,
pub latency: Duration,
pub status: EventStatus,
}

pub enum EventStatus { Solved, Skipped, BudgetRefused, Errored }
```

Pipeline per event

  1. `delta_below_solve_threshold` skip gate (if cached coherence/min_diag supplied) → `Skipped` status.
  2. `PlanBudget::try_consume` (if supplied) → `BudgetRefused` ends the iterator.
  3. `contrastive_solve_on_change_sublinear_auto` for top-k anomalies → `Solved`.

Composition

```rust
for event in event_stream_iter(matrix, prev, events, &cfg, Some(&mut budget)) {
match event.status {
EventStatus::Skipped => continue,
EventStatus::BudgetRefused => break,
EventStatus::Solved => process(event.anomalies),
EventStatus::Errored => log_and_continue(),
}
}
```

Native composition with `.filter()`, `.take()`, `.collect()`, etc.

Test plan

  • 5 `stream::tests::*` covering empty input, one-event-per-input with `Solved` status, skip-gate short-circuit on tiny delta, budget-exhausted termination, op-class compile-time
  • `cargo test --lib -- stream::` → 5/5 pass
  • `.github/complexity-baseline.txt` regenerated with `EventStreamOp` (SubLinear)
  • Full CI

🤖 Generated with claude-flow

The single-event call (auto-tuned contrastive orchestrator + coherence
gate + PlanBudget) was complete after PR #40. Real callers — RuView
agents watching sensor / log / metric streams, Cognitum reflex loops
over a learning system — process an *iterator* of events, not a
single one. This module lifts the single-event call into a stdlib
Iterator adapter.

Lands:

  src/stream.rs (new, 280 LOC)
    pub fn event_stream_iter<'a, I>(matrix, prev, events, config, budget)
        -> EventStreamIter<'a, I>
        where I: Iterator<Item = (SparseDelta, Vec<Precision>)>

    pub struct ProcessedEvent {
        event_idx, anomalies, latency, status
    }

    pub enum EventStatus {
        Solved, Skipped, BudgetRefused, Errored
    }

    pub struct EventStreamConfig {
        tolerance, k, skip_threshold (optional gate inputs)
    }

    pub struct EventStreamOp; // SubLinear class marker

  Pipeline per event:
    1. delta_below_solve_threshold skip gate (if configured)
    2. PlanBudget::try_consume (if supplied) — ends iterator on refusal
    3. contrastive_solve_on_change_sublinear_auto for top-k

  Native composition with stdlib iterators:
    for event in event_stream_iter(matrix, prev, events, &cfg, Some(&mut budget)) {
        if event.status == EventStatus::Skipped { continue; }
        if event.status == EventStatus::BudgetRefused { break; }
        // process event.anomalies
    }

  src/lib.rs
    Re-exports event_stream_iter, ProcessedEvent, EventStatus,
    EventStreamConfig, EventStreamOp.

  .github/complexity-baseline.txt
    Adds EventStreamOp (SubLinear).

Tests: 5 covering empty input, one-event-per-input with Solved
status, skip-gate short-circuit on tiny delta, budget-exhausted
terminating the stream, and op class compile-time check.

Co-Authored-By: claude-flow <ruv@ruv.net>
@ruvnet
ruvnet merged commit 10ae3bd into main May 19, 2026
11 of 12 checks passed
@ruvnet
ruvnet deleted the adr/event-stream branch May 19, 2026 16:29
ruvnet added a commit that referenced this pull request May 19, 2026
Companion to event_driven_anomaly.rs. The earlier example called
the per-event primitives directly; this one uses the iterator
adapter (PR #45) and composes it with stdlib's iterator chain —
the native Rust pattern RuView / Cognitum / Ruflo would adopt.

Demonstrates:
  - One event stream → one iterator chain → reports.
  - Coherence-gated skip on tiny deltas (latency 0).
  - PlanBudget cap on cumulative ops; iterator ends gracefully
    when budget exhausted.
  - Per-event latency tracked by the iterator.

Sample output (10-event stream, budget=4 SubLinear ops):

  event #  status       latency_us top-anomaly
  ────────────────────────────────────────────
  0        Solved        73        row 7
  1        Skipped        0        row —
  2        Solved        43        row 22
  3        Skipped        0        row —
  4        Solved        41        row 40
  5        Solved        42        row 51
  6        Skipped        0        row —
  7        BudgetRefused  0        row —

  Summary: 4 solved, 3 skipped by gate, 1 budget-refused

The architecture line is fully realised: streaming surface with
stdlib iterator composition, per-event SubLinear cost, O(|δ|) gate,
O(1) budget consume.

Co-Authored-By: claude-flow <ruv@ruv.net>
ruvnet added a commit that referenced this pull request May 19, 2026
docs(adr-001): refresh primitive catalogue with PRs #45-#57
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant