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/src/brainlayer/backfill.py b/src/brainlayer/backfill.py
new file mode 100644
index 00000000..c06f02f5
--- /dev/null
+++ b/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),
+        )
diff --git a/src/brainlayer/cli/__init__.py b/src/brainlayer/cli/__init__.py
index a171e155..23d395cd 100644
--- a/src/brainlayer/cli/__init__.py
+++ b/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)
diff --git a/src/brainlayer/ingest_denylist.py b/src/brainlayer/ingest_denylist.py
index 43ff6ba4..a9818f1b 100644
--- a/src/brainlayer/ingest_denylist.py
+++ b/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
diff --git a/src/brainlayer/watcher.py b/src/brainlayer/watcher.py
index 40bb3f30..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,12 +133,12 @@ def normalize_provider_entry(entry: dict[str, Any], provider: str) -> dict[str,
     if not text:
         return None
 
+    source_timestamp = _first_timestamp(entry, candidate)
     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,
     }
 
@@ -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
 
@@ -499,6 +522,10 @@ def __init__(
         health_path: str | Path | None = None,
         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,
+        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]
@@ -521,6 +548,10 @@ 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.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()
@@ -543,7 +574,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 +588,14 @@ def provider_for_file(self, filepath: str) -> str:
                 return root.provider
         return "unknown"
 
+    def _is_denylisted(self, filepath: str) -> bool:
+        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."""
         discovered: list[tuple[float, str, str]] = []
@@ -574,7 +613,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
@@ -593,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
@@ -775,20 +833,21 @@ 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)
                     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
new file mode 100644
index 00000000..70a2dfb2
--- /dev/null
+++ b/tests/test_backfill.py
@@ -0,0 +1,210 @@
+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
+
+
+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_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,
+        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_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)
+    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")
+
+    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_ingest_denylist.py b/tests/test_ingest_denylist.py
index 79a1a6d0..7f7ec46b 100644
--- a/tests/test_ingest_denylist.py
+++ b/tests/test_ingest_denylist.py
@@ -2,7 +2,11 @@
 from pathlib import Path
 
 import brainlayer.ingest_denylist as denylist
-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:
@@ -195,3 +199,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_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_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()
diff --git a/tests/test_watch_backfill_cli.py b/tests/test_watch_backfill_cli.py
index bf5c7088..e1e29abc 100644
--- a/tests/test_watch_backfill_cli.py
+++ b/tests/test_watch_backfill_cli.py
@@ -64,6 +64,155 @@ 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
+    assert "scope=legacy-excluded-only" in result.output
+    assert "-legacy-excluded-only.json" in result.output
+
+
+def test_watch_backfill_legacy_scope_keeps_current_denylist(tmp_path, monkeypatch):
+    monkeypatch.delenv("BRAINLAYER_INGEST_DENYLIST", raising=False)
+    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(
+            {
+                "type": "assistant",
+                "timestamp": "2026-07-12T12:00:00Z",
+                "message": {
+                    "role": "assistant",
+                    "content": [
+                        {
+                            "type": "text",
+                            "text": (
+                                "Configured denylist backfill sentinel records the durable implementation decision, "
+                                "the exact migration window, and the verification contract for a safe replay."
+                            ),
+                        }
+                    ],
+                },
+            }
+        )
+        + "\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_INGEST_DENYLIST": "~/.codex/sessions/**/blocked/**",
+            "BRAINLAYER_QUEUE_DIR": str(queue_dir),
+        },
+    )
+
+    assert result.exit_code == 0, result.output
+    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):
     monkeypatch.delenv("BRAINLAYER_INGEST_DENYLIST", raising=False)
     transcript = (
@@ -123,3 +272,124 @@ 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_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,
+        [
+            "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