diff --git a/scripts/run_tests.sh b/scripts/run_tests.sh index 351d2769..70aefc84 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/watcher.py) test_path="$TEST_ROOT/test_jsonl_watcher.py" if [ -f "$test_path" ]; then diff --git a/src/brainlayer/backfill.py b/src/brainlayer/backfill.py new file mode 100644 index 00000000..1e94b40f --- /dev/null +++ b/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), + ) diff --git a/src/brainlayer/cli/__init__.py b/src/brainlayer/cli/__init__.py index 4af6abc7..1ba67ebb 100644 --- a/src/brainlayer/cli/__init__.py +++ b/src/brainlayer/cli/__init__.py @@ -3491,17 +3491,51 @@ 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( @@ -3509,6 +3543,10 @@ 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, + 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] = {} @@ -3516,26 +3554,78 @@ 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 - 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) diff --git a/src/brainlayer/ingest_denylist.py b/src/brainlayer/ingest_denylist.py index f3e31aab..79eea475 100644 --- a/src/brainlayer/ingest_denylist.py +++ b/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 diff --git a/src/brainlayer/watcher.py b/src/brainlayer/watcher.py index 9b53184c..4bc34170 100644 --- a/src/brainlayer/watcher.py +++ b/src/brainlayer/watcher.py @@ -96,6 +96,26 @@ 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 = _valid_timestamp(entry.get(key)) + if value is not None: + 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 @@ -103,6 +123,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 @@ -111,9 +134,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 @@ -134,12 +157,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, } @@ -198,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: @@ -615,17 +642,30 @@ 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, + 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 [] @@ -643,6 +683,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(_ProgressMarker(_line_end_offset=self.offset, _watermark_only=True)) continue try: @@ -650,8 +692,22 @@ 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( + _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( + _ProgressMarker( + _line_end_offset=self.offset + nl_idx + 1, + _watermark_only=True, + ) + ) self.offset += nl_idx + 1 @@ -692,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]): @@ -761,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: @@ -829,6 +887,13 @@ 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, + 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] @@ -851,6 +916,13 @@ 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.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() @@ -863,7 +935,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(): @@ -874,7 +946,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) @@ -888,6 +960,17 @@ 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 _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]] = [] @@ -903,16 +986,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 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) @@ -924,10 +1008,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 isinstance(line, _ProgressMarker): + 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 @@ -948,6 +1051,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 @@ -1114,20 +1238,22 @@ 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 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) - 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, + max_read_bytes=self.max_read_bytes_per_file, + ) # 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..1651413b --- /dev/null +++ b/tests/test_backfill.py @@ -0,0 +1,242 @@ +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 + + +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_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) + 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") + 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_ingest_denylist.py b/tests/test_ingest_denylist.py index 6d5094bf..35d789fa 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: @@ -224,3 +228,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 7330edd7..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" @@ -954,6 +1005,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 938b698e..01eb1686 100644 --- a/tests/test_run_tests_script.py +++ b/tests/test_run_tests_script.py @@ -217,6 +217,37 @@ 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 + 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 + + def test_changed_only_scope_maps_watcher_source_to_jsonl_watcher_tests(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..b86447f6 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): @@ -64,6 +75,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 +283,309 @@ 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 + + +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 + + +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