# Firing Policies + Ingestion Merge — Design Spec

**Date:** 2026-06-03
**Status:** Draft — awaiting user spec review
**Authors:** orchestrator + Claude

## Goal

Make the reactive semantics of the engine *sit right from the start*: a node
re-evaluates only when its inputs warrant it, driven by **two firing policies**
that must both ship and both be fully tested:

- **A — fire-on-any-fresh + hold** (as-of join): the node fires whenever any of
  its A-inputs is fresh this cycle; stale inputs contribute their last held value
  (C5 sample-and-hold).
- **B — all-fresh barrier** (synchronizing join): the node fires only when every
  member of a barrier group shares the current cycle's timestamp — e.g. O/H/L/C
  assembled from four separate sources, complete only when all four have ticked
  at the same bar time.

Firing is only *observable* when inputs tick at heterogeneous rates, so this
cycle also delivers the enabling substrate: a **k-way merge of multiple sources**
into one chronological cycle stream (C3), the **monotonic cycle_id** clock (C4),
and **freshness-gated recompute** (C5). Per the chosen scope, sources are
in-memory `(timestamp, value)` streams declared by the test/caller; the merge is
the permanent ingestion core, while the `Source` trait + data-server IO are a
later, orthogonal cycle.

This is the first place the engine departs from RustAst rather than sharpening
it: RustAst had one clock per source and a `cycle_id`-equality barrier that only
synchronizes a single source's fan-out; it never merged heterogeneous sources and
never implemented mode A (see the design notes below).

## Architecture

Three contracts compose into one mechanism that turns out to need **two** read
clocks, not one:

- **`cycle_id` is the freshness epoch (C4 → C5).** The merge advances a monotonic
  `cycle_id` per emitted record. Each push into an input slot stamps that slot
  with the current `cycle_id`. A slot is *fresh this cycle* iff its stamp equals
  the current `cycle_id`. This is the genuine first reader of C4's `cycle_id` —
  it is not a write-only field.
- **`timestamp` is the barrier token (C4 → C6).** Each push also stamps the slot
  with the cycle's data timestamp `T`. A barrier group is complete iff all its
  members carry `last_ts == T`. This is RustAst's `last_cycle_per_input.all(==)`
  barrier with the synchronization token changed from `cycle_id` to `timestamp` —
  the change forced by C4's tie rule (see design notes).
- **Push *is* freshness propagation.** A node that fires pushes its output into
  consumers' input columns, stamping them fresh; a node that does not fire pushes
  nothing, so consumers keep seeing the held value via `window[0]`. Sample-and-
  hold (C5) falls out of the existing push model with no new storage — only
  engine-side per-slot bookkeeping (`fresh_at`, `last_ts`).

The graph wiring (flat node array, index edge table, Kahn topo order), the
node-owns-its-input-columns shape (cycle 0002), and `Ctx` are all unchanged. What
changes: `bootstrap` takes many sources instead of one; `run` takes many streams
and merges them; each node carries per-input firing + freshness state; the eval
step is gated by the firing predicate instead of running unconditionally.

## Concrete code shapes

### User-facing: the two rails, authored as nodes

The two firing modes are exercised by two fixture nodes that are **identical
except for the firing annotation in their schema** — a 2-input f64 sum. This is
the empirical evidence for the feature-acceptance criterion: the only thing a
node author writes to choose a rail is `Firing::Any` vs `Firing::Barrier(g)`.

```rust
// Mode A — as-of join: fires whenever either input is fresh; holds the other.
struct AsOfSum;
impl Node for AsOfSum {
    fn schema(&self) -> NodeSchema {
        NodeSchema {
            inputs: vec![
                InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
                InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
            ],
            output: ScalarKind::F64,
        }
    }
    fn eval(&mut self, ctx: Ctx<'_>) -> Option<Scalar> {
        let a = ctx.f64_in(0);
        let b = ctx.f64_in(1);
        if a.is_empty() || b.is_empty() { return None; } // warm-up
        Some(Scalar::F64(a[0] + b[0]))
    }
}

// Mode B — barrier: fires only when both inputs share the cycle timestamp T.
struct BarrierSum;
impl Node for BarrierSum {
    fn schema(&self) -> NodeSchema {
        NodeSchema {
            inputs: vec![
                InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
                InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
            ],
            output: ScalarKind::F64,
        }
    }
    fn eval(&mut self, ctx: Ctx<'_>) -> Option<Scalar> {
        Some(Scalar::F64(ctx.f64_in(0)[0] + ctx.f64_in(1)[0]))
    }
}
```

### User-facing: wiring two heterogeneous sources and running

```rust
// source 0 ticks at t=1,2,3,4; source 1 ticks at t=2,4 (slower).
let s0 = vec![(Timestamp(1), Scalar::F64(10.0)), (Timestamp(2), Scalar::F64(20.0)),
              (Timestamp(3), Scalar::F64(30.0)), (Timestamp(4), Scalar::F64(40.0))];
let s1 = vec![(Timestamp(2), Scalar::F64(100.0)), (Timestamp(4), Scalar::F64(200.0))];

let mut h = Harness::bootstrap(
    vec![Box::new(AsOfSum)],
    vec![
        SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] },
        SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] },
    ],
    vec![],            // no producer->producer edges
    0,                 // observe node 0
).expect("valid");

let out = h.run(vec![s0, s1]);
// Mode A:      [None, None, Some(120.0), Some(130.0), Some(140.0), Some(240.0)]
//   ^ holds s1=100 across t=3 and t=4-on-s0; emits on every tick.
// Mode B (same wiring, BarrierSum instead of AsOfSum):
//              [None, None, Some(120.0), None,        None,        Some(240.0)]
//   ^ emits ONLY at t=2 and t=4 where both inputs share the timestamp; holds (no
//     emit) at t=1, t=3, and the t=4 s0-cycle until s1 completes the bar.
```

The merged cycle order (one record = one cycle, ties by source index, C4) is:
`(t1,s0) (t2,s0) (t2,s1) (t3,s0) (t4,s0) (t4,s1)` — six cycles. The two emitted
vectors above differ exactly at the held cycles, which *is* the difference
between the two rails.

### Implementation shape (secondary)

**aura-core — `Firing` enum + `InputSpec.firing` (before → after):**

```rust
// before
pub struct InputSpec { pub kind: ScalarKind, pub lookback: usize }

// after
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Firing {
    Any,           // mode A: fires the node whenever this input is fresh this cycle
    Barrier(u8),   // mode B: member of barrier group N; the group fires when all
                   //         its members share the current cycle's timestamp
}
pub struct InputSpec { pub kind: ScalarKind, pub lookback: usize, pub firing: Firing }
```

`Sma` (1 input) and `Sub` (2 inputs) in aura-std gain `firing: Firing::Any` on
each declared input — under a single source ticking every cycle, an `Any` input
is fresh every cycle, so their behavior is unchanged (backward compatible).

**aura-engine — `SourceSpec`, per-slot state, signature changes (before → after):**

```rust
// new: a declared source — its scalar kind and the input slots it feeds.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SourceSpec { pub kind: ScalarKind, pub targets: Vec<Target> }

// per-input engine bookkeeping (not visible to nodes).
struct SlotState { fresh_at: u64, last_ts: Timestamp } // init { 0, Timestamp(i64::MIN) }

// NodeBox gains the lifted firing vector + the per-input slot state.
struct NodeBox {
    node: Box<dyn Node>,
    inputs: Vec<AnyColumn>,
    firing: Vec<Firing>,   // lifted from schema at bootstrap
    slots:  Vec<SlotState>,
}

// bootstrap: many sources instead of (source_targets, source_kind).
// before: bootstrap(nodes, source_targets: Vec<Target>, edges, source_kind: ScalarKind, observe)
// after:
pub fn bootstrap(
    nodes: Vec<Box<dyn Node>>,
    sources: Vec<SourceSpec>,
    edges: Vec<Edge>,
    observe: usize,
) -> Result<Harness, BootstrapError>

// run: many streams instead of one record iterator.
// before: run(&mut self, records: impl Iterator<Item = (Timestamp, Scalar)>)
// after:
pub fn run(&mut self, streams: Vec<Vec<(Timestamp, Scalar)>>) -> Vec<Option<Scalar>>
```

The `Harness` struct's `source_targets: Vec<Target>` field becomes
`sources: Vec<SourceSpec>`; no other field is added (`cycle_id` is a loop local,
slot state lives in `NodeBox`).

**The firing predicate (shape):**

```rust
fn fires(firing: &[Firing], slots: &[SlotState], cycle_id: u64, ts: Timestamp) -> bool {
    // mode A: any A-input fresh this cycle
    for (i, f) in firing.iter().enumerate() {
        if matches!(f, Firing::Any) && slots[i].fresh_at == cycle_id { return true; }
    }
    // mode B: for each distinct barrier group, fire iff >=1 member fresh this
    // cycle AND all members carry last_ts == ts. The ">=1 fresh" clause is the
    // once-per-T guard: a group already completed in a prior cycle has no fresh
    // member now, so it does not re-fire.
    for g in distinct_barrier_groups(firing) {
        let members = barrier_members(firing, g);
        let any_fresh = members.iter().any(|&i| slots[i].fresh_at == cycle_id);
        let all_at_ts = members.iter().all(|&i| slots[i].last_ts == ts);
        if any_fresh && all_at_ts { return true; }
    }
    false
}
```

**The run loop (shape):** maintain a cursor per source; per cycle, pick the live
source head with the smallest `(timestamp, source_index)` (C4 k-way merge);
increment `cycle_id`; push the value into that source's target slots (stamping
`fresh_at = cycle_id`, `last_ts = ts`); walk nodes in topo order, and for each
node that `fires(...)`, `eval` it, capture the observed node's result, and
forward each `Some` output into consumers' slots (stamping them likewise). A node
that does not fire is skipped (held). Allocation-free on the per-cycle path
(cursors + counters only).

## Components

- **`Firing` (aura-core)** — the per-input policy enum; part of `InputSpec`, so
  it is declared in `schema()` exactly where C8 places it ("declares each input's
  … firing group").
- **`SourceSpec` (aura-engine)** — a declared source: its kind + the input slots
  it feeds. The multi-source generalization of cycle 0003's `(source_targets,
  source_kind)` pair.
- **`SlotState` (aura-engine, private)** — per-input `{ fresh_at, last_ts }`; the
  two read clocks (freshness epoch + barrier token).
- **`Harness::bootstrap` / `run` (aura-engine)** — extended to many sources +
  the merge + the firing gate. Topo order, edge kind-checks, and cycle rejection
  are unchanged from 0003.
- **`AsOfSum` / `BarrierSum` (test fixtures in aura-engine tests)** — the two
  rails as authored nodes; not library nodes (no speculative aura-std surface
  until a real strategy needs a reusable join).

## Data flow

1. The caller hands `run` one `(Timestamp, Scalar)` stream per source, each
   ascending in time (guaranteed by C3 ingestion; the test supplies sorted vecs).
2. The k-way merge emits one record per cycle in global `(timestamp, source
   index)` order, advancing `cycle_id` monotonically (C3/C4).
3. The record's value is pushed into its source's target slots; those slots are
   stamped `fresh_at = cycle_id`, `last_ts = ts`.
4. Nodes evaluate in topo order. Each node's firing predicate reads its slots'
   `fresh_at` (freshness, C5) and `last_ts` (barrier, C6). A firing node evals
   and forwards `Some` outputs into consumers (stamping them fresh); a non-firing
   node holds.
5. The observed node's emitted value (`Some` when it fired and produced output,
   `None` when it held or filtered) is collected per cycle and returned.

## Error handling

- **Bootstrap (`BootstrapError`, unchanged variants):**
  - `KindMismatch { producer, consumer }` — a source target's slot kind ≠ the
    source's `kind`, or an edge's producer output ≠ consumer slot kind. Checked
    per source and per edge, as in 0003.
  - `BadIndex` — a source target, edge endpoint, or `observe` index out of range.
  - `Cycle` — the wiring contains a directed cycle (Kahn leftover).
- **`run` precondition:** `streams.len()` must equal `sources.len()`, and each
  stream must be ascending by timestamp. A mismatch in length is a caller bug; the
  shape (panic with a clear message vs. a typed error) is a planner decision, but
  it is *not* a silent no-op. Stream monotonicity is a documented precondition of
  the C3 ingestion boundary, not re-validated in the hot path.
- **Cold barrier member:** a slot never pushed has `last_ts = Timestamp(i64::MIN)`,
  which equals no real timestamp, so a barrier with an un-ticked member never
  completes — correct warm-up behavior.
- **Zero-input node:** never has a fresh input, so never fires; harmless (no such
  node exists today). Noted, not special-cased.

## Testing strategy

All on both rails, all deterministic (C1):

1. **Mode A — as-of hold** (`aura-engine`): the worked `AsOfSum` example above;
   assert the emitted vector `[None, None, 120, 130, 140, 240]`, proving the held
   s1 value (100) is reused across t=3 and the t=4 s0-cycle.
2. **Mode B — barrier wait** (`aura-engine`): same wiring with `BarrierSum`;
   assert `[None, None, 120, None, None, 240]`, proving the node emits only when
   both inputs share the timestamp and holds (no emit) otherwise.
3. **Determinism** (C1): a second identical `run` of each rail is bit-identical to
   the first.
4. **Mixed A+B on one node:** a 3-input fixture `[Barrier(0), Barrier(0), Any]`
   fed by three sources; assert it fires both when the barrier pair completes
   (held A) and when the A-input ticks (held barrier pair) — the OR-combine.
5. **Backward-compat fan-out/join** (single source, all `Any`): the 0003
   `source -> {Sma(2), Sma(4)} -> Sub` DAG re-expressed on the new API still emits
   `[None, None, None, 2, 2, 2]`, proving the firing gate is transparent when one
   source ticks every cycle.
6. **Bootstrap rejections** (adapted to `SourceSpec`): kind-mismatch (i64 source
   into an f64 slot), bad index (observe out of range), and a directed cycle —
   the 0003 reject tests re-expressed.
7. **aura-core:** `InputSpec`/schema carry `firing`; `Sma`/`Sub` declare
   `Firing::Any`; existing aura-core/aura-std tests stay green.

## Acceptance criteria

Per aura's feature-acceptance criterion (the audience reaches for it; it improves
correctness; it reintroduces no failure class the core constraints forbid):

- **Reached-for:** authoring a strategy that combines a fast price with a held
  daily bias (mode A) or assembles a bar from separate O/H/L/C feeds (mode B) is
  exactly C6's named motivation; the only authoring act is the `Firing`
  annotation, shown above.
- **Correctness / no look-ahead (C2):** freshness reads only past pushes; the
  barrier waits for completeness rather than peeking ahead; held values are
  explicit last values, never future ones.
- **Determinism (C1):** the merge tie-break (source order) and the firing
  predicate (pure function of `fresh_at`/`last_ts`, which evolve deterministically)
  make every run reproducible — asserted in test 3.
- **No new failure class:** no interior concurrency, no per-cycle allocation, no
  merge inside the graph (C3 — the single merge is at ingestion), no `dyn Any` /
  `Rc` / `RefCell` on the surface.
- Gates green: `cargo build/test/clippy --workspace -- -D warnings`; surface-
  purity grep clean.

## Design notes — why aura diverges from RustAst here

RustAst (`src/ast/rtl/streams/`) is the only prior implementation, and studying it
fixed two decisions:

- **Barrier token: `timestamp`, not `cycle_id`.** RustAst's `PipeStream` barrier
  is `last_cycle_per_input.iter().all(|&c| c == cycle_id)` — all inputs reported
  the same `cycle_id`. That synchronizes one source's fan-out/rejoin (every path
  carries the originating tick's `cycle_id`), but it **cannot** synchronize
  independent sources: under C4, four sources at the same bar timestamp become
  four *different* `cycle_id`s (ties broken by source order), so `all(== cycle_id)`
  would never fire for C6's own O/H/L/C example. Substituting `timestamp` for
  `cycle_id` is the minimal faithful generalization: it fires the diamond rejoin
  (paths share the cycle timestamp) *and* the multi-source bar (separate cycles,
  shared timestamp).
- **Mode A is new.** RustAst implemented only the barrier (C6: "RustAst
  implemented only B"); fire-on-any-fresh + hold has no prior art and is built
  here from the freshness primitive.
- **The merge is new.** RustAst gives each source its own `RootStream` clock
  (`data_streams.rs`) with no k-way merge; aura's single merged timeline (C3) with
  one shared `cycle_id`/`timestamp` sequence is precisely what makes a cross-source
  barrier meaningful. RustAst's `total_count` push counter is aura's already-built
  `Column::run_count` (glossary: run-count, Avoid total_count).
