fix(history): stop reprocessing the slice that was just appended to

`add_events_with_prior` advanced `k` past the slice it had written when it
created a new one, but not when it appended to an existing one. The trailing
forward-refresh loop therefore started *on* the slice just modified and ran
`new_forward_info` over it again.

That is not merely redundant work. The loop immediately above it sets each
agent's message to `forward * likelihood` for that slice, and
`new_forward_info` then assigns `skill.forward = message.forget(drift)` —
folding the slice's own likelihood back into its own forward prior. The
skills it produced depended on how events had been batched.

Ingesting one event at a time now converges to the same fixed point as
ingesting the same events in a single call, which it previously did not:
for five events sharing a timestamp, competitor `a` converged to
mu=7.44 sigma=3.90 batched versus mu=7.99 sigma=3.10 incrementally. Both
runs had converged; the gap was not a convergence residual.

The numerical goldens never caught this because they all ingest in one call
with a distinct timestamp per event, so the append-to-existing-slice branch
is never taken. `tests/ingestion_equivalence.rs` covers it directly, and
asserts convergence before comparing so that a residual cannot be mistaken
for agreement.

Removing the redundant re-inference also removes the dominant cost of
incremental ingestion, which was quadratic in the number of events already
in the slice:

    events   before     after    speedup
       500   45.8ms     1.1ms       42x
      1000  179.5ms     2.8ms       64x
      2000  721.8ms     9.9ms       73x
      4000    2.9s     35.4ms       82x

Ingesting one at a time is now 1.8x a single batched call, down from 148x.

Two supporting changes are included:

- Color groups are rebuilt lazily rather than on every append. Nothing
  reads the partition between an append and the next full sweep, so the
  per-append rebuild was pure waste.
- `ColorGroups::groups_are_contiguous` is asserted after each rebuild and
  in `color_range`. The parallel sweep derives one `&mut` sub-slice per
  color from those ranges and relies on them being disjoint; that invariant
  was established by construction but never checked.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DnsaJg74eNSva3PJjK2eej

src/color_group.rs

@@ -49,16 +49,53 @@ impl ColorGroups {
 
     /// Contiguous index range for one color after events have been reordered
     /// into color-contiguous positions by `TimeSlice::recompute_color_groups`.
-    #[allow(dead_code)]
     pub(crate) fn color_range(&self, color_idx: usize) -> std::ops::Range<usize> {
         let group = &self.groups[color_idx];
         if group.is_empty() {
             return 0..0;
         }
+
         let start = *group.first().unwrap();
         let end = *group.last().unwrap() + 1;
+
+        debug_assert_eq!(
+            end - start,
+            group.len(),
+            "color {color_idx} is not contiguous; its range would overlap other colors"
+        );
+
         start..end
     }
+
+    /// Whether every color occupies a contiguous, ascending range of event
+    /// indices, and no two colors overlap.
+    ///
+    /// The parallel sweep derives one `&mut` sub-slice per color from these
+    /// ranges and relies on them being disjoint. That disjointness is what
+    /// makes concurrent writes to distinct skills sound, so it is checked
+    /// rather than assumed.
+    pub(crate) fn groups_are_contiguous(&self) -> bool {
+        let mut expected_start = 0;
+
+        for group in &self.groups {
+            if group.is_empty() {
+                continue;
+            }
+
+            let ascending_run = group
+                .iter()
+                .enumerate()
+                .all(|(offset, &idx)| idx == group[0] + offset);
+
+            if !ascending_run || group[0] != expected_start {
+                return false;
+            }
+
+            expected_start += group.len();
+        }
+
+        true
+    }
 }
 
 /// Compute color groups greedily.

src/history.rs

@@ -646,6 +646,8 @@ impl<T: Time, D: Drift<T>, O: Observer<T>, K: Eq + Hash + Clone> History<T, D, O
                     agent.last_time = Some(t);
                     agent.message = time_slice.forward_prior_out(&agent_idx);
                 }
+
+                k += 1;
             } else {
                 let mut time_slice = TimeSlice::new(t, self.p_draw, self.convergence);
                 time_slice.add_events(composition, results, weights, kinds_chunk, &self.agents);

src/key_table.rs

@@ -12,59 +12,68 @@ use crate::Index;
 /// crate. Power users can promote `&K` to `Index` via `get_or_create` and
 /// skip the lookup on subsequent hot-path calls.
 #[derive(Debug)]
-pub struct KeyTable<K>(HashMap<K, Index>);
+pub struct KeyTable<K> {
+    forward: HashMap<K, Index>,
+    /// Reverse mapping, indexed by `Index.0`.
+    ///
+    /// Indices are handed out densely and sequentially, so position *is* the
+    /// index and `key()` is a lookup rather than a scan over every entry.
+    reverse: Vec<K>,
+}
 
 impl<K> KeyTable<K>
 where
-    K: Eq + Hash,
+    K: Eq + Hash + Clone,
 {
     pub fn new() -> Self {
-        Self(HashMap::new())
+        Self {
+            forward: HashMap::new(),
+            reverse: Vec::new(),
+        }
     }
 
     pub fn get<Q: ?Sized + Hash + Eq>(&self, k: &Q) -> Option<Index>
     where
         K: Borrow<Q>,
     {
-        self.0.get(k).cloned()
+        self.forward.get(k).cloned()
     }
 
     pub fn get_or_create<Q: ?Sized + Hash + Eq + ToOwned<Owned = K>>(&mut self, k: &Q) -> Index
     where
         K: Borrow<Q>,
     {
-        if let Some(idx) = self.0.get(k) {
+        if let Some(idx) = self.forward.get(k) {
             *idx
         } else {
-            let idx = Index::from(self.0.len());
-            self.0.insert(k.to_owned(), idx);
+            let idx = Index::from(self.reverse.len());
+            let owned = k.to_owned();
+            self.reverse.push(owned.clone());
+            self.forward.insert(owned, idx);
             idx
         }
     }
 
     pub fn key(&self, idx: Index) -> Option<&K> {
-        self.0
-            .iter()
-            .find(|&(_, value)| *value == idx)
-            .map(|(key, _)| key)
+        self.reverse.get(idx.0)
     }
 
     pub fn keys(&self) -> impl Iterator<Item = &K> {
-        self.0.keys()
+        self.forward.keys()
     }
 
     pub fn len(&self) -> usize {
-        self.0.len()
+        self.reverse.len()
     }
 
     pub fn is_empty(&self) -> bool {
-        self.0.is_empty()
+        self.reverse.is_empty()
     }
 }
 
 impl<K> Default for KeyTable<K>
 where
-    K: Eq + Hash,
+    K: Eq + Hash + Clone,
 {
     fn default() -> Self {
         KeyTable::new()

src/time_slice.rs

@@ -174,6 +174,14 @@ pub struct TimeSlice<T: Time = i64> {
     pub(crate) convergence: crate::ConvergenceOptions,
     arena: ScratchArena,
     pub(crate) color_groups: ColorGroups,
+    /// Whether `color_groups` still reflects `events`.
+    ///
+    /// Coloring is rebuilt lazily, on the first full sweep after an append,
+    /// rather than eagerly per append: the partition is thrown away and
+    /// recomputed wholesale either way, so doing it per append made ingesting
+    /// n events O(n^2) with no benefit — nothing reads the partition between
+    /// an append and the next full sweep.
+    color_groups_dirty: bool,
 }
 
 impl<T: Time> TimeSlice<T> {
@@ -186,6 +194,7 @@ impl<T: Time> TimeSlice<T> {
             convergence,
             arena: ScratchArena::new(),
             color_groups: ColorGroups::new(),
+            color_groups_dirty: false,
         }
     }
 
@@ -198,6 +207,7 @@ impl<T: Time> TimeSlice<T> {
         let n = self.events.len();
         if n == 0 {
             self.color_groups = ColorGroups::new();
+            self.color_groups_dirty = false;
             return;
         }
 
@@ -221,6 +231,12 @@ impl<T: Time> TimeSlice<T> {
 
         self.events = reordered;
         self.color_groups = ColorGroups { groups: new_groups };
+        self.color_groups_dirty = false;
+
+        debug_assert!(
+            self.color_groups.groups_are_contiguous(),
+            "color groups must occupy contiguous event ranges"
+        );
     }
 
     pub fn add_events<D: Drift<T>>(
@@ -306,8 +322,9 @@ impl<T: Time> TimeSlice<T> {
 
         self.events.extend(events);
 
+        self.color_groups_dirty = true;
+
         self.iteration(from, agents);
-        self.recompute_color_groups();
     }
 
     pub(crate) fn posteriors(&self) -> HashMap<Index, Gaussian> {
@@ -318,6 +335,10 @@ impl<T: Time> TimeSlice<T> {
     }
 
     pub fn iteration<D: Drift<T>>(&mut self, from: usize, agents: &CompetitorStore<T, D>) {
+        if from == 0 && self.color_groups_dirty {
+            self.recompute_color_groups();
+        }
+
         if from > 0 || self.color_groups.is_empty() {
             // Initial pass (add_events) or no color groups yet: simple sequential sweep.
             for event in self.events.iter_mut().skip(from) {