feat(research): add episode density-contrast gate

docs/adr/0001-episode-semantics-boundaries.md

@@ -183,6 +183,40 @@ multiple dense peaks. It blocks production adoption without a calibrated
 density-contrast rule or an explicit alert-volume budget; it does not estimate
 a real-world false-positive rate.
 
+A follow-up research diagnostic tests the minimum evidence needed for a density
+contrast decision without changing candidate selection. For each selected
+window, its internal mean gap is its exact timestamp span divided by one fewer
+than its event count. For each adjacent selected pair, the bridge mean gap is
+the span from the left window's last event to the right window's first event,
+divided by the number of strictly intervening events plus one. The pair passes
+when:
+
+```text
+bridge_mean_gap >= 2 * max(left_internal_mean_gap, right_internal_mean_gap)
+```
+
+All arithmetic uses integer microseconds and rational comparison. Event lookup
+and bridge counting use ordered indexes and binary search, so this diagnostic
+adds `O(E log E + S log E)` work after receiving `E` events and `S` selected
+windows. Invalid candidate references, boundaries, ordering, overlap, or reused
+event evidence fail closed. A single selected window passes because there is no
+split contrast to evaluate.
+
+On `continuous_background_two_peaks`, candidate v1 selects `line:1` through
+`line:5` and `line:11` through `line:15`. Each internal mean gap is 30 seconds;
+the five bridge events divide the 3,240-second boundary span into a 540-second
+bridge mean gap, giving an 18x contrast. The same calculation on the fourteen
+uniform events gives a 150-second bridge mean gap and 150-second internal means,
+so its contrast is 1x and it fails. Reversing both event and selection input
+does not change either result.
+
+This accepts the 2x mean-gap diagnostic only as a prerequisite for a candidate
+v2 research design: it separates the existing positive fixture from the known
+uniform false split while preserving the single-episode case. The stopping rule
+is therefore met, and no cadence sweep is warranted. The result does not
+calibrate a production threshold, estimate a false-positive rate, or authorize
+wiring the diagnostic into `Detector::analyze()`.
+
 Together, the two fixtures and focused tie/shared-evidence/background controls
 accept the recovery, null-control, deterministic tie-break, and publication
 single-consumption hypotheses for their bounded cases. The background control

scripts/episode_candidate_core.py

@@ -2,8 +2,10 @@
 
 from __future__ import annotations
 
+from bisect import bisect_left, bisect_right
 from dataclasses import dataclass
 from datetime import datetime
+from fractions import Fraction
 from typing import Any, Sequence
 
 
@@ -152,6 +154,88 @@ def select_window_separated_candidates(
     )
 
 
+def _duration_microseconds(start: datetime, end: datetime) -> int:
+    delta = end - start
+    return (
+        (delta.days * 86_400 + delta.seconds) * 1_000_000
+        + delta.microseconds
+    )
+
+
+def selection_has_minimum_gap_contrast(
+    events: Sequence[dict[str, Any]],
+    selected: Sequence[CandidateWindow],
+    minimum_ratio: int | Fraction = 2,
+) -> bool:
+    """Return whether every selected-window gap meets an exact mean-gap ratio."""
+    if (
+        isinstance(minimum_ratio, bool)
+        or not isinstance(minimum_ratio, (int, Fraction))
+        or minimum_ratio <= 0
+    ):
+        raise ValueError("minimum_ratio must be a positive integer or Fraction")
+
+    ordered = ordered_events(events)
+    timestamps = [parse_timestamp(str(event["timestamp"])) for event in ordered]
+    event_positions = {
+        str(event["event_id"]): index for index, event in enumerate(ordered)
+    }
+    chronological = sorted(
+        selected,
+        key=lambda candidate: (
+            candidate.first_seen,
+            candidate.last_seen,
+            candidate.event_ids,
+        ),
+    )
+
+    consumed: set[str] = set()
+    for candidate in chronological:
+        if candidate.event_count < 2:
+            raise ValueError("selected candidates must contain at least two events")
+        if len(candidate.event_ids) != len(set(candidate.event_ids)):
+            raise ValueError("selected candidate event IDs must be unique")
+        if any(event_id not in event_positions for event_id in candidate.event_ids):
+            raise ValueError("selected candidate references an unknown event")
+        positions = [event_positions[event_id] for event_id in candidate.event_ids]
+        if positions != list(range(positions[0], positions[-1] + 1)):
+            raise ValueError("selected candidate events must be contiguous and ordered")
+        if (
+            candidate.first_seen != timestamps[positions[0]]
+            or candidate.last_seen != timestamps[positions[-1]]
+        ):
+            raise ValueError("selected candidate boundaries must match its events")
+        if candidate.threshold_crossing_event_id not in candidate.event_ids:
+            raise ValueError("threshold crossing must belong to its candidate")
+        if consumed.intersection(candidate.event_ids):
+            raise ValueError("selected candidates reuse event evidence")
+        consumed.update(candidate.event_ids)
+
+    for left, right in zip(chronological, chronological[1:]):
+        if right.first_seen <= left.last_seen:
+            raise ValueError("selected candidate windows must not overlap")
+
+        left_mean_gap = Fraction(
+            _duration_microseconds(left.first_seen, left.last_seen),
+            left.event_count - 1,
+        )
+        right_mean_gap = Fraction(
+            _duration_microseconds(right.first_seen, right.last_seen),
+            right.event_count - 1,
+        )
+        bridge_event_count = bisect_left(timestamps, right.first_seen) - bisect_right(
+            timestamps, left.last_seen
+        )
+        bridge_mean_gap = Fraction(
+            _duration_microseconds(left.last_seen, right.first_seen),
+            bridge_event_count + 1,
+        )
+        if bridge_mean_gap < minimum_ratio * max(left_mean_gap, right_mean_gap):
+            return False
+
+    return True
+
+
 def activity_segments(
     events: Sequence[dict[str, Any]], window_seconds: int
 ) -> list[list[dict[str, Any]]]:

tests/test_episode_candidate_core.py

@@ -10,6 +10,7 @@
 from scripts.episode_candidate_core import (  # noqa: E402
     activity_segments,
     enumerate_candidate_windows,
+    selection_has_minimum_gap_contrast,
     select_window_separated_candidates,
 )
 
@@ -108,6 +109,49 @@ def test_uniform_threshold_rate_can_multiply_episodes_in_one_segment(self) -> No
             ],
         )
 
+    def test_gap_contrast_separates_dense_peaks_from_uniform_background(self) -> None:
+        dense_peaks = make_events(
+            [0, 30, 60, 90, 120, 660, 1200, 1740, 2280, 2820, 3360, 3390, 3420, 3450, 3480]
+        )
+        uniform_background = make_events([150 * offset for offset in range(14)])
+
+        for events, expected_ids, expected_contrast in (
+            (dense_peaks, ((1, 5), (11, 15)), True),
+            (uniform_background, ((1, 5), (10, 14)), False),
+        ):
+            selected = select_window_separated_candidates(
+                enumerate_candidate_windows(events, 5, 600), 600
+            )
+
+            self.assertEqual(
+                [candidate.event_ids for candidate in selected],
+                [
+                    tuple(f"line:{index}" for index in range(start, end + 1))
+                    for start, end in expected_ids
+                ],
+            )
+            self.assertEqual(
+                selection_has_minimum_gap_contrast(events, selected), expected_contrast
+            )
+            self.assertEqual(
+                selection_has_minimum_gap_contrast(
+                    list(reversed(events)), list(reversed(selected))
+                ),
+                expected_contrast,
+            )
+
+    def test_gap_contrast_accepts_a_single_selected_episode(self) -> None:
+        events = make_events([0, 30, 60, 90, 120])
+        selected = select_window_separated_candidates(
+            enumerate_candidate_windows(events, 5, 600), 600
+        )
+
+        self.assertTrue(selection_has_minimum_gap_contrast(events, selected))
+
+    def test_gap_contrast_requires_a_positive_ratio(self) -> None:
+        with self.assertRaisesRegex(ValueError, "minimum_ratio"):
+            selection_has_minimum_gap_contrast([], [], minimum_ratio=0)
+
     def test_equal_score_windows_prefer_chronological_key(self) -> None:
         candidates = enumerate_candidate_windows(
             make_events([0, 1, 2, 3, 600, 601]), 5, 600