From 84ef3e852ad94720a147aae56672ae3c434e2b1f Mon Sep 17 00:00:00 2001 From: stacknil Date: Tue, 25 Aug 2026 11:12:16 +0800 Subject: [PATCH 1/4] feat(research): materialize episode candidate oracle --- scripts/evaluate_episode_candidate.py | 431 ++++++++++++++++++++++++++ 1 file changed, 431 insertions(+) create mode 100644 scripts/evaluate_episode_candidate.py diff --git a/scripts/evaluate_episode_candidate.py b/scripts/evaluate_episode_candidate.py new file mode 100644 index 0000000..3fbb1e0 --- /dev/null +++ b/scripts/evaluate_episode_candidate.py @@ -0,0 +1,431 @@ +#!/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, Sequence + +if __package__: + from .episode_candidate_core import ( + CandidateWindow, + activity_segments, + enumerate_candidate_windows, + ordered_events, + parse_timestamp, + select_window_separated_candidates, + ) +else: + from episode_candidate_core import ( + CandidateWindow, + activity_segments, + enumerate_candidate_windows, + ordered_events, + parse_timestamp, + select_window_separated_candidates, + ) + +RULE_FIELDS = ( + "rule_id", + "grouping_key", + "subject", + "threshold", + "window_seconds", + "window_boundary", +) + + +def _hash_append(value: int, text: str) -> int: + for byte in text.encode("utf-8"): + value ^= byte + value = (value * 1099511628211) & ((1 << 64) - 1) + value ^= 0xFF + return (value * 1099511628211) & ((1 << 64) - 1) + + +def _finding_id(rule: dict[str, Any], candidate: CandidateWindow) -> str: + value = 14695981039346656037 + fields = [ + str(rule["rule_id"]), + str(rule["grouping_key"]), + str(rule["subject"]), + candidate.first_seen.strftime("%Y-%m-%d %H:%M:%S"), + candidate.last_seen.strftime("%Y-%m-%d %H:%M:%S"), + str(rule["threshold"]), + str(candidate.event_count), + str(candidate.event_count), + *candidate.event_ids, + ] + for field in fields: + value = _hash_append(value, field) + return f"finding:{rule['rule_id']}:{value:016x}" + + +def _candidate_json( + candidate: CandidateWindow, + candidates: Sequence[CandidateWindow], + selected_ids: set[str], +) -> dict[str, Any]: + overlaps = sorted( + { + event_id + for other in candidates + if other.candidate_id != candidate.candidate_id + for event_id in set(candidate.event_ids).intersection(other.event_ids) + } + ) + selected = candidate.candidate_id in selected_ids + return { + "candidate_id": candidate.candidate_id, + "event_ids": list(candidate.event_ids), + "first_seen": candidate.first_seen.isoformat().replace("+00:00", "Z"), + "last_seen": candidate.last_seen.isoformat().replace("+00:00", "Z"), + "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 _rule_projection(rule: dict[str, Any]) -> dict[str, Any]: + return {key: rule[key] for key in RULE_FIELDS} + + +def _validate_input_pair( + fixture: dict[str, Any], + baseline: dict[str, Any], + activity_segments: Sequence[Sequence[dict[str, Any]]], +) -> int: + if fixture.get("format") != "loglens.episode_research_fixture.v1": + raise ValueError("unsupported research fixture format") + if baseline.get("format") != "loglens.episode_baseline_expected.v1": + raise ValueError("unsupported baseline format") + if baseline.get("fixture_id") != fixture.get("fixture_id"): + raise ValueError("baseline fixture_id does not match the research fixture") + + rule = fixture["rule"] + if rule.get("window_boundary") != "inclusive": + raise ValueError( + "candidate algorithm v1 supports only inclusive window boundaries" + ) + if _rule_projection(baseline["rule"]) != _rule_projection(rule): + raise ValueError("baseline rule does not match the research fixture") + + baseline_segments = baseline["derived_input"]["activity_segments"] + if len(baseline_segments) != len(activity_segments): + raise ValueError("baseline activity segment count does not match derived input") + for index, (events, expected) in enumerate( + zip(activity_segments, baseline_segments), start=1 + ): + event_ids = [str(event["event_id"]) for event in events] + if ( + expected.get("segment_id") != f"segment:{index}" + or expected.get("event_ids") != event_ids + ): + raise ValueError( + f"derived segment segment:{index} does not match baseline.expected.json" + ) + episode_count = int(baseline["expected_output"]["episode_count"]) + if episode_count != len(baseline["expected_output"]["findings"]): + raise ValueError("baseline episode count does not match its findings") + return episode_count + + +def evaluate_fixture( + fixture: dict[str, Any], + baseline: dict[str, Any], + baseline_reference: str = "baseline.expected.json", +) -> dict[str, Any]: + rule = fixture["rule"] + window_seconds = int(rule["window_seconds"]) + threshold = int(rule["threshold"]) + if ( + not baseline_reference + or "/" in baseline_reference + or "\\" in baseline_reference + ): + raise ValueError("baseline_reference must be a privacy-safe file name") + segments = activity_segments(fixture["events"], window_seconds) + baseline_episode_count = _validate_input_pair(fixture, baseline, segments) + 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} + 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": candidate.first_seen.isoformat().replace( + "+00:00", "Z" + ), + "last_seen": candidate.last_seen.isoformat().replace("+00:00", "Z"), + "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 = sorted( + candidate.candidate_id + for candidate in candidates + if event_id in candidate.event_ids + ) + 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": str(events[0]["timestamp"]), + "last_seen": str(events[-1]["timestamp"]), + "candidate_windows": [ + _candidate_json(candidate, candidates, 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.", + ], + }, + } + validate_oracle(fixture, oracle) + return oracle + + +def validate_oracle(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 + } + 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 = sorted( + candidate_id + for candidate_id, candidate in candidates.items() + if decision["event_id"] in candidate["event_ids"] + ) + if decision.get("candidate_ids", []) != expected_candidates: + raise ValueError( + "event decision candidate_ids must match candidate evidence" + ) + for candidate in candidate_records: + if not set(candidate["event_ids"]).issubset(events): + raise ValueError( + "candidate evidence must stay inside its baseline segment" + ) + 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 _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()) From 1a0db3132306438b86bb229e01c1d58063ffbf35 Mon Sep 17 00:00:00 2001 From: stacknil Date: Tue, 25 Aug 2026 11:12:24 +0800 Subject: [PATCH 2/4] test(research): lock candidate oracle semantics --- tests/test_episode_candidate.py | 92 +++++++++++++++++++++++++++++++++ 1 file changed, 92 insertions(+) create mode 100644 tests/test_episode_candidate.py diff --git a/tests/test_episode_candidate.py b/tests/test_episode_candidate.py new file mode 100644 index 0000000..b7a1d9e --- /dev/null +++ b/tests/test_episode_candidate.py @@ -0,0 +1,92 @@ +import copy +import json +import sys +import unittest +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, +) + + +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, 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, oracle) + + +if __name__ == "__main__": + unittest.main() From fccc176d5b10d7a6155d908c5d4beb776865d00b Mon Sep 17 00:00:00 2001 From: stacknil Date: Tue, 25 Aug 2026 11:12:36 +0800 Subject: [PATCH 3/4] docs(episodes): record candidate oracle outcome --- docs/adr/0001-episode-semantics-boundaries.md | 28 +- ...andidate.window-separated-v1.expected.json | 272 ++++++++++++++++++ 2 files changed, 296 insertions(+), 4 deletions(-) create mode 100644 tests/fixtures/episode_semantics_v0.7/continuous_background_two_peaks/candidate.window-separated-v1.expected.json diff --git a/docs/adr/0001-episode-semantics-boundaries.md b/docs/adr/0001-episode-semantics-boundaries.md index b58e893..d9519c3 100644 --- a/docs/adr/0001-episode-semantics-boundaries.md +++ b/docs/adr/0001-episode-semantics-boundaries.md @@ -110,10 +110,27 @@ rejects more than 200 fixture events or 1,000 candidates per segment. These limits make research execution bounded; they do not make the algorithm suitable for `Detector::analyze()`. -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 binds the evaluator to the committed baseline and fails closed +when fixture identity, rule configuration, inclusive boundary semantics, or the +ordered activity-segment partition differs. It also 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 @@ -148,3 +165,6 @@ remain unchanged. - [`Detector tests`](../../tests/test_detector.cpp) - [`Candidate selection core`](../../scripts/episode_candidate_core.py) - [`Candidate core tests`](../../tests/test_episode_candidate_core.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/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." + ] + } +} From a52d8829c7fa6972e0a386ed7da58452f720f2c6 Mon Sep 17 00:00:00 2001 From: stacknil Date: Wed, 26 Aug 2026 10:11:05 +0800 Subject: [PATCH 4/4] fix(research): bind oracle validation to source artifacts --- scripts/evaluate_episode_candidate.py | 226 ++++++++++++-------------- tests/test_episode_candidate.py | 38 ++++- 2 files changed, 136 insertions(+), 128 deletions(-) diff --git a/scripts/evaluate_episode_candidate.py b/scripts/evaluate_episode_candidate.py index 3fbb1e0..48c429c 100644 --- a/scripts/evaluate_episode_candidate.py +++ b/scripts/evaluate_episode_candidate.py @@ -8,82 +8,66 @@ import sys from datetime import datetime from pathlib import Path -from typing import Any, Sequence +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, - activity_segments, 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, - activity_segments, enumerate_candidate_windows, ordered_events, parse_timestamp, select_window_separated_candidates, ) -RULE_FIELDS = ( - "rule_id", - "grouping_key", - "subject", - "threshold", - "window_seconds", - "window_boundary", -) - - -def _hash_append(value: int, text: str) -> int: - for byte in text.encode("utf-8"): - value ^= byte - value = (value * 1099511628211) & ((1 << 64) - 1) - value ^= 0xFF - return (value * 1099511628211) & ((1 << 64) - 1) - - -def _finding_id(rule: dict[str, Any], candidate: CandidateWindow) -> str: - value = 14695981039346656037 - fields = [ - str(rule["rule_id"]), - str(rule["grouping_key"]), - str(rule["subject"]), - candidate.first_seen.strftime("%Y-%m-%d %H:%M:%S"), - candidate.last_seen.strftime("%Y-%m-%d %H:%M:%S"), - str(rule["threshold"]), - str(candidate.event_count), - str(candidate.event_count), - *candidate.event_ids, - ] - for field in fields: - value = _hash_append(value, field) - return f"finding:{rule['rule_id']}:{value:016x}" +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, - candidates: Sequence[CandidateWindow], + memberships: dict[str, tuple[str, ...]], selected_ids: set[str], ) -> dict[str, Any]: overlaps = sorted( - { - event_id - for other in candidates - if other.candidate_id != candidate.candidate_id - for event_id in set(candidate.event_ids).intersection(other.event_ids) - } + 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": candidate.first_seen.isoformat().replace("+00:00", "Z"), - "last_seen": candidate.last_seen.isoformat().replace("+00:00", "Z"), + "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": { @@ -106,66 +90,24 @@ def _candidate_json( } -def _rule_projection(rule: dict[str, Any]) -> dict[str, Any]: - return {key: rule[key] for key in RULE_FIELDS} - - -def _validate_input_pair( - fixture: dict[str, Any], - baseline: dict[str, Any], - activity_segments: Sequence[Sequence[dict[str, Any]]], -) -> int: - if fixture.get("format") != "loglens.episode_research_fixture.v1": - raise ValueError("unsupported research fixture format") - if baseline.get("format") != "loglens.episode_baseline_expected.v1": - raise ValueError("unsupported baseline format") - if baseline.get("fixture_id") != fixture.get("fixture_id"): - raise ValueError("baseline fixture_id does not match the research fixture") - - rule = fixture["rule"] - if rule.get("window_boundary") != "inclusive": - raise ValueError( - "candidate algorithm v1 supports only inclusive window boundaries" - ) - if _rule_projection(baseline["rule"]) != _rule_projection(rule): - raise ValueError("baseline rule does not match the research fixture") - - baseline_segments = baseline["derived_input"]["activity_segments"] - if len(baseline_segments) != len(activity_segments): - raise ValueError("baseline activity segment count does not match derived input") - for index, (events, expected) in enumerate( - zip(activity_segments, baseline_segments), start=1 +def _validate_baseline_reference(baseline_reference: str) -> None: + if ( + not baseline_reference + or "/" in baseline_reference + or "\\" in baseline_reference ): - event_ids = [str(event["event_id"]) for event in events] - if ( - expected.get("segment_id") != f"segment:{index}" - or expected.get("event_ids") != event_ids - ): - raise ValueError( - f"derived segment segment:{index} does not match baseline.expected.json" - ) - episode_count = int(baseline["expected_output"]["episode_count"]) - if episode_count != len(baseline["expected_output"]["findings"]): - raise ValueError("baseline episode count does not match its findings") - return episode_count + raise ValueError("baseline_reference must be a privacy-safe file name") -def evaluate_fixture( +def _build_oracle( fixture: dict[str, Any], - baseline: dict[str, Any], - baseline_reference: str = "baseline.expected.json", + 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"]) - if ( - not baseline_reference - or "/" in baseline_reference - or "\\" in baseline_reference - ): - raise ValueError("baseline_reference must be a privacy-safe file name") - segments = activity_segments(fixture["events"], window_seconds) - baseline_episode_count = _validate_input_pair(fixture, baseline, segments) output_segments: list[dict[str, Any]] = [] episode_index = 0 @@ -175,6 +117,9 @@ def evaluate_fixture( 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 } @@ -184,24 +129,18 @@ def evaluate_fixture( episodes.append( { "episode_index": episode_index, - "finding_id": _finding_id(rule, candidate), + "finding_id": finding_id(rule, candidate), "candidate_id": candidate.candidate_id, "event_ids": list(candidate.event_ids), - "first_seen": candidate.first_seen.isoformat().replace( - "+00:00", "Z" - ), - "last_seen": candidate.last_seen.isoformat().replace("+00:00", "Z"), + "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 = sorted( - candidate.candidate_id - for candidate in candidates - if event_id in candidate.event_ids - ) + containing = list(memberships.get(event_id, ())) included = event_id in selected_event_ids event_decisions.append( { @@ -226,10 +165,14 @@ def evaluate_fixture( { "segment_id": segment_id, "event_ids": event_ids, - "first_seen": str(events[0]["timestamp"]), - "last_seen": str(events[-1]["timestamp"]), + "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, candidates, selected_ids) + _candidate_json(candidate, memberships, selected_ids) for candidate in candidates ], "selected_episodes": episodes, @@ -245,7 +188,7 @@ def evaluate_fixture( "version": "1", "status": "candidate", }, - "rule": _rule_projection(rule), + "rule": rule_projection(rule), "segments": output_segments, "comparison": { "baseline_reference": baseline_reference, @@ -260,14 +203,29 @@ def evaluate_fixture( ], }, } - validate_oracle(fixture, oracle) return oracle -def validate_oracle(fixture: dict[str, Any], oracle: dict[str, Any]) -> None: +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"]): + if oracle.get("rule") != rule_projection(fixture["rule"]): raise ValueError("oracle rule must match the research fixture") fixture_ids = [ @@ -291,6 +249,15 @@ def validate_oracle(fixture: dict[str, Any], oracle: dict[str, Any]) -> None: 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() @@ -319,7 +286,7 @@ def validate_oracle(fixture: dict[str, Any], oracle: dict[str, Any]) -> None: last_seen=parse_timestamp(candidate["last_seen"]), threshold_crossing_event_id="", ) - if episode["finding_id"] != _finding_id(fixture["rule"], materialized): + if episode["finding_id"] != finding_id(fixture["rule"], materialized): raise ValueError( "selected episode finding_id must match its candidate evidence" ) @@ -351,20 +318,11 @@ def validate_oracle(fixture: dict[str, Any], oracle: dict[str, Any]) -> None: "included event decisions must equal selected episode evidence" ) for decision in segment["event_decisions"]: - expected_candidates = sorted( - candidate_id - for candidate_id, candidate in candidates.items() - if decision["event_id"] in candidate["event_ids"] - ) + 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" ) - for candidate in candidate_records: - if not set(candidate["event_ids"]).issubset(events): - raise ValueError( - "candidate evidence must stay inside its baseline segment" - ) 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)): @@ -373,6 +331,22 @@ def validate_oracle(fixture: dict[str, Any], oracle: dict[str, Any]) -> None: 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")) diff --git a/tests/test_episode_candidate.py b/tests/test_episode_candidate.py index b7a1d9e..43197cb 100644 --- a/tests/test_episode_candidate.py +++ b/tests/test_episode_candidate.py @@ -2,6 +2,7 @@ import json import sys import unittest +from datetime import timedelta, timezone from pathlib import Path @@ -12,6 +13,7 @@ evaluate_fixture, validate_oracle, ) +from scripts.episode_candidate_core import parse_timestamp # noqa: E402 FIXTURE_ROOT = ( @@ -75,7 +77,7 @@ def test_generated_oracle_matches_committed_fixture_and_cross_references( ) ) - validate_oracle(self.fixture, oracle) + validate_oracle(self.fixture, self.baseline, oracle) self.assertEqual(oracle, expected) def test_validator_rejects_duplicate_event_decisions(self) -> None: @@ -85,8 +87,40 @@ def test_validator_rejects_duplicate_event_decisions(self) -> None: ) with self.assertRaisesRegex(ValueError, "exactly one event decision"): - validate_oracle(self.fixture, oracle) + 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()