# Firing Policies + Ingestion Merge — Implementation Plan

> **Parent spec:** `docs/specs/0004-firing-policies-and-merge.md`
>
> **For agentic workers:** REQUIRED SUB-SKILL: use the `implement` skill to run
> this plan. Steps use `- [ ]` checkboxes for tracking.

**Goal:** Ship both firing policies (C6 — A fire-on-any-fresh+hold, B all-fresh
barrier) on top of a k-way ingestion merge (C3), the monotonic cycle_id clock
(C4), and freshness-gated recompute (C5), fully tested on both rails.

**Architecture:** `Firing` is a per-input enum on `InputSpec` (aura-core). The
engine merges N in-memory timestamped sources into one chronological cycle stream
(linear-scan k-way, ties by source index), stamping each input slot per push with
two read clocks — `fresh_at` (cycle_id, freshness epoch C5) and `last_ts`
(timestamp, barrier token C6). A node fires when any `Any` input is fresh or any
`Barrier` group has all members at the current timestamp; a non-firing node holds
(no push → consumers see the held value via `window[0]`).

**Tech Stack:** aura-core (`Firing`, `InputSpec`), aura-std (`Sma`/`Sub` firing
annotation), aura-engine (`SourceSpec`, `SlotState`, `Harness::bootstrap`/`run`,
the `fires` predicate). Gates: `cargo build/test/clippy --workspace -- -D
warnings`, surface-purity grep.

**Note on validation gates:** every inlined code body below is Rust (the project's
source language), so the per-task `cargo build`/`cargo test` gates are its
validation — there is no separate surface-language parse gate (the profile
declares no `spec_validation`; the parse-every-block gate is a documented no-op).

**Task ordering rationale (compile-gate discipline):** Task 1 adds a field to
`InputSpec`, which breaks `Sma`/`Sub` — both callers are threaded inside Task 1,
so it ends on a full green workspace. Task 2 changes the `bootstrap`/`run`
signatures, which breaks the in-module tests; Task 2 therefore ends on a
**partial** gate (`cargo build -p aura-engine --lib`, which excludes `#[cfg(test)]`),
and Task 3 finishes threading by rewriting the tests, ending on the full
workspace gate.

---

## Files this plan creates or modifies

- Modify: `crates/aura-core/src/node.rs` — add `Firing` enum + `firing` field on `InputSpec`; fix module doc.
- Modify: `crates/aura-core/src/lib.rs:41` — re-export `Firing`; update roadmap doc.
- Modify: `crates/aura-std/src/sma.rs:6,24` — import `Firing`; declare `Firing::Any`.
- Modify: `crates/aura-std/src/sub.rs:6,23-25` — import `Firing`; declare `Firing::Any` (×2).
- Modify: `crates/aura-engine/src/harness.rs` — `SourceSpec`, `SlotState`, `NodeBox`/`Harness` fields, `Debug`, `bootstrap`, `run`, `fires`/`group_id`, module doc; rewrite the test module.
- Modify: `crates/aura-engine/src/lib.rs:24` — re-export `SourceSpec`; update module doc.
- Test: `crates/aura-core/src/node.rs` — `input_spec_carries_firing`.
- Test: `crates/aura-engine/src/harness.rs` — 8 tests (chain, fan-out compat+determinism, mode A, mode B, mixed A+B, three bootstrap rejects).

---

## Task 1: aura-core `Firing` + `InputSpec.firing`, threaded through aura-std

**Files:**
- Modify: `crates/aura-core/src/node.rs`
- Modify: `crates/aura-core/src/lib.rs:41`
- Modify: `crates/aura-std/src/sma.rs:6,24`
- Modify: `crates/aura-std/src/sub.rs:6,23-25`

- [ ] **Step 1: Add the `Firing` enum and the `firing` field, fix the module doc**

In `crates/aura-core/src/node.rs`, replace the module-doc lines 1-4:

```rust
//! The node contract (C8): the interface every node implements. A node declares
//! its inputs and output kind via `schema`, and computes one cycle's output via
//! `eval`. Firing policy (C6) and tunable params (C12/C19) are deliberately not
//! part of the schema yet — see spec 0002's "Out of scope".
```

with:

```rust
//! The node contract (C8): the interface every node implements. A node declares
//! its inputs (each with its scalar kind, lookback depth, and firing policy, C6)
//! and its output kind via `schema`, and computes one cycle's output via `eval`.
//! Tunable params (C12/C19) are deliberately not part of the schema yet — see
//! spec 0002's "Out of scope".
```

Then replace the `InputSpec` definition (lines 8-14):

```rust
/// One declared input of a node: its scalar kind and the lookback depth the
/// engine must pre-size for it (must be >= 1).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct InputSpec {
    pub kind: ScalarKind,
    pub lookback: usize,
}
```

with the new `Firing` enum followed by the extended `InputSpec`:

```rust
/// The firing policy of one input (C6): how the engine decides whether a node
/// re-evaluates this cycle. A node fires when *any* of its input groups fires
/// (OR over groups).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Firing {
    /// Mode A — fire-on-any-fresh + hold (as-of join): fires the node whenever
    /// this input is fresh this cycle; a stale input contributes its held value.
    Any,
    /// Mode B — all-fresh barrier (synchronizing join): this input is a member of
    /// barrier group `N`; the group fires the node only when every member shares
    /// the current cycle's timestamp.
    Barrier(u8),
}

/// One declared input of a node: its scalar kind, the lookback depth the engine
/// must pre-size for it (must be >= 1), and its firing policy (C6).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct InputSpec {
    pub kind: ScalarKind,
    pub lookback: usize,
    pub firing: Firing,
}
```

- [ ] **Step 2: Add the `input_spec_carries_firing` test to node.rs**

Append to `crates/aura-core/src/node.rs` (the file currently ends after the
`Node` trait at line 31, with no test module):

```rust

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn input_spec_carries_firing() {
        let a = InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any };
        let b = InputSpec { kind: ScalarKind::F64, lookback: 2, firing: Firing::Barrier(0) };
        assert_eq!(a.firing, Firing::Any);
        assert_eq!(b.firing, Firing::Barrier(0));
        assert_ne!(a.firing, b.firing);
    }
}
```

- [ ] **Step 3: Re-export `Firing` and update the aura-core roadmap doc**

In `crates/aura-core/src/lib.rs`, replace line 41:

```rust
pub use node::{InputSpec, Node, NodeSchema};
```

with:

```rust
pub use node::{Firing, InputSpec, Node, NodeSchema};
```

Then replace the roadmap doc (lines 26-28):

```rust
//! Still to come (subsequent cycles): the firing policies (A: fire-on-any-fresh +
//! hold; B: all-fresh barrier), the deterministic sim loop, sources, and
//! ingestion.
```

with:

```rust
//! Still to come (subsequent cycles): the `Source` trait + data-server ingestion
//! and source-native time normalization (C3/C11), the broker-independent
//! position-event output and downstream broker nodes (C10), and the run registry
//! (C18/C22).
```

- [ ] **Step 4: Thread `Firing::Any` through `Sma`**

In `crates/aura-std/src/sma.rs`, replace the import line 6:

```rust
use aura_core::{Ctx, InputSpec, Node, NodeSchema, Scalar, ScalarKind};
```

with:

```rust
use aura_core::{Ctx, Firing, InputSpec, Node, NodeSchema, Scalar, ScalarKind};
```

Then replace the `schema` inputs line 24:

```rust
            inputs: vec![InputSpec { kind: ScalarKind::F64, lookback: self.length }],
```

with:

```rust
            inputs: vec![InputSpec {
                kind: ScalarKind::F64,
                lookback: self.length,
                firing: Firing::Any,
            }],
```

- [ ] **Step 5: Thread `Firing::Any` through `Sub`**

In `crates/aura-std/src/sub.rs`, replace the import line 6:

```rust
use aura_core::{Ctx, InputSpec, Node, NodeSchema, Scalar, ScalarKind};
```

with:

```rust
use aura_core::{Ctx, Firing, InputSpec, Node, NodeSchema, Scalar, ScalarKind};
```

Then replace the two `InputSpec` constructions (lines 23-26):

```rust
            inputs: vec![
                InputSpec { kind: ScalarKind::F64, lookback: 1 },
                InputSpec { kind: ScalarKind::F64, lookback: 1 },
            ],
```

with:

```rust
            inputs: vec![
                InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
                InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
            ],
```

- [ ] **Step 6: Build and test the full workspace**

Run: `cargo build --workspace`
Expected: 0 errors (aura-engine still uses the old single-source API, untouched
by this task — it reads `schema.inputs[i].kind`/`.lookback`, not constructing
`InputSpec`, so the new field does not break it).

Run: `cargo test --workspace`
Expected: PASS. aura-core gains `input_spec_carries_firing` (19 passed); aura-std
`Sma`/`Sub` tests unchanged (3 passed); aura-engine's 5 existing tests still pass
(5 passed). 27 total, 0 failed.

Run: `cargo clippy --workspace --all-targets -- -D warnings`
Expected: 0 warnings.

---

## Task 2: aura-engine production — `SourceSpec`, merge, firing gate

**Files:**
- Modify: `crates/aura-engine/src/harness.rs` (production code only; the test module is rewritten in Task 3)
- Modify: `crates/aura-engine/src/lib.rs:24`

- [ ] **Step 1: Rewrite the harness.rs module doc + import `Firing`**

In `crates/aura-engine/src/harness.rs`, replace the module-doc block (lines 1-11):

```rust
//! The harness — the closed root graph that runs (glossary: `harness`, C20) —
//! and its deterministic single-source run loop. A `Harness` is a bootstrapped,
//! frozen root graph: a flat node array plus an index edge table, topologically
//! ordered, driven cycle by cycle by one source. It is the flat, monomorphized
//! sharpening of RustAst's reference-counted, interior-mutable observer push
//! graph: no reference counting, no interior mutability, no per-cycle allocation
//! (C1/C7). Each node owns its input columns (the cycle-0002 shape), so `Ctx` is
//! unchanged; a producer's `eval`
//! output is forwarded into its consumers' input columns, and a `None` forwards
//! nothing (the structural seed of sample-and-hold, realized with freshness in a
//! later cycle).
```

with:

```rust
//! The harness — the closed root graph that runs (glossary: `harness`, C20) —
//! and its deterministic run loop. A `Harness` is a bootstrapped, frozen root
//! graph: a flat node array plus an index edge table, topologically ordered,
//! driven cycle by cycle by a k-way merge of timestamped sources (C3/C4). It is
//! the flat, monomorphized sharpening of RustAst's reference-counted,
//! interior-mutable observer push graph: no reference counting, no interior
//! mutability, no per-cycle allocation (C1/C7). Each node owns its input columns
//! (the cycle-0002 shape), so `Ctx` is unchanged.
//!
//! Firing (C5/C6) gates re-evaluation. Two read clocks drive it, both stamped per
//! input slot on every push: `fresh_at` (the `cycle_id` of the last push —
//! freshness epoch, C5) and `last_ts` (the data timestamp of that push — barrier
//! token, C6). A node fires when any `Firing::Any` input is fresh this cycle, or
//! when a `Firing::Barrier` group has every member at the current timestamp; a
//! node that does not fire pushes nothing, so consumers keep seeing the held
//! value via `window[0]` (sample-and-hold falls out of the push model). The
//! barrier token is the timestamp, not the `cycle_id`: under C4 four same-time
//! sources are four cycles, so RustAst's `cycle_id`-equality barrier could never
//! fire across sources — the timestamp generalizes it faithfully.
```

Then replace the import line 13:

```rust
use aura_core::{AnyColumn, Ctx, Node, Scalar, ScalarKind, Timestamp};
```

with:

```rust
use aura_core::{AnyColumn, Ctx, Firing, Node, Scalar, ScalarKind, Timestamp};
```

- [ ] **Step 2: Add `SourceSpec` after the `Target` struct**

In `crates/aura-engine/src/harness.rs`, immediately after the `Target` struct
(which ends at line 28 with its closing `}`), insert:

```rust

/// A declared source: the scalar kind it produces and the input slots each of
/// its records is forwarded into. The multi-source generalization of cycle
/// 0003's `(source_targets, source_kind)` pair — sources are k-way-merged by
/// timestamp at ingestion (C3); there is no merge inside the graph.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SourceSpec {
    pub kind: ScalarKind,
    pub targets: Vec<Target>,
}
```

- [ ] **Step 3: Add `SlotState` and extend `NodeBox`**

In `crates/aura-engine/src/harness.rs`, replace the `NodeBox` struct (lines 43-46):

```rust
struct NodeBox {
    node: Box<dyn Node>,
    inputs: Vec<AnyColumn>,
}
```

with the `SlotState` declaration followed by the extended `NodeBox`:

```rust
/// Per-input engine bookkeeping (invisible to nodes): the two read clocks of the
/// firing machinery. `fresh_at` is the `cycle_id` of the last push into this slot
/// (freshness epoch, C5: fresh this cycle iff `fresh_at == cycle_id`); `last_ts`
/// is the data timestamp of that push (barrier token, C6: a barrier group is
/// complete iff all members carry `last_ts == T`). A never-pushed slot keeps the
/// cold sentinel `Timestamp(i64::MIN)`, which no real timestamp equals.
struct SlotState {
    fresh_at: u64,
    last_ts: Timestamp,
}

struct NodeBox {
    node: Box<dyn Node>,
    inputs: Vec<AnyColumn>,
    firing: Vec<Firing>,
    slots: Vec<SlotState>,
}
```

- [ ] **Step 4: Rename the `Harness.source_targets` field and its `Debug` print**

In `crates/aura-engine/src/harness.rs`, replace the `Harness` struct (lines 48-55):

```rust
pub struct Harness {
    nodes: Vec<NodeBox>,
    topo: Vec<usize>,
    out_edges: Vec<Vec<Edge>>,
    source_targets: Vec<Target>,
    observe: usize,
}
```

with:

```rust
pub struct Harness {
    nodes: Vec<NodeBox>,
    topo: Vec<usize>,
    out_edges: Vec<Vec<Edge>>,
    sources: Vec<SourceSpec>,
    observe: usize,
}
```

Then in the manual `Debug` impl, replace the `source_targets` field-print line 67:

```rust
            .field("source_targets", &self.source_targets)
```

with:

```rust
            .field("sources", &self.sources)
```

- [ ] **Step 5: Rewrite `bootstrap` to take `sources: Vec<SourceSpec>`**

In `crates/aura-engine/src/harness.rs`, replace the entire `bootstrap` function
(lines 73-159, from `    /// Bind nodes + wiring` through the closing `    }` of
`bootstrap`) with:

```rust
    /// Bind nodes + wiring into a frozen, runnable graph. Sizes each node's input
    /// columns from its `schema`, lifts each input's firing policy, initializes
    /// per-slot freshness state, kind-checks every source target and edge, and
    /// topologically orders the nodes (Kahn), rejecting any directed cycle.
    pub fn bootstrap(
        nodes: Vec<Box<dyn Node>>,
        sources: Vec<SourceSpec>,
        edges: Vec<Edge>,
        observe: usize,
    ) -> Result<Harness, BootstrapError> {
        let n = nodes.len();
        if observe >= n {
            return Err(BootstrapError::BadIndex);
        }

        let schemas: Vec<_> = nodes.iter().map(|nd| nd.schema()).collect();

        // size each node's input columns from its schema; lift firing; init slots
        let mut boxes: Vec<NodeBox> = Vec::with_capacity(n);
        for (nd, schema) in nodes.into_iter().zip(schemas.iter()) {
            let inputs: Vec<AnyColumn> = schema
                .inputs
                .iter()
                .map(|spec| AnyColumn::with_capacity(spec.kind, spec.lookback))
                .collect();
            let firing: Vec<Firing> = schema.inputs.iter().map(|spec| spec.firing).collect();
            let slots: Vec<SlotState> = schema
                .inputs
                .iter()
                .map(|_| SlotState { fresh_at: 0, last_ts: Timestamp(i64::MIN) })
                .collect();
            boxes.push(NodeBox { node: nd, inputs, firing, slots });
        }

        // source targets: each source's value must match each of its target slots' kind
        for src in &sources {
            for t in &src.targets {
                let s = schemas.get(t.node).ok_or(BootstrapError::BadIndex)?;
                let slot = s.inputs.get(t.slot).ok_or(BootstrapError::BadIndex)?;
                if slot.kind != src.kind {
                    return Err(BootstrapError::KindMismatch {
                        producer: src.kind,
                        consumer: slot.kind,
                    });
                }
            }
        }

        // edges: indices in range, producer output kind == consumer slot kind
        let mut out_edges: Vec<Vec<Edge>> = vec![Vec::new(); n];
        for &e in &edges {
            let from = schemas.get(e.from).ok_or(BootstrapError::BadIndex)?;
            let to = schemas.get(e.to).ok_or(BootstrapError::BadIndex)?;
            let slot = to.inputs.get(e.slot).ok_or(BootstrapError::BadIndex)?;
            if from.output != slot.kind {
                return Err(BootstrapError::KindMismatch {
                    producer: from.output,
                    consumer: slot.kind,
                });
            }
            out_edges[e.from].push(e);
        }

        // Kahn topological sort; a leftover node means a cycle
        let mut indeg = vec![0usize; n];
        for &e in &edges {
            indeg[e.to] += 1;
        }
        let mut queue: Vec<usize> = (0..n).filter(|&i| indeg[i] == 0).collect();
        let mut topo: Vec<usize> = Vec::with_capacity(n);
        let mut head = 0;
        while head < queue.len() {
            let u = queue[head];
            head += 1;
            topo.push(u);
            for e in &out_edges[u] {
                indeg[e.to] -= 1;
                if indeg[e.to] == 0 {
                    queue.push(e.to);
                }
            }
        }
        if topo.len() != n {
            return Err(BootstrapError::Cycle);
        }

        Ok(Harness {
            nodes: boxes,
            topo,
            out_edges,
            sources,
            observe,
        })
    }
```

- [ ] **Step 6: Rewrite `run` as the merge + firing loop**

In `crates/aura-engine/src/harness.rs`, replace the entire `run` function
(lines 161-202, from `    /// Drive the records` through the closing `    }` of
`run`) with:

```rust
    /// Drive the sources, k-way-merged in timestamp order (ties by source index,
    /// C4); returns the observed node's per-cycle emission (`Some` when it fired
    /// and produced output, `None` when it held or filtered). One stream per
    /// source, each ascending in timestamp (C3 ingestion precondition). Allocates
    /// nothing per cycle beyond the output vector.
    pub fn run(&mut self, streams: Vec<Vec<(Timestamp, Scalar)>>) -> Vec<Option<Scalar>> {
        assert_eq!(
            streams.len(),
            self.sources.len(),
            "run: one stream per source required (got {} streams for {} sources)",
            streams.len(),
            self.sources.len()
        );

        // disjoint field borrows so the topo walk can read topo/out_edges/sources
        // while mutating nodes
        let Harness { nodes, topo, out_edges, sources, observe } = self;
        let observe = *observe;

        let mut cursor: Vec<usize> = vec![0; streams.len()];
        let mut cycle_id: u64 = 0;
        let mut out = Vec::new();

        loop {
            // pick the live source head with the smallest (timestamp, source index)
            let mut pick: Option<usize> = None;
            for (s, stream) in streams.iter().enumerate() {
                if cursor[s] < stream.len() {
                    match pick {
                        None => pick = Some(s),
                        Some(p) => {
                            if stream[cursor[s]].0 < streams[p][cursor[p]].0 {
                                pick = Some(s);
                            }
                        }
                    }
                }
            }
            let s = match pick {
                Some(s) => s,
                None => break, // all streams exhausted
            };
            let (ts, value) = streams[s][cursor[s]];
            cursor[s] += 1;
            cycle_id += 1;

            // forward the source value into its target slots, stamping freshness
            for t in sources[s].targets.iter() {
                let nb = &mut nodes[t.node];
                nb.inputs[t.slot].push(value).expect("source kind checked at wiring");
                nb.slots[t.slot] = SlotState { fresh_at: cycle_id, last_ts: ts };
            }

            // evaluate in topological order; gate by firing; forward Some outputs
            let mut observed = None;
            for &nidx in topo.iter() {
                let fired = {
                    let nb = &nodes[nidx];
                    fires(&nb.firing, &nb.slots, cycle_id, ts)
                };
                if !fired {
                    continue; // hold: no eval, no push
                }
                let result = {
                    let nb = &mut nodes[nidx];
                    nb.node.eval(Ctx::new(&nb.inputs))
                };
                if nidx == observe {
                    observed = result;
                }
                if let Some(v) = result {
                    for e in out_edges[nidx].iter() {
                        let nb = &mut nodes[e.to];
                        nb.inputs[e.slot].push(v).expect("edge kind checked at wiring");
                        nb.slots[e.slot] = SlotState { fresh_at: cycle_id, last_ts: ts };
                    }
                }
            }
            out.push(observed);
        }
        out
    }
```

- [ ] **Step 7: Add the `fires` predicate and `group_id` helper**

In `crates/aura-engine/src/harness.rs`, immediately after the closing `}` of the
`impl Harness { ... }` block (after `run`), insert the two free functions:

```rust

/// The firing predicate (C5/C6): does this node re-evaluate this cycle? A node
/// fires when *any* of its input groups fires (OR). A `Firing::Any` input fires
/// the node when it is fresh this cycle (`fresh_at == cycle_id`). A barrier group
/// fires when >=1 member is fresh this cycle AND every member carries
/// `last_ts == ts` — the ">=1 fresh" clause is the once-per-timestamp guard (a
/// group completed in a prior cycle has no fresh member now, so it does not
/// re-fire).
fn fires(firing: &[Firing], slots: &[SlotState], cycle_id: u64, ts: Timestamp) -> bool {
    // mode A: any fire-on-any-fresh input that is 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: each distinct barrier group fires when complete this cycle
    if let Some(max_group) = firing.iter().filter_map(group_id).max() {
        for g in 0..=max_group {
            let mut has_member = false;
            let mut any_fresh = false;
            let mut all_at_ts = true;
            for (i, f) in firing.iter().enumerate() {
                if group_id(f) == Some(g) {
                    has_member = true;
                    if slots[i].fresh_at == cycle_id {
                        any_fresh = true;
                    }
                    if slots[i].last_ts != ts {
                        all_at_ts = false;
                    }
                }
            }
            if has_member && any_fresh && all_at_ts {
                return true;
            }
        }
    }
    false
}

/// The barrier group id of a firing policy, or `None` for mode A.
fn group_id(f: &Firing) -> Option<u8> {
    match f {
        Firing::Any => None,
        Firing::Barrier(g) => Some(*g),
    }
}
```

- [ ] **Step 8: Re-export `SourceSpec` and update the aura-engine module doc**

In `crates/aura-engine/src/lib.rs`, replace the module-doc lines 3-18:

```rust
//! Delivered in cycle 0003 — the harness and its deterministic single-source
//! run loop:
//!
//! - [`Harness`] — the closed root graph that runs (a flat node array + an index
//!   edge table, topologically ordered) and its deterministic `run` loop: one
//!   source driven through a wired DAG of nodes, cycle by cycle, each output
//!   forwarded into its consumers' input columns (C1/C4/C7/C8/C9).
//! - [`Edge`] / [`Target`] — producer->consumer and source->consumer wiring.
//! - [`BootstrapError`] — wiring faults caught once, at bootstrap (kind
//!   mismatch, bad index, directed cycle).
//!
//! Still to come (subsequent cycles): freshness-gated recompute / sample-and-hold
//! (C5), the ingestion boundary (k-way merge of timestamped sources, C3), the
//! broker-independent position-event output and downstream broker nodes (C10),
//! and the atomic sim unit `(topology + params + data-window + seed) -> metrics`
//! that the sweep / optimize / walk-forward / Monte-Carlo axes orchestrate.
```

with:

```rust
//! Delivered across cycles 0003-0004 — the harness and its deterministic run
//! loop:
//!
//! - [`Harness`] — the closed root graph that runs (a flat node array + an index
//!   edge table, topologically ordered) and its deterministic `run` loop: a
//!   k-way merge of timestamped sources (C3/C4) driven through a wired DAG of
//!   nodes, cycle by cycle, with freshness-gated recompute and the two firing
//!   policies (C5/C6) deciding when each node re-evaluates and what it holds.
//! - [`Edge`] / [`Target`] / [`SourceSpec`] — producer->consumer wiring,
//!   source->consumer wiring, and a declared source (its kind + target slots).
//! - [`BootstrapError`] — wiring faults caught once, at bootstrap (kind
//!   mismatch, bad index, directed cycle).
//!
//! Still to come (subsequent cycles): the `Source` trait + data-server ingestion
//! and source-native time normalization (C3/C11), the broker-independent
//! position-event output and downstream broker nodes (C10), and the atomic sim
//! unit `(topology + params + data-window + seed) -> metrics` that the sweep /
//! optimize / walk-forward / Monte-Carlo axes orchestrate.
```

Then replace the re-export line 24:

```rust
pub use harness::{BootstrapError, Edge, Harness, Target};
```

with:

```rust
pub use harness::{BootstrapError, Edge, Harness, SourceSpec, Target};
```

- [ ] **Step 9: Partial build gate (library only; the test module is rewritten in Task 3)**

Run: `cargo build -p aura-engine --lib`
Expected: 0 errors. This compiles the aura-engine library without its
`#[cfg(test)]` module (which still references the old `bootstrap`/`run` signatures
and is rewritten in Task 3). Do **not** run `cargo test` or
`cargo build --all-targets` in this task — the test module does not compile yet,
by design.

Run: `cargo clippy -p aura-engine --lib -- -D warnings`
Expected: 0 warnings.

---

## Task 3: aura-engine tests — rewrite the 5 existing onto the new API, add the firing rails

**Files:**
- Modify: `crates/aura-engine/src/harness.rs` (the `#[cfg(test)] mod tests` block, lines 205-328)

- [ ] **Step 1: Replace the entire test module**

In `crates/aura-engine/src/harness.rs`, replace the whole `#[cfg(test)] mod tests`
block (lines 205-328, from `#[cfg(test)]` to the final closing `}` of the module)
with:

```rust
#[cfg(test)]
mod tests {
    use super::*;
    // InputSpec / NodeSchema are not imported by harness.rs production code (it
    // only reads `nd.schema()` fields, never naming the types), so they are not
    // brought in by `use super::*` — the fixtures construct them, so import here.
    use aura_core::{InputSpec, NodeSchema};
    use aura_std::{Sma, Sub};

    /// Build an f64 source stream from (timestamp, value) points.
    fn f64_stream(points: &[(i64, f64)]) -> Vec<(Timestamp, Scalar)> {
        points.iter().map(|&(t, v)| (Timestamp(t), Scalar::F64(v))).collect()
    }

    // --- firing-policy fixtures (test-local; not library nodes — C9: examples
    // for the engine's own tests, no speculative aura-std surface) ---

    /// Mode A as-of join: a 2-input f64 sum that fires whenever either input is
    /// fresh, holding the other. Warm-up returns None until both have a value.
    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;
            }
            Some(Scalar::F64(a[0] + b[0]))
        }
    }

    /// Mode B barrier join: a 2-input f64 sum that fires only when both inputs
    /// share the current cycle timestamp (both warm by construction when it fires).
    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]))
        }
    }

    /// Mixed A+B: barrier pair (inputs 0,1 in group 0) plus an as-of input
    /// (input 2). Fires when the pair completes (holding input 2) OR when input 2
    /// ticks (holding the pair) — the OR-combine.
    struct MixedSum;
    impl Node for MixedSum {
        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) },
                    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);
            let c = ctx.f64_in(2);
            if a.is_empty() || b.is_empty() || c.is_empty() {
                return None;
            }
            Some(Scalar::F64(a[0] + b[0] + c[0]))
        }
    }

    #[test]
    fn chain_source_sma_runs() {
        // node 0 = SMA(3); one source -> SMA(3).in0; observe node 0
        let mut h = Harness::bootstrap(
            vec![Box::new(Sma::new(3))],
            vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
            vec![],
            0,
        )
        .expect("valid");
        let out = h.run(vec![f64_stream(&[(1, 1.0), (2, 2.0), (3, 3.0), (4, 4.0), (5, 5.0)])]);
        assert_eq!(
            out,
            vec![
                None,
                None,
                Some(Scalar::F64(2.0)),
                Some(Scalar::F64(3.0)),
                Some(Scalar::F64(4.0)),
            ]
        );
    }

    #[test]
    fn fan_out_join_dag_runs_deterministically() {
        // 0 = SMA(2), 1 = SMA(4), 2 = Sub; one source fans into both SMAs (all
        // Any), SMAs join into Sub — the 0003 compat baseline on the new API.
        let build = || {
            Harness::bootstrap(
                vec![Box::new(Sma::new(2)), Box::new(Sma::new(4)), Box::new(Sub::new())],
                vec![SourceSpec {
                    kind: ScalarKind::F64,
                    targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
                }],
                vec![Edge { from: 0, to: 2, slot: 0 }, Edge { from: 1, to: 2, slot: 1 }],
                2,
            )
            .expect("valid DAG")
        };
        let prices = f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0), (6, 20.0)]);

        let mut h = build();
        let out = h.run(vec![prices.clone()]);
        // Sub fires once SMA(4) is warm (cycle 4): 15-13, 17-15, 19-17 -> 2.
        assert_eq!(
            out,
            vec![
                None,
                None,
                None,
                Some(Scalar::F64(2.0)),
                Some(Scalar::F64(2.0)),
                Some(Scalar::F64(2.0)),
            ]
        );

        // determinism (C1): a second identical run is bit-identical
        let mut h2 = build();
        assert_eq!(h2.run(vec![prices]), out);
    }

    #[test]
    fn mode_a_as_of_fires_on_any_fresh_and_holds() {
        // source 0 ticks t=1,2,3,4; source 1 ticks t=2,4 (slower); both inputs Any.
        let build = || {
            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![],
                0,
            )
            .expect("valid")
        };
        let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]);
        let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]);

        let mut h = build();
        let out = h.run(vec![s0.clone(), s1.clone()]);
        // holds s1=100 across t=3 and the t=4 s0-cycle; emits on every tick once warm.
        assert_eq!(
            out,
            vec![
                None,
                None,
                Some(Scalar::F64(120.0)),
                Some(Scalar::F64(130.0)),
                Some(Scalar::F64(140.0)),
                Some(Scalar::F64(240.0)),
            ]
        );

        let mut h2 = build();
        assert_eq!(h2.run(vec![s0, s1]), out); // deterministic
    }

    #[test]
    fn mode_b_barrier_fires_only_on_timestamp_coincidence() {
        // identical wiring to mode A, but both inputs are Barrier(0).
        let build = || {
            Harness::bootstrap(
                vec![Box::new(BarrierSum)],
                vec![
                    SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] },
                    SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] },
                ],
                vec![],
                0,
            )
            .expect("valid")
        };
        let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]);
        let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]);

        let mut h = build();
        let out = h.run(vec![s0.clone(), s1.clone()]);
        // emits ONLY at t=2 and t=4 where both inputs share the timestamp; holds otherwise.
        assert_eq!(
            out,
            vec![
                None,
                None,
                Some(Scalar::F64(120.0)),
                None,
                None,
                Some(Scalar::F64(240.0)),
            ]
        );

        let mut h2 = build();
        assert_eq!(h2.run(vec![s0, s1]), out); // deterministic
    }

    #[test]
    fn mixed_a_and_b_or_combine_on_one_node() {
        // in0,in1 = barrier group 0 (sources 0,1); in2 = as-of (source 2).
        let mut h = Harness::bootstrap(
            vec![Box::new(MixedSum)],
            vec![
                SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] },
                SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] },
                SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 2 }] },
            ],
            vec![],
            0,
        )
        .expect("valid");
        let s0 = f64_stream(&[(2, 20.0), (5, 50.0)]); // in0 (barrier)
        let s1 = f64_stream(&[(2, 200.0)]);           // in1 (barrier)
        let s2 = f64_stream(&[(1, 1.0), (3, 3.0)]);   // in2 (as-of)

        let out = h.run(vec![s0, s1, s2]);
        // cycle order (ts, source idx): (1,s2) (2,s0) (2,s1) (3,s2) (5,s0).
        // c3: barrier pair completes at t=2, holds c=1 -> 20+200+1 = 221.
        // c4: as-of input ticks at t=3, holds the pair -> 20+200+3 = 223.
        // c1 fires on the as-of input but filters (pair cold); c2,c5 hold.
        assert_eq!(
            out,
            vec![
                None,
                None,
                Some(Scalar::F64(221.0)),
                Some(Scalar::F64(223.0)),
                None,
            ]
        );
    }

    #[test]
    fn bootstrap_rejects_a_cycle() {
        // two SMA(1) nodes wired a -> b -> a
        let err = Harness::bootstrap(
            vec![Box::new(Sma::new(1)), Box::new(Sma::new(1))],
            vec![],
            vec![Edge { from: 0, to: 1, slot: 0 }, Edge { from: 1, to: 0, slot: 0 }],
            0,
        )
        .unwrap_err();
        assert_eq!(err, BootstrapError::Cycle);
    }

    #[test]
    fn bootstrap_rejects_a_kind_mismatch() {
        // SMA(1) declares an f64 input; an i64 source mismatches
        let err = Harness::bootstrap(
            vec![Box::new(Sma::new(1))],
            vec![SourceSpec { kind: ScalarKind::I64, targets: vec![Target { node: 0, slot: 0 }] }],
            vec![],
            0,
        )
        .unwrap_err();
        assert_eq!(
            err,
            BootstrapError::KindMismatch { producer: ScalarKind::I64, consumer: ScalarKind::F64 }
        );
    }

    #[test]
    fn bootstrap_rejects_a_bad_index() {
        // observe node 5 does not exist
        let err = Harness::bootstrap(
            vec![Box::new(Sma::new(1))],
            vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
            vec![],
            5,
        )
        .unwrap_err();
        assert_eq!(err, BootstrapError::BadIndex);
    }
}
```

- [ ] **Step 2: Full workspace build**

Run: `cargo build --workspace --all-targets`
Expected: 0 errors (the test module now compiles against the new API).

- [ ] **Step 3: Full workspace test suite**

Run: `cargo test --workspace`
Expected: PASS, 0 failed. Per-crate: aura-core 19, aura-std 3, aura-engine 8
(30 total). The 8 aura-engine tests are: `chain_source_sma_runs`,
`fan_out_join_dag_runs_deterministically`,
`mode_a_as_of_fires_on_any_fresh_and_holds`,
`mode_b_barrier_fires_only_on_timestamp_coincidence`,
`mixed_a_and_b_or_combine_on_one_node`, `bootstrap_rejects_a_cycle`,
`bootstrap_rejects_a_kind_mismatch`, `bootstrap_rejects_a_bad_index`.

- [ ] **Step 4: Lint gate**

Run: `cargo clippy --workspace --all-targets -- -D warnings`
Expected: 0 warnings.

- [ ] **Step 5: Surface-purity grep**

Run: `grep -rnE 'RefCell|Rc<|dyn Any' crates/*/src`
Expected: no output (exit code 1) — the harness module doc phrases RustAst's model
as "reference-counted, interior-mutable" without the literal `Rc<`/`RefCell`/`dyn
Any` tokens, and no production code uses them.
