R7 (partial): offline detector calibration — entropy + bridge-correlation AUC

Land only the offline-validatable half of R7 (instrument-validation gate items
3 and 4): the detectors the gate calibrates on synthetic fixtures, never on
confirmatory-cell data. Pure stdlib functions over in-memory series — no pcaps,
no engine, no traffic, no VM fabric.

- analysis/detectors.py: shannon_entropy_bits (RQ2 anonymity-set entropy),
  pearson/score_matrix/auc/linkage_auc, bridge_correlation_auc (RQ1
  linkability scorer), and synthetic_bridge_fixture (seed-deterministic
  known-linked / known-unlinked ground truth via the R1 SorRng).

Calibration green: entropy returns exactly log2(N) for N equiprobable senders
(gate item 4); a known-linked control pair scores AUC=1.0 and the unlinked
estimator is unbiased at chance (ensemble mean over 40 seeds = 0.498 ~ 0.5,
gate item 3). Detectors are calibrated on synthetic fixtures only — no fitting.

The traffic-moving R7 pieces (churn.py VM spin/kill, live selector rebuild loop,
metrics.json emission) are HELD for R4/R6 + a live grid and are absent here.
Python SOR suite 66 passed.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

cmd_chat/sor/analysis/__init__.py

@@ -0,0 +1,18 @@
+"""R7 (partial) — offline analysis calibration primitives.
+
+This package holds only the *offline-validatable* half of R7: the detectors that
+the instrument-validation gate calibrates on **synthetic fixtures** (never on
+confirmatory-cell data — CLAUDE.md build discipline). Specifically:
+
+- ``detectors.shannon_entropy_bits`` — the RQ2 anonymity-set entropy metric; gate
+  item 4 requires it to return ``log2(N)`` for ``N`` equiprobable senders.
+- ``detectors.bridge_correlation_auc`` — the RQ1 bridge-linkability scorer; gate
+  item 3 requires AUC ≈ 1 on a known-linked control pair and ≈ 0.5 on a
+  known-unlinked pair.
+
+These are pure functions over in-memory series/distributions — they read no
+pcaps, spawn no engine, move no traffic, and touch no VM fabric. The
+traffic-moving R7 pieces (``churn.py`` seeded VM spin/kill; the live selector
+rebuild loop; the confirmatory battery that writes ``metrics.json``) are NOT
+here — they are HELD pending R4/R6 + a live grid (see OVERSEER-STATUS.md).
+"""

cmd_chat/sor/analysis/detectors.py

@@ -0,0 +1,132 @@
+"""R7 (partial) — offline detector primitives + seeded synthetic fixtures.
+
+Pure, stdlib-only calibration instruments (gate items 3 and 4). No I/O, no
+engine, no pcaps: they operate on in-memory numeric series/distributions so they
+can be validated entirely offline against synthetic ground truth. Determinism is
+inherited from the R1 ``SorRng`` so a calibration fixture is reproducible from
+its seed alone.
+"""
+
+from __future__ import annotations
+
+import math
+from typing import Dict, List, Sequence, Union
+
+from cmd_chat.sor.config import Domain, SorRng
+
+Number = Union[int, float]
+Series = Sequence[Number]
+
+
+# --------------------------------------------------------------------------- #
+# RQ2 — anonymity-set entropy (gate item 4).
+# --------------------------------------------------------------------------- #
+def shannon_entropy_bits(weights: Union[Dict[object, Number], Sequence[Number]]) -> float:
+    """Shannon entropy in **bits** of an observed sender distribution.
+
+    ``weights`` is either a mapping ``sender -> count`` or a sequence of
+    non-negative weights. For ``N`` equiprobable senders this returns exactly
+    ``log2(N)`` (the maximum for an N-set), which is the gate item 4 predicate.
+    An empty or all-zero distribution has entropy 0."""
+    vals = list(weights.values()) if isinstance(weights, dict) else list(weights)
+    total = float(sum(vals))
+    if total <= 0:
+        return 0.0
+    h = 0.0
+    for c in vals:
+        if c > 0:
+            p = c / total
+            h -= p * math.log2(p)
+    return h
+
+
+# --------------------------------------------------------------------------- #
+# RQ1 — bridge-linkability correlation scorer (gate item 3).
+# --------------------------------------------------------------------------- #
+def pearson(xs: Series, ys: Series) -> float:
+    """Pearson correlation coefficient of two equal-length numeric series.
+    Returns 0.0 for empty, mismatched-length, or zero-variance inputs (a flat
+    series carries no linkage signal)."""
+    n = len(xs)
+    if n == 0 or n != len(ys):
+        return 0.0
+    mx = sum(xs) / n
+    my = sum(ys) / n
+    num = sum((x - mx) * (y - my) for x, y in zip(xs, ys))
+    dx = math.sqrt(sum((x - mx) ** 2 for x in xs))
+    dy = math.sqrt(sum((y - my) ** 2 for y in ys))
+    if dx == 0.0 or dy == 0.0:
+        return 0.0
+    return num / (dx * dy)
+
+
+def score_matrix(ingress: Sequence[Series], egress: Sequence[Series]) -> List[List[float]]:
+    """Full pairwise linkage-score matrix ``S[i][j] = pearson(ingress[i],
+    egress[j])``. The candidate true pairing is the diagonal (``i == j``)."""
+    return [[pearson(a, b) for b in egress] for a in ingress]
+
+
+def auc(pos: Series, neg: Series) -> float:
+    """Rank AUC = P(pos ranked above neg) with ties counted as 0.5 (Mann-Whitney
+    U normalized). AUC 1.0 = perfect separation, 0.5 = chance. Empty side -> 0.5."""
+    if not pos or not neg:
+        return 0.5
+    wins = 0.0
+    for p in pos:
+        for q in neg:
+            if p > q:
+                wins += 1.0
+            elif p == q:
+                wins += 0.5
+    return wins / (len(pos) * len(neg))
+
+
+def linkage_auc(scores: Sequence[Sequence[float]]) -> float:
+    """AUC of the diagonal (linked) scores against the off-diagonal (unlinked)
+    scores of a square score matrix — the RQ1 bridge-correlation AUC."""
+    n = len(scores)
+    pos = [scores[i][i] for i in range(n)]
+    neg = [scores[i][j] for i in range(n) for j in range(n) if i != j]
+    return auc(pos, neg)
+
+
+def bridge_correlation_auc(ingress: Sequence[Series], egress: Sequence[Series]) -> float:
+    """RQ1 scorer: how well the correlator links each ingress flow to its true
+    egress flow, as AUC over all candidate pairings. ≈1 for a linked bridge,
+    ≈0.5 for an unlinked one (gate item 3)."""
+    return linkage_auc(score_matrix(ingress, egress))
+
+
+# --------------------------------------------------------------------------- #
+# Seeded synthetic calibration fixtures (ground truth known by construction).
+# --------------------------------------------------------------------------- #
+def synthetic_bridge_fixture(
+    seed: int,
+    n_flows: int = 8,
+    bins: int = 64,
+    jitter: int = 5,
+    linked: bool = True,
+) -> tuple[List[List[int]], List[List[int]]]:
+    """Deterministically build ``(ingress, egress)`` flow sets of per-bin byte
+    counts from ``seed`` alone (reuses the R1 ``SorRng`` streams).
+
+    - ``linked=True``: each egress flow is its ingress flow plus small independent
+      padding/latency ``jitter`` — the *known-linked* control (diagonal
+      correlation dominates -> AUC ≈ 1).
+    - ``linked=False``: egress flows are drawn from an independent domain stream,
+      uncorrelated with ingress — the *known-unlinked* control (no diagonal
+      advantage -> AUC ≈ 0.5).
+    """
+    rng = SorRng(seed)
+    sig = rng.stream(Domain.PATH)  # ingress signal
+    ingress = [[sig.next_below(1000) for _ in range(bins)] for _ in range(n_flows)]
+    if linked:
+        noise = rng.stream(Domain.PADDING)
+        span = 2 * jitter + 1
+        egress = [
+            [v + (noise.next_below(span) - jitter) for v in flow] for flow in ingress
+        ]
+    else:
+        alt = rng.stream(Domain.CHURN)  # independent of the PATH signal
+        egress = [[alt.next_below(1000) for _ in range(bins)] for _ in range(n_flows)]
+    return ingress, egress

tests/test_sor_analysis.py

@@ -0,0 +1,112 @@
+"""R7 (partial) — offline detector calibration checks (gate items 3 & 4).
+
+Acceptance predicate (roadmap R7, the offline-verifiable half):
+  - entropy metric returns log2(N) for N equiprobable senders   (gate item 4);
+  - the bridge-correlation scorer gives AUC ≈ 1 on a known-linked control pair
+    and ≈ 0.5 on a known-unlinked pair                          (gate item 3).
+
+Calibrated on synthetic fixtures only (CLAUDE.md: detectors are never fit to
+confirmatory-cell data). These detectors read no pcaps, spawn no engine, and
+move no traffic — fully offline. The traffic-moving R7 pieces (churn VM
+spin/kill, live selector rebuild, metrics.json emission) are HELD for R4/R6 + a
+live grid and are intentionally absent.
+"""
+
+import math
+import statistics
+
+import pytest
+
+from cmd_chat.sor.analysis.detectors import (
+    auc,
+    bridge_correlation_auc,
+    pearson,
+    shannon_entropy_bits,
+    synthetic_bridge_fixture,
+)
+
+
+# --------------------------------------------------------------------------- #
+# Gate item 4 — anonymity-set entropy.
+# --------------------------------------------------------------------------- #
+@pytest.mark.parametrize("n", [1, 2, 3, 4, 5, 8, 16, 17])
+def test_entropy_equiprobable_is_log2n(n):
+    # N equiprobable senders -> exactly log2(N) bits (the gate item 4 predicate).
+    h = shannon_entropy_bits({f"sender{i}": 1 for i in range(n)})
+    assert h == pytest.approx(math.log2(n), abs=1e-12)
+
+
+def test_entropy_is_maximized_by_the_uniform_distribution():
+    n = 8
+    uniform = shannon_entropy_bits([1] * n)
+    skewed = shannon_entropy_bits([10, 1, 1, 1, 1, 1, 1, 1])
+    assert skewed < uniform == pytest.approx(3.0)
+
+
+def test_entropy_degenerate_cases():
+    assert shannon_entropy_bits({"only": 5}) == 0.0  # one sender -> 0 bits
+    assert shannon_entropy_bits([]) == 0.0
+    assert shannon_entropy_bits([0, 0]) == 0.0
+
+
+def test_entropy_known_skewed_value():
+    # 3:1 split -> H = 0.75*log2(4/3) + 0.25*log2(4) = 0.811278...
+    assert shannon_entropy_bits([3, 1]) == pytest.approx(0.8112781, abs=1e-6)
+
+
+# --------------------------------------------------------------------------- #
+# Correlation + AUC primitives.
+# --------------------------------------------------------------------------- #
+def test_pearson_identities():
+    assert pearson([1, 2, 3, 4], [1, 2, 3, 4]) == pytest.approx(1.0)
+    assert pearson([1, 2, 3, 4], [4, 3, 2, 1]) == pytest.approx(-1.0)
+    assert pearson([1, 2, 3], [5, 5, 5]) == 0.0  # zero variance -> no signal
+    assert pearson([], []) == 0.0
+    assert pearson([1, 2], [1, 2, 3]) == 0.0  # length mismatch
+
+
+def test_auc_basics():
+    assert auc([3, 4, 5], [0, 1, 2]) == 1.0  # perfectly separated
+    assert auc([0, 1], [2, 3]) == 0.0  # perfectly inverted
+    assert auc([1, 1], [1, 1]) == 0.5  # all ties -> chance
+    assert auc([], [1]) == 0.5
+
+
+# --------------------------------------------------------------------------- #
+# Gate item 3 — bridge-correlation AUC calibration.
+# --------------------------------------------------------------------------- #
+def test_known_linked_pair_scores_auc_near_one():
+    # A linked bridge (egress = ingress + small jitter): the diagonal dominates.
+    ingress, egress = synthetic_bridge_fixture(42, linked=True)
+    assert bridge_correlation_auc(ingress, egress) == pytest.approx(1.0)
+    # Robust across seeds — never below near-perfect.
+    for seed in (1, 123456789, 777, 2024):
+        ig, eg = synthetic_bridge_fixture(seed, linked=True)
+        assert bridge_correlation_auc(ig, eg) >= 0.99
+
+
+def test_known_unlinked_pair_scores_auc_near_half():
+    # A single unlinked pair is noisy (finite flows); the estimator is *unbiased*,
+    # so the ensemble mean over many seeds sits at chance (0.5).
+    aucs = [
+        bridge_correlation_auc(*synthetic_bridge_fixture(seed, n_flows=24, linked=False))
+        for seed in range(40)
+    ]
+    assert statistics.mean(aucs) == pytest.approx(0.5, abs=0.05)
+    # A representative single pair is within a chance band (not linkable).
+    ig, eg = synthetic_bridge_fixture(42, n_flows=24, linked=False)
+    assert 0.35 <= bridge_correlation_auc(ig, eg) <= 0.65
+
+
+def test_linked_separates_from_unlinked():
+    linked = bridge_correlation_auc(*synthetic_bridge_fixture(7, n_flows=24, linked=True))
+    unlinked = bridge_correlation_auc(*synthetic_bridge_fixture(7, n_flows=24, linked=False))
+    assert linked > unlinked
+
+
+def test_fixture_is_seed_deterministic():
+    a = synthetic_bridge_fixture(99, linked=True)
+    b = synthetic_bridge_fixture(99, linked=True)
+    assert a == b
+    # Different ground truth from the same seed is actually different data.
+    assert synthetic_bridge_fixture(99, linked=True) != synthetic_bridge_fixture(99, linked=False)