[rescue] rescue/2026-08-09-review-599 — preserve the PR #599 backfill review snapshot

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"

src/brainlayer/backfill.py

@@ -0,0 +1,135 @@
+"""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, 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)}"
+
+
+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
+        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

@@ -3482,16 +3482,42 @@ 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 ..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)
 
     if dry_run:
@@ -3500,33 +3526,64 @@ def watch_backfill(
             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,
+            preserve_raw_progress=window is not None,
         )
         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)
             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,
+        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()
-        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)

src/brainlayer/ingest_denylist.py

@@ -11,6 +11,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)
@@ -62,6 +68,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
@@ -132,12 +147,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