TECH-3501 FRD: stream_mixing + ledger observability metrics

Why the naive join fails. The ledger keys on group_id: int (extracted by the processor at processors/sc_multimodal/processor.py:172 via element.get_group_id() — downstream of the dataloader). The dataset keys on stream_id: int (positional index into stream_paths) and the path basename string. There is no stream_id → group_id mapping; one stream contains many cells contains many gids.

The inverted idea. Don't try to track "currently active gids" — that's hard and expensive. Instead, track "recently evicted gids" on the ledger side. When a fresh-write happens (the dataset sent us a gid the ledger hadn't seen, or had seen and evicted), check whether we just evicted that gid. If yes → false eviction.

Where the join happens. One single instrumentation site: _store_state at recurrent_encoder.py:1115:

prev = self.group_state_ledger.get(group_id)
+ if prev is None:
+     mode = self._recently_evicted.pop(group_id, None)
+     if mode is not None:
+         self._false_evict_counts[mode] += 1
# ... existing fresh-init logic continues ...

Owner-locality. _store_state is owner-only — enforced at L1108-1114 (raise if non-owner attempts a store). The three eviction sites (LRU at L1193, STALE at L660, full-wipe at L635) all operate on the local shard. So eviction and re-store both happen on the owning rank for any given gid. The intra-rank join is exact, not an approximation; there is no cross-rank gap.

Why _lookup_local at L893 is not the instrumentation site. The miss path at L893 (if group_id not in self.group_state_ledger) fires on every rank for every non-owned gid, since non-owners never hold the entry. Using L893 would over-count by world_size and conflate "non-owner doesn't shard this gid" with "owner evicted it." _store_state is owner-gated by construction, which is exactly what we want.

Memory. _recently_evicted grows on eviction, pops on re-store, clears on full-wipe. Bounded by "currently-evicted but not yet re-encountered" gids — at most unique_gids_owned − len(ledger). Empirically a few hundred ints in steady state.

Distributed semantics. The counters are per-rank-owned-gid quantities. After SUM-mode reduction in reduce_loss_metrics, the wandb keys show cluster-wide totals (the natural reading: "the cluster suffered N false evictions this step"). The paired _max shows the worst single rank — useful for hotspot diagnosis (one rank with a hostile gid distribution).

Mirrors DataStatsCollector.aggregate_distributed at data_stats.py:170. Pseudocode:

def aggregate_step_metrics(local, world_size, device):
    if world_size == 1:
        return local                              # identity, no collectives

    # Pack scalars by op-kind into typed tensors of fixed shape.
    # Modality count is sparse — flatten over the project's known modality enum
    # (matches data_stats.py:209) so all ranks have identical shapes.
    all_modalities = sorted(MODALITY_TO_INT.keys())

    mins = torch.tensor([local["remaining_min"],
                         local["remaining_fraction_min"]],
                        dtype=torch.float32, device=device)
    maxs = torch.tensor([local["remaining_max"],
                         local["remaining_fraction_max"],
                         local["steps_since_pick_max"]],
                        dtype=torch.float32, device=device)
    sums = torch.zeros(len(all_modalities) + 1,  # +1 for exhaust_events
                       dtype=torch.float32, device=device)
    for i, m in enumerate(all_modalities):
        sums[i] = local["modalities"].get(m, 0)
    sums[-1] = local["exhaust_events"]

    # Empty-pool ranks contribute sentinels (per data_stats.py:232).
    if local_active_empty:
        mins.fill_(float("inf"))
        maxs.fill_(0.0)

    dist.all_reduce(mins, op=dist.ReduceOp.MIN)
    dist.all_reduce(maxs, op=dist.ReduceOp.MAX)
    dist.all_reduce(sums, op=dist.ReduceOp.SUM)
    # Unpack + emit only non-zero modality fan-outs.
    ...

Three collectives per log step. Each is a small fixed-shape tensor; total bytes << gradient all-reduces happening every step. Negligible.

Per-step additions to the pick site:

• remaining_picks.min/max on a size-K array (K ≤ 8): < 1 µs.
• remaining_fraction per reader (K subtractions + divisions): < 2 µs.
• Counter(_modality_for_reader(r) for r in active) with K ≤ 8: < 5 µs.
• self._last_picked_step[stream_id] = current_step + max-over-active: < 1 µs.
• Dict population (≈ 10 keys): < 1 µs.
• Aggregator call (only on log steps, ≈ every 10 steps): three small all_reduces, amortised < 5 µs/step.
• Ledger eviction sites: one dict insert per evicted gid. STALE sweep amortised over expected_ledger_wipe_frequency steps.
• Ledger fresh-write check at L1115: one dict pop (O(1)). Sub-microsecond.

Total << 50 µs budget per step. No tensor ops in the hot path beyond what PR #820 already does. No GPU sync, no CPU↔GPU transfers.

The dataset already documents at dataset.py:79-82 that mid-epoch resume requires num_workers=0. The drain_step_metrics() + aggregate_step_metrics() APIs inherit this — both work because the dataset lives in the main process and the trainer reads its attributes directly. If num_workers > 0 support is ever added, drain semantics will need a queue or batch-dict piggyback design (out of scope for v1).

Two new test files (see FR14, FR15) and one deletion (FR7 removes three obsolete tests).