From 86cafc2dadda102d0945bb3a42a6868a7e10bab4 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Fri, 17 Jul 2026 11:44:07 +0300 Subject: [PATCH 01/10] feat: add idempotent transcript window backfill --- src/brainlayer/backfill.py | 113 +++++++++++++++++++++++++++++++ src/brainlayer/cli/__init__.py | 61 +++++++++++++++-- tests/test_backfill.py | 88 ++++++++++++++++++++++++ tests/test_watch_backfill_cli.py | 87 ++++++++++++++++++++++++ 4 files changed, 343 insertions(+), 6 deletions(-) create mode 100644 src/brainlayer/backfill.py create mode 100644 tests/test_backfill.py diff --git a/src/brainlayer/backfill.py b/src/brainlayer/backfill.py new file mode 100644 index 00000000..191c7ff1 --- /dev/null +++ b/src/brainlayer/backfill.py @@ -0,0 +1,113 @@ +"""Idempotent time-window filtering for transcript watcher backfills.""" + +from __future__ import annotations + +from collections.abc import Callable +from datetime import UTC, datetime +from pathlib import Path + +from .watcher_bridge import FlushWatermarks + + +def _parse_utc(value: str) -> datetime: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=UTC) + return parsed.astimezone(UTC) + + +def parse_backfill_window(since: str | None, until: str | None) -> tuple[datetime, datetime]: + """Parse a required half-open UTC interval and reject empty ranges.""" + if not since or not until: + raise ValueError("--since and --until must be provided together") + try: + parsed_since = _parse_utc(since) + parsed_until = _parse_utc(until) + except ValueError as exc: + raise ValueError("--since and --until must be ISO 8601 timestamps") from exc + if parsed_since >= parsed_until: + raise ValueError("--since must be earlier than --until") + return parsed_since, parsed_until + + +def window_registry_suffix(since: datetime, until: datetime) -> str: + """Return a stable filesystem-safe name for a backfill interval.""" + return f"{since:%Y%m%dT%H%M%SZ}-{until:%Y%m%dT%H%M%SZ}" + + +def _contains_ordered(parts: tuple[str, ...], expected: tuple[str, ...]) -> bool: + position = 0 + for part in parts: + if part == expected[position]: + position += 1 + if position == len(expected): + return True + return False + + +def is_legacy_excluded_path(path: str | Path) -> bool: + """Identify roots blocked by the blanket denylist retired in July 2026.""" + parts = Path(path).expanduser().parts + return any( + _contains_ordered(parts, expected) + for expected in ( + (".claude", "projects", "subagents"), + (".codex", "sessions"), + (".cursor", "agent-transcripts"), + (".gemini", "sessions"), + ) + ) + + +class WindowedFlush: + """Filter normalized watcher entries while confirming every scanned offset.""" + + def __init__( + self, + downstream: Callable[[list[dict]], dict[str, int] | None], + *, + since: datetime, + until: datetime, + source_predicate: Callable[[str | Path], bool] | None = None, + ) -> None: + self.downstream = downstream + self.since = since + self.until = until + self.source_predicate = source_predicate + self.scanned_entries = 0 + self.matched_entries = 0 + self.inserted_chunks = 0 + + def _matches(self, entry: dict) -> bool: + source_file = entry.get("_source_file") + if self.source_predicate and (not isinstance(source_file, str) or not self.source_predicate(source_file)): + return False + timestamp = entry.get("timestamp") + if not isinstance(timestamp, str): + return False + try: + parsed = _parse_utc(timestamp) + except ValueError: + return False + return self.since <= parsed < self.until + + def __call__(self, entries: list[dict]) -> FlushWatermarks: + matched = [entry for entry in entries if self._matches(entry)] + downstream_result = self.downstream(matched) if matched else FlushWatermarks() + watermarks = dict(downstream_result or {}) + for entry in entries: + source_file = entry.get("_source_file") + offset = entry.get("_line_end_offset") + if isinstance(source_file, str) and isinstance(offset, int): + watermarks[source_file] = max(watermarks.get(source_file, 0), offset) + + inserted = int(getattr(downstream_result, "inserted", len(matched))) + downstream_skipped = int(getattr(downstream_result, "skipped", 0)) + self.scanned_entries += len(entries) + self.matched_entries += len(matched) + self.inserted_chunks += inserted + return FlushWatermarks( + watermarks, + inserted=inserted, + skipped=downstream_skipped + len(entries) - len(matched), + ) diff --git a/src/brainlayer/cli/__init__.py b/src/brainlayer/cli/__init__.py index a171e155..a14946dd 100644 --- a/src/brainlayer/cli/__init__.py +++ b/src/brainlayer/cli/__init__.py @@ -3482,16 +3482,40 @@ def watch_backfill( "--registry", help="Offset registry path. Defaults to the live BrainLayer offsets file.", ), + since: Optional[str] = typer.Option(None, "--since", help="Inclusive ISO 8601 entry timestamp."), + until: Optional[str] = typer.Option(None, "--until", help="Exclusive ISO 8601 entry timestamp."), + legacy_excluded_only: bool = typer.Option( + False, + "--legacy-excluded-only", + help="Replay only transcript roots blocked by the retired blanket denylist.", + ), dry_run: bool = typer.Option(False, "--dry-run", help="Report files that would be replayed without writing."), max_cycles: int = typer.Option(100, "--max-cycles", min=1, help="Maximum poll cycles to run."), ) -> None: """One-shot replay for watched JSONL roots using the durable queue writer path.""" + from ..backfill import WindowedFlush, is_legacy_excluded_path, parse_backfill_window, window_registry_suffix from ..paths import get_db_path from ..watcher import JSONLWatcher, WatchRoot, default_watch_roots from ..watcher_bridge import create_flush_callback db_path = get_db_path() - registry_path = registry.expanduser() if registry else db_path.parent / "offsets.json" + window = None + if since is not None or until is not None: + try: + window = parse_backfill_window(since, until) + except ValueError as exc: + raise typer.BadParameter(str(exc), param_hint="--since/--until") from exc + if legacy_excluded_only and window is None: + raise typer.BadParameter( + "--legacy-excluded-only requires --since and --until", + param_hint="--legacy-excluded-only", + ) + if registry: + registry_path = registry.expanduser() + elif window: + registry_path = db_path.parent / f"backfill-offsets-{window_registry_suffix(*window)}.json" + else: + registry_path = db_path.parent / "offsets.json" watch_roots = [WatchRoot("custom", item) for item in source] if source else default_watch_roots(home=home) if dry_run: @@ -3507,26 +3531,51 @@ def watch_backfill( provider = watcher.provider_for_file(path) provider_counts[provider] = provider_counts.get(provider, 0) + 1 provider_summary = " ".join(f"{provider}={count}" for provider, count in sorted(provider_counts.items())) - rprint(f"candidate_files={len(files)} {provider_summary} processed_entries=0 registry={registry_path}") + window_summary = f" window=[{window[0].isoformat()},{window[1].isoformat()})" if window else "" + scope_summary = " scope=legacy-excluded-only" if legacy_excluded_only else "" + rprint( + f"candidate_files={len(files)} {provider_summary} processed_entries=0{window_summary}{scope_summary} " + f"registry={registry_path}" + ) return + downstream_flush = create_flush_callback(db_path, arbitrated=True) + windowed_flush = ( + WindowedFlush( + downstream_flush, + since=window[0], + until=window[1], + source_predicate=is_legacy_excluded_path if legacy_excluded_only else None, + ) + if window + else None + ) watcher = JSONLWatcher( watch_roots=watch_roots, registry_path=registry_path, - on_flush=create_flush_callback(db_path, arbitrated=True), + on_flush=windowed_flush or downstream_flush, db_path=db_path, ) processed = 0 cycles = 0 while cycles < max_cycles: cycles += 1 + offsets_before = {path: tailer.offset for path, tailer in watcher._tailers.items()} count = watcher.poll_once() - if count == 0: - break processed += count + made_progress = any(tailer.offset > offsets_before.get(path, 0) for path, tailer in watcher._tailers.items()) + if not made_progress: + break watcher.indexer.flush() watcher.registry.flush() - rprint(f"processed_entries={processed} cycles={cycles} registry={registry_path}") + if windowed_flush: + rprint( + f"processed_entries={processed} scanned_entries={windowed_flush.scanned_entries} " + f"matched_entries={windowed_flush.matched_entries} queued_chunks={windowed_flush.inserted_chunks} " + f"cycles={cycles} registry={registry_path}" + ) + else: + rprint(f"processed_entries={processed} cycles={cycles} registry={registry_path}") @app.command("index-fast", hidden=True) diff --git a/tests/test_backfill.py b/tests/test_backfill.py new file mode 100644 index 00000000..160f8ac7 --- /dev/null +++ b/tests/test_backfill.py @@ -0,0 +1,88 @@ +from datetime import UTC, datetime + +from brainlayer.backfill import WindowedFlush, is_legacy_excluded_path, parse_backfill_window +from brainlayer.watcher_bridge import FlushWatermarks + + +def _entry(timestamp: str, offset: int) -> dict: + return { + "type": "assistant", + "timestamp": timestamp, + "_source_file": "/tmp/session.jsonl", + "_line_end_offset": offset, + } + + +def test_windowed_flush_filters_half_open_interval_and_confirms_scanned_offsets(): + received = [] + + def downstream(entries): + received.extend(entries) + return FlushWatermarks( + {"/tmp/session.jsonl": entries[-1]["_line_end_offset"]}, + inserted=len(entries), + ) + + windowed = WindowedFlush( + downstream, + since=datetime(2026, 7, 10, tzinfo=UTC), + until=datetime(2026, 7, 16, tzinfo=UTC), + ) + + result = windowed( + [ + _entry("2026-07-09T23:59:59Z", 100), + _entry("2026-07-10T00:00:00Z", 200), + _entry("2026-07-15T23:59:59Z", 300), + _entry("2026-07-16T00:00:00Z", 400), + ] + ) + + assert [entry["_line_end_offset"] for entry in received] == [200, 300] + assert result == {"/tmp/session.jsonl": 400} + assert result.inserted == 2 + assert result.skipped == 2 + assert windowed.scanned_entries == 4 + assert windowed.matched_entries == 2 + assert windowed.inserted_chunks == 2 + + +def test_windowed_flush_excludes_invalid_timestamps_but_confirms_them(): + windowed = WindowedFlush( + lambda entries: FlushWatermarks(inserted=len(entries)), + since=datetime(2026, 7, 10, tzinfo=UTC), + until=datetime(2026, 7, 16, tzinfo=UTC), + ) + + result = windowed([_entry("not-a-date", 50)]) + + assert result == {"/tmp/session.jsonl": 50} + assert windowed.matched_entries == 0 + + +def test_parse_backfill_window_is_utc_and_rejects_empty_or_reversed_ranges(): + since, until = parse_backfill_window("2026-07-10", "2026-07-16") + + assert since == datetime(2026, 7, 10, tzinfo=UTC) + assert until == datetime(2026, 7, 16, tzinfo=UTC) + + for invalid in ((None, "2026-07-16"), ("2026-07-16", None), ("2026-07-16", "2026-07-10")): + try: + parse_backfill_window(*invalid) + except ValueError: + pass + else: + raise AssertionError(f"expected invalid window: {invalid}") + + +def test_legacy_excluded_path_selects_only_roots_blocked_by_old_policy(tmp_path): + assert is_legacy_excluded_path(tmp_path / ".codex" / "sessions" / "worker.jsonl") + assert is_legacy_excluded_path( + tmp_path / ".cursor" / "projects" / "repo" / "agent-transcripts" / "session" / "worker.jsonl" + ) + assert is_legacy_excluded_path(tmp_path / ".gemini" / "sessions" / "worker.jsonl") + assert is_legacy_excluded_path( + tmp_path / ".claude" / "projects" / "repo" / "session" / "subagents" / "agent-worker.jsonl" + ) + assert not is_legacy_excluded_path(tmp_path / ".claude" / "projects" / "repo" / "direct.jsonl") + assert not is_legacy_excluded_path(tmp_path / ".cursor" / "projects" / "repo" / "state.jsonl") diff --git a/tests/test_watch_backfill_cli.py b/tests/test_watch_backfill_cli.py index bf5c7088..cb7040e6 100644 --- a/tests/test_watch_backfill_cli.py +++ b/tests/test_watch_backfill_cli.py @@ -123,3 +123,90 @@ def test_watch_backfill_indexes_cursor_agent_transcripts_once(tmp_path, monkeypa assert second.exit_code == 0, second.output assert "processed_entries=0" in second.output assert list(queue_dir.glob("watcher-*.jsonl")) == queue_files + + +def test_watch_backfill_indexes_only_requested_window_and_is_idempotent(tmp_path, monkeypatch): + monkeypatch.delenv("BRAINLAYER_INGEST_DENYLIST", raising=False) + transcript = tmp_path / ".claude" / "projects" / "repo" / "session.jsonl" + registry = tmp_path / "window-offsets.json" + queue_dir = tmp_path / "queue" + transcript.parent.mkdir(parents=True) + + def entry(timestamp: str, token: str) -> str: + return json.dumps( + { + "type": "assistant", + "timestamp": timestamp, + "message": { + "role": "assistant", + "content": [ + { + "type": "text", + "text": ( + f"{token} is a substantive backfill sentinel with enough durable technical " + "context to pass classification and chunking thresholds." + ), + } + ], + }, + } + ) + + transcript.write_text( + "\n".join( + [ + entry("2026-07-09T23:59:59Z", "BEFOREWINDOW"), + entry("2026-07-12T12:00:00Z", "INSIDEWINDOW"), + entry("2026-07-16T00:00:00Z", "AFTERWINDOW"), + ] + ) + + "\n" + ) + + args = [ + "watch-backfill", + "--home", + str(tmp_path), + "--registry", + str(registry), + "--since", + "2026-07-10", + "--until", + "2026-07-16", + ] + first = CliRunner().invoke( + app, + args, + env={"BRAINLAYER_QUEUE_DIR": str(queue_dir)}, + ) + + assert first.exit_code == 0, first.output + assert "matched_entries=1" in first.output + assert len(list(queue_dir.glob("watcher-*.jsonl"))) == 1 + + second = CliRunner().invoke( + app, + args, + env={"BRAINLAYER_QUEUE_DIR": str(queue_dir)}, + ) + + assert second.exit_code == 0, second.output + assert "matched_entries=0" in second.output + assert len(list(queue_dir.glob("watcher-*.jsonl"))) == 1 + + +def test_watch_backfill_rejects_legacy_scope_without_window(tmp_path): + result = CliRunner().invoke( + app, + [ + "watch-backfill", + "--home", + str(tmp_path), + "--legacy-excluded-only", + "--dry-run", + ], + ) + + assert result.exit_code != 0 + assert "--legacy-excluded-only requires" in result.output + assert "--since and --until" in result.output From 09eb2ea05ed91b48dc834ab1beb20b752c96487c Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Fri, 17 Jul 2026 11:55:02 +0300 Subject: [PATCH 02/10] test: map CLI entrypoint to CLI suites --- scripts/run_tests.sh | 8 ++++++++ tests/test_run_tests_script.py | 29 +++++++++++++++++++++++++++++ 2 files changed, 37 insertions(+) diff --git a/scripts/run_tests.sh b/scripts/run_tests.sh index a76fb621..a169c1b5 100755 --- a/scripts/run_tests.sh +++ b/scripts/run_tests.sh @@ -142,6 +142,14 @@ map_changed_files_to_pytests() { changed_files_seen=1 mapped=0 case "$changed" in + src/brainlayer/cli/__init__.py) + for test_path in "$TEST_ROOT"/test_cli*.py "$TEST_ROOT"/test_watch_backfill_cli.py; do + if [ -f "$test_path" ] && ! is_real_db_test_file "$test_path"; then + append_unique "$test_path" + mapped=1 + fi + done + ;; src/brainlayer/mcp/store_handler.py|src/brainlayer/queue_io.py|src/brainlayer/drain.py|src/brainlayer/store.py) for rel in test_store_handler.py test_write_queue.py test_brainstore.py; do test_path="$TEST_ROOT/$rel" diff --git a/tests/test_run_tests_script.py b/tests/test_run_tests_script.py index 49b19620..df6e7984 100644 --- a/tests/test_run_tests_script.py +++ b/tests/test_run_tests_script.py @@ -217,6 +217,35 @@ def test_changed_only_scope_maps_changed_source_to_targeted_tests(tmp_path: Path assert f"{test_root}/ -v" not in logged +def test_changed_only_scope_maps_cli_entrypoint_to_all_cli_tests(tmp_path: Path) -> None: + test_root = tmp_path / "tests" + test_root.mkdir() + (test_root / "test_cli_commands.py").write_text("test placeholder\n") + (test_root / "test_watch_backfill_cli.py").write_text("test placeholder\n") + (test_root / "test_think_recall_integration.py").write_text("test placeholder\n") + + pytest_log, bun_log = _make_stub_bin(tmp_path, pytest_exit=0, bun_exit=0) + + env = _script_env() + env["PATH"] = f"{tmp_path / 'bin'}:{env['PATH']}" + env["BRAINLAYER_TEST_ROOT"] = str(test_root) + env["BRAINLAYER_USE_UV"] = "0" + env["BRAINLAYER_PREPUSH"] = "1" + env["BRAINLAYER_PREPUSH_SCOPE"] = "changed-only" + env["BRAINLAYER_CHANGED_FILES"] = "src/brainlayer/cli/__init__.py" + env["PYTEST_LOG"] = str(pytest_log) + env["BUN_LOG"] = str(bun_log) + + result = subprocess.run(["bash", str(SCRIPT_PATH)], capture_output=True, text=True, env=env) + + assert result.returncode == 0 + logged = pytest_log.read_text() + assert str(test_root / "test_cli_commands.py") in logged + assert str(test_root / "test_watch_backfill_cli.py") in logged + assert "falling back to full pytest unit suite" not in result.stdout + assert f"{test_root}/ -v" not in logged + + def test_changed_only_scope_falls_back_when_mapped_and_unmapped_sources_change(tmp_path: Path) -> None: test_root = tmp_path / "tests" test_root.mkdir() From f32ecb87e46dae1a4a999fd7bedbb49e983b2a86 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Fri, 17 Jul 2026 12:08:41 +0300 Subject: [PATCH 03/10] fix: preserve backfill replay guarantees --- src/brainlayer/backfill.py | 15 +++++++++++---- src/brainlayer/cli/__init__.py | 2 ++ tests/test_backfill.py | 33 +++++++++++++++++++++++++++++++- tests/test_watch_backfill_cli.py | 29 ++++++++++++++++++++++++++++ 4 files changed, 74 insertions(+), 5 deletions(-) diff --git a/src/brainlayer/backfill.py b/src/brainlayer/backfill.py index 191c7ff1..db0862f2 100644 --- a/src/brainlayer/backfill.py +++ b/src/brainlayer/backfill.py @@ -32,7 +32,12 @@ def parse_backfill_window(since: str | None, until: str | None) -> tuple[datetim def window_registry_suffix(since: datetime, until: datetime) -> str: """Return a stable filesystem-safe name for a backfill interval.""" - return f"{since:%Y%m%dT%H%M%SZ}-{until:%Y%m%dT%H%M%SZ}" + def format_timestamp(value: datetime) -> str: + base = f"{value:%Y%m%dT%H%M%S}" + fraction = f"{value.microsecond:06d}" if value.microsecond else "" + return f"{base}{fraction}Z" + + return f"{format_timestamp(since)}-{format_timestamp(until)}" def _contains_ordered(parts: tuple[str, ...], expected: tuple[str, ...]) -> bool: @@ -91,9 +96,13 @@ def _matches(self, entry: dict) -> bool: return False return self.since <= parsed < self.until - def __call__(self, entries: list[dict]) -> FlushWatermarks: + def __call__(self, entries: list[dict]) -> FlushWatermarks | None: matched = [entry for entry in entries if self._matches(entry)] downstream_result = self.downstream(matched) if matched else FlushWatermarks() + self.scanned_entries += len(entries) + self.matched_entries += len(matched) + if downstream_result is None: + return None watermarks = dict(downstream_result or {}) for entry in entries: source_file = entry.get("_source_file") @@ -103,8 +112,6 @@ def __call__(self, entries: list[dict]) -> FlushWatermarks: inserted = int(getattr(downstream_result, "inserted", len(matched))) downstream_skipped = int(getattr(downstream_result, "skipped", 0)) - self.scanned_entries += len(entries) - self.matched_entries += len(matched) self.inserted_chunks += inserted return FlushWatermarks( watermarks, diff --git a/src/brainlayer/cli/__init__.py b/src/brainlayer/cli/__init__.py index a14946dd..29a4ca35 100644 --- a/src/brainlayer/cli/__init__.py +++ b/src/brainlayer/cli/__init__.py @@ -3526,6 +3526,8 @@ def watch_backfill( db_path=db_path, ) files = watcher._discover_jsonl_files() + if legacy_excluded_only: + files = [path for path in files if is_legacy_excluded_path(path)] provider_counts: dict[str, int] = {} for path in files: provider = watcher.provider_for_file(path) diff --git a/tests/test_backfill.py b/tests/test_backfill.py index 160f8ac7..5e120694 100644 --- a/tests/test_backfill.py +++ b/tests/test_backfill.py @@ -1,6 +1,6 @@ from datetime import UTC, datetime -from brainlayer.backfill import WindowedFlush, is_legacy_excluded_path, parse_backfill_window +from brainlayer.backfill import WindowedFlush, is_legacy_excluded_path, parse_backfill_window, window_registry_suffix from brainlayer.watcher_bridge import FlushWatermarks @@ -60,6 +60,37 @@ def test_windowed_flush_excludes_invalid_timestamps_but_confirms_them(): assert windowed.matched_entries == 0 +def test_windowed_flush_does_not_confirm_offsets_when_downstream_returns_none(): + windowed = WindowedFlush( + lambda _entries: None, + since=datetime(2026, 7, 10, tzinfo=UTC), + until=datetime(2026, 7, 16, tzinfo=UTC), + ) + + result = windowed([_entry("2026-07-12T00:00:00Z", 50)]) + + assert result is None + assert windowed.scanned_entries == 1 + assert windowed.matched_entries == 1 + assert windowed.inserted_chunks == 0 + + +def test_window_registry_suffix_preserves_fractional_seconds_without_changing_whole_seconds(): + whole_since = datetime(2026, 7, 10, tzinfo=UTC) + whole_until = datetime(2026, 7, 16, tzinfo=UTC) + first = window_registry_suffix( + whole_since.replace(microsecond=100_000), + whole_until.replace(microsecond=200_000), + ) + second = window_registry_suffix( + whole_since.replace(microsecond=300_000), + whole_until.replace(microsecond=400_000), + ) + + assert window_registry_suffix(whole_since, whole_until) == "20260710T000000Z-20260716T000000Z" + assert first != second + + def test_parse_backfill_window_is_utc_and_rejects_empty_or_reversed_ranges(): since, until = parse_backfill_window("2026-07-10", "2026-07-16") diff --git a/tests/test_watch_backfill_cli.py b/tests/test_watch_backfill_cli.py index cb7040e6..66902926 100644 --- a/tests/test_watch_backfill_cli.py +++ b/tests/test_watch_backfill_cli.py @@ -64,6 +64,35 @@ def test_watch_backfill_dry_run_includes_codex_and_gemini_sessions(tmp_path, mon assert not (tmp_path / "offsets.json").exists() +def test_watch_backfill_legacy_dry_run_counts_only_legacy_excluded_roots(tmp_path, monkeypatch): + monkeypatch.delenv("BRAINLAYER_INGEST_DENYLIST", raising=False) + direct = tmp_path / ".claude" / "projects" / "repo" / "direct.jsonl" + legacy = tmp_path / ".codex" / "sessions" / "2026" / "07" / "worker.jsonl" + for path in (direct, legacy): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"role": "user", "content": "transcript"}) + "\n") + + result = CliRunner().invoke( + app, + [ + "watch-backfill", + "--home", + str(tmp_path), + "--since", + "2026-07-10", + "--until", + "2026-07-16", + "--legacy-excluded-only", + "--dry-run", + ], + ) + + assert result.exit_code == 0, result.output + assert "candidate_files=1" in result.output + assert "codex=1" in result.output + assert "claude=" not in result.output + + def test_watch_backfill_indexes_cursor_agent_transcripts_once(tmp_path, monkeypatch): monkeypatch.delenv("BRAINLAYER_INGEST_DENYLIST", raising=False) transcript = ( From ae0f85b553648f96a462f90077be414c960f3bc8 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Fri, 17 Jul 2026 12:43:01 +0300 Subject: [PATCH 04/10] fix: isolate legacy backfill replay --- src/brainlayer/backfill.py | 6 ++-- src/brainlayer/cli/__init__.py | 5 ++- src/brainlayer/watcher.py | 19 ++++++---- tests/test_backfill.py | 38 ++++++++++++++++++++ tests/test_watch_backfill_cli.py | 61 ++++++++++++++++++++++++++++++++ 5 files changed, 119 insertions(+), 10 deletions(-) diff --git a/src/brainlayer/backfill.py b/src/brainlayer/backfill.py index db0862f2..2ff458dd 100644 --- a/src/brainlayer/backfill.py +++ b/src/brainlayer/backfill.py @@ -23,7 +23,7 @@ def parse_backfill_window(since: str | None, until: str | None) -> tuple[datetim try: parsed_since = _parse_utc(since) parsed_until = _parse_utc(until) - except ValueError as exc: + except (ValueError, OverflowError) as exc: raise ValueError("--since and --until must be ISO 8601 timestamps") from exc if parsed_since >= parsed_until: raise ValueError("--since must be earlier than --until") @@ -87,12 +87,14 @@ def _matches(self, entry: dict) -> bool: source_file = entry.get("_source_file") if self.source_predicate and (not isinstance(source_file, str) or not self.source_predicate(source_file)): return False + if entry.get("_timestamp_synthesized") is True: + return False timestamp = entry.get("timestamp") if not isinstance(timestamp, str): return False try: parsed = _parse_utc(timestamp) - except ValueError: + except (ValueError, OverflowError): return False return self.since <= parsed < self.until diff --git a/src/brainlayer/cli/__init__.py b/src/brainlayer/cli/__init__.py index 29a4ca35..8ee4c0cd 100644 --- a/src/brainlayer/cli/__init__.py +++ b/src/brainlayer/cli/__init__.py @@ -3513,7 +3513,8 @@ def watch_backfill( if registry: registry_path = registry.expanduser() elif window: - registry_path = db_path.parent / f"backfill-offsets-{window_registry_suffix(*window)}.json" + scope_suffix = "-legacy-excluded-only" if legacy_excluded_only else "" + registry_path = db_path.parent / f"backfill-offsets-{window_registry_suffix(*window)}{scope_suffix}.json" else: registry_path = db_path.parent / "offsets.json" watch_roots = [WatchRoot("custom", item) for item in source] if source else default_watch_roots(home=home) @@ -3524,6 +3525,7 @@ def watch_backfill( registry_path=registry_path, on_flush=lambda items: None, db_path=db_path, + respect_denylist=not legacy_excluded_only, ) files = watcher._discover_jsonl_files() if legacy_excluded_only: @@ -3557,6 +3559,7 @@ def watch_backfill( registry_path=registry_path, on_flush=windowed_flush or downstream_flush, db_path=db_path, + respect_denylist=not legacy_excluded_only, ) processed = 0 cycles = 0 diff --git a/src/brainlayer/watcher.py b/src/brainlayer/watcher.py index 40bb3f30..f27edea1 100644 --- a/src/brainlayer/watcher.py +++ b/src/brainlayer/watcher.py @@ -121,12 +121,12 @@ def normalize_provider_entry(entry: dict[str, Any], provider: str) -> dict[str, if not text: return None + source_timestamp = candidate.get("timestamp") or candidate.get("created_at") return { "type": role, "message": {"role": role, "content": [{"type": "text", "text": text}]}, - "timestamp": candidate.get("timestamp") - or candidate.get("created_at") - or datetime.now(timezone.utc).isoformat(), + "timestamp": source_timestamp or datetime.now(timezone.utc).isoformat(), + "_timestamp_synthesized": not isinstance(source_timestamp, str) or not source_timestamp.strip(), "_provider": provider, } @@ -499,6 +499,7 @@ def __init__( health_path: str | Path | None = None, coverage_watchdog: CoverageWatchdog | None = None, max_lines_per_file: int = 100, + respect_denylist: bool = True, ): if watch_roots is not None: self.watch_roots = [WatchRoot(root.provider, root.path, root.glob_pattern) for root in watch_roots] @@ -521,6 +522,7 @@ def __init__( self.poll_interval_s = poll_interval_s self.registry_flush_interval_s = registry_flush_interval_s self.max_lines_per_file = max(1, max_lines_per_file) + self.respect_denylist = respect_denylist self._tailers: dict[str, JSONLTailer] = {} self._file_providers: dict[str, str] = {} self._stop = threading.Event() @@ -543,7 +545,7 @@ def _advance_confirmed_offsets(self, watermarks: dict[str, int]) -> None: self.registry.set(filepath, offset, inode) def provider_for_file(self, filepath: str) -> str: - if is_denylisted(filepath): + if self._is_denylisted(filepath): return "unknown" provider = self._file_providers.get(filepath) @@ -557,6 +559,9 @@ def provider_for_file(self, filepath: str) -> str: return root.provider return "unknown" + def _is_denylisted(self, filepath: str) -> bool: + return self.respect_denylist and is_denylisted(filepath) + def _discover_jsonl_files(self) -> list[str]: """Find all .jsonl files under each watched project, including nested session artifacts.""" discovered: list[tuple[float, str, str]] = [] @@ -574,7 +579,7 @@ def _discover_jsonl_files(self) -> list[str]: for f in files: if f.is_file(): path = str(f) - if is_denylisted(path): + if self._is_denylisted(path): continue try: mtime = f.stat().st_mtime @@ -775,13 +780,13 @@ def poll_once(self) -> int: files = self._discover_jsonl_files() for filepath in list(self._tailers): - if is_denylisted(filepath): + if self._is_denylisted(filepath): self._tailers.pop(filepath, None) self._file_providers.pop(filepath, None) self.registry.remove(filepath) for filepath in files: - if is_denylisted(filepath): + if self._is_denylisted(filepath): self._tailers.pop(filepath, None) self._file_providers.pop(filepath, None) self.registry.remove(filepath) diff --git a/tests/test_backfill.py b/tests/test_backfill.py index 5e120694..b49e3cbe 100644 --- a/tests/test_backfill.py +++ b/tests/test_backfill.py @@ -1,6 +1,7 @@ from datetime import UTC, datetime from brainlayer.backfill import WindowedFlush, is_legacy_excluded_path, parse_backfill_window, window_registry_suffix +from brainlayer.watcher import normalize_provider_entry from brainlayer.watcher_bridge import FlushWatermarks @@ -60,6 +61,43 @@ def test_windowed_flush_excludes_invalid_timestamps_but_confirms_them(): assert windowed.matched_entries == 0 +def test_windowed_flush_excludes_overflowing_timezone_timestamps(): + windowed = WindowedFlush( + lambda entries: FlushWatermarks(inserted=len(entries)), + since=datetime(2026, 7, 10, tzinfo=UTC), + until=datetime(2026, 7, 16, tzinfo=UTC), + ) + + result = windowed([_entry("0001-01-01T00:00:00+23:59", 50)]) + + assert result == {"/tmp/session.jsonl": 50} + assert windowed.matched_entries == 0 + + +def test_windowed_flush_rejects_timestamps_synthesized_during_normalization(): + normalized = normalize_provider_entry( + {"role": "user", "content": "historical undated transcript"}, + "codex", + ) + assert normalized is not None + normalized.update( + { + "_source_file": "/tmp/session.jsonl", + "_line_end_offset": 50, + } + ) + windowed = WindowedFlush( + lambda entries: FlushWatermarks(inserted=len(entries)), + since=datetime(2020, 1, 1, tzinfo=UTC), + until=datetime(2100, 1, 1, tzinfo=UTC), + ) + + result = windowed([normalized]) + + assert result == {"/tmp/session.jsonl": 50} + assert windowed.matched_entries == 0 + + def test_windowed_flush_does_not_confirm_offsets_when_downstream_returns_none(): windowed = WindowedFlush( lambda _entries: None, diff --git a/tests/test_watch_backfill_cli.py b/tests/test_watch_backfill_cli.py index 66902926..4eebb8e1 100644 --- a/tests/test_watch_backfill_cli.py +++ b/tests/test_watch_backfill_cli.py @@ -91,6 +91,67 @@ def test_watch_backfill_legacy_dry_run_counts_only_legacy_excluded_roots(tmp_pat assert "candidate_files=1" in result.output assert "codex=1" in result.output assert "claude=" not in result.output + assert "scope=legacy-excluded-only" in result.output + assert "-legacy-excluded-only.json" in result.output + + +def test_watch_backfill_legacy_scope_bypasses_current_subagent_denylist(tmp_path, monkeypatch): + monkeypatch.delenv("BRAINLAYER_INGEST_DENYLIST", raising=False) + transcript = ( + tmp_path + / ".claude" + / "projects" + / "repo" + / "session" + / "subagents" + / "agent-worker.jsonl" + ) + transcript.parent.mkdir(parents=True) + transcript.write_text( + json.dumps( + { + "type": "assistant", + "timestamp": "2026-07-12T12:00:00Z", + "message": { + "role": "assistant", + "content": [ + { + "type": "text", + "text": ( + "Legacy subagent backfill sentinel records the durable implementation decision, " + "the reason the previous indexing policy excluded this transcript, the exact " + "migration window, and the verification contract needed to replay it safely " + "without discarding the raw JSONL source or advancing offsets before persistence." + ), + } + ], + }, + } + ) + + "\n" + ) + queue_dir = tmp_path / "queue" + + result = CliRunner().invoke( + app, + [ + "watch-backfill", + "--home", + str(tmp_path), + "--since", + "2026-07-10", + "--until", + "2026-07-16", + "--legacy-excluded-only", + "--max-cycles", + "5", + ], + env={"BRAINLAYER_QUEUE_DIR": str(queue_dir)}, + ) + + assert result.exit_code == 0, result.output + assert "matched_entries=1" in result.output + assert len(list(queue_dir.glob("watcher-*.jsonl"))) == 1 def test_watch_backfill_indexes_cursor_agent_transcripts_once(tmp_path, monkeypatch): From 8cbd5afb1c70db6d387318baab53de4de748769d Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Fri, 17 Jul 2026 12:49:44 +0300 Subject: [PATCH 05/10] style: satisfy backfill format gate --- src/brainlayer/backfill.py | 1 + tests/test_watch_backfill_cli.py | 10 +--------- 2 files changed, 2 insertions(+), 9 deletions(-) diff --git a/src/brainlayer/backfill.py b/src/brainlayer/backfill.py index 2ff458dd..0b79c8d2 100644 --- a/src/brainlayer/backfill.py +++ b/src/brainlayer/backfill.py @@ -32,6 +32,7 @@ def parse_backfill_window(since: str | None, until: str | None) -> tuple[datetim def window_registry_suffix(since: datetime, until: datetime) -> str: """Return a stable filesystem-safe name for a backfill interval.""" + def format_timestamp(value: datetime) -> str: base = f"{value:%Y%m%dT%H%M%S}" fraction = f"{value.microsecond:06d}" if value.microsecond else "" diff --git a/tests/test_watch_backfill_cli.py b/tests/test_watch_backfill_cli.py index 4eebb8e1..9d140b2a 100644 --- a/tests/test_watch_backfill_cli.py +++ b/tests/test_watch_backfill_cli.py @@ -97,15 +97,7 @@ def test_watch_backfill_legacy_dry_run_counts_only_legacy_excluded_roots(tmp_pat def test_watch_backfill_legacy_scope_bypasses_current_subagent_denylist(tmp_path, monkeypatch): monkeypatch.delenv("BRAINLAYER_INGEST_DENYLIST", raising=False) - transcript = ( - tmp_path - / ".claude" - / "projects" - / "repo" - / "session" - / "subagents" - / "agent-worker.jsonl" - ) + transcript = tmp_path / ".claude" / "projects" / "repo" / "session" / "subagents" / "agent-worker.jsonl" transcript.parent.mkdir(parents=True) transcript.write_text( json.dumps( From bf8874871508db6dd76e4d0fd7f8f3d7c00e0db3 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Fri, 17 Jul 2026 13:47:06 +0300 Subject: [PATCH 06/10] fix: retain active exclusions during legacy replay --- src/brainlayer/cli/__init__.py | 5 +- src/brainlayer/ingest_denylist.py | 34 +++++++++++-- src/brainlayer/watcher.py | 11 +++- tests/test_ingest_denylist.py | 41 ++++++++++++++- tests/test_watch_backfill_cli.py | 85 +++++++++++++++++++++++++++---- 5 files changed, 158 insertions(+), 18 deletions(-) diff --git a/src/brainlayer/cli/__init__.py b/src/brainlayer/cli/__init__.py index 8ee4c0cd..34bd2cba 100644 --- a/src/brainlayer/cli/__init__.py +++ b/src/brainlayer/cli/__init__.py @@ -3494,6 +3494,7 @@ def watch_backfill( ) -> None: """One-shot replay for watched JSONL roots using the durable queue writer path.""" from ..backfill import WindowedFlush, is_legacy_excluded_path, parse_backfill_window, window_registry_suffix + from ..ingest_denylist import is_legacy_backfill_denylisted from ..paths import get_db_path from ..watcher import JSONLWatcher, WatchRoot, default_watch_roots from ..watcher_bridge import create_flush_callback @@ -3525,7 +3526,7 @@ def watch_backfill( registry_path=registry_path, on_flush=lambda items: None, db_path=db_path, - respect_denylist=not legacy_excluded_only, + denylist_predicate=is_legacy_backfill_denylisted if legacy_excluded_only else None, ) files = watcher._discover_jsonl_files() if legacy_excluded_only: @@ -3559,7 +3560,7 @@ def watch_backfill( registry_path=registry_path, on_flush=windowed_flush or downstream_flush, db_path=db_path, - respect_denylist=not legacy_excluded_only, + denylist_predicate=is_legacy_backfill_denylisted if legacy_excluded_only else None, ) processed = 0 cycles = 0 diff --git a/src/brainlayer/ingest_denylist.py b/src/brainlayer/ingest_denylist.py index 945e286d..b125bd27 100644 --- a/src/brainlayer/ingest_denylist.py +++ b/src/brainlayer/ingest_denylist.py @@ -10,6 +10,12 @@ BRAINLAYER_INGEST_DENYLIST_ENV = "BRAINLAYER_INGEST_DENYLIST" DEFAULT_INGEST_DENYLIST = ("~/.claude/projects/**/wf_*/**",) +RETIRED_BLANKET_INGEST_DENYLIST = ( + "~/.claude/projects/*/**/subagents/**", + "~/.codex/sessions/**", + "~/.cursor/**/agent-transcripts/**", + "~/.gemini/sessions/**", +) _SUBAGENT_ATTRIBUTION_CACHE: dict[str, tuple[int, int, str | None]] = {} @@ -50,6 +56,15 @@ def _match_parts(path_parts: tuple[str, ...], pattern_parts: tuple[str, ...]) -> return fnmatch.fnmatchcase(path_parts[0], pattern_parts[0]) and _match_parts(path_parts[1:], pattern_parts[1:]) +def _matches_patterns(candidate: Path, patterns: tuple[str, ...]) -> bool: + homes = _inferred_homes(candidate) + return any( + _match_parts(candidate.parts, expanded_pattern.parts) + for pattern in patterns + for expanded_pattern in _expand_globs(pattern, homes) + ) + + def _is_claude_subagent(path: Path) -> bool: parts = path.parts return ".claude" in parts and "projects" in parts and "subagents" in parts @@ -91,12 +106,21 @@ def _claude_subagent_attribution(path: Path) -> str | None: def is_denylisted(path: str | Path, *, unknown_subagent_is_denylisted: bool = True) -> bool: """Return True when a source path is under an ingest-denylisted transcript root.""" candidate = Path(os.path.abspath(os.path.expanduser(str(path)))) - homes = _inferred_homes(candidate) - for pattern in _configured_patterns(): - for expanded_pattern in _expand_globs(pattern, homes): - if _match_parts(candidate.parts, expanded_pattern.parts): - return True + if _matches_patterns(candidate, _configured_patterns()): + return True if BRAINLAYER_INGEST_DENYLIST_ENV not in os.environ and _is_claude_subagent(candidate): attribution = _claude_subagent_attribution(candidate) return (attribution is None and unknown_subagent_is_denylisted) or attribution == "brain-worker" return False + + +def is_legacy_backfill_denylisted(path: str | Path) -> bool: + """Apply current safety exclusions while retiring only the old blanket roots.""" + candidate = Path(os.path.abspath(os.path.expanduser(str(path)))) + configured = tuple(pattern for pattern in _configured_patterns() if pattern not in RETIRED_BLANKET_INGEST_DENYLIST) + active_patterns = tuple(dict.fromkeys((*DEFAULT_INGEST_DENYLIST, *configured))) + if _matches_patterns(candidate, active_patterns): + return True + if _is_claude_subagent(candidate): + return _claude_subagent_attribution(candidate) == "brain-worker" + return False diff --git a/src/brainlayer/watcher.py b/src/brainlayer/watcher.py index f27edea1..ca02848a 100644 --- a/src/brainlayer/watcher.py +++ b/src/brainlayer/watcher.py @@ -500,6 +500,8 @@ def __init__( coverage_watchdog: CoverageWatchdog | None = None, max_lines_per_file: int = 100, respect_denylist: bool = True, + unknown_subagent_is_denylisted: bool = True, + denylist_predicate: Callable[[str | Path], bool] | None = None, ): if watch_roots is not None: self.watch_roots = [WatchRoot(root.provider, root.path, root.glob_pattern) for root in watch_roots] @@ -523,6 +525,8 @@ def __init__( self.registry_flush_interval_s = registry_flush_interval_s self.max_lines_per_file = max(1, max_lines_per_file) self.respect_denylist = respect_denylist + self.unknown_subagent_is_denylisted = unknown_subagent_is_denylisted + self.denylist_predicate = denylist_predicate self._tailers: dict[str, JSONLTailer] = {} self._file_providers: dict[str, str] = {} self._stop = threading.Event() @@ -560,7 +564,12 @@ def provider_for_file(self, filepath: str) -> str: return "unknown" def _is_denylisted(self, filepath: str) -> bool: - return self.respect_denylist and is_denylisted(filepath) + if self.denylist_predicate is not None: + return self.respect_denylist and self.denylist_predicate(filepath) + return self.respect_denylist and is_denylisted( + filepath, + unknown_subagent_is_denylisted=self.unknown_subagent_is_denylisted, + ) def _discover_jsonl_files(self) -> list[str]: """Find all .jsonl files under each watched project, including nested session artifacts.""" diff --git a/tests/test_ingest_denylist.py b/tests/test_ingest_denylist.py index 65a10966..ce6329dd 100644 --- a/tests/test_ingest_denylist.py +++ b/tests/test_ingest_denylist.py @@ -1,7 +1,11 @@ import json from pathlib import Path -from brainlayer.ingest_denylist import BRAINLAYER_INGEST_DENYLIST_ENV, is_denylisted +from brainlayer.ingest_denylist import ( + BRAINLAYER_INGEST_DENYLIST_ENV, + is_denylisted, + is_legacy_backfill_denylisted, +) def _write_subagent(path: Path, attribution: str | None) -> Path: @@ -166,3 +170,38 @@ def test_explicit_environment_override_can_deny_an_otherwise_allowed_provider(mo assert is_denylisted(tmp_path / ".codex" / "sessions" / "worker.jsonl") assert not is_denylisted(tmp_path / ".gemini" / "sessions" / "worker.jsonl") + + +def test_legacy_backfill_retires_only_blanket_patterns(monkeypatch, tmp_path): + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv( + BRAINLAYER_INGEST_DENYLIST_ENV, + ",".join( + [ + "~/.claude/projects/*/**/subagents/**", + "~/.claude/projects/**/wf_*/**", + "~/.codex/sessions/**", + "~/.cursor/**/agent-transcripts/**", + "~/.gemini/sessions/**", + "~/.codex/sessions/**/private/**", + ] + ), + ) + ordinary = _write_subagent( + tmp_path / ".claude" / "projects" / "repo" / "session" / "subagents" / "agent-ordinary.jsonl", + None, + ) + brain_worker = _write_subagent( + tmp_path / ".claude" / "projects" / "repo" / "session" / "subagents" / "agent-brain.jsonl", + "brain-worker", + ) + workflow = _write_subagent( + tmp_path / ".claude" / "projects" / "repo" / "wf_test" / "agent-workflow.jsonl", + "workflow-worker", + ) + + assert not is_legacy_backfill_denylisted(ordinary) + assert is_legacy_backfill_denylisted(brain_worker) + assert is_legacy_backfill_denylisted(workflow) + assert not is_legacy_backfill_denylisted(tmp_path / ".codex" / "sessions" / "worker.jsonl") + assert is_legacy_backfill_denylisted(tmp_path / ".codex" / "sessions" / "private" / "worker.jsonl") diff --git a/tests/test_watch_backfill_cli.py b/tests/test_watch_backfill_cli.py index 9d140b2a..8f48ca2a 100644 --- a/tests/test_watch_backfill_cli.py +++ b/tests/test_watch_backfill_cli.py @@ -95,9 +95,74 @@ def test_watch_backfill_legacy_dry_run_counts_only_legacy_excluded_roots(tmp_pat assert "-legacy-excluded-only.json" in result.output -def test_watch_backfill_legacy_scope_bypasses_current_subagent_denylist(tmp_path, monkeypatch): +def test_watch_backfill_legacy_scope_keeps_current_denylist(tmp_path, monkeypatch): monkeypatch.delenv("BRAINLAYER_INGEST_DENYLIST", raising=False) - transcript = tmp_path / ".claude" / "projects" / "repo" / "session" / "subagents" / "agent-worker.jsonl" + subagents = tmp_path / ".claude" / "projects" / "repo" / "session" / "subagents" + allowed = subagents / "agent-ordinary.jsonl" + unattributed = subagents / "agent-historical.jsonl" + denied = subagents / "agent-brain-worker.jsonl" + workflow = tmp_path / ".claude" / "projects" / "repo" / "wf_test" / "worker.jsonl" + for path in (allowed, unattributed, denied, workflow): + path.parent.mkdir(parents=True, exist_ok=True) + + def entry(attribution: str | None, label: str | None = None) -> str: + worker = label or attribution or "historical-worker" + return json.dumps( + { + "type": "assistant", + "timestamp": "2026-07-12T12:00:00Z", + **({"attributionAgent": attribution} if attribution else {}), + "message": { + "role": "assistant", + "content": [ + { + "type": "text", + "text": ( + f"Legacy {worker} backfill sentinel records the durable implementation " + "decision, why the retired blanket policy excluded this transcript, the exact " + "migration window, and the verification contract needed to replay it safely." + ), + } + ], + }, + } + ) + + allowed.write_text(entry("repo-worker") + "\n") + unattributed.write_text(entry(None, "historical-worker") + "\n") + denied.write_text(entry("brain-worker") + "\n") + workflow.write_text(entry("workflow-worker") + "\n") + queue_dir = tmp_path / "queue" + + result = CliRunner().invoke( + app, + [ + "watch-backfill", + "--home", + str(tmp_path), + "--since", + "2026-07-10", + "--until", + "2026-07-16", + "--legacy-excluded-only", + "--max-cycles", + "5", + ], + env={"BRAINLAYER_QUEUE_DIR": str(queue_dir)}, + ) + + assert result.exit_code == 0, result.output + assert "matched_entries=2" in result.output + queued = "".join(path.read_text() for path in queue_dir.glob("watcher-*.jsonl")) + assert queued.count('"kind": "watcher_chunk"') == 2 + assert "repo-worker" in queued + assert "historical-worker" in queued + assert "brain-worker" not in queued + assert "workflow-worker" not in queued + + +def test_watch_backfill_legacy_scope_honors_configured_denylist(tmp_path, monkeypatch): + transcript = tmp_path / ".codex" / "sessions" / "2026" / "07" / "blocked" / "worker.jsonl" transcript.parent.mkdir(parents=True) transcript.write_text( json.dumps( @@ -110,10 +175,8 @@ def test_watch_backfill_legacy_scope_bypasses_current_subagent_denylist(tmp_path { "type": "text", "text": ( - "Legacy subagent backfill sentinel records the durable implementation decision, " - "the reason the previous indexing policy excluded this transcript, the exact " - "migration window, and the verification contract needed to replay it safely " - "without discarding the raw JSONL source or advancing offsets before persistence." + "Configured denylist backfill sentinel records the durable implementation decision, " + "the exact migration window, and the verification contract for a safe replay." ), } ], @@ -138,12 +201,16 @@ def test_watch_backfill_legacy_scope_bypasses_current_subagent_denylist(tmp_path "--max-cycles", "5", ], - env={"BRAINLAYER_QUEUE_DIR": str(queue_dir)}, + env={ + "BRAINLAYER_INGEST_DENYLIST": "~/.codex/sessions/**/blocked/**", + "BRAINLAYER_QUEUE_DIR": str(queue_dir), + }, ) assert result.exit_code == 0, result.output - assert "matched_entries=1" in result.output - assert len(list(queue_dir.glob("watcher-*.jsonl"))) == 1 + assert "processed_entries=0" in result.output + assert "matched_entries=0" in result.output + assert not list(queue_dir.glob("watcher-*.jsonl")) def test_watch_backfill_indexes_cursor_agent_transcripts_once(tmp_path, monkeypatch): From 37b790576389e964a1662bfe2aceedeafbdf8e24 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Fri, 17 Jul 2026 14:30:40 +0300 Subject: [PATCH 07/10] fix: preserve backfill progress integrity --- src/brainlayer/backfill.py | 18 ++++++++-- src/brainlayer/cli/__init__.py | 2 ++ src/brainlayer/watcher.py | 59 ++++++++++++++++++++++++++++---- tests/test_backfill.py | 53 ++++++++++++++++++++++++++++ tests/test_jsonl_watcher.py | 22 ++++++++++++ tests/test_watch_backfill_cli.py | 34 ++++++++++++++++++ 6 files changed, 178 insertions(+), 10 deletions(-) diff --git a/src/brainlayer/backfill.py b/src/brainlayer/backfill.py index 0b79c8d2..c06f02f5 100644 --- a/src/brainlayer/backfill.py +++ b/src/brainlayer/backfill.py @@ -100,18 +100,30 @@ def _matches(self, entry: dict) -> bool: return self.since <= parsed < self.until def __call__(self, entries: list[dict]) -> FlushWatermarks | None: - matched = [entry for entry in entries if self._matches(entry)] + match_flags = [self._matches(entry) for entry in entries] + matched = [entry for entry, matches in zip(entries, match_flags, strict=True) if matches] downstream_result = self.downstream(matched) if matched else FlushWatermarks() self.scanned_entries += len(entries) self.matched_entries += len(matched) if downstream_result is None: return None watermarks = dict(downstream_result or {}) - for entry in entries: + by_source: dict[str, list[tuple[int, bool]]] = {} + for entry, matches in zip(entries, match_flags, strict=True): source_file = entry.get("_source_file") offset = entry.get("_line_end_offset") if isinstance(source_file, str) and isinstance(offset, int): - watermarks[source_file] = max(watermarks.get(source_file, 0), offset) + by_source.setdefault(source_file, []).append((offset, matches)) + for source_file, source_entries in by_source.items(): + confirmed = int(watermarks.get(source_file, 0)) + for offset, matches in sorted(source_entries): + if offset <= confirmed: + continue + if matches: + break + confirmed = offset + if confirmed > 0: + watermarks[source_file] = confirmed inserted = int(getattr(downstream_result, "inserted", len(matched))) downstream_skipped = int(getattr(downstream_result, "skipped", 0)) diff --git a/src/brainlayer/cli/__init__.py b/src/brainlayer/cli/__init__.py index 34bd2cba..23d395cd 100644 --- a/src/brainlayer/cli/__init__.py +++ b/src/brainlayer/cli/__init__.py @@ -3527,6 +3527,7 @@ def watch_backfill( on_flush=lambda items: None, db_path=db_path, denylist_predicate=is_legacy_backfill_denylisted if legacy_excluded_only else None, + preserve_raw_progress=window is not None, ) files = watcher._discover_jsonl_files() if legacy_excluded_only: @@ -3561,6 +3562,7 @@ def watch_backfill( on_flush=windowed_flush or downstream_flush, db_path=db_path, denylist_predicate=is_legacy_backfill_denylisted if legacy_excluded_only else None, + preserve_raw_progress=window is not None, ) processed = 0 cycles = 0 diff --git a/src/brainlayer/watcher.py b/src/brainlayer/watcher.py index ca02848a..022b86c8 100644 --- a/src/brainlayer/watcher.py +++ b/src/brainlayer/watcher.py @@ -83,6 +83,15 @@ def _mapping_value(value: Any) -> dict[str, Any]: return value if isinstance(value, dict) else {} +def _first_timestamp(*entries: dict[str, Any]) -> str | None: + for entry in entries: + for key in ("timestamp", "created_at"): + value = entry.get(key) + if isinstance(value, str) and value.strip(): + return value + return None + + def normalize_provider_entry(entry: dict[str, Any], provider: str) -> dict[str, Any] | None: if not isinstance(entry, dict): return None @@ -90,6 +99,9 @@ def normalize_provider_entry(entry: dict[str, Any], provider: str) -> dict[str, entry_type = entry.get("type") if entry_type in {"user", "assistant"} and isinstance(entry.get("message"), dict): normalized = dict(entry) + source_timestamp = _first_timestamp(entry) + normalized["timestamp"] = source_timestamp or datetime.now(timezone.utc).isoformat() + normalized["_timestamp_synthesized"] = source_timestamp is None normalized["_provider"] = provider return normalized @@ -98,9 +110,9 @@ def normalize_provider_entry(entry: dict[str, Any], provider: str) -> dict[str, if provider == "codex" and entry_type == "response_item": if not payload_entry or payload_entry.get("type") != "message": return None - candidate = {**payload_entry, "timestamp": entry.get("timestamp")} + candidate = dict(payload_entry) elif payload_entry: - candidate = {**payload_entry, "timestamp": entry.get("timestamp") or payload_entry.get("timestamp")} + candidate = dict(payload_entry) else: candidate = entry @@ -121,7 +133,7 @@ def normalize_provider_entry(entry: dict[str, Any], provider: str) -> dict[str, if not text: return None - source_timestamp = candidate.get("timestamp") or candidate.get("created_at") + source_timestamp = _first_timestamp(entry, candidate) return { "type": role, "message": {"role": role, "content": [{"type": "text", "text": text}]}, @@ -285,7 +297,12 @@ def check_rewind(self) -> bool: return True return False - def read_new_lines(self, max_lines: int | None = None) -> list[dict]: + def read_new_lines( + self, + max_lines: int | None = None, + *, + include_progress_markers: bool = False, + ) -> list[dict]: """Read any new complete lines since last call. Returns parsed JSON dicts.""" # Check for rewind before reading self.check_rewind() @@ -313,6 +330,8 @@ def read_new_lines(self, max_lines: int | None = None) -> list[dict]: if not line_data.strip(): self.offset += nl_idx + 1 + if include_progress_markers: + lines.append({"_line_end_offset": self.offset, "_watermark_only": True}) continue try: @@ -320,8 +339,12 @@ def read_new_lines(self, max_lines: int | None = None) -> list[dict]: if isinstance(parsed, dict): parsed["_line_end_offset"] = self.offset + nl_idx + 1 lines.append(parsed) + elif include_progress_markers: + lines.append({"_line_end_offset": self.offset + nl_idx + 1, "_watermark_only": True}) except (json.JSONDecodeError, UnicodeDecodeError): logger.debug("Skipping corrupt JSONL line at offset %d", self.offset) + if include_progress_markers: + lines.append({"_line_end_offset": self.offset + nl_idx + 1, "_watermark_only": True}) self.offset += nl_idx + 1 @@ -502,6 +525,7 @@ def __init__( respect_denylist: bool = True, unknown_subagent_is_denylisted: bool = True, denylist_predicate: Callable[[str | Path], bool] | None = None, + preserve_raw_progress: bool = False, ): if watch_roots is not None: self.watch_roots = [WatchRoot(root.provider, root.path, root.glob_pattern) for root in watch_roots] @@ -527,6 +551,7 @@ def __init__( self.respect_denylist = respect_denylist self.unknown_subagent_is_denylisted = unknown_subagent_is_denylisted self.denylist_predicate = denylist_predicate + self.preserve_raw_progress = preserve_raw_progress self._tailers: dict[str, JSONLTailer] = {} self._file_providers: dict[str, str] = {} self._stop = threading.Event() @@ -607,10 +632,29 @@ def _normalize_lines(self, filepath: str, new_lines: list[dict]) -> list[dict]: provider = self.provider_for_file(filepath) normalized = [] for line in new_lines: + if line.get("_watermark_only") is True: + normalized.append( + { + "_source_file": filepath, + "_provider": provider, + "_line_end_offset": line["_line_end_offset"], + "_watermark_only": True, + } + ) + continue entry = normalize_provider_entry(line, provider) if not entry and provider == "claude": entry = dict(line) if not entry: + if self.preserve_raw_progress and isinstance(line.get("_line_end_offset"), int): + normalized.append( + { + "_source_file": filepath, + "_provider": provider, + "_line_end_offset": line["_line_end_offset"], + "_watermark_only": True, + } + ) continue entry["_source_file"] = filepath entry["_provider"] = provider @@ -792,17 +836,18 @@ def poll_once(self) -> int: if self._is_denylisted(filepath): self._tailers.pop(filepath, None) self._file_providers.pop(filepath, None) - self.registry.remove(filepath) for filepath in files: if self._is_denylisted(filepath): self._tailers.pop(filepath, None) self._file_providers.pop(filepath, None) - self.registry.remove(filepath) continue try: tailer = self._ensure_tailer(filepath) - new_lines = tailer.read_new_lines(max_lines=self.max_lines_per_file) + new_lines = tailer.read_new_lines( + max_lines=self.max_lines_per_file, + include_progress_markers=self.preserve_raw_progress, + ) # Handle rewind detection (checkpoint restore) if tailer.rewound: diff --git a/tests/test_backfill.py b/tests/test_backfill.py index b49e3cbe..70a2dfb2 100644 --- a/tests/test_backfill.py +++ b/tests/test_backfill.py @@ -113,6 +113,59 @@ def test_windowed_flush_does_not_confirm_offsets_when_downstream_returns_none(): assert windowed.inserted_chunks == 0 +def test_windowed_flush_stops_before_unconfirmed_matched_entry(): + windowed = WindowedFlush( + lambda _entries: FlushWatermarks(inserted=0), + since=datetime(2026, 7, 10, tzinfo=UTC), + until=datetime(2026, 7, 16, tzinfo=UTC), + ) + + result = windowed( + [ + _entry("2026-07-09T23:59:59Z", 100), + _entry("2026-07-12T00:00:00Z", 200), + _entry("2026-07-16T00:00:00Z", 300), + ] + ) + + assert result == {"/tmp/session.jsonl": 100} + + +def test_windowed_flush_advances_exclusions_after_confirmed_match_until_next_match(): + windowed = WindowedFlush( + lambda _entries: FlushWatermarks({"/tmp/session.jsonl": 200}, inserted=1), + since=datetime(2026, 7, 10, tzinfo=UTC), + until=datetime(2026, 7, 16, tzinfo=UTC), + ) + + result = windowed( + [ + _entry("2026-07-12T00:00:00Z", 200), + _entry("2026-07-16T00:00:00Z", 300), + _entry("2026-07-13T00:00:00Z", 400), + _entry("2026-07-16T00:00:00Z", 500), + ] + ) + + assert result == {"/tmp/session.jsonl": 300} + + +def test_normalize_provider_entry_uses_valid_created_at_for_canonical_entry(): + normalized = normalize_provider_entry( + { + "type": "assistant", + "timestamp": {"invalid": True}, + "created_at": "2026-07-12T12:00:00Z", + "message": {"role": "assistant", "content": "durable decision"}, + }, + "claude", + ) + + assert normalized is not None + assert normalized["timestamp"] == "2026-07-12T12:00:00Z" + assert normalized["_timestamp_synthesized"] is False + + def test_window_registry_suffix_preserves_fractional_seconds_without_changing_whole_seconds(): whole_since = datetime(2026, 7, 10, tzinfo=UTC) whole_until = datetime(2026, 7, 16, tzinfo=UTC) diff --git a/tests/test_jsonl_watcher.py b/tests/test_jsonl_watcher.py index e6599abd..7d7e4bca 100644 --- a/tests/test_jsonl_watcher.py +++ b/tests/test_jsonl_watcher.py @@ -412,6 +412,28 @@ def crash_one_file(filepath, new_lines): payload = json.loads(health_path.read_text()) assert payload["poll_count"] == 1 + def test_newly_denylisted_file_preserves_confirmed_registry_offset(self, monkeypatch, tmp_path): + monkeypatch.setenv("BRAINLAYER_INGEST_DENYLIST", "") + project = self._make_project_dir(tmp_path) + transcript = project / "session.jsonl" + transcript.write_text('{"type":"user","message":{"role":"user","content":"remember this"}}\n') + + watcher = JSONLWatcher( + watch_dir=tmp_path / "projects", + registry_path=tmp_path / "offsets.json", + on_flush=lambda items: {str(transcript): items[-1]["_line_end_offset"]}, + batch_size=1, + ) + assert watcher.poll_once() == 1 + confirmed = watcher.registry.get(str(transcript)) + assert confirmed[0] == transcript.stat().st_size + + monkeypatch.setenv("BRAINLAYER_INGEST_DENYLIST", str(transcript)) + assert watcher.poll_once() == 0 + + assert str(transcript) not in watcher._tailers + assert watcher.registry.get(str(transcript)) == confirmed + def test_offset_survives_restart(self, tmp_path): project = self._make_project_dir(tmp_path) f = project / "s1.jsonl" diff --git a/tests/test_watch_backfill_cli.py b/tests/test_watch_backfill_cli.py index 8f48ca2a..e1e29abc 100644 --- a/tests/test_watch_backfill_cli.py +++ b/tests/test_watch_backfill_cli.py @@ -344,6 +344,40 @@ def entry(timestamp: str, token: str) -> str: assert len(list(queue_dir.glob("watcher-*.jsonl"))) == 1 +def test_watch_backfill_persists_progress_for_rejected_and_malformed_lines(tmp_path, monkeypatch): + monkeypatch.delenv("BRAINLAYER_INGEST_DENYLIST", raising=False) + transcript = tmp_path / ".codex" / "sessions" / "2026" / "07" / "unsupported.jsonl" + registry = tmp_path / "window-offsets.json" + queue_dir = tmp_path / "queue" + transcript.parent.mkdir(parents=True) + transcript.write_text('{"type":"unsupported"}\nnot-json\n', encoding="utf-8") + args = [ + "watch-backfill", + "--home", + str(tmp_path), + "--registry", + str(registry), + "--since", + "2026-07-10", + "--until", + "2026-07-16", + ] + + first = CliRunner().invoke(app, args, env={"BRAINLAYER_QUEUE_DIR": str(queue_dir)}) + + assert first.exit_code == 0, first.output + assert "processed_entries=2" in first.output + assert "matched_entries=0" in first.output + assert "queued_chunks=0" in first.output + registry_payload = json.loads(registry.read_text()) + assert registry_payload[str(transcript)]["offset"] == transcript.stat().st_size + + second = CliRunner().invoke(app, args, env={"BRAINLAYER_QUEUE_DIR": str(queue_dir)}) + + assert second.exit_code == 0, second.output + assert "processed_entries=0" in second.output + + def test_watch_backfill_rejects_legacy_scope_without_window(tmp_path): result = CliRunner().invoke( app, From 39abf1e550bfab10f2c317392071d6278621c39b Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Fri, 17 Jul 2026 18:10:43 +0300 Subject: [PATCH 08/10] test: use production tier0 alert budget --- tests/test_tier0_drills.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/test_tier0_drills.py b/tests/test_tier0_drills.py index a52d33a2..a2e9e559 100644 --- a/tests/test_tier0_drills.py +++ b/tests/test_tier0_drills.py @@ -48,6 +48,7 @@ def _run_drill( last_alert_epoch: int | None = None, last_alert_reason: str = "", repeat_alert_seconds: int = 1_800, + alert_timeout_seconds: int = 3, state_contents: str = "{}\n", ) -> DrillResult: fake_bin = tmp_path / "bin" @@ -128,7 +129,7 @@ def _run_drill( **os.environ, "FAKE_LAUNCHCTL_PRINT_EXIT": "0" if label_loaded else "113", "FAKE_STATE_MTIME": str(state_mtime or 0), - "TIER0_ALERT_TIMEOUT_SECONDS": "1", + "TIER0_ALERT_TIMEOUT_SECONDS": str(alert_timeout_seconds), "TIER0_ALERT_STATE_PATH": str(alert_state_path), "TIER0_CURL": str(fake_bin / "curl"), "TIER0_DOMAIN": DOMAIN, @@ -229,6 +230,7 @@ def test_d3_hanging_notify_endpoint_cannot_suppress_local_alert_or_heal(tmp_path label_loaded=True, state_mtime=NOW_EPOCH - STALE_SECONDS - 1, curl_hangs=True, + alert_timeout_seconds=1, ) assert result.process.returncode == 1, result.process.stdout + result.process.stderr @@ -304,6 +306,7 @@ def test_alert_fanout_uses_one_shared_deadline(tmp_path: Path) -> None: curl_hangs=True, osascript_hangs=True, use_fake_wait_sleep=True, + alert_timeout_seconds=1, ) assert result.process.returncode == 1, result.process.stdout + result.process.stderr From e43e1fb21c9ed8c72cc95d4bef95f57bbd144812 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Sat, 18 Jul 2026 01:47:57 +0300 Subject: [PATCH 09/10] fix: harden transcript backfill safety Co-Authored-By: Claude Fable 5 --- src/brainlayer/backfill.py | 57 ++++++++---- src/brainlayer/cli/__init__.py | 101 ++++++++++++-------- src/brainlayer/watcher.py | 113 ++++++++++++++++++----- tests/test_backfill.py | 32 +++++++ tests/test_jsonl_watcher.py | 51 +++++++++++ tests/test_run_tests_script.py | 2 + tests/test_watch_backfill_cli.py | 152 +++++++++++++++++++++++++++++++ 7 files changed, 431 insertions(+), 77 deletions(-) diff --git a/src/brainlayer/backfill.py b/src/brainlayer/backfill.py index c06f02f5..1e94b40f 100644 --- a/src/brainlayer/backfill.py +++ b/src/brainlayer/backfill.py @@ -2,9 +2,13 @@ from __future__ import annotations +import errno +import fcntl from collections.abc import Callable +from contextlib import contextmanager from datetime import UTC, datetime from pathlib import Path +from typing import Iterator from .watcher_bridge import FlushWatermarks @@ -41,28 +45,47 @@ def format_timestamp(value: datetime) -> str: return f"{format_timestamp(since)}-{format_timestamp(until)}" -def _contains_ordered(parts: tuple[str, ...], expected: tuple[str, ...]) -> bool: - position = 0 - for part in parts: - if part == expected[position]: - position += 1 - if position == len(expected): - return True - return False +class BackfillAlreadyRunning(RuntimeError): + """Raised when another process owns the same registry-scoped backfill.""" + + +@contextmanager +def backfill_run_lock(registry_path: str | Path) -> Iterator[None]: + """Serialize scan, enqueue, and offset persistence for one registry.""" + registry = Path(registry_path).expanduser() + registry.parent.mkdir(parents=True, exist_ok=True) + lock_path = registry.with_name(f"{registry.name}.backfill.lock") + with lock_path.open("a+b") as lock_file: + try: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError as exc: + if exc.errno not in {errno.EACCES, errno.EAGAIN}: + raise + raise BackfillAlreadyRunning(f"another backfill is using registry {registry}") from exc + try: + yield + finally: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + + +def _adjacent_pair_indices(parts: tuple[str, ...], first: str, second: str) -> list[int]: + return [index for index in range(len(parts) - 1) if parts[index : index + 2] == (first, second)] def is_legacy_excluded_path(path: str | Path) -> bool: """Identify roots blocked by the blanket denylist retired in July 2026.""" parts = Path(path).expanduser().parts - return any( - _contains_ordered(parts, expected) - for expected in ( - (".claude", "projects", "subagents"), - (".codex", "sessions"), - (".cursor", "agent-transcripts"), - (".gemini", "sessions"), - ) - ) + if _adjacent_pair_indices(parts, ".codex", "sessions"): + return True + if _adjacent_pair_indices(parts, ".gemini", "sessions"): + return True + for cursor_index, part in enumerate(parts): + if part == ".cursor" and "agent-transcripts" in parts[cursor_index + 1 :]: + return True + for claude_index in _adjacent_pair_indices(parts, ".claude", "projects"): + if "subagents" in parts[claude_index + 3 :]: + return True + return False class WindowedFlush: diff --git a/src/brainlayer/cli/__init__.py b/src/brainlayer/cli/__init__.py index 54303f14..2ff19008 100644 --- a/src/brainlayer/cli/__init__.py +++ b/src/brainlayer/cli/__init__.py @@ -3502,7 +3502,14 @@ def watch_backfill( max_cycles: int = typer.Option(100, "--max-cycles", min=1, help="Maximum poll cycles to run."), ) -> None: """One-shot replay for watched JSONL roots using the durable queue writer path.""" - from ..backfill import WindowedFlush, is_legacy_excluded_path, parse_backfill_window, window_registry_suffix + from ..backfill import ( + BackfillAlreadyRunning, + WindowedFlush, + backfill_run_lock, + is_legacy_excluded_path, + parse_backfill_window, + window_registry_suffix, + ) from ..ingest_denylist import is_legacy_backfill_denylisted from ..paths import get_db_path from ..watcher import JSONLWatcher, WatchRoot, default_watch_roots @@ -3528,6 +3535,7 @@ def watch_backfill( else: registry_path = db_path.parent / "offsets.json" watch_roots = [WatchRoot("custom", item) for item in source] if source else default_watch_roots(home=home) + discovery_predicate = is_legacy_excluded_path if legacy_excluded_only else None if dry_run: watcher = JSONLWatcher( @@ -3536,11 +3544,11 @@ def watch_backfill( on_flush=lambda items: None, db_path=db_path, denylist_predicate=is_legacy_backfill_denylisted if legacy_excluded_only else None, + discovery_predicate=discovery_predicate, preserve_raw_progress=window is not None, + prune_missing_offsets=not legacy_excluded_only, ) files = watcher._discover_jsonl_files() - if legacy_excluded_only: - files = [path for path in files if is_legacy_excluded_path(path)] provider_counts: dict[str, int] = {} for path in files: provider = watcher.provider_for_file(path) @@ -3554,45 +3562,66 @@ def watch_backfill( ) return - downstream_flush = create_flush_callback(db_path, arbitrated=True) - windowed_flush = ( - WindowedFlush( - downstream_flush, - since=window[0], - until=window[1], - source_predicate=is_legacy_excluded_path if legacy_excluded_only else None, - ) - if window - else None - ) - watcher = JSONLWatcher( - watch_roots=watch_roots, - registry_path=registry_path, - on_flush=windowed_flush or downstream_flush, - db_path=db_path, - denylist_predicate=is_legacy_backfill_denylisted if legacy_excluded_only else None, - preserve_raw_progress=window is not None, - ) - processed = 0 - cycles = 0 - while cycles < max_cycles: - cycles += 1 - offsets_before = {path: tailer.offset for path, tailer in watcher._tailers.items()} - count = watcher.poll_once() - processed += count - made_progress = any(tailer.offset > offsets_before.get(path, 0) for path, tailer in watcher._tailers.items()) - if not made_progress: - break - watcher.indexer.flush() - watcher.registry.flush() + try: + with backfill_run_lock(registry_path): + downstream_flush = create_flush_callback(db_path, arbitrated=True) + windowed_flush = ( + WindowedFlush( + downstream_flush, + since=window[0], + until=window[1], + source_predicate=discovery_predicate, + ) + if window + else None + ) + watcher = JSONLWatcher( + watch_roots=watch_roots, + registry_path=registry_path, + on_flush=windowed_flush or downstream_flush, + db_path=db_path, + denylist_predicate=is_legacy_backfill_denylisted if legacy_excluded_only else None, + discovery_predicate=discovery_predicate, + preserve_raw_progress=window is not None, + max_read_bytes_per_file=1_048_576, + prune_missing_offsets=not legacy_excluded_only, + ) + processed = 0 + cycles = 0 + while cycles < max_cycles: + cycles += 1 + positions_before = { + path: (tailer.offset, tailer.offset + len(tailer._buffer)) + for path, tailer in watcher._tailers.items() + } + count = watcher.poll_once() + processed += count + made_progress = any( + tailer.offset > positions_before.get(path, (0, 0))[0] + or tailer.offset + len(tailer._buffer) > positions_before.get(path, (0, 0))[1] + for path, tailer in watcher._tailers.items() + ) + if not made_progress: + break + watcher.indexer.flush() + incomplete = watcher.has_pending_input() or watcher.indexer.retained_failed_input_count() > 0 + registry_flushed = watcher.registry.flush() + incomplete = incomplete or not registry_flushed + except BackfillAlreadyRunning as exc: + rprint(f"[red]{exc}[/]") + raise typer.Exit(code=2) from exc + + incomplete_summary = f" incomplete={'true' if incomplete else 'false'}" if windowed_flush: rprint( f"processed_entries={processed} scanned_entries={windowed_flush.scanned_entries} " f"matched_entries={windowed_flush.matched_entries} queued_chunks={windowed_flush.inserted_chunks} " - f"cycles={cycles} registry={registry_path}" + f"cycles={cycles}{incomplete_summary} registry={registry_path}" ) else: - rprint(f"processed_entries={processed} cycles={cycles} registry={registry_path}") + rprint(f"processed_entries={processed} cycles={cycles}{incomplete_summary} registry={registry_path}") + if incomplete: + raise typer.Exit(code=1) @app.command("index-fast", hidden=True) diff --git a/src/brainlayer/watcher.py b/src/brainlayer/watcher.py index ba3dc372..034794d6 100644 --- a/src/brainlayer/watcher.py +++ b/src/brainlayer/watcher.py @@ -96,11 +96,22 @@ def _mapping_value(value: Any) -> dict[str, Any]: return value if isinstance(value, dict) else {} +def _valid_timestamp(value: Any) -> str | None: + if not isinstance(value, str) or not value.strip(): + return None + candidate = value.strip() + try: + datetime.fromisoformat(candidate.replace("Z", "+00:00")) + except (ValueError, OverflowError): + return None + return candidate + + def _first_timestamp(*entries: dict[str, Any]) -> str | None: for entry in entries: for key in ("timestamp", "created_at"): - value = entry.get(key) - if isinstance(value, str) and value.strip(): + value = _valid_timestamp(entry.get(key)) + if value is not None: return value return None @@ -210,6 +221,10 @@ def evaluate( _OFFSET_TOMBSTONE_RETENTION_S = 24 * 60 * 60 +class _ProgressMarker(dict): + """Internal marker type that cannot be spoofed by parsed JSON content.""" + + def _lock_offset_registry_file(lock_file) -> None: """Acquire an exclusive advisory lock on POSIX or Windows.""" if fcntl is not None: @@ -632,17 +647,25 @@ def read_new_lines( max_lines: int | None = None, *, include_progress_markers: bool = False, + max_read_bytes: int | None = None, ) -> list[dict]: """Read any new complete lines since last call. Returns parsed JSON dicts.""" # Check for rewind before reading self.check_rewind() - try: - with open(self.filepath, "rb") as f: - f.seek(self.offset + len(self._buffer)) - new_data = f.read() - except OSError: - return [] + new_data = b"" + if b"\n" not in self._buffer: + try: + with open(self.filepath, "rb") as f: + f.seek(self.offset + len(self._buffer)) + if max_read_bytes is None: + new_data = f.read() + else: + read_budget = max(1, max_read_bytes) + remaining_budget = read_budget - len(self._buffer) + new_data = f.read(read_budget if remaining_budget <= 0 else remaining_budget) + except OSError: + return [] if not new_data and b"\n" not in self._buffer: return [] @@ -661,7 +684,7 @@ def read_new_lines( if not line_data.strip(): self.offset += nl_idx + 1 if include_progress_markers: - lines.append({"_line_end_offset": self.offset, "_watermark_only": True}) + lines.append(_ProgressMarker(_line_end_offset=self.offset, _watermark_only=True)) continue try: @@ -670,11 +693,21 @@ def read_new_lines( parsed["_line_end_offset"] = self.offset + nl_idx + 1 lines.append(parsed) elif include_progress_markers: - lines.append({"_line_end_offset": self.offset + nl_idx + 1, "_watermark_only": True}) + lines.append( + _ProgressMarker( + _line_end_offset=self.offset + nl_idx + 1, + _watermark_only=True, + ) + ) except (json.JSONDecodeError, UnicodeDecodeError): logger.debug("Skipping corrupt JSONL line at offset %d", self.offset) if include_progress_markers: - lines.append({"_line_end_offset": self.offset + nl_idx + 1, "_watermark_only": True}) + lines.append( + _ProgressMarker( + _line_end_offset=self.offset + nl_idx + 1, + _watermark_only=True, + ) + ) self.offset += nl_idx + 1 @@ -855,7 +888,10 @@ def __init__( respect_denylist: bool = True, unknown_subagent_is_denylisted: bool = True, denylist_predicate: Callable[[str | Path], bool] | None = None, + discovery_predicate: Callable[[str | Path], bool] | None = None, preserve_raw_progress: bool = False, + max_read_bytes_per_file: int | None = None, + prune_missing_offsets: bool = True, ): if watch_roots is not None: self.watch_roots = [WatchRoot(root.provider, root.path, root.glob_pattern) for root in watch_roots] @@ -881,7 +917,10 @@ def __init__( self.respect_denylist = respect_denylist self.unknown_subagent_is_denylisted = unknown_subagent_is_denylisted self.denylist_predicate = denylist_predicate + self.discovery_predicate = discovery_predicate self.preserve_raw_progress = preserve_raw_progress + self.max_read_bytes_per_file = max(1, max_read_bytes_per_file) if max_read_bytes_per_file is not None else None + self.prune_missing_offsets = prune_missing_offsets self._tailers: dict[str, JSONLTailer] = {} self._file_providers: dict[str, str] = {} self._stop = threading.Event() @@ -894,7 +933,7 @@ def __init__( self._health_entries_seen = 0 self._health_output_at_start = 0 self.poll_count = 0 - self._offset_prune_complete = False + self._offset_prune_complete = not prune_missing_offsets def _advance_confirmed_offsets(self, watermarks: dict[str, int]) -> None: for filepath, offset in watermarks.items(): @@ -927,6 +966,9 @@ def _is_denylisted(self, filepath: str) -> bool: unknown_subagent_is_denylisted=self.unknown_subagent_is_denylisted, ) + def _is_discovery_candidate(self, filepath: str) -> bool: + return self.discovery_predicate is None or self.discovery_predicate(filepath) + def _discover_jsonl_files(self) -> list[str]: """Find all .jsonl files under each watched project, including nested session artifacts.""" discovered: list[tuple[float, str, str]] = [] @@ -942,16 +984,17 @@ def _discover_jsonl_files(self) -> list[str]: for base in bases: files = base.glob(root.glob_pattern) for f in files: - if f.is_file(): - path = str(f) - if self._is_denylisted(path): - continue - try: - mtime = f.stat().st_mtime - except OSError as e: - logger.debug("Skipping JSONL file during discovery after stat failure: %s: %s", path, e) - continue - discovered.append((mtime, path, root.provider)) + path = str(f) + if not self._is_discovery_candidate(path) or self._is_denylisted(path): + continue + if not f.is_file(): + continue + try: + mtime = f.stat().st_mtime + except OSError as e: + logger.debug("Skipping JSONL file during discovery after stat failure: %s: %s", path, e) + continue + discovered.append((mtime, path, root.provider)) except OSError: continue discovered.sort(key=lambda item: item[0], reverse=True) @@ -963,7 +1006,7 @@ def _normalize_lines(self, filepath: str, new_lines: list[dict]) -> list[dict]: provider = self.provider_for_file(filepath) normalized = [] for line in new_lines: - if line.get("_watermark_only") is True: + if isinstance(line, _ProgressMarker): normalized.append( { "_source_file": filepath, @@ -1006,6 +1049,27 @@ def _max_offset_lag_bytes(self, files: list[str]) -> int: max_lag = max(max_lag, max(size - offset, 0)) return max_lag + def has_pending_input(self) -> bool: + """Return whether any discovered source still has a complete unread line.""" + for filepath in self._discover_jsonl_files(): + tailer = self._tailers.get(filepath) + if tailer is None: + offset = self.registry.get(filepath)[0] + try: + if os.path.getsize(filepath) > offset: + return True + except OSError: + continue + continue + if b"\n" in tailer._buffer: + return True + try: + if os.path.getsize(filepath) > tailer.offset + len(tailer._buffer): + return True + except OSError: + continue + return False + def _db_realtime_inserts_since_window_start(self) -> int | None: if not self.db_path: return None @@ -1172,7 +1236,7 @@ def poll_once(self) -> int: self._offset_prune_complete = self.registry.flush() and self.registry.last_prune_complete for filepath in list(self._tailers): - if self._is_denylisted(filepath): + if not self._is_discovery_candidate(filepath) or self._is_denylisted(filepath): self._tailers.pop(filepath, None) self._file_providers.pop(filepath, None) @@ -1186,6 +1250,7 @@ def poll_once(self) -> int: new_lines = tailer.read_new_lines( max_lines=self.max_lines_per_file, include_progress_markers=self.preserve_raw_progress, + max_read_bytes=self.max_read_bytes_per_file, ) # Handle rewind detection (checkpoint restore) diff --git a/tests/test_backfill.py b/tests/test_backfill.py index 70a2dfb2..1651413b 100644 --- a/tests/test_backfill.py +++ b/tests/test_backfill.py @@ -1,5 +1,8 @@ from datetime import UTC, datetime +import pytest + +import brainlayer.backfill as backfill from brainlayer.backfill import WindowedFlush, is_legacy_excluded_path, parse_backfill_window, window_registry_suffix from brainlayer.watcher import normalize_provider_entry from brainlayer.watcher_bridge import FlushWatermarks @@ -166,6 +169,22 @@ def test_normalize_provider_entry_uses_valid_created_at_for_canonical_entry(): assert normalized["_timestamp_synthesized"] is False +def test_normalize_provider_entry_falls_back_from_malformed_timestamp_to_created_at(): + normalized = normalize_provider_entry( + { + "type": "assistant", + "timestamp": "not-an-iso-timestamp", + "created_at": "2026-07-12T12:00:00Z", + "message": {"role": "assistant", "content": "durable decision"}, + }, + "claude", + ) + + assert normalized is not None + assert normalized["timestamp"] == "2026-07-12T12:00:00Z" + assert normalized["_timestamp_synthesized"] is False + + def test_window_registry_suffix_preserves_fractional_seconds_without_changing_whole_seconds(): whole_since = datetime(2026, 7, 10, tzinfo=UTC) whole_until = datetime(2026, 7, 16, tzinfo=UTC) @@ -208,3 +227,16 @@ def test_legacy_excluded_path_selects_only_roots_blocked_by_old_policy(tmp_path) ) assert not is_legacy_excluded_path(tmp_path / ".claude" / "projects" / "repo" / "direct.jsonl") assert not is_legacy_excluded_path(tmp_path / ".cursor" / "projects" / "repo" / "state.jsonl") + assert not is_legacy_excluded_path(tmp_path / ".codex" / "archive" / "sessions" / "worker.jsonl") + assert not is_legacy_excluded_path( + tmp_path / ".claude" / "cache" / "projects" / "repo" / "subagents" / "worker.jsonl" + ) + + +def test_backfill_run_lock_rejects_concurrent_registry_owner(tmp_path): + registry = tmp_path / "offsets.json" + + with backfill.backfill_run_lock(registry): + with pytest.raises(backfill.BackfillAlreadyRunning): + with backfill.backfill_run_lock(registry): + raise AssertionError("concurrent backfill unexpectedly acquired the run lock") diff --git a/tests/test_jsonl_watcher.py b/tests/test_jsonl_watcher.py index 6eeb501a..36b36009 100644 --- a/tests/test_jsonl_watcher.py +++ b/tests/test_jsonl_watcher.py @@ -539,6 +539,17 @@ def test_read_complete_lines(self, tmp_path): assert lines[0]["id"] == "1" assert lines[1]["id"] == "2" + def test_read_new_lines_bounds_offset_zero_read_bytes(self, tmp_path): + transcript = tmp_path / "large.jsonl" + transcript.write_bytes(b'{"type":"msg","payload":"xxxxxxxxxxxxxxxx"}\n' * 1_000) + tailer = JSONLTailer(str(transcript)) + + for _ in range(10): + lines = tailer.read_new_lines(max_lines=1, max_read_bytes=256) + + assert len(lines) == 1 + assert len(tailer._buffer) <= 256 + def test_partial_line_buffered(self, tmp_path): f = tmp_path / "test.jsonl" f.write_bytes(b'{"type":"msg","id":"1"}\n{"partial":') @@ -716,6 +727,46 @@ def test_discover_jsonl_files(self, tmp_path): assert len(files) == 2 assert all(f.endswith(".jsonl") for f in files) + def test_discovery_predicate_filters_before_tailer_reads(self, tmp_path): + project = self._make_project_dir(tmp_path) + legacy = project / "legacy.jsonl" + ordinary = project / "ordinary.jsonl" + legacy.write_text('{"id":"legacy"}\n') + ordinary.write_text('{"id":"ordinary"}\n') + watcher = JSONLWatcher( + watch_dir=tmp_path / "projects", + registry_path=tmp_path / "offsets.json", + on_flush=lambda items: None, + discovery_predicate=lambda path: Path(path).name == "legacy.jsonl", + ) + + assert watcher._discover_jsonl_files() == [str(legacy)] + + @pytest.mark.parametrize("preserve_raw_progress", [False, True]) + def test_external_watermark_field_remains_transcript_content(self, tmp_path, preserve_raw_progress): + source = tmp_path / "projects" / "project" / "session.jsonl" + source.parent.mkdir(parents=True) + watcher = JSONLWatcher( + watch_dir=tmp_path / "projects", + registry_path=tmp_path / "offsets.json", + on_flush=lambda items: None, + preserve_raw_progress=preserve_raw_progress, + ) + + normalized = watcher._normalize_lines( + str(source), + [ + { + "role": "user", + "content": "real transcript content", + "_watermark_only": True, + "_line_end_offset": 123, + } + ], + ) + + assert normalized[0]["message"]["content"][0]["text"] == "real transcript content" + def test_first_poll_prunes_and_flushes_deleted_offset_entries(self, tmp_path): project = self._make_project_dir(tmp_path) existing = project / "existing.jsonl" diff --git a/tests/test_run_tests_script.py b/tests/test_run_tests_script.py index f29ed305..01eb1686 100644 --- a/tests/test_run_tests_script.py +++ b/tests/test_run_tests_script.py @@ -242,6 +242,8 @@ def test_changed_only_scope_maps_cli_entrypoint_to_all_cli_tests(tmp_path: Path) logged = pytest_log.read_text() assert str(test_root / "test_cli_commands.py") in logged assert str(test_root / "test_watch_backfill_cli.py") in logged + targeted_invocation = next(line for line in logged.splitlines() if str(test_root / "test_cli_commands.py") in line) + assert str(test_root / "test_think_recall_integration.py") not in targeted_invocation assert "falling back to full pytest unit suite" not in result.stdout assert f"{test_root}/ -v" not in logged diff --git a/tests/test_watch_backfill_cli.py b/tests/test_watch_backfill_cli.py index e1e29abc..4d64d544 100644 --- a/tests/test_watch_backfill_cli.py +++ b/tests/test_watch_backfill_cli.py @@ -1,8 +1,19 @@ import json +import pytest from typer.testing import CliRunner from brainlayer.cli import app +from brainlayer.paths import get_db_path + + +@pytest.fixture(autouse=True) +def _isolate_brainlayer_db(tmp_path, monkeypatch): + monkeypatch.setenv("BRAINLAYER_DB", str(tmp_path / "brainlayer.db")) + + +def test_watch_backfill_tests_use_isolated_database(tmp_path): + assert get_db_path() == tmp_path / "brainlayer.db" def test_watch_backfill_dry_run_includes_cursor_agent_transcripts(tmp_path, monkeypatch): @@ -393,3 +404,144 @@ def test_watch_backfill_rejects_legacy_scope_without_window(tmp_path): assert result.exit_code != 0 assert "--legacy-excluded-only requires" in result.output assert "--since and --until" in result.output + + +def test_watch_backfill_legacy_scope_never_reads_or_advances_nonlegacy_files(tmp_path, monkeypatch): + monkeypatch.delenv("BRAINLAYER_INGEST_DENYLIST", raising=False) + direct = tmp_path / ".claude" / "projects" / "repo" / "direct.jsonl" + legacy = tmp_path / ".codex" / "sessions" / "2026" / "07" / "worker.jsonl" + direct.parent.mkdir(parents=True) + legacy.parent.mkdir(parents=True) + direct.write_text('{"type":"unsupported"}\n') + legacy.write_text('{"type":"unsupported"}\n') + registry = tmp_path / "shared-offsets.json" + registry.write_text( + json.dumps( + { + str(direct): { + "offset": 1, + "inode": direct.stat().st_ino, + "mtime": direct.stat().st_mtime, + } + } + ) + ) + read_paths = [] + from brainlayer.watcher import JSONLTailer + + original_read = JSONLTailer.read_new_lines + + def tracking_read(self, *args, **kwargs): + read_paths.append(self.filepath) + return original_read(self, *args, **kwargs) + + monkeypatch.setattr(JSONLTailer, "read_new_lines", tracking_read) + + result = CliRunner().invoke( + app, + [ + "watch-backfill", + "--home", + str(tmp_path), + "--registry", + str(registry), + "--since", + "2026-07-10", + "--until", + "2026-07-16", + "--legacy-excluded-only", + "--max-cycles", + "5", + ], + env={"BRAINLAYER_QUEUE_DIR": str(tmp_path / "queue")}, + ) + + assert result.exit_code == 0, result.output + assert str(direct) not in read_paths + assert str(legacy) in read_paths + assert json.loads(registry.read_text())[str(direct)]["offset"] == 1 + + +def test_watch_backfill_max_cycles_exits_nonzero_while_complete_lines_remain(tmp_path, monkeypatch): + monkeypatch.delenv("BRAINLAYER_INGEST_DENYLIST", raising=False) + transcript = tmp_path / ".codex" / "sessions" / "2026" / "07" / "worker.jsonl" + transcript.parent.mkdir(parents=True) + transcript.write_text('{"type":"unsupported"}\n' * 101) + + result = CliRunner().invoke( + app, + [ + "watch-backfill", + "--home", + str(tmp_path), + "--registry", + str(tmp_path / "offsets.json"), + "--since", + "2026-07-10", + "--until", + "2026-07-16", + "--max-cycles", + "1", + ], + env={"BRAINLAYER_QUEUE_DIR": str(tmp_path / "queue")}, + ) + + assert result.exit_code == 1, result.output + assert "incomplete=true" in result.output + + +def test_watch_backfill_caps_each_file_read(tmp_path, monkeypatch): + monkeypatch.delenv("BRAINLAYER_INGEST_DENYLIST", raising=False) + transcript = tmp_path / ".codex" / "sessions" / "2026" / "07" / "worker.jsonl" + transcript.parent.mkdir(parents=True) + transcript.write_text('{"type":"unsupported"}\n') + from brainlayer.watcher import JSONLTailer + + observed_read_limits = [] + original_read = JSONLTailer.read_new_lines + + def tracking_read(self, *args, **kwargs): + observed_read_limits.append(kwargs.get("max_read_bytes")) + return original_read(self, *args, **kwargs) + + monkeypatch.setattr(JSONLTailer, "read_new_lines", tracking_read) + + result = CliRunner().invoke( + app, + [ + "watch-backfill", + "--home", + str(tmp_path), + "--registry", + str(tmp_path / "offsets.json"), + "--max-cycles", + "2", + ], + env={"BRAINLAYER_QUEUE_DIR": str(tmp_path / "queue")}, + ) + + assert result.exit_code == 0, result.output + assert observed_read_limits + assert set(observed_read_limits) == {1_048_576} + + +def test_watch_backfill_rejects_concurrent_run_for_same_registry(tmp_path): + from brainlayer.backfill import backfill_run_lock + + registry = tmp_path / "offsets.json" + with backfill_run_lock(registry): + result = CliRunner().invoke( + app, + [ + "watch-backfill", + "--home", + str(tmp_path), + "--registry", + str(registry), + "--max-cycles", + "1", + ], + ) + + assert result.exit_code == 2 + assert "another backfill is using registry" in result.output From 90ac44bea8d2b672ba33017ca4d0ac36df24b2a4 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Sat, 18 Jul 2026 02:23:56 +0300 Subject: [PATCH 10/10] fix: fail backfill after quarantine Co-Authored-By: Claude Fable 5 --- src/brainlayer/cli/__init__.py | 6 ++++- src/brainlayer/watcher.py | 2 ++ tests/test_watch_backfill_cli.py | 44 ++++++++++++++++++++++++++++++++ 3 files changed, 51 insertions(+), 1 deletion(-) diff --git a/src/brainlayer/cli/__init__.py b/src/brainlayer/cli/__init__.py index 2ff19008..1ba67ebb 100644 --- a/src/brainlayer/cli/__init__.py +++ b/src/brainlayer/cli/__init__.py @@ -3604,7 +3604,11 @@ def watch_backfill( if not made_progress: break watcher.indexer.flush() - incomplete = watcher.has_pending_input() or watcher.indexer.retained_failed_input_count() > 0 + incomplete = ( + watcher.has_pending_input() + or watcher.indexer.retained_failed_input_count() > 0 + or watcher.indexer.total_quarantined_inputs > 0 + ) registry_flushed = watcher.registry.flush() incomplete = incomplete or not registry_flushed except BackfillAlreadyRunning as exc: diff --git a/src/brainlayer/watcher.py b/src/brainlayer/watcher.py index 034794d6..4bc34170 100644 --- a/src/brainlayer/watcher.py +++ b/src/brainlayer/watcher.py @@ -748,6 +748,7 @@ def __init__( self.total_flushed = 0 self.total_outputs = 0 self.total_failed_inputs = 0 + self.total_quarantined_inputs = 0 self._flush_failures = 0 def add(self, items: list[dict]): @@ -817,6 +818,7 @@ def _quarantine_entries(self, entries: list[dict], reason: Exception) -> None: with path.open("w", encoding="utf-8") as handle: for entry in entries: handle.write(json.dumps({"reason": str(reason), "entry": entry}, sort_keys=True) + "\n") + self.total_quarantined_inputs += len(entries) logger.critical("Quarantined %d watcher flush entries at %s after repeated failures", len(entries), path) def _isolate_failed_flush(self, batch: list[dict], reason: Exception) -> int: diff --git a/tests/test_watch_backfill_cli.py b/tests/test_watch_backfill_cli.py index 4d64d544..b86447f6 100644 --- a/tests/test_watch_backfill_cli.py +++ b/tests/test_watch_backfill_cli.py @@ -545,3 +545,47 @@ def test_watch_backfill_rejects_concurrent_run_for_same_registry(tmp_path): assert result.exit_code == 2 assert "another backfill is using registry" in result.output + + +def test_watch_backfill_exits_nonzero_when_flush_failure_is_quarantined(tmp_path, monkeypatch): + monkeypatch.delenv("BRAINLAYER_INGEST_DENYLIST", raising=False) + transcript = tmp_path / ".codex" / "sessions" / "2026" / "07" / "worker.jsonl" + transcript.parent.mkdir(parents=True) + transcript.write_text( + json.dumps( + { + "role": "user", + "content": "quarantine this backfill entry after a synthetic queue failure", + "timestamp": "2026-07-12T12:00:00Z", + } + ) + + "\n" + ) + monkeypatch.setenv("BRAINLAYER_WATCHER_FLUSH_RETAIN_LIMIT", "1") + monkeypatch.setenv("BRAINLAYER_WATCHER_QUARANTINE_DIR", str(tmp_path / "quarantine")) + + flush_calls = [] + + def failing_flush(entries): + flush_calls.append(entries) + raise RuntimeError("synthetic queue failure") + + monkeypatch.setattr("brainlayer.watcher_bridge.create_flush_callback", lambda *_args, **_kwargs: failing_flush) + + result = CliRunner().invoke( + app, + [ + "watch-backfill", + "--home", + str(tmp_path), + "--registry", + str(tmp_path / "offsets.json"), + "--max-cycles", + "2", + ], + ) + + assert flush_calls, result.output + assert list((tmp_path / "quarantine").glob("watcher-flush-*.jsonl")) + assert result.exit_code == 1, result.output + assert "incomplete=true" in result.output