From 0ec667c9c9d77fffa0b96feb2c0ae6759590b1f0 Mon Sep 17 00:00:00 2001 From: stacknil Date: Wed, 26 Aug 2026 10:04:49 +0800 Subject: [PATCH 1/5] feat(research): bind candidate inputs to v0.6 baseline --- scripts/episode_baseline_contract.py | 322 +++++++++++++++++++++++++++ 1 file changed, 322 insertions(+) create mode 100644 scripts/episode_baseline_contract.py diff --git a/scripts/episode_baseline_contract.py b/scripts/episode_baseline_contract.py new file mode 100644 index 0000000..c8c30d7 --- /dev/null +++ b/scripts/episode_baseline_contract.py @@ -0,0 +1,322 @@ +"""Fail-closed binding between a research fixture and its v0.6 baseline.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, Sequence + +if __package__: + from .episode_candidate_core import ( + CandidateWindow, + activity_segments, + ordered_events, + parse_timestamp, + ) +else: + from episode_candidate_core import ( + CandidateWindow, + activity_segments, + ordered_events, + parse_timestamp, + ) + + +RULE_FIELDS = ( + "rule_id", + "grouping_key", + "subject", + "threshold", + "window_seconds", + "window_boundary", +) +SUPPORTED_RULE = { + "rule_id": "brute_force", + "grouping_key": "source_ip", + "window_boundary": "inclusive", + "signal_kind": "ssh_failed_password", + "counts_as_terminal_auth_failure": True, +} +BASELINE_ALGORITHM = { + "id": "v0.6.activity_segments_best_count_window", + "implementation": "src/detector.cpp", + "segment_boundary": "split when an adjacent event gap is greater than the rule window", + "selection_tie_break": "keep the first maximum encountered in chronological scan order", +} + + +@dataclass(frozen=True) +class BaselineContext: + segments: list[list[dict[str, Any]]] + selections: tuple[CandidateWindow, ...] + episode_count: int + + +def rule_projection(rule: dict[str, Any]) -> dict[str, Any]: + return {key: rule[key] for key in RULE_FIELDS} + + +def canonical_oracle_timestamp(timestamp: datetime) -> str: + return timestamp.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + + +def canonical_finding_timestamp(timestamp: datetime) -> str: + return timestamp.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") + + +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"]), + canonical_finding_timestamp(candidate.first_seen), + canonical_finding_timestamp(candidate.last_seen), + 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 _validate_fixture(fixture: dict[str, Any]) -> None: + if fixture.get("format") != "loglens.episode_research_fixture.v1": + raise ValueError("unsupported research fixture format") + rule = fixture["rule"] + if any(rule.get(field) != value for field, value in SUPPORTED_RULE.items()): + raise ValueError( + "candidate v1 supports only brute_force source_ip terminal-failure rules" + ) + subject = str(rule.get("subject", "")) + if not subject: + raise ValueError("candidate v1 requires a non-empty source_ip subject") + if int(rule.get("threshold", 0)) < 1 or int(rule.get("window_seconds", 0)) < 1: + raise ValueError("candidate v1 threshold and window_seconds must be positive") + if any( + event.get("event_type") != SUPPORTED_RULE["signal_kind"] + or str(event.get("source_ip", "")) != subject + for event in fixture["events"] + ): + raise ValueError( + "candidate v1 fixture events must match the brute_force rule subject" + ) + + +def _baseline_selections( + segments: Sequence[Sequence[dict[str, Any]]], + threshold: int, + window_seconds: int, +) -> tuple[CandidateWindow, ...]: + selections: list[CandidateWindow] = [] + for events in segments: + start = 0 + best_start = 0 + best_end = 0 + best_count = 0 + for end, event in enumerate(events): + end_timestamp = parse_timestamp(str(event["timestamp"])) + while ( + start < end + and ( + end_timestamp + - parse_timestamp(str(events[start]["timestamp"])) + ).total_seconds() + > window_seconds + ): + start += 1 + count = end - start + 1 + if count > best_count: + best_start, best_end, best_count = start, end, count + if best_count < threshold: + continue + selected = events[best_start : best_end + 1] + selections.append( + CandidateWindow( + event_ids=tuple(str(event["event_id"]) for event in selected), + first_seen=parse_timestamp(str(selected[0]["timestamp"])), + last_seen=parse_timestamp(str(selected[-1]["timestamp"])), + threshold_crossing_event_id=str( + selected[threshold - 1]["event_id"] + ), + ) + ) + return tuple(selections) + + +def _expected_findings( + rule: dict[str, Any], selections: Sequence[CandidateWindow] +) -> list[dict[str, Any]]: + return [ + { + "finding_id": finding_id(rule, candidate), + "episode_index": index, + "rule_id": "brute_force", + "subject_kind": "source_ip", + "subject": str(rule["subject"]), + "grouping_key": "source_ip", + "threshold": int(rule["threshold"]), + "observed_count": candidate.event_count, + "event_count": candidate.event_count, + "window_start": canonical_finding_timestamp(candidate.first_seen), + "window_end": canonical_finding_timestamp(candidate.last_seen), + "evidence_event_ids": list(candidate.event_ids), + "verdict_boundary": "triage_signal_not_compromise_or_attribution", + } + for index, candidate in enumerate(selections, start=1) + ] + + +def _contiguous_events( + event_ids: Sequence[str], segments: Sequence[Sequence[dict[str, Any]]] +) -> list[dict[str, Any]] | None: + target = list(event_ids) + for events in segments: + segment_ids = [str(event["event_id"]) for event in events] + for start in range(len(segment_ids) - len(target) + 1): + if segment_ids[start : start + len(target)] == target: + return list(events[start : start + len(target)]) + return None + + +def _validate_candidate_windows( + baseline: dict[str, Any], + segments: Sequence[Sequence[dict[str, Any]]], + selections: Sequence[CandidateWindow], + threshold: int, + window_seconds: int, +) -> None: + candidate_ids: set[str] = set() + event_sequences: set[tuple[str, ...]] = set() + declared_selected: set[tuple[str, ...]] = set() + expected_selected = {candidate.event_ids for candidate in selections} + for record in baseline["candidate_windows"]: + candidate_id = str(record.get("candidate_id", "")) + event_ids = tuple(str(item) for item in record.get("event_ids", [])) + if not candidate_id or candidate_id in candidate_ids: + raise ValueError("baseline candidate IDs must be non-empty and unique") + if not event_ids or event_ids in event_sequences: + raise ValueError("baseline candidate event sequences must be unique") + candidate_ids.add(candidate_id) + event_sequences.add(event_ids) + events = _contiguous_events(event_ids, segments) + if events is None: + raise ValueError("baseline candidate must be contiguous inside one segment") + first_seen = parse_timestamp(str(events[0]["timestamp"])) + last_seen = parse_timestamp(str(events[-1]["timestamp"])) + try: + recorded_first = parse_timestamp(str(record["first_seen"])) + recorded_last = parse_timestamp(str(record["last_seen"])) + except (KeyError, TypeError, ValueError) as error: + raise ValueError("baseline candidate timestamps are invalid") from error + if ( + len(event_ids) < threshold + or (last_seen - first_seen).total_seconds() > window_seconds + or recorded_first != first_seen + or recorded_last != last_seen + or record.get("event_count") != len(event_ids) + or record.get("threshold_crossing_event_id") + != event_ids[threshold - 1] + or record.get("threshold_met") is not True + ): + raise ValueError("baseline candidate evidence does not match fixture events") + decision = record.get("decision") + if decision == "selected" and event_ids in expected_selected: + declared_selected.add(event_ids) + elif decision != "not_selected" or event_ids in expected_selected: + raise ValueError("baseline candidate decision disagrees with v0.6 selection") + if not str(record.get("decision_reason", "")): + raise ValueError("baseline candidate decision requires a reason") + if declared_selected != expected_selected: + raise ValueError("baseline selected windows do not match v0.6 selection") + + +def validate_fixture_baseline( + fixture: dict[str, Any], baseline: dict[str, Any] +) -> BaselineContext: + _validate_fixture(fixture) + rule = fixture["rule"] + window_seconds = int(rule["window_seconds"]) + threshold = int(rule["threshold"]) + segments = activity_segments(fixture["events"], window_seconds) + 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") + if baseline.get("rule") != rule_projection(rule): + raise ValueError("baseline rule does not match the research fixture") + if baseline.get("algorithm") != BASELINE_ALGORITHM: + raise ValueError("baseline algorithm contract is stale or unsupported") + + ordered = ordered_events(fixture["events"]) + derived = baseline["derived_input"] + records = derived["activity_segments"] + if ( + derived.get("event_count") != len(ordered) + or derived.get("activity_segment_count") != len(segments) + or len(records) != len(segments) + ): + raise ValueError("baseline activity segment count does not match derived input") + for index, (events, record) in enumerate(zip(segments, records), start=1): + event_ids = [str(event["event_id"]) for event in events] + gaps = [ + ( + parse_timestamp(str(events[position]["timestamp"])) + - parse_timestamp(str(events[position - 1]["timestamp"])) + ).total_seconds() + for position in range(1, len(events)) + ] + try: + recorded_first = parse_timestamp(str(record["first_seen"])) + recorded_last = parse_timestamp(str(record["last_seen"])) + except (KeyError, TypeError, ValueError) as error: + raise ValueError("baseline segment timestamps are invalid") from error + if ( + record.get("segment_id") != f"segment:{index}" + or record.get("event_ids") != event_ids + or recorded_first != parse_timestamp(str(events[0]["timestamp"])) + or recorded_last != parse_timestamp(str(events[-1]["timestamp"])) + or record.get("max_adjacent_gap_seconds") != int(max(gaps, default=0)) + ): + raise ValueError("baseline derived segment does not match fixture events") + + selections = _baseline_selections(segments, threshold, window_seconds) + _validate_candidate_windows( + baseline, segments, selections, threshold, window_seconds + ) + expected_output = baseline["expected_output"] + expected_findings = _expected_findings(rule, selections) + if ( + expected_output.get("episode_count") != len(expected_findings) + or expected_output.get("findings") != expected_findings + ): + raise ValueError("baseline findings do not match the v0.6 selection") + selected_ids = { + event_id for selection in selections for event_id in selection.event_ids + } + excluded_ids = [ + str(event["event_id"]) + for event in ordered + if str(event["event_id"]) not in selected_ids + ] + decisions = expected_output.get("excluded_event_decisions", []) + if [item.get("event_id") for item in decisions] != excluded_ids or any( + item.get("decision") != "excluded" or not str(item.get("reason_code", "")) + for item in decisions + ): + raise ValueError("baseline excluded decisions do not partition fixture events") + return BaselineContext( + segments=segments, + selections=selections, + episode_count=len(expected_findings), + ) From 52d204524e815a245a7f6539afb5647fdeb34811 Mon Sep 17 00:00:00 2001 From: stacknil Date: Wed, 26 Aug 2026 10:04:50 +0800 Subject: [PATCH 2/5] test(research): reject stale baseline contracts --- ...est_episode_candidate_baseline_contract.py | 118 ++++++++++++++++++ 1 file changed, 118 insertions(+) create mode 100644 tests/test_episode_candidate_baseline_contract.py diff --git a/tests/test_episode_candidate_baseline_contract.py b/tests/test_episode_candidate_baseline_contract.py new file mode 100644 index 0000000..81a7a26 --- /dev/null +++ b/tests/test_episode_candidate_baseline_contract.py @@ -0,0 +1,118 @@ +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.episode_baseline_contract import ( # noqa: E402 + canonical_oracle_timestamp, + finding_id, + validate_fixture_baseline, +) +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 EpisodeBaselineContractTests(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_committed_baseline_replays_v06_selection_and_identity(self) -> None: + context = validate_fixture_baseline(self.fixture, self.baseline) + + self.assertEqual(context.episode_count, 1) + self.assertEqual(len(context.selections), 1) + self.assertEqual( + finding_id(self.fixture["rule"], context.selections[0]), + "finding:brute_force:584fd14b544a7959", + ) + + def test_equivalent_timezone_offsets_preserve_baseline_identity(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() + + original = validate_fixture_baseline(self.fixture, self.baseline) + shifted = validate_fixture_baseline(offset_fixture, self.baseline) + + self.assertEqual( + [canonical_oracle_timestamp(item.first_seen) for item in shifted.selections], + [canonical_oracle_timestamp(item.first_seen) for item in original.selections], + ) + self.assertEqual( + [finding_id(offset_fixture["rule"], item) for item in shifted.selections], + [finding_id(self.fixture["rule"], item) for item in original.selections], + ) + + def test_unsupported_rule_contracts_fail_closed(self) -> None: + mutations = { + "rule": ("rule_id", "multi_user_probing"), + "grouping": ("grouping_key", "username"), + "boundary": ("window_boundary", "exclusive"), + "signal": ("signal_kind", "sudo_command"), + } + + for label, (field, value) in mutations.items(): + with self.subTest(label=label): + fixture = copy.deepcopy(self.fixture) + baseline = copy.deepcopy(self.baseline) + fixture["rule"][field] = value + if field in baseline["rule"]: + baseline["rule"][field] = value + with self.assertRaisesRegex(ValueError, "candidate v1 supports only"): + validate_fixture_baseline(fixture, baseline) + + def test_fixture_events_must_match_the_rule_subject(self) -> None: + fixture = copy.deepcopy(self.fixture) + fixture["events"][0]["source_ip"] = "203.0.113.99" + + with self.assertRaisesRegex(ValueError, "fixture events"): + validate_fixture_baseline(fixture, self.baseline) + + def test_semantically_stale_baseline_fails_closed(self) -> None: + mutations = { + "algorithm": lambda baseline: baseline["algorithm"].__setitem__( + "id", "stale.algorithm" + ), + "finding identity": lambda baseline: baseline["expected_output"][ + "findings" + ][0].__setitem__("finding_id", "finding:brute_force:stale"), + "selected window": lambda baseline: baseline["candidate_windows"][ + 0 + ].__setitem__("last_seen", "2026-03-10T09:01:30Z"), + "segment evidence": lambda baseline: baseline["derived_input"][ + "activity_segments" + ][0].__setitem__("max_adjacent_gap_seconds", 539), + } + + for label, mutate in mutations.items(): + with self.subTest(label=label): + baseline = copy.deepcopy(self.baseline) + mutate(baseline) + with self.assertRaisesRegex(ValueError, "baseline"): + validate_fixture_baseline(self.fixture, baseline) + + +if __name__ == "__main__": + unittest.main() From c9c92e22fb296ebcade3ada94e8763a0db29254a Mon Sep 17 00:00:00 2001 From: stacknil Date: Wed, 26 Aug 2026 10:04:52 +0800 Subject: [PATCH 3/5] docs(episodes): define fail-closed baseline binding --- docs/adr/0001-episode-semantics-boundaries.md | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/docs/adr/0001-episode-semantics-boundaries.md b/docs/adr/0001-episode-semantics-boundaries.md index b58e893..713d003 100644 --- a/docs/adr/0001-episode-semantics-boundaries.md +++ b/docs/adr/0001-episode-semantics-boundaries.md @@ -110,6 +110,15 @@ 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 fail-closed baseline contract is a separate prerequisite for oracle +materialization. Candidate v1 accepts only the `brute_force` / `source_ip` / +inclusive-window fixture contract, replays the v0.6 best-count selection, and +requires the declared segments, selected window, finding identity, and excluded +event partition to match that replay. Timestamp comparison uses instants and +finding identity uses canonical UTC, so equivalent offsets cannot create a new +episode identity. 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` @@ -148,3 +157,5 @@ 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) +- [`Baseline contract`](../../scripts/episode_baseline_contract.py) +- [`Baseline contract tests`](../../tests/test_episode_candidate_baseline_contract.py) From cbeb97b48995b575bc2243e61c80f9660677fa5d Mon Sep 17 00:00:00 2001 From: stacknil Date: Thu, 27 Aug 2026 00:38:18 +0800 Subject: [PATCH 4/5] fix(research): reject noncanonical rule scalar types --- scripts/episode_baseline_contract.py | 16 +++++++++++++--- ...test_episode_candidate_baseline_contract.py | 18 ++++++++++++++++++ 2 files changed, 31 insertions(+), 3 deletions(-) diff --git a/scripts/episode_baseline_contract.py b/scripts/episode_baseline_contract.py index c8c30d7..1661a81 100644 --- a/scripts/episode_baseline_contract.py +++ b/scripts/episode_baseline_contract.py @@ -94,18 +94,28 @@ def _validate_fixture(fixture: dict[str, Any]) -> None: if fixture.get("format") != "loglens.episode_research_fixture.v1": raise ValueError("unsupported research fixture format") rule = fixture["rule"] + subject = rule.get("subject") + threshold = rule.get("threshold") + window_seconds = rule.get("window_seconds") + terminal_failure = rule.get("counts_as_terminal_auth_failure") + if ( + type(subject) is not str + or type(threshold) is not int + or type(window_seconds) is not int + or type(terminal_failure) is not bool + ): + raise ValueError("candidate v1 rule requires canonical scalar types") if any(rule.get(field) != value for field, value in SUPPORTED_RULE.items()): raise ValueError( "candidate v1 supports only brute_force source_ip terminal-failure rules" ) - subject = str(rule.get("subject", "")) if not subject: raise ValueError("candidate v1 requires a non-empty source_ip subject") - if int(rule.get("threshold", 0)) < 1 or int(rule.get("window_seconds", 0)) < 1: + if threshold < 1 or window_seconds < 1: raise ValueError("candidate v1 threshold and window_seconds must be positive") if any( event.get("event_type") != SUPPORTED_RULE["signal_kind"] - or str(event.get("source_ip", "")) != subject + or event.get("source_ip") != subject for event in fixture["events"] ): raise ValueError( diff --git a/tests/test_episode_candidate_baseline_contract.py b/tests/test_episode_candidate_baseline_contract.py index 81a7a26..4aec78f 100644 --- a/tests/test_episode_candidate_baseline_contract.py +++ b/tests/test_episode_candidate_baseline_contract.py @@ -90,6 +90,24 @@ def test_fixture_events_must_match_the_rule_subject(self) -> None: with self.assertRaisesRegex(ValueError, "fixture events"): validate_fixture_baseline(fixture, self.baseline) + def test_rule_scalar_types_fail_closed(self) -> None: + mutations = { + "string threshold": ("threshold", "5"), + "string window": ("window_seconds", "600"), + "numeric subject": ("subject", 77), + "numeric terminal flag": ("counts_as_terminal_auth_failure", 1), + } + + for label, (field, value) in mutations.items(): + with self.subTest(label=label): + fixture = copy.deepcopy(self.fixture) + baseline = copy.deepcopy(self.baseline) + fixture["rule"][field] = value + if field in baseline["rule"]: + baseline["rule"][field] = value + with self.assertRaisesRegex(ValueError, "canonical scalar types"): + validate_fixture_baseline(fixture, baseline) + def test_semantically_stale_baseline_fails_closed(self) -> None: mutations = { "algorithm": lambda baseline: baseline["algorithm"].__setitem__( From 8e23d902fbf2c36095d7ed9b18a1a42a5a703987 Mon Sep 17 00:00:00 2001 From: stacknil Date: Thu, 27 Aug 2026 00:47:19 +0800 Subject: [PATCH 5/5] docs(episodes): define canonical rule scalar types --- docs/adr/0001-episode-semantics-boundaries.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/docs/adr/0001-episode-semantics-boundaries.md b/docs/adr/0001-episode-semantics-boundaries.md index 713d003..aca5443 100644 --- a/docs/adr/0001-episode-semantics-boundaries.md +++ b/docs/adr/0001-episode-semantics-boundaries.md @@ -116,8 +116,10 @@ inclusive-window fixture contract, replays the v0.6 best-count selection, and requires the declared segments, selected window, finding identity, and excluded event partition to match that replay. Timestamp comparison uses instants and finding identity uses canonical UTC, so equivalent offsets cannot create a new -episode identity. Unsupported rules or semantically stale baselines are rejected -before candidate selection. +episode identity. Rule scalars are type-strict: the subject is a string, threshold +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