feat(research): evaluate minimum-span threshold cores

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

@@ -241,6 +241,38 @@ between a densest threshold-sized core, quantile, trimmed-gap, or other robust
 statistic. It only requires a future design to separate dense-core evidence from
 maximal-window coverage before production calibration.
 
+A minimum-span threshold-core control tests the smallest follow-up hypothesis:
+inside each selected maximal window, choose the contiguous sequence of exactly
+`threshold` events with the shortest timestamp span, breaking equal-span ties by
+the chronological event key. The selected maximal windows still define candidate
+coverage, while these cores supply only the internal-density evidence used by the
+contrast diagnostic. All events strictly between adjacent core boundaries remain
+part of the bridge calculation, including maximal-window padding that the core
+does not retain.
+
+On the padding control, both the original and padded streams choose the same two
+120-second dense cores. The base stream retains its `38/3x` contrast. In the
+padded stream, the event at offset 600 is excluded from the first core but counts
+as a third bridge event; the bridge mean is therefore 285 seconds and the core
+contrast is `19/2x`. Both streams pass the 2x diagnostic. Replacing the minimum
+span with the maximum span makes the focused padding test fail, confirming that
+the control observes core selection rather than episode count alone.
+
+The same core calculation preserves the two independent controls: the existing
+continuous-background fixture remains an `18x` positive, while the fourteen-event
+uniform stream remains a `1x` negative. Reversing event and selection input does
+not change the ordered cores or verdict. When two threshold-sized cores have the
+same span, the earlier chronological core wins. Invalid thresholds and selected
+windows that cannot supply a threshold-sized core fail closed.
+
+This accepts the minimum-span threshold-sized core as the candidate-v2 evidence
+abstraction for these three bounded controls. The stopping rule is met, so there
+is no quantile, trimmed-gap, position, cadence, or ratio sweep in this slice. The
+result does not calibrate the 2x ratio, prove robustness to arbitrary background
+processes, change candidate-v1 selection, or authorize a detector, CLI, report,
+oracle-schema, or evaluator change. Any candidate-v2 materialization remains a
+separate compatibility decision.
+
 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

@@ -162,19 +162,14 @@ def _duration_microseconds(start: datetime, end: datetime) -> int:
     )
 
 
-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")
-
+def _validated_selection(
+    events: Sequence[dict[str, Any]], selected: Sequence[CandidateWindow]
+) -> tuple[
+    list[dict[str, Any]],
+    list[datetime],
+    dict[str, int],
+    list[CandidateWindow],
+]:
     ordered = ordered_events(events)
     timestamps = [parse_timestamp(str(event["timestamp"])) for event in ordered]
     event_positions = {
@@ -215,6 +210,25 @@ def selection_has_minimum_gap_contrast(
         if right.first_seen <= left.last_seen:
             raise ValueError("selected candidate windows must not overlap")
 
+    return ordered, timestamps, event_positions, chronological
+
+
+def _validated_minimum_ratio(minimum_ratio: int | Fraction) -> int | Fraction:
+    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")
+    return minimum_ratio
+
+
+def _windows_have_minimum_gap_contrast(
+    timestamps: Sequence[datetime],
+    chronological: Sequence[CandidateWindow],
+    minimum_ratio: int | Fraction,
+) -> bool:
+    for left, right in zip(chronological, chronological[1:]):
         left_mean_gap = Fraction(
             _duration_microseconds(left.first_seen, left.last_seen),
             left.event_count - 1,
@@ -236,6 +250,76 @@ def selection_has_minimum_gap_contrast(
     return True
 
 
+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."""
+    ratio = _validated_minimum_ratio(minimum_ratio)
+    _, timestamps, _, chronological = _validated_selection(events, selected)
+    return _windows_have_minimum_gap_contrast(timestamps, chronological, ratio)
+
+
+def minimum_span_threshold_cores(
+    events: Sequence[dict[str, Any]],
+    selected: Sequence[CandidateWindow],
+    threshold: int,
+) -> list[CandidateWindow]:
+    """Choose each selected window's shortest contiguous threshold-sized core."""
+    if isinstance(threshold, bool) or not isinstance(threshold, int) or threshold < 2:
+        raise ValueError("threshold must be an integer of at least two")
+
+    _, timestamps, event_positions, chronological = _validated_selection(
+        events, selected
+    )
+    cores: list[CandidateWindow] = []
+    for candidate in chronological:
+        if candidate.event_count < threshold:
+            raise ValueError("selected candidate contains fewer events than threshold")
+        positions = [event_positions[event_id] for event_id in candidate.event_ids]
+        core_start = min(
+            range(candidate.event_count - threshold + 1),
+            key=lambda start: (
+                _duration_microseconds(
+                    timestamps[positions[start]],
+                    timestamps[positions[start + threshold - 1]],
+                ),
+                timestamps[positions[start]],
+                timestamps[positions[start + threshold - 1]],
+                candidate.event_ids[start : start + threshold],
+            ),
+        )
+        core_positions = positions[core_start : core_start + threshold]
+        core_event_ids = candidate.event_ids[
+            core_start : core_start + threshold
+        ]
+        cores.append(
+            CandidateWindow(
+                event_ids=core_event_ids,
+                first_seen=timestamps[core_positions[0]],
+                last_seen=timestamps[core_positions[-1]],
+                threshold_crossing_event_id=core_event_ids[-1],
+            )
+        )
+    return cores
+
+
+def selection_has_minimum_core_gap_contrast(
+    events: Sequence[dict[str, Any]],
+    selected: Sequence[CandidateWindow],
+    threshold: int,
+    minimum_ratio: int | Fraction = 2,
+) -> bool:
+    """Evaluate exact mean-gap contrast on minimum-span threshold cores."""
+    ratio = _validated_minimum_ratio(minimum_ratio)
+    cores = minimum_span_threshold_cores(events, selected, threshold)
+    timestamps = [
+        parse_timestamp(str(event["timestamp"])) for event in ordered_events(events)
+    ]
+    return _windows_have_minimum_gap_contrast(timestamps, cores, ratio)
+
+
 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,8 @@
 from scripts.episode_candidate_core import (  # noqa: E402
     activity_segments,
     enumerate_candidate_windows,
+    minimum_span_threshold_cores,
+    selection_has_minimum_core_gap_contrast,
     selection_has_minimum_gap_contrast,
     select_window_separated_candidates,
 )
@@ -185,6 +187,102 @@ def test_gap_contrast_is_not_monotone_under_maximal_window_padding(self) -> None
             selection_has_minimum_gap_contrast(with_padding, padded_selected)
         )
 
+    def test_minimum_span_core_is_stable_under_maximal_window_padding(self) -> None:
+        without_padding = make_events(
+            [0, 30, 60, 90, 120, 650, 700, 1260, 1290, 1320, 1350, 1380]
+        )
+        with_padding = make_events(
+            [0, 30, 60, 90, 120, 600, 650, 700, 1260, 1290, 1320, 1350, 1380]
+        )
+        selected_without_padding = select_window_separated_candidates(
+            enumerate_candidate_windows(without_padding, 5, 600), 600
+        )
+        selected_with_padding = select_window_separated_candidates(
+            enumerate_candidate_windows(with_padding, 5, 600), 600
+        )
+
+        cores_without_padding = minimum_span_threshold_cores(
+            without_padding, selected_without_padding, 5
+        )
+        cores_with_padding = minimum_span_threshold_cores(
+            with_padding, selected_with_padding, 5
+        )
+
+        self.assertEqual(
+            [core.event_ids for core in cores_without_padding],
+            [
+                tuple(f"line:{index}" for index in range(1, 6)),
+                tuple(f"line:{index}" for index in range(8, 13)),
+            ],
+        )
+        self.assertEqual(
+            [core.event_ids for core in cores_with_padding],
+            [
+                tuple(f"line:{index}" for index in range(1, 6)),
+                tuple(f"line:{index}" for index in range(9, 14)),
+            ],
+        )
+        self.assertEqual([core.span_seconds for core in cores_with_padding], [120, 120])
+        self.assertTrue(
+            selection_has_minimum_core_gap_contrast(
+                without_padding, selected_without_padding, 5
+            )
+        )
+        self.assertTrue(
+            selection_has_minimum_core_gap_contrast(
+                with_padding, selected_with_padding, 5
+            )
+        )
+        self.assertEqual(
+            minimum_span_threshold_cores(
+                list(reversed(with_padding)),
+                list(reversed(selected_with_padding)),
+                5,
+            ),
+            cores_with_padding,
+        )
+
+    def test_minimum_span_core_preserves_positive_and_uniform_controls(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_contrast in (
+            (dense_peaks, True),
+            (uniform_background, False),
+        ):
+            selected = select_window_separated_candidates(
+                enumerate_candidate_windows(events, 5, 600), 600
+            )
+
+            self.assertEqual(
+                selection_has_minimum_core_gap_contrast(events, selected, 5),
+                expected_contrast,
+            )
+            self.assertEqual(
+                selection_has_minimum_core_gap_contrast(
+                    list(reversed(events)), list(reversed(selected)), 5
+                ),
+                expected_contrast,
+            )
+
+    def test_minimum_span_core_breaks_ties_chronologically(self) -> None:
+        events = make_events([0, 1, 2, 3, 4, 5])
+        selected = select_window_separated_candidates(
+            enumerate_candidate_windows(events, 5, 600), 600
+        )
+
+        cores = minimum_span_threshold_cores(events, selected, 5)
+
+        self.assertEqual(
+            [core.event_ids for core in cores],
+            [tuple(f"line:{index}" for index in range(1, 6))],
+        )
+        for invalid_threshold in (True, 1, 7):
+            with self.assertRaisesRegex(ValueError, "threshold"):
+                minimum_span_threshold_cores(events, selected, invalid_threshold)
+
     def test_gap_contrast_requires_a_positive_ratio(self) -> None:
         with self.assertRaisesRegex(ValueError, "minimum_ratio"):
             selection_has_minimum_gap_contrast([], [], minimum_ratio=0)