diff --git a/docs/adr/0001-episode-semantics-boundaries.md b/docs/adr/0001-episode-semantics-boundaries.md index f297105..2144e45 100644 --- a/docs/adr/0001-episode-semantics-boundaries.md +++ b/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 diff --git a/scripts/episode_candidate_core.py b/scripts/episode_candidate_core.py index 3706708..58a7ddc 100644 --- a/scripts/episode_candidate_core.py +++ b/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]]]: diff --git a/tests/test_episode_candidate_core.py b/tests/test_episode_candidate_core.py index 6ae6fc1..77fa490 100644 --- a/tests/test_episode_candidate_core.py +++ b/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)