diff --git a/docs/adr/0001-episode-semantics-boundaries.md b/docs/adr/0001-episode-semantics-boundaries.md index aca5443..0e610c7 100644 --- a/docs/adr/0001-episode-semantics-boundaries.md +++ b/docs/adr/0001-episode-semantics-boundaries.md @@ -121,10 +121,25 @@ and window_seconds are integers, and the terminal-failure flag is a Boolean. Coercible strings and numeric Boolean equivalents fail closed. Unsupported rules or semantically stale baselines are rejected before candidate selection. -A separate oracle slice must bind this core to the committed baseline, validate -all candidate-to-episode evidence references, and record the measured fixture -outcome before the hypothesis is accepted. The runtime and `loglens.report.v3` -remain unchanged. +The oracle slice consumes that baseline contract and requires candidate, +episode, and event-decision references to agree before materializing output. + +On `continuous_background_two_peaks`, the v0.6 baseline emits one episode. The +candidate emits two: `line:1` through `line:5` and `line:11` through `line:15`. +It covers ten dense-peak events exactly once, excludes the five bridge events, +and materializes these deterministic IDs: + +- `finding:brute_force:584fd14b544a7959` +- `finding:brute_force:a885dcb623777120` + +Reversing the fixture input produces the same ordered oracle. Boundary tests +also retain one episode at an exact 600-second cooldown gap and permit two at +601 seconds. + +This accepts the hypothesis for the bounded fixture only. Isolated bursts, +maximal-window ties, shared evidence, background calibration, and a production +complexity design still require independent evidence. `Detector::analyze()` and +`loglens.report.v3` remain unchanged. ## Alternatives considered @@ -161,3 +176,6 @@ remain unchanged. - [`Candidate core tests`](../../tests/test_episode_candidate_core.py) - [`Baseline contract`](../../scripts/episode_baseline_contract.py) - [`Baseline contract tests`](../../tests/test_episode_candidate_baseline_contract.py) +- [`Candidate evaluator`](../../scripts/evaluate_episode_candidate.py) +- [`Candidate regression tests`](../../tests/test_episode_candidate.py) +- [`Candidate oracle`](../../tests/fixtures/episode_semantics_v0.7/continuous_background_two_peaks/candidate.window-separated-v1.expected.json) diff --git a/scripts/evaluate_episode_candidate.py b/scripts/evaluate_episode_candidate.py new file mode 100644 index 0000000..48c429c --- /dev/null +++ b/scripts/evaluate_episode_candidate.py @@ -0,0 +1,405 @@ +#!/usr/bin/env python3 +"""Evaluate a research-only episode candidate against committed fixtures.""" + +from __future__ import annotations + +import argparse +import json +import sys +from datetime import datetime +from pathlib import Path +from typing import Any, Iterable, Sequence + +if __package__: + from .episode_baseline_contract import ( + canonical_oracle_timestamp, + finding_id, + rule_projection, + validate_fixture_baseline, + ) + from .episode_candidate_core import ( + CandidateWindow, + enumerate_candidate_windows, + ordered_events, + parse_timestamp, + select_window_separated_candidates, + ) +else: + from episode_baseline_contract import ( + canonical_oracle_timestamp, + finding_id, + rule_projection, + validate_fixture_baseline, + ) + from episode_candidate_core import ( + CandidateWindow, + enumerate_candidate_windows, + ordered_events, + parse_timestamp, + select_window_separated_candidates, + ) + +def _candidate_membership_index( + records: Iterable[tuple[str, Sequence[str]]], +) -> dict[str, tuple[str, ...]]: + memberships: dict[str, list[str]] = {} + for candidate_id, event_ids in records: + for event_id in event_ids: + memberships.setdefault(str(event_id), []).append(str(candidate_id)) + return { + event_id: tuple(sorted(candidate_ids)) + for event_id, candidate_ids in memberships.items() + } + + +def _candidate_json( + candidate: CandidateWindow, + memberships: dict[str, tuple[str, ...]], + selected_ids: set[str], +) -> dict[str, Any]: + overlaps = sorted( + event_id + for event_id in candidate.event_ids + if len(memberships.get(event_id, ())) > 1 + ) + selected = candidate.candidate_id in selected_ids + return { + "candidate_id": candidate.candidate_id, + "event_ids": list(candidate.event_ids), + "first_seen": canonical_oracle_timestamp(candidate.first_seen), + "last_seen": canonical_oracle_timestamp(candidate.last_seen), + "threshold_crossing_event_ids": [candidate.threshold_crossing_event_id], + "overlap_event_ids": overlaps, + "score": { + "metric": "event_count_then_compactness", + "value": candidate.event_count, + "details": { + "event_count": candidate.event_count, + "span_seconds": candidate.span_seconds, + }, + }, + "decision": { + "selected": selected, + "reason_code": "selected" if selected else "not_selected", + "reason": ( + "Selected by maximum covered events, then minimum total span and episode count." + if selected + else "Rejected by overlap/cooldown compatibility or the deterministic global objective." + ), + }, + } + + +def _validate_baseline_reference(baseline_reference: str) -> None: + if ( + not baseline_reference + or "/" in baseline_reference + or "\\" in baseline_reference + ): + raise ValueError("baseline_reference must be a privacy-safe file name") + + +def _build_oracle( + fixture: dict[str, Any], + segments: Sequence[Sequence[dict[str, Any]]], + baseline_episode_count: int, + baseline_reference: str, +) -> dict[str, Any]: + rule = fixture["rule"] + window_seconds = int(rule["window_seconds"]) + threshold = int(rule["threshold"]) + output_segments: list[dict[str, Any]] = [] + episode_index = 0 + + for segment_number, events in enumerate(segments, start=1): + segment_id = f"segment:{segment_number}" + event_ids = [str(event["event_id"]) for event in events] + candidates = enumerate_candidate_windows(events, threshold, window_seconds) + selected = select_window_separated_candidates(candidates, window_seconds) + selected_ids = {candidate.candidate_id for candidate in selected} + memberships = _candidate_membership_index( + (candidate.candidate_id, candidate.event_ids) for candidate in candidates + ) + selected_event_ids = { + event_id for candidate in selected for event_id in candidate.event_ids + } + episodes = [] + for candidate in selected: + episode_index += 1 + episodes.append( + { + "episode_index": episode_index, + "finding_id": finding_id(rule, candidate), + "candidate_id": candidate.candidate_id, + "event_ids": list(candidate.event_ids), + "first_seen": canonical_oracle_timestamp(candidate.first_seen), + "last_seen": canonical_oracle_timestamp(candidate.last_seen), + "inclusion_reason": "Window is part of the optimal window-separated evidence set.", + } + ) + event_decisions = [] + for event in events: + event_id = str(event["event_id"]) + containing = list(memberships.get(event_id, ())) + included = event_id in selected_event_ids + event_decisions.append( + { + "event_id": event_id, + "decision": "included" if included else "excluded", + "reason_code": "selected_window" + if included + else ( + "bridge_background" + if event.get("role") == "bridge_background" + else "not_selected" + ), + "candidate_ids": containing, + "reason": ( + "Included exactly once by a selected candidate window." + if included + else "Retained as background evidence outside every selected dense window." + ), + } + ) + output_segments.append( + { + "segment_id": segment_id, + "event_ids": event_ids, + "first_seen": canonical_oracle_timestamp( + parse_timestamp(str(events[0]["timestamp"])) + ), + "last_seen": canonical_oracle_timestamp( + parse_timestamp(str(events[-1]["timestamp"])) + ), + "candidate_windows": [ + _candidate_json(candidate, memberships, selected_ids) + for candidate in candidates + ], + "selected_episodes": episodes, + "event_decisions": event_decisions, + } + ) + + oracle = { + "format": "loglens.episode_candidate_oracle.v1", + "fixture_id": fixture["fixture_id"], + "algorithm": { + "id": "research.window_separated_weighted_intervals", + "version": "1", + "status": "candidate", + }, + "rule": rule_projection(rule), + "segments": output_segments, + "comparison": { + "baseline_reference": baseline_reference, + "baseline_episode_count": baseline_episode_count, + "candidate_episode_count": episode_index, + "continuous_segment_split": episode_index > baseline_episode_count, + "notes": [ + "Research-only candidate; Detector::analyze() and loglens.report.v3 are unchanged.", + "Selected windows require a gap strictly greater than one rule window; exact-boundary gaps remain one cooldown episode.", + "Selection maximizes covered evidence count, then minimizes total span and episode count, with a chronological final tie-break.", + "Exhaustive candidate materialization and overlap reporting have super-quadratic worst-case cost; hard limits keep this fixture tool bounded, not production-ready.", + ], + }, + } + return oracle + + +def evaluate_fixture( + fixture: dict[str, Any], + baseline: dict[str, Any], + baseline_reference: str = "baseline.expected.json", +) -> dict[str, Any]: + _validate_baseline_reference(baseline_reference) + context = validate_fixture_baseline(fixture, baseline) + oracle = _build_oracle( + fixture, context.segments, context.episode_count, baseline_reference + ) + _validate_oracle_cross_references(fixture, oracle) + return oracle + + +def _validate_oracle_cross_references( + fixture: dict[str, Any], oracle: dict[str, Any] +) -> None: + if oracle.get("fixture_id") != fixture.get("fixture_id"): + raise ValueError("oracle fixture_id must match the research fixture") + if oracle.get("rule") != rule_projection(fixture["rule"]): + raise ValueError("oracle rule must match the research fixture") + + fixture_ids = [ + str(event["event_id"]) for event in ordered_events(fixture["events"]) + ] + segment_ids: list[str] = [] + selected_count = 0 + episode_indexes: list[int] = [] + for segment in oracle["segments"]: + events = list(segment["event_ids"]) + segment_ids.extend(events) + decisions = [decision["event_id"] for decision in segment["event_decisions"]] + if len(decisions) != len(set(decisions)) or set(decisions) != set(events): + raise ValueError( + "exactly one event decision is required for every segment event" + ) + candidate_records = segment["candidate_windows"] + candidate_ids = [candidate["candidate_id"] for candidate in candidate_records] + if len(candidate_ids) != len(set(candidate_ids)): + raise ValueError("candidate IDs must be unique inside a segment") + candidates = { + candidate["candidate_id"]: candidate for candidate in candidate_records + } + for candidate in candidate_records: + if not set(candidate["event_ids"]).issubset(events): + raise ValueError( + "candidate evidence must stay inside its baseline segment" + ) + memberships = _candidate_membership_index( + (candidate_id, candidate["event_ids"]) + for candidate_id, candidate in candidates.items() + ) + selected_candidate_ids = { + candidate_id + for candidate_id, candidate in candidates.items() + if candidate["decision"]["selected"] + } + selected_events: set[str] = set() + episode_candidate_ids: list[str] = [] + episode_order: list[tuple[datetime, datetime, str]] = [] + for episode in segment["selected_episodes"]: + candidate_id = episode["candidate_id"] + if ( + candidate_id not in candidates + or not candidates[candidate_id]["decision"]["selected"] + ): + raise ValueError("selected episode must reference a selected candidate") + candidate = candidates[candidate_id] + if ( + episode["event_ids"] != candidate["event_ids"] + or episode["first_seen"] != candidate["first_seen"] + or episode["last_seen"] != candidate["last_seen"] + ): + raise ValueError("selected episode evidence must match its candidate") + materialized = CandidateWindow( + event_ids=tuple(candidate["event_ids"]), + first_seen=parse_timestamp(candidate["first_seen"]), + last_seen=parse_timestamp(candidate["last_seen"]), + threshold_crossing_event_id="", + ) + if episode["finding_id"] != finding_id(fixture["rule"], materialized): + raise ValueError( + "selected episode finding_id must match its candidate evidence" + ) + overlap = selected_events.intersection(episode["event_ids"]) + if overlap: + raise ValueError("selected episodes reuse event evidence") + selected_events.update(episode["event_ids"]) + selected_count += 1 + episode_indexes.append(int(episode["episode_index"])) + episode_candidate_ids.append(candidate_id) + episode_order.append( + (materialized.first_seen, materialized.last_seen, candidate_id) + ) + if len(episode_candidate_ids) != len(set(episode_candidate_ids)): + raise ValueError("a candidate may materialize at most one selected episode") + if set(episode_candidate_ids) != selected_candidate_ids: + raise ValueError( + "selected candidates and materialized episodes must match exactly" + ) + if episode_order != sorted(episode_order): + raise ValueError("selected episodes must be chronological") + included = { + decision["event_id"] + for decision in segment["event_decisions"] + if decision["decision"] == "included" + } + if included != selected_events: + raise ValueError( + "included event decisions must equal selected episode evidence" + ) + for decision in segment["event_decisions"]: + expected_candidates = list(memberships.get(decision["event_id"], ())) + if decision.get("candidate_ids", []) != expected_candidates: + raise ValueError( + "event decision candidate_ids must match candidate evidence" + ) + if segment_ids != fixture_ids or len(segment_ids) != len(set(segment_ids)): + raise ValueError("oracle segments must partition fixture events exactly once") + if episode_indexes != list(range(1, selected_count + 1)): + raise ValueError("episode indexes must be chronological and contiguous") + if oracle["comparison"]["candidate_episode_count"] != selected_count: + raise ValueError("comparison candidate count must match selected episodes") + + +def validate_oracle( + fixture: dict[str, Any], + baseline: dict[str, Any], + oracle: dict[str, Any], + baseline_reference: str = "baseline.expected.json", +) -> None: + _validate_baseline_reference(baseline_reference) + context = validate_fixture_baseline(fixture, baseline) + _validate_oracle_cross_references(fixture, oracle) + expected = _build_oracle( + fixture, context.segments, context.episode_count, baseline_reference + ) + if oracle != expected: + raise ValueError("oracle does not match the canonical fixture derivation") + + +def _load(path: Path) -> dict[str, Any]: + return json.loads(path.read_text(encoding="utf-8")) + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--fixture", type=Path, required=True) + parser.add_argument("--baseline", type=Path, required=True) + parser.add_argument( + "--check", + type=Path, + help="Require generated JSON to match this committed oracle.", + ) + parser.add_argument( + "--output", type=Path, help="Write generated JSON; otherwise print it." + ) + args = parser.parse_args(argv) + + try: + oracle = evaluate_fixture( + _load(args.fixture), + _load(args.baseline), + baseline_reference=args.baseline.name, + ) + rendered = json.dumps(oracle, indent=2, ensure_ascii=False) + "\n" + if ( + args.check is not None + and args.check.read_text(encoding="utf-8") != rendered + ): + print(f"candidate oracle drift: {args.check.name}", file=sys.stderr) + return 1 + if args.output is not None: + args.output.write_text(rendered, encoding="utf-8", newline="\n") + elif args.check is None: + sys.stdout.write(rendered) + return 0 + except json.JSONDecodeError as error: + print( + f"episode candidate error: invalid JSON at line {error.lineno}, column {error.colno}", + file=sys.stderr, + ) + except OSError as error: + print( + f"episode candidate error: I/O failure ({error.strerror or 'operation failed'})", + file=sys.stderr, + ) + except (KeyError, TypeError, ValueError): + print( + "episode candidate error: invalid fixture, baseline, or oracle data", + file=sys.stderr, + ) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/fixtures/episode_semantics_v0.7/continuous_background_two_peaks/candidate.window-separated-v1.expected.json b/tests/fixtures/episode_semantics_v0.7/continuous_background_two_peaks/candidate.window-separated-v1.expected.json new file mode 100644 index 0000000..a5e5278 --- /dev/null +++ b/tests/fixtures/episode_semantics_v0.7/continuous_background_two_peaks/candidate.window-separated-v1.expected.json @@ -0,0 +1,272 @@ +{ + "format": "loglens.episode_candidate_oracle.v1", + "fixture_id": "episode_semantics_v0.7.continuous_background_two_peaks", + "algorithm": { + "id": "research.window_separated_weighted_intervals", + "version": "1", + "status": "candidate" + }, + "rule": { + "rule_id": "brute_force", + "grouping_key": "source_ip", + "subject": "203.0.113.77", + "threshold": 5, + "window_seconds": 600, + "window_boundary": "inclusive" + }, + "segments": [ + { + "segment_id": "segment:1", + "event_ids": [ + "line:1", + "line:2", + "line:3", + "line:4", + "line:5", + "line:6", + "line:7", + "line:8", + "line:9", + "line:10", + "line:11", + "line:12", + "line:13", + "line:14", + "line:15" + ], + "first_seen": "2026-03-10T09:00:00Z", + "last_seen": "2026-03-10T09:58:00Z", + "candidate_windows": [ + { + "candidate_id": "candidate:line:1..line:5", + "event_ids": [ + "line:1", + "line:2", + "line:3", + "line:4", + "line:5" + ], + "first_seen": "2026-03-10T09:00:00Z", + "last_seen": "2026-03-10T09:02:00Z", + "threshold_crossing_event_ids": [ + "line:5" + ], + "overlap_event_ids": [], + "score": { + "metric": "event_count_then_compactness", + "value": 5, + "details": { + "event_count": 5, + "span_seconds": 120 + } + }, + "decision": { + "selected": true, + "reason_code": "selected", + "reason": "Selected by maximum covered events, then minimum total span and episode count." + } + }, + { + "candidate_id": "candidate:line:11..line:15", + "event_ids": [ + "line:11", + "line:12", + "line:13", + "line:14", + "line:15" + ], + "first_seen": "2026-03-10T09:56:00Z", + "last_seen": "2026-03-10T09:58:00Z", + "threshold_crossing_event_ids": [ + "line:15" + ], + "overlap_event_ids": [], + "score": { + "metric": "event_count_then_compactness", + "value": 5, + "details": { + "event_count": 5, + "span_seconds": 120 + } + }, + "decision": { + "selected": true, + "reason_code": "selected", + "reason": "Selected by maximum covered events, then minimum total span and episode count." + } + } + ], + "selected_episodes": [ + { + "episode_index": 1, + "finding_id": "finding:brute_force:584fd14b544a7959", + "candidate_id": "candidate:line:1..line:5", + "event_ids": [ + "line:1", + "line:2", + "line:3", + "line:4", + "line:5" + ], + "first_seen": "2026-03-10T09:00:00Z", + "last_seen": "2026-03-10T09:02:00Z", + "inclusion_reason": "Window is part of the optimal window-separated evidence set." + }, + { + "episode_index": 2, + "finding_id": "finding:brute_force:a885dcb623777120", + "candidate_id": "candidate:line:11..line:15", + "event_ids": [ + "line:11", + "line:12", + "line:13", + "line:14", + "line:15" + ], + "first_seen": "2026-03-10T09:56:00Z", + "last_seen": "2026-03-10T09:58:00Z", + "inclusion_reason": "Window is part of the optimal window-separated evidence set." + } + ], + "event_decisions": [ + { + "event_id": "line:1", + "decision": "included", + "reason_code": "selected_window", + "candidate_ids": [ + "candidate:line:1..line:5" + ], + "reason": "Included exactly once by a selected candidate window." + }, + { + "event_id": "line:2", + "decision": "included", + "reason_code": "selected_window", + "candidate_ids": [ + "candidate:line:1..line:5" + ], + "reason": "Included exactly once by a selected candidate window." + }, + { + "event_id": "line:3", + "decision": "included", + "reason_code": "selected_window", + "candidate_ids": [ + "candidate:line:1..line:5" + ], + "reason": "Included exactly once by a selected candidate window." + }, + { + "event_id": "line:4", + "decision": "included", + "reason_code": "selected_window", + "candidate_ids": [ + "candidate:line:1..line:5" + ], + "reason": "Included exactly once by a selected candidate window." + }, + { + "event_id": "line:5", + "decision": "included", + "reason_code": "selected_window", + "candidate_ids": [ + "candidate:line:1..line:5" + ], + "reason": "Included exactly once by a selected candidate window." + }, + { + "event_id": "line:6", + "decision": "excluded", + "reason_code": "bridge_background", + "candidate_ids": [], + "reason": "Retained as background evidence outside every selected dense window." + }, + { + "event_id": "line:7", + "decision": "excluded", + "reason_code": "bridge_background", + "candidate_ids": [], + "reason": "Retained as background evidence outside every selected dense window." + }, + { + "event_id": "line:8", + "decision": "excluded", + "reason_code": "bridge_background", + "candidate_ids": [], + "reason": "Retained as background evidence outside every selected dense window." + }, + { + "event_id": "line:9", + "decision": "excluded", + "reason_code": "bridge_background", + "candidate_ids": [], + "reason": "Retained as background evidence outside every selected dense window." + }, + { + "event_id": "line:10", + "decision": "excluded", + "reason_code": "bridge_background", + "candidate_ids": [], + "reason": "Retained as background evidence outside every selected dense window." + }, + { + "event_id": "line:11", + "decision": "included", + "reason_code": "selected_window", + "candidate_ids": [ + "candidate:line:11..line:15" + ], + "reason": "Included exactly once by a selected candidate window." + }, + { + "event_id": "line:12", + "decision": "included", + "reason_code": "selected_window", + "candidate_ids": [ + "candidate:line:11..line:15" + ], + "reason": "Included exactly once by a selected candidate window." + }, + { + "event_id": "line:13", + "decision": "included", + "reason_code": "selected_window", + "candidate_ids": [ + "candidate:line:11..line:15" + ], + "reason": "Included exactly once by a selected candidate window." + }, + { + "event_id": "line:14", + "decision": "included", + "reason_code": "selected_window", + "candidate_ids": [ + "candidate:line:11..line:15" + ], + "reason": "Included exactly once by a selected candidate window." + }, + { + "event_id": "line:15", + "decision": "included", + "reason_code": "selected_window", + "candidate_ids": [ + "candidate:line:11..line:15" + ], + "reason": "Included exactly once by a selected candidate window." + } + ] + } + ], + "comparison": { + "baseline_reference": "baseline.expected.json", + "baseline_episode_count": 1, + "candidate_episode_count": 2, + "continuous_segment_split": true, + "notes": [ + "Research-only candidate; Detector::analyze() and loglens.report.v3 are unchanged.", + "Selected windows require a gap strictly greater than one rule window; exact-boundary gaps remain one cooldown episode.", + "Selection maximizes covered evidence count, then minimizes total span and episode count, with a chronological final tie-break.", + "Exhaustive candidate materialization and overlap reporting have super-quadratic worst-case cost; hard limits keep this fixture tool bounded, not production-ready." + ] + } +} diff --git a/tests/test_episode_candidate.py b/tests/test_episode_candidate.py new file mode 100644 index 0000000..43197cb --- /dev/null +++ b/tests/test_episode_candidate.py @@ -0,0 +1,126 @@ +import copy +import json +import sys +import unittest +from datetime import timedelta, timezone +from pathlib import Path + + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT)) + +from scripts.evaluate_episode_candidate import ( # noqa: E402 + evaluate_fixture, + validate_oracle, +) +from scripts.episode_candidate_core import parse_timestamp # noqa: E402 + + +FIXTURE_ROOT = ( + REPO_ROOT + / "tests" + / "fixtures" + / "episode_semantics_v0.7" + / "continuous_background_two_peaks" +) + + +class ContinuousBackgroundFixtureTests(unittest.TestCase): + def setUp(self) -> None: + self.fixture = json.loads( + (FIXTURE_ROOT / "fixture.json").read_text(encoding="utf-8") + ) + self.baseline = json.loads( + (FIXTURE_ROOT / "baseline.expected.json").read_text(encoding="utf-8") + ) + + def test_candidate_recovers_two_peaks_without_reusing_events(self) -> None: + oracle = evaluate_fixture(self.fixture, self.baseline) + segment = oracle["segments"][0] + + self.assertEqual(oracle["comparison"]["baseline_episode_count"], 1) + self.assertEqual(oracle["comparison"]["candidate_episode_count"], 2) + self.assertEqual( + [episode["finding_id"] for episode in segment["selected_episodes"]], + [ + "finding:brute_force:584fd14b544a7959", + "finding:brute_force:a885dcb623777120", + ], + ) + included = [ + decision["event_id"] + for decision in segment["event_decisions"] + if decision["decision"] == "included" + ] + self.assertEqual( + included, + [f"line:{i}" for i in range(1, 6)] + [f"line:{i}" for i in range(11, 16)], + ) + self.assertEqual(len(included), len(set(included))) + + def test_input_permutation_does_not_change_oracle(self) -> None: + shuffled = copy.deepcopy(self.fixture) + shuffled["events"].reverse() + + self.assertEqual( + evaluate_fixture(shuffled, self.baseline), + evaluate_fixture(self.fixture, self.baseline), + ) + + def test_generated_oracle_matches_committed_fixture_and_cross_references( + self, + ) -> None: + oracle = evaluate_fixture(self.fixture, self.baseline) + expected = json.loads( + (FIXTURE_ROOT / "candidate.window-separated-v1.expected.json").read_text( + encoding="utf-8" + ) + ) + + validate_oracle(self.fixture, self.baseline, oracle) + self.assertEqual(oracle, expected) + + def test_validator_rejects_duplicate_event_decisions(self) -> None: + oracle = evaluate_fixture(self.fixture, self.baseline) + oracle["segments"][0]["event_decisions"].append( + copy.deepcopy(oracle["segments"][0]["event_decisions"][0]) + ) + + with self.assertRaisesRegex(ValueError, "exactly one event decision"): + validate_oracle(self.fixture, self.baseline, oracle) + + def test_validator_rejects_fixture_inconsistent_derived_fields(self) -> None: + mutations = { + "segment identity": lambda oracle: oracle["segments"][0].__setitem__( + "segment_id", "segment:tampered" + ), + "candidate score": lambda oracle: oracle["segments"][0][ + "candidate_windows" + ][0]["score"]["details"].__setitem__("span_seconds", 999), + "comparison": lambda oracle: oracle["comparison"].__setitem__( + "continuous_segment_split", False + ), + } + + for label, mutate in mutations.items(): + with self.subTest(label=label): + oracle = evaluate_fixture(self.fixture, self.baseline) + mutate(oracle) + with self.assertRaisesRegex(ValueError, "canonical fixture derivation"): + validate_oracle(self.fixture, self.baseline, oracle) + + def test_equivalent_timezone_offsets_produce_the_same_oracle(self) -> None: + offset_fixture = copy.deepcopy(self.fixture) + offset = timezone(timedelta(hours=1)) + for event in offset_fixture["events"]: + event["timestamp"] = parse_timestamp(event["timestamp"]).astimezone( + offset + ).isoformat() + + self.assertEqual( + evaluate_fixture(offset_fixture, self.baseline), + evaluate_fixture(self.fixture, self.baseline), + ) + +if __name__ == "__main__": + unittest.main()