fix(std): snap time-bucket/session reads to the nominal minute

Provider M1 stamps carry sub-second jitter around minute boundaries
(~55% of days stamp the 09:00 Berlin bar at 08:59:59.xxx), so any
consumer dividing or truncating the raw stamp mis-buckets boundary
bars: Resample and VolTfStop fold the bar into the previous bucket
instead of rolling over, Session demotes the first in-session bar to
at-open.

Fix per the decision logged on the issue: one shared
Timestamp::snap_to_nearest_minute in aura-core (round-half-up,
rem_euclid so negative epochs snap correctly), applied consumer-side
at the three membership computations. The recorded stream stays
byte-verbatim — no ingestion rewrite, record-then-replay untouched;
an ingestion-time rewrite was rejected because it would hide the raw
time axis from every consumer and could reorder the k-way merge
around boundaries.

Verified: the three formerly-RED tests pass individually; full
workspace suite green; clippy clean.

closes #280

crates/aura-core/src/scalar.rs

@@ -11,6 +11,32 @@ use crate::cell::Cell;
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize)]
 pub struct Timestamp(pub i64);
 
+impl Timestamp {
+    /// Nanoseconds per minute — the one scale factor `snap_to_nearest_minute`
+    /// rounds against.
+    const NS_PER_MINUTE: i64 = 60 * 1_000_000_000;
+
+    /// Snap this instant to the nearest whole-minute boundary (round-half-up,
+    /// ties toward the later minute) — the sub-second-provider-jitter
+    /// correction (#280): a bar's nominal minute can be a bucket/session
+    /// boundary while the raw provider stamp lands a fraction of a second
+    /// early or late. Consumers deriving bucket membership or wall-clock
+    /// minutes from `ctx.now()` snap the read here, once, rather than
+    /// dividing/truncating the raw jittered stamp — the recorded stream
+    /// itself stays byte-verbatim (record-then-replay, C6); only the
+    /// consumer-side read is corrected.
+    pub fn snap_to_nearest_minute(self) -> Timestamp {
+        let m = Self::NS_PER_MINUTE;
+        let rem = self.0.rem_euclid(m);
+        let floor = self.0 - rem;
+        if rem * 2 >= m {
+            Timestamp(floor + m)
+        } else {
+            Timestamp(floor)
+        }
+    }
+}
+
 /// The kind tag of a scalar / column, used for the edge-time type check (C7).
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
 pub enum ScalarKind {

crates/aura-std/src/resample.rs

@@ -104,7 +104,7 @@ impl Node for Resample {
     fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
         // Barrier(0) guarantees all four arrive co-fresh; read the newest M1.
         let (o, h, l, c) = (ctx.f64_in(0)[0], ctx.f64_in(1)[0], ctx.f64_in(2)[0], ctx.f64_in(3)[0]);
-        let bucket = ctx.now().0 / self.period_ns;
+        let bucket = ctx.now().snap_to_nearest_minute().0 / self.period_ns;
 
         match self.acc.take() {
             // first ever sample: start the accumulator, nothing complete yet.

crates/aura-std/src/session.rs

@@ -118,7 +118,7 @@ impl Node for Session {
         // tz-aware LOCAL wall-clock — chrono-tz applies the correct
         // DST offset for this date, so the same local minute reads the same bar
         // index in summer (CEST) and winter (CET).
-        let local = self.tz.timestamp_nanos(ctx.now().0);
+        let local = self.tz.timestamp_nanos(ctx.now().snap_to_nearest_minute().0);
         let mins = local.hour() as i64 * 60 + local.minute() as i64 - self.open_minutes;
         // i64 division: in-session closes land on exact period boundaries, so the
         // value is the bar index; pre-open is <= 0 (non-actionable).

crates/aura-std/src/vol_tf_stop.rs

@@ -90,7 +90,7 @@ impl Node for VolTfStop {
             return None; // fire with price (warm-up filter)
         }
         let price = w[0];
-        let bucket = ctx.now().0 / self.period_ns;
+        let bucket = ctx.now().snap_to_nearest_minute().0 / self.period_ns;
         match self.bucket {
             // First ever sample: open the accumulator, nothing complete yet.
             None => {