feat: add idempotent transcript window backfill

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/watcher.py)
         test_path="$TEST_ROOT/test_jsonl_watcher.py"
         if [ -f "$test_path" ]; then

src/brainlayer/backfill.py

@@ -0,0 +1,158 @@
+"""Idempotent time-window filtering for transcript watcher backfills."""
+
+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
+
+
+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, 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")
+    return parsed_since, parsed_until
+
+
+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 ""
+        return f"{base}{fraction}Z"
+
+    return f"{format_timestamp(since)}-{format_timestamp(until)}"
+
+
+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
+    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:
+    """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
+        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, OverflowError):
+            return False
+        return self.since <= parsed < self.until
+
+    def __call__(self, entries: list[dict]) -> FlushWatermarks | None:
+        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 {})
+        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):
+                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))
+        self.inserted_chunks += inserted
+        return FlushWatermarks(
+            watermarks,
+            inserted=inserted,
+            skipped=downstream_skipped + len(entries) - len(matched),
+        )

src/brainlayer/cli/__init__.py

@@ -3491,51 +3491,141 @@ 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 (
+        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
     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:
+        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)
+    discovery_predicate = is_legacy_excluded_path if legacy_excluded_only else None
 
     if dry_run:
         watcher = JSONLWatcher(
             watch_roots=watch_roots,
             registry_path=registry_path,
             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()
         provider_counts: dict[str, int] = {}
         for path in files:
             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
 
-    watcher = JSONLWatcher(
-        watch_roots=watch_roots,
-        registry_path=registry_path,
-        on_flush=create_flush_callback(db_path, arbitrated=True),
-        db_path=db_path,
-    )
-    processed = 0
-    cycles = 0
-    while cycles < max_cycles:
-        cycles += 1
-        count = watcher.poll_once()
-        if count == 0:
-            break
-        processed += count
-    watcher.indexer.flush()
-    watcher.registry.flush()
-    rprint(f"processed_entries={processed} cycles={cycles} registry={registry_path}")
+    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
+                or watcher.indexer.total_quarantined_inputs > 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}{incomplete_summary} registry={registry_path}"
+        )
+    else:
+        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)

src/brainlayer/ingest_denylist.py

@@ -12,6 +12,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/**",
+)
 
 
 @dataclass(frozen=True)
@@ -64,6 +70,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
@@ -156,12 +171,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