research(episodes): add bounded window-separated candidate core

.github/workflows/ci.yml

@@ -22,6 +22,11 @@ jobs:
       - name: Checkout
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0  # v7.0.0
 
+      - name: Set up Python
+        uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1  # v6.3.0
+        with:
+          python-version: "3.14"
+
       - name: Configure
         run: >-
           cmake -S . -B build

.gitignore

@@ -4,5 +4,7 @@ out/
 report.md
 report.json
 *.exe
+__pycache__/
+*.py[cod]
 !tests/fixtures/report_contracts/**/report.md
 !tests/fixtures/report_contracts/**/report.json

CMakeLists.txt

@@ -113,6 +113,25 @@ if(BUILD_TESTING)
             $<TARGET_FILE:loglens>
             ${CMAKE_CURRENT_BINARY_DIR}/report_contract_output
     )
+
+    find_package(Python3 3.9 COMPONENTS Interpreter QUIET)
+    if(Python3_Interpreter_FOUND)
+        add_test(
+            NAME episode_candidate_research
+            COMMAND
+                ${Python3_EXECUTABLE}
+                -B
+                -m
+                unittest
+                discover
+                -s
+                ${CMAKE_CURRENT_SOURCE_DIR}/tests
+                -p
+                test_episode_candidate*.py
+        )
+    else()
+        message(STATUS "Python 3.9+ not found; skipping research fixture tests")
+    endif()
 endif()
 
 if(LOGLENS_BUILD_FUZZERS)

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

@@ -85,6 +85,36 @@ candidate windows, selected windows, episode indexes, and the reason an event
 was included or excluded. The fixture must not claim compromise, intent,
 attribution, or an incident boundary.
 
+## Candidate experiment: window-separated weighted intervals
+
+The 2026-08-25 experiment tests one narrow hypothesis: after v0.6 activity
+segmentation, a deterministic global selection over dense windows can recover
+both peaks in the continuous-background fixture without reusing event evidence.
+It does not select the v0.7 production algorithm.
+
+The research evaluator enumerates every contiguous window that meets the
+threshold inside the inclusive rule window. It then selects a compatible set
+with these ordered objectives:
+
+1. A later selected window must start strictly more than one rule window after
+   the previous selected window ends. An exact-boundary gap remains one cooldown
+   episode.
+2. Maximize the total number of covered events.
+3. Minimize total selected-window span, then episode count.
+4. Resolve a remaining tie by the chronological window key.
+
+This first slice adds only the bounded selection core and boundary tests.
+Exhaustive window materialization copies many event-ID sequences, and selection
+scans candidate predecessors. Worst-case cost is super-quadratic, so the core
+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.
+
 ## Alternatives considered
 
 - **Keep adjacent-gap segmentation as the final model** - simple and
@@ -116,3 +146,5 @@ attribution, or an incident boundary.
 - [`v0.6 Episode Policy`](../release-v0.6.0.md#episode-policy)
 - [`Detector implementation`](../../src/detector.cpp)
 - [`Detector tests`](../../tests/test_detector.cpp)
+- [`Candidate selection core`](../../scripts/episode_candidate_core.py)
+- [`Candidate core tests`](../../tests/test_episode_candidate_core.py)

scripts/episode_candidate_core.py

@@ -0,0 +1,171 @@
+"""Bounded research core for deterministic episode candidate selection."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any, Sequence
+
+
+MAX_FIXTURE_EVENTS = 200
+MAX_CANDIDATES_PER_SEGMENT = 1_000
+
+
+@dataclass(frozen=True)
+class CandidateWindow:
+    event_ids: tuple[str, ...]
+    first_seen: datetime
+    last_seen: datetime
+    threshold_crossing_event_id: str
+
+    @property
+    def candidate_id(self) -> str:
+        return f"candidate:{self.event_ids[0]}..{self.event_ids[-1]}"
+
+    @property
+    def event_count(self) -> int:
+        return len(self.event_ids)
+
+    @property
+    def span_seconds(self) -> int:
+        return int((self.last_seen - self.first_seen).total_seconds())
+
+
+def parse_timestamp(value: str) -> datetime:
+    try:
+        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError as error:
+        raise ValueError("timestamp must use ISO 8601 syntax") from error
+    if parsed.tzinfo is None:
+        raise ValueError("timestamp must include an offset")
+    return parsed
+
+
+def ordered_events(events: Sequence[dict[str, Any]]) -> list[dict[str, Any]]:
+    if len(events) > MAX_FIXTURE_EVENTS:
+        raise ValueError(
+            f"research evaluator accepts at most {MAX_FIXTURE_EVENTS} events"
+        )
+    event_ids = [str(event["event_id"]) for event in events]
+    if len(event_ids) != len(set(event_ids)):
+        raise ValueError("event IDs must be unique")
+    return sorted(
+        events,
+        key=lambda event: (
+            parse_timestamp(str(event["timestamp"])),
+            int(event["line_number"]),
+            str(event["event_id"]),
+        ),
+    )
+
+
+def enumerate_candidate_windows(
+    events: Sequence[dict[str, Any]], threshold: int, window_seconds: int
+) -> list[CandidateWindow]:
+    """Enumerate every contiguous threshold window inside the inclusive rule window."""
+    if threshold < 1 or window_seconds < 1:
+        raise ValueError("threshold and window_seconds must be positive")
+    ordered = ordered_events(events)
+    candidates: list[CandidateWindow] = []
+    for start in range(len(ordered)):
+        first_seen = parse_timestamp(str(ordered[start]["timestamp"]))
+        for end in range(start, len(ordered)):
+            last_seen = parse_timestamp(str(ordered[end]["timestamp"]))
+            if (last_seen - first_seen).total_seconds() > window_seconds:
+                break
+            if end - start + 1 < threshold:
+                continue
+            candidates.append(
+                CandidateWindow(
+                    event_ids=tuple(
+                        str(event["event_id"]) for event in ordered[start : end + 1]
+                    ),
+                    first_seen=first_seen,
+                    last_seen=last_seen,
+                    threshold_crossing_event_id=str(
+                        ordered[start + threshold - 1]["event_id"]
+                    ),
+                )
+            )
+            if len(candidates) > MAX_CANDIDATES_PER_SEGMENT:
+                raise ValueError(
+                    "research evaluator candidate limit exceeded "
+                    f"({MAX_CANDIDATES_PER_SEGMENT} per segment)"
+                )
+    return candidates
+
+
+def _selection_score(selection: Sequence[CandidateWindow]) -> tuple[int, int, int]:
+    return (
+        sum(candidate.event_count for candidate in selection),
+        -sum(candidate.span_seconds for candidate in selection),
+        -len(selection),
+    )
+
+
+def _selection_key(selection: Sequence[CandidateWindow]) -> tuple[tuple[Any, ...], ...]:
+    return tuple(
+        (candidate.first_seen, candidate.last_seen, candidate.event_ids)
+        for candidate in selection
+    )
+
+
+def _prefer(
+    left: list[CandidateWindow], right: list[CandidateWindow]
+) -> list[CandidateWindow]:
+    left_score = _selection_score(left)
+    right_score = _selection_score(right)
+    if left_score != right_score:
+        return left if left_score > right_score else right
+    return left if _selection_key(left) <= _selection_key(right) else right
+
+
+def select_window_separated_candidates(
+    candidates: Sequence[CandidateWindow], window_seconds: int
+) -> list[CandidateWindow]:
+    """Select an optimal set whose adjacent windows are more than one rule window apart."""
+    ordered = sorted(
+        candidates,
+        key=lambda candidate: (
+            candidate.last_seen,
+            candidate.first_seen,
+            candidate.event_ids,
+        ),
+    )
+    best: list[list[CandidateWindow]] = [[]]
+    for index, candidate in enumerate(ordered):
+        predecessor = -1
+        for prior in range(index - 1, -1, -1):
+            gap = (candidate.first_seen - ordered[prior].last_seen).total_seconds()
+            if gap > window_seconds:
+                predecessor = prior
+                break
+        take = best[predecessor + 1] + [candidate]
+        best.append(_prefer(take, best[index]))
+    return sorted(
+        best[-1],
+        key=lambda candidate: (
+            candidate.first_seen,
+            candidate.last_seen,
+            candidate.event_ids,
+        ),
+    )
+
+
+def activity_segments(
+    events: Sequence[dict[str, Any]], window_seconds: int
+) -> list[list[dict[str, Any]]]:
+    ordered = ordered_events(events)
+    if not ordered:
+        return []
+    segments: list[list[dict[str, Any]]] = [[ordered[0]]]
+    for event in ordered[1:]:
+        previous = segments[-1][-1]
+        gap = (
+            parse_timestamp(str(event["timestamp"]))
+            - parse_timestamp(str(previous["timestamp"]))
+        ).total_seconds()
+        if gap > window_seconds:
+            segments.append([])
+        segments[-1].append(event)
+    return segments

tests/test_episode_candidate_core.py

@@ -0,0 +1,80 @@
+import sys
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO_ROOT))
+
+from scripts.episode_candidate_core import (  # noqa: E402
+    enumerate_candidate_windows,
+    select_window_separated_candidates,
+)
+
+
+def make_events(offsets: list[int]) -> list[dict[str, object]]:
+    origin = datetime(2026, 3, 10, 9, 0, tzinfo=timezone.utc)
+    return [
+        {
+            "event_id": f"line:{index}",
+            "line_number": index,
+            "timestamp": (origin + timedelta(seconds=offset))
+            .isoformat()
+            .replace("+00:00", "Z"),
+            "event_type": "ssh_failed_password",
+            "source_ip": "203.0.113.77",
+        }
+        for index, offset in enumerate(offsets, start=1)
+    ]
+
+
+class WindowSeparatedSelectionTests(unittest.TestCase):
+    def test_threshold_and_window_boundaries_are_inclusive(self) -> None:
+        exact = make_events([0, 597, 598, 599, 600])
+        below = exact[:-1]
+
+        self.assertEqual(len(enumerate_candidate_windows(exact, 5, 600)), 1)
+        self.assertEqual(enumerate_candidate_windows(below, 5, 600), [])
+
+    def test_candidate_cooldown_requires_more_than_one_rule_window(self) -> None:
+        exact_gap = make_events([0, 1, 2, 3, 4, 604, 605, 606, 607, 608])
+        over_gap = make_events([0, 1, 2, 3, 4, 605, 606, 607, 608, 609])
+
+        exact_selected = select_window_separated_candidates(
+            enumerate_candidate_windows(exact_gap, 5, 600), 600
+        )
+        over_selected = select_window_separated_candidates(
+            enumerate_candidate_windows(over_gap, 5, 600), 600
+        )
+
+        self.assertEqual(
+            [candidate.event_ids for candidate in exact_selected],
+            [tuple(f"line:{i}" for i in range(1, 6))],
+        )
+        self.assertEqual(
+            [candidate.event_ids for candidate in over_selected],
+            [
+                tuple(f"line:{i}" for i in range(1, 6)),
+                tuple(f"line:{i}" for i in range(6, 11)),
+            ],
+        )
+
+    def test_overlapping_candidates_materialize_one_maximal_window(self) -> None:
+        candidates = enumerate_candidate_windows(
+            make_events([0, 1, 2, 3, 4, 5]), 5, 600
+        )
+        selected = select_window_separated_candidates(candidates, 600)
+
+        self.assertEqual(len(selected), 1)
+        self.assertEqual(selected[0].event_ids, tuple(f"line:{i}" for i in range(1, 7)))
+
+    def test_research_size_limits_fail_closed(self) -> None:
+        with self.assertRaisesRegex(ValueError, "at most 200 events"):
+            enumerate_candidate_windows(make_events(list(range(201))), 5, 600)
+        with self.assertRaisesRegex(ValueError, "candidate limit exceeded"):
+            enumerate_candidate_windows(make_events(list(range(50))), 5, 600)
+
+
+if __name__ == "__main__":
+    unittest.main()