diff --git a/src/brainlayer/cli/__init__.py b/src/brainlayer/cli/__init__.py index c9ff3ead..f94bfb33 100644 --- a/src/brainlayer/cli/__init__.py +++ b/src/brainlayer/cli/__init__.py @@ -3494,9 +3494,9 @@ def watch_backfill( while cycles < max_cycles: cycles += 1 count = watcher.poll_once() - if count == 0: - break processed += count + if count == 0 and not watcher.last_poll_made_progress: + break watcher.indexer.flush() watcher.registry.flush() rprint(f"processed_entries={processed} cycles={cycles} registry={registry_path}") diff --git a/src/brainlayer/watcher.py b/src/brainlayer/watcher.py index dbeed8ef..a540da59 100644 --- a/src/brainlayer/watcher.py +++ b/src/brainlayer/watcher.py @@ -15,6 +15,7 @@ """ import errno +import hashlib import json import logging import math @@ -47,9 +48,15 @@ _WATCH_MAX_FILE_BYTES_ENV = "BRAINLAYER_WATCH_MAX_FILE_BYTES" _DEFAULT_WATCH_MAX_FILE_BYTES = 100 * 1024 * 1024 +_WATCH_MAX_RECORD_BYTES_ENV = "BRAINLAYER_WATCH_MAX_RECORD_BYTES" +_DEFAULT_WATCH_MAX_RECORD_BYTES = 128 * 1024 * 1024 +_WATCH_READ_CHUNK_BYTES = 64 * 1024 +_MAX_HEALTH_FAILURE_DETAILS = 100 +_MAX_HEALTH_QUARANTINE_DETAILS = 100 -def _watch_max_file_bytes() -> int: +def _watch_read_window_bytes() -> int: + """Load the per-file, per-poll read window from the legacy environment name.""" raw_value = os.environ.get(_WATCH_MAX_FILE_BYTES_ENV, str(_DEFAULT_WATCH_MAX_FILE_BYTES)) try: parsed_value = int(raw_value) @@ -61,7 +68,7 @@ def _watch_max_file_bytes() -> int: _DEFAULT_WATCH_MAX_FILE_BYTES, ) return _DEFAULT_WATCH_MAX_FILE_BYTES - if parsed_value < 0: + if parsed_value <= 0: logger.warning( "Invalid %s=%r; using default %d", _WATCH_MAX_FILE_BYTES_ENV, @@ -72,6 +79,24 @@ def _watch_max_file_bytes() -> int: return parsed_value +def _watch_max_record_bytes() -> int: + """Load the hard in-memory ceiling for one JSONL record.""" + raw_value = os.environ.get(_WATCH_MAX_RECORD_BYTES_ENV, str(_DEFAULT_WATCH_MAX_RECORD_BYTES)) + try: + parsed_value = int(raw_value) + except ValueError: + parsed_value = 0 + if parsed_value <= 0: + logger.warning( + "Invalid %s=%r; using default %d", + _WATCH_MAX_RECORD_BYTES_ENV, + raw_value, + _DEFAULT_WATCH_MAX_RECORD_BYTES, + ) + return _DEFAULT_WATCH_MAX_RECORD_BYTES + return parsed_value + + @dataclass(frozen=True) class WatchRoot: provider: str @@ -336,13 +361,27 @@ def get(self, filepath: str) -> tuple[int, int]: entry = self._data.get(filepath, {}) return entry.get("offset", 0), entry.get("inode", 0) - def set(self, filepath: str, offset: int, inode: int): - """Update offset for a file.""" - generation = self._entry_generation(self._data.get(filepath)) + def generation(self, filepath: str) -> int: + """Return the current rewind generation for a file.""" + return self._entry_generation(self._data.get(filepath)) + + def set(self, filepath: str, offset: int, inode: int, *, generation: int | None = None): + """Update offset for a file, optionally preserving its validated generation.""" + current_generation = self._entry_generation(self._data.get(filepath)) + preserve_generation = generation is not None + if generation is None: + generation = current_generation + elif not isinstance(generation, int) or isinstance(generation, bool) or generation < 0: + raise ValueError("generation must be a non-negative integer") + elif generation < current_generation: + return tombstone = self._removed.get(filepath) valid_inode = isinstance(inode, int) and not isinstance(inode, bool) and inode > 0 if tombstone is not None and valid_inode: - generation = max(generation, int(tombstone["generation"]) + 1, time.time_ns()) + if preserve_generation and generation <= int(tombstone["generation"]): + return + if not preserve_generation: + generation = max(generation, int(tombstone["generation"]) + 1, time.time_ns()) self._data[filepath] = { "offset": offset, "inode": inode, @@ -602,21 +641,30 @@ def prune_missing_files( # ── JSONL Tailer ───────────────────────────────────────────────────────────── +class OversizedJSONLRecordError(ValueError): + """A single JSONL record exceeded the configured in-memory safety ceiling.""" + + class JSONLTailer: """Tail-follows a single JSONL file from a stored offset. Handles partial writes: buffers incomplete lines until a newline arrives. - Validates JSON before yielding — skips corrupt lines silently. + Validates JSON before yielding and stops before corrupt records so callers + can surface the failure without checkpointing unparsed bytes. Detects file rewinds (checkpoint restore) when file shrinks. """ - def __init__(self, filepath: str, offset: int = 0): + def __init__(self, filepath: str, offset: int = 0, max_record_bytes: int | None = None): self.filepath = filepath self.offset = offset + self.max_record_bytes = max_record_bytes self._buffer = b"" self.rewound = False # Set to True when rewind detected self.rewind_old_offset = 0 self.rewind_new_offset = 0 + self.last_error: OSError | json.JSONDecodeError | UnicodeDecodeError | OversizedJSONLRecordError | None = None + self.failed_record: bytes | None = None + self.observed_inode = self.get_inode() def check_rewind(self) -> bool: """Check if file has shrunk (checkpoint restore). Returns True if rewound.""" @@ -641,23 +689,62 @@ def check_rewind(self) -> bool: return True return False - def read_new_lines(self, max_lines: int | None = None) -> list[dict]: - """Read any new complete lines since last call. Returns parsed JSON dicts.""" + def read_new_lines( + self, + max_lines: int | None = None, + max_bytes: int | None = None, + ) -> list[dict]: + """Read a bounded window of new bytes and return complete parsed JSON dicts.""" + self.last_error = None + self.failed_record = None # Check for rewind before reading self.check_rewind() + if self.max_record_bytes is not None and self._partial_record_bytes() > self.max_record_bytes: + return self.read_buffered_lines(max_lines=max_lines) + try: with open(self.filepath, "rb") as f: f.seek(self.offset + len(self._buffer)) - new_data = f.read() - except OSError: + remaining_bytes = max_bytes if max_bytes is not None and max_bytes > 0 else None + complete_lines = self._buffer.count(b"\n") + current_record_bytes = self._partial_record_bytes() + combined = bytearray(self._buffer) + while remaining_bytes is None or remaining_bytes > 0: + if max_lines is not None and complete_lines >= max_lines: + break + chunk_bytes = _WATCH_READ_CHUNK_BYTES + if remaining_bytes is not None: + chunk_bytes = min(chunk_bytes, remaining_bytes) + if self.max_record_bytes is not None: + record_capacity = self.max_record_bytes - current_record_bytes + 1 + if record_capacity <= 0: + break + chunk_bytes = min(chunk_bytes, record_capacity) + new_data = f.read(chunk_bytes) + if not new_data: + break + combined.extend(new_data) + if remaining_bytes is not None: + remaining_bytes -= len(new_data) + + cursor = 0 + while True: + newline_index = new_data.find(b"\n", cursor) + if newline_index < 0: + current_record_bytes += len(new_data) - cursor + break + current_record_bytes += newline_index - cursor + complete_lines += 1 + current_record_bytes = 0 + cursor = newline_index + 1 + self._buffer = bytes(combined) + except OSError as error: + self.last_error = error return [] - if not new_data and b"\n" not in self._buffer: + if b"\n" not in self._buffer and not self._buffer: return [] - - if new_data: - self._buffer += new_data return self.read_buffered_lines(max_lines=max_lines) def has_complete_buffered_line(self) -> bool: @@ -667,30 +754,67 @@ def has_complete_buffered_line(self) -> bool: def read_buffered_lines(self, max_lines: int | None = None) -> list[dict]: """Parse complete buffered records without reading more bytes from disk.""" lines = [] + self.last_error = None + self.failed_record = None + consumed_bytes = 0 + starting_offset = self.offset - while b"\n" in self._buffer: + while True: if max_lines is not None and len(lines) >= max_lines: break - nl_idx = self._buffer.index(b"\n") - line_data = self._buffer[:nl_idx] - self._buffer = self._buffer[nl_idx + 1 :] + nl_idx = self._buffer.find(b"\n", consumed_bytes) + if nl_idx < 0: + break + line_data = self._buffer[consumed_bytes:nl_idx] if not line_data.strip(): - self.offset += nl_idx + 1 + consumed_bytes = nl_idx + 1 continue + if self.max_record_bytes is not None and len(line_data) > self.max_record_bytes: + self.last_error = OversizedJSONLRecordError(f"JSONL record exceeds {self.max_record_bytes} bytes") + break + try: parsed = json.loads(line_data) - if isinstance(parsed, dict): - parsed["_line_end_offset"] = self.offset + nl_idx + 1 - lines.append(parsed) - except (json.JSONDecodeError, UnicodeDecodeError): - logger.debug("Skipping corrupt JSONL line at offset %d", self.offset) + except (json.JSONDecodeError, UnicodeDecodeError) as error: + self.last_error = error + break + + line_end_offset = starting_offset + nl_idx + 1 + consumed_bytes = nl_idx + 1 + if isinstance(parsed, dict): + parsed["_line_end_offset"] = line_end_offset + lines.append(parsed) - self.offset += nl_idx + 1 + if consumed_bytes: + self._buffer = self._buffer[consumed_bytes:] + self.offset = starting_offset + consumed_bytes + + if self.last_error is not None and b"\n" in self._buffer: + failed_end = self._buffer.index(b"\n") + 1 + self.failed_record = self._buffer[:failed_end] + elif self.max_record_bytes is not None and self._partial_record_bytes() > self.max_record_bytes: + self.last_error = OversizedJSONLRecordError(f"JSONL record exceeds {self.max_record_bytes} bytes") return lines + def discard_failed_record(self) -> tuple[int, int, bytes] | None: + """Advance over a failed complete record after the caller durably quarantines it.""" + if self.failed_record is None or not self._buffer.startswith(self.failed_record): + return None + start_offset = self.offset + record = self.failed_record + self._buffer = self._buffer[len(record) :] + self.offset += len(record) + self.failed_record = None + self.last_error = None + return start_offset, self.offset, record + + def _partial_record_bytes(self) -> int: + last_newline = self._buffer.rfind(b"\n") + return len(self._buffer) if last_newline < 0 else len(self._buffer) - last_newline - 1 + def get_inode(self) -> int: """Return the inode of the file, or 0 if not accessible.""" try: @@ -714,12 +838,12 @@ def __init__( on_flush: Callable[[list[dict]], dict[str, int] | None], batch_size: int = 10, flush_interval_ms: int = 100, - on_confirm_offsets: Callable[[dict[str, int]], None] | None = None, + on_confirm_batch: Callable[[dict[str, int], list[dict]], None] | None = None, ): self.on_flush = on_flush self.batch_size = batch_size self.flush_interval_ms = flush_interval_ms - self.on_confirm_offsets = on_confirm_offsets + self.on_confirm_batch = on_confirm_batch self._buffer: list[dict] = [] self._lock = threading.Lock() self._last_flush = time.monotonic() @@ -765,8 +889,8 @@ def _do_flush(self): try: result = self.on_flush(batch) watermarks = self._confirmed_watermarks(batch, result) - if watermarks and self.on_confirm_offsets: - self.on_confirm_offsets(watermarks) + if watermarks and self.on_confirm_batch: + self.on_confirm_batch(watermarks, batch) self._buffer = [] # Clear only after successful flush self._flush_failures = 0 self.total_flushed += count @@ -812,6 +936,7 @@ def _isolate_failed_flush(self, batch: list[dict], reason: Exception) -> int: return len(batch) retained: list[dict] = [] + confirmed_items: list[dict] = [] confirmed: dict[str, int] = {} outputs = 0 for item in batch: @@ -823,10 +948,11 @@ def _isolate_failed_flush(self, batch: list[dict], reason: Exception) -> int: item_watermarks = self._confirmed_watermarks([item], result) for source_file, offset in item_watermarks.items(): confirmed[source_file] = max(confirmed.get(source_file, 0), offset) + confirmed_items.append(item) outputs += getattr(result, "inserted", 1) - if confirmed and self.on_confirm_offsets: - self.on_confirm_offsets(confirmed) + if confirmed and self.on_confirm_batch: + self.on_confirm_batch(confirmed, confirmed_items) self.total_outputs += outputs self.total_flushed += len(batch) - len(retained) self._buffer = retained @@ -887,15 +1013,19 @@ def __init__( on_flush=on_flush or (lambda _items: None), batch_size=batch_size, flush_interval_ms=flush_interval_ms, - on_confirm_offsets=self._advance_confirmed_offsets, + on_confirm_batch=self._advance_confirmed_batch, ) 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.max_file_bytes = _watch_max_file_bytes() + self.max_read_bytes_per_file = _watch_read_window_bytes() + self.max_record_bytes = _watch_max_record_bytes() self._tailers: dict[str, JSONLTailer] = {} self._file_providers: dict[str, str] = {} - self._oversized_files: set[str] = set() + self._file_ingestion_failures: dict[str, dict[str, Any]] = {} + self._quarantined_record_count_total = 0 + self._quarantined_records: list[dict[str, Any]] = [] + self._pending_quarantined_offsets: dict[str, list[tuple[int, int, int, int]]] = {} self._stop = threading.Event() self._last_registry_flush = time.monotonic() self.health_path = Path(health_path).expanduser() if health_path else None @@ -906,15 +1036,62 @@ def __init__( self._health_entries_seen = 0 self._health_output_at_start = 0 self.poll_count = 0 + self.last_poll_made_progress = False self._offset_prune_complete = False - def _advance_confirmed_offsets(self, watermarks: dict[str, int]) -> None: - for filepath, offset in watermarks.items(): - tailer = self._tailers.get(filepath) - inode = tailer.get_inode() if tailer else self.registry.get(filepath)[1] + def _advance_confirmed_offsets(self, confirmations: dict[str, tuple[int, int, int]]) -> None: + for filepath, (offset, source_inode, source_generation) in confirmations.items(): current_offset, _current_inode = self.registry.get(filepath) if offset >= current_offset: - self.registry.set(filepath, offset, inode) + self.registry.set(filepath, offset, source_inode, generation=source_generation) + self._advance_quarantined_offsets(filepath, source_inode, source_generation) + + def _advance_confirmed_batch(self, watermarks: dict[str, int], batch: list[dict]) -> None: + """Confirm only offsets produced by the file generation currently being tailed.""" + current_confirmations: dict[str, tuple[int, int, int]] = {} + for filepath, reported_offset in watermarks.items(): + tailer = self._tailers.get(filepath) + if tailer is None: + continue + current_inode = tailer.observed_inode + current_generation = self.registry.generation(filepath) + eligible_offsets = [ + item["_line_end_offset"] + for item in batch + if item.get("_source_file") == filepath + and item.get("_source_inode") == current_inode + and item.get("_source_generation") == current_generation + and isinstance(item.get("_line_end_offset"), int) + and item["_line_end_offset"] <= reported_offset + ] + if eligible_offsets: + current_confirmations[filepath] = (max(eligible_offsets), current_inode, current_generation) + self._advance_confirmed_offsets(current_confirmations) + + def _advance_quarantined_offsets(self, filepath: str, source_inode: int, source_generation: int) -> None: + """Advance over consecutive durably quarantined records after prior bytes confirm.""" + pending = self._pending_quarantined_offsets.get(filepath) + if not pending: + return + current_offset, current_inode = self.registry.get(filepath) + current_generation = self.registry.generation(filepath) + if current_generation != source_generation or current_inode not in (0, source_inode): + return + remaining: list[tuple[int, int, int, int]] = [] + for start_offset, end_offset, pending_inode, pending_generation in sorted(pending): + if (pending_inode, pending_generation) != (source_inode, source_generation): + remaining.append((start_offset, end_offset, pending_inode, pending_generation)) + continue + if current_offset < start_offset: + remaining.append((start_offset, end_offset, pending_inode, pending_generation)) + continue + if end_offset > current_offset: + self.registry.set(filepath, end_offset, pending_inode, generation=pending_generation) + current_offset = end_offset + if remaining: + self._pending_quarantined_offsets[filepath] = remaining + else: + self._pending_quarantined_offsets.pop(filepath, None) def provider_for_file(self, filepath: str) -> str: if is_denylisted(filepath): @@ -985,6 +1162,8 @@ def _checkpoint_discarded_progress( read_start_offset: int, read_end_offset: int, normalized_lines: list[dict], + source_inode: int, + source_generation: int, ) -> None: """Confirm intentionally discarded bytes without crossing indexable work.""" required_confirmed_offset = max( @@ -998,7 +1177,150 @@ def _checkpoint_discarded_progress( confirmed_offset, _confirmed_inode = self.registry.get(filepath) if confirmed_offset < required_confirmed_offset: return - self._advance_confirmed_offsets({filepath: read_end_offset}) + self._advance_confirmed_offsets( + { + filepath: ( + read_end_offset, + source_inode, + source_generation, + ) + } + ) + + def _record_file_ingestion_failure( + self, + filepath: str, + error: BaseException, + extra_context: dict[str, Any] | None = None, + ) -> None: + """Raise once per distinct failure and retain it on the health surface.""" + tailer = self._tailers.get(filepath) + confirmed_offset, confirmed_inode = self.registry.get(filepath) + try: + file_size = os.path.getsize(filepath) + except OSError: + file_size = None + context = { + "file_path": filepath, + "error_type": type(error).__name__, + "error": str(error), + "confirmed_offset": confirmed_offset, + "confirmed_inode": confirmed_inode, + "read_offset": tailer.offset if tailer else confirmed_offset, + "file_size_bytes": file_size, + "observed_at": datetime.now(timezone.utc).isoformat(), + } + if extra_context: + context.update(extra_context) + fingerprint = ( + context["error_type"], + context["error"], + context["confirmed_offset"], + context["read_offset"], + context.get("disposition"), + context.get("quarantine_path"), + ) + previous = self._file_ingestion_failures.get(filepath) + self._file_ingestion_failures[filepath] = {**context, "_fingerprint": fingerprint} + if previous and previous.get("_fingerprint") == fingerprint: + return + try: + raise_alarm( + "watcher_file_ingestion_failed", + "watcher deferred a JSONL file because bytes could not be safely parsed", + context, + ) + except BrainLayerAlarm as alarm: + logger.error("Watcher file-ingestion alarm emitted without stopping watcher: %s", alarm) + + def _clear_file_ingestion_failure(self, filepath: str) -> None: + self._file_ingestion_failures.pop(filepath, None) + + def _quarantine_failed_record( + self, + filepath: str, + tailer: JSONLTailer, + source_inode: int, + source_generation: int, + ) -> bool: + """Durably preserve one malformed record, alarm, and advance over only those bytes.""" + if tailer.failed_record is None or not isinstance( + tailer.last_error, + (json.JSONDecodeError, UnicodeDecodeError), + ): + return False + + record = tailer.failed_record + start_offset = tailer.offset + end_offset = start_offset + len(record) + digest = hashlib.sha256(record).hexdigest() + quarantine_dir = Path( + os.environ.get("BRAINLAYER_WATCHER_QUARANTINE_DIR", "~/.brainlayer/quarantine") + ).expanduser() + quarantine_path = quarantine_dir / ( + f"watcher-parse-{Path(filepath).stem}-{start_offset}-{digest[:16]}.jsonl.bad" + ) + temp_path: Path | None = None + try: + quarantine_dir.mkdir(parents=True, exist_ok=True) + if not quarantine_path.exists(): + with tempfile.NamedTemporaryFile( + mode="wb", + dir=quarantine_dir, + prefix=f".{quarantine_path.name}.", + delete=False, + ) as handle: + temp_path = Path(handle.name) + handle.write(record) + handle.flush() + os.fsync(handle.fileno()) + temp_path.replace(quarantine_path) + temp_path = None + directory_fd = os.open(quarantine_dir, os.O_RDONLY) + try: + os.fsync(directory_fd) + finally: + os.close(directory_fd) + elif quarantine_path.read_bytes() != record: + raise OSError(f"quarantine collision at {quarantine_path}") + except OSError as error: + if temp_path is not None: + temp_path.unlink(missing_ok=True) + tailer.last_error = error + tailer.failed_record = None + return False + + parse_error = tailer.last_error + event = { + "file_path": filepath, + "start_offset": start_offset, + "end_offset": end_offset, + "record_bytes": len(record), + "sha256": digest, + "quarantine_path": str(quarantine_path), + "observed_at": datetime.now(timezone.utc).isoformat(), + } + self._quarantined_record_count_total += 1 + self._quarantined_records.append(event) + self._quarantined_records = self._quarantined_records[-_MAX_HEALTH_QUARANTINE_DETAILS:] + self._record_file_ingestion_failure( + filepath, + parse_error, + { + "disposition": "quarantined", + "quarantine_path": str(quarantine_path), + "quarantined_start_offset": start_offset, + "quarantined_end_offset": end_offset, + }, + ) + discarded = tailer.discard_failed_record() + if discarded is None: + raise RuntimeError("quarantined record no longer matches the tailer buffer") + self._pending_quarantined_offsets.setdefault(filepath, []).append( + (start_offset, end_offset, source_inode, source_generation) + ) + self._advance_quarantined_offsets(filepath, source_inode, source_generation) + return True def _max_offset_lag_bytes(self, files: list[str]) -> int: max_lag = 0 @@ -1075,6 +1397,24 @@ def _write_health_snapshot(self, files: list[str]): realtime_inserts_per_minute=watchdog_inserts_per_min, max_offset_lag_bytes=max_lag, ) + all_failure_payloads = [ + {key: value for key, value in failure.items() if key != "_fingerprint"} + for _filepath, failure in sorted(self._file_ingestion_failures.items()) + ] + failure_payloads = all_failure_payloads[:_MAX_HEALTH_FAILURE_DETAILS] + failure_overflow_count = len(all_failure_payloads) - len(failure_payloads) + alert_reasons = list(watchdog.get("alert_reasons", [])) + if all_failure_payloads and "file_ingestion_failure" not in alert_reasons: + alert_reasons.append("file_ingestion_failure") + if self._quarantined_record_count_total and "quarantined_record" not in alert_reasons: + alert_reasons.append("quarantined_record") + watchdog = { + **watchdog, + "alerting": bool(watchdog.get("alerting")) + or bool(all_failure_payloads) + or bool(self._quarantined_record_count_total), + "alert_reasons": alert_reasons, + } coverage_degraded = ( active_entries_per_min > 0 and watchdog_inserts_per_min / active_entries_per_min < self.coverage_watchdog.coverage_ratio_threshold @@ -1093,6 +1433,11 @@ def _write_health_snapshot(self, files: list[str]): "failed_flush_inputs_per_minute": failed_flush_inputs_per_min, "watcher_chunks_output_per_minute": outputs_per_min, "max_offset_lag_bytes": max_lag, + "file_ingestion_failure_count": len(all_failure_payloads), + "file_ingestion_failures": failure_payloads, + "file_ingestion_failures_overflow_count": failure_overflow_count, + "quarantined_record_count_total": self._quarantined_record_count_total, + "quarantined_records": list(self._quarantined_records), **watchdog, } try: @@ -1143,19 +1488,27 @@ def _ensure_tailer(self, filepath: str) -> JSONLTailer: # Check inode hasn't changed (file replaced) current_inode = tailer.get_inode() stored_offset, stored_inode = self.registry.get(filepath) - if stored_inode != 0 and current_inode != stored_inode: + inode_changed = current_inode != 0 and ( + (stored_inode != 0 and current_inode != stored_inode) + or (tailer.observed_inode != 0 and current_inode != tailer.observed_inode) + ) + if inode_changed: # File was replaced — reset offset - tailer = JSONLTailer(filepath, offset=0) + self._pending_quarantined_offsets.pop(filepath, None) + self.registry.mark_rewind(filepath, current_inode) + tailer = JSONLTailer(filepath, offset=0, max_record_bytes=self.max_record_bytes) self._tailers[filepath] = tailer return tailer stored_offset, stored_inode = self.registry.get(filepath) - tailer = JSONLTailer(filepath, offset=stored_offset) + tailer = JSONLTailer(filepath, offset=stored_offset, max_record_bytes=self.max_record_bytes) # Verify inode matches current_inode = tailer.get_inode() - if stored_inode != 0 and current_inode != stored_inode: - tailer = JSONLTailer(filepath, offset=0) + if stored_inode != 0 and current_inode != 0 and current_inode != stored_inode: + self._pending_quarantined_offsets.pop(filepath, None) + self.registry.mark_rewind(filepath, current_inode) + tailer = JSONLTailer(filepath, offset=0, max_record_bytes=self.max_record_bytes) self._tailers[filepath] = tailer return tailer @@ -1169,6 +1522,7 @@ def _handle_rewind( ) -> None: """Persist a rewind and notify archival consumers.""" session_id = Path(filepath).stem + self._pending_quarantined_offsets.pop(filepath, None) self.registry.mark_rewind(filepath, inode) logger.warning( "Checkpoint restore: %s (offset %d → %d)", @@ -1198,81 +1552,26 @@ def _handle_rewind( except Exception as e: logger.error("Rewind callback failed: %s", e) - def _skip_oversized_file(self, filepath: str) -> bool: - if self.max_file_bytes <= 0: - self._oversized_files.discard(filepath) - return False - - try: - file_stat = os.stat(filepath) - except OSError: - return False - - tailer = self._tailers.get(filepath) - registry_offset, registry_inode = self.registry.get(filepath) - tailer_offset = tailer.offset if tailer else registry_offset - rewind_old_offset = tailer_offset - file_rewound = file_stat.st_size < tailer_offset - inode_changed = registry_inode != 0 and registry_inode != file_stat.st_ino - offset = 0 if inode_changed or file_rewound else registry_offset - pending_bytes = max(file_stat.st_size - offset, 0) - if pending_bytes <= self.max_file_bytes: - self._oversized_files.discard(filepath) - return False - - if tailer is not None and not inode_changed and not file_rewound and tailer_offset > registry_offset: - if self.indexer.has_buffered_source(filepath): - self.indexer.flush() - confirmed_offset, confirmed_inode = self.registry.get(filepath) - if confirmed_inode != file_stat.st_ino or confirmed_offset < tailer_offset: - if filepath not in self._oversized_files: - logger.error( - "Oversized JSONL checkpoint deferred for unconfirmed entries: %s", - filepath, - ) - self._oversized_files.add(filepath) - return True - offset = confirmed_offset - pending_bytes = max(file_stat.st_size - offset, 0) - if pending_bytes <= self.max_file_bytes: - self._oversized_files.discard(filepath) - return False - - self._tailers.pop(filepath, None) - if file_rewound: - self._handle_rewind( - filepath, - rewind_old_offset, - file_stat.st_size, - file_stat.st_ino, - ) - self.registry.set(filepath, file_stat.st_size, file_stat.st_ino) - if not self.registry.flush(): - logger.error( - "Oversized JSONL checkpoint could not be persisted immediately: %s", - filepath, - ) - if filepath not in self._oversized_files: - logger.warning( - "Oversized JSONL checkpointed and skipped: %s pending_bytes=%d max_file_bytes=%d offset=%d size=%d", - filepath, - pending_bytes, - self.max_file_bytes, - offset, - file_stat.st_size, - ) - self._oversized_files.add(filepath) - return True - def poll_once(self) -> int: """Run one poll cycle. Returns number of new lines found.""" total_new = 0 files: list[str] = [] self.poll_count += 1 + self.last_poll_made_progress = False try: files = self._discover_jsonl_files() - self._oversized_files.intersection_update(filepath for filepath in files if not is_denylisted(filepath)) + live_files = {filepath for filepath in files if not is_denylisted(filepath)} + self._file_ingestion_failures = { + filepath: failure + for filepath, failure in self._file_ingestion_failures.items() + if filepath in live_files + } + self._pending_quarantined_offsets = { + filepath: pending + for filepath, pending in self._pending_quarantined_offsets.items() + if filepath in live_files + } if not self._offset_prune_complete: pruned = self.registry.prune_missing_files( [root.resolved_path for root in self.watch_roots], @@ -1286,20 +1585,31 @@ def poll_once(self) -> int: if is_denylisted(filepath): self._tailers.pop(filepath, None) self._file_providers.pop(filepath, None) + self._pending_quarantined_offsets.pop(filepath, None) self.registry.remove(filepath) for filepath in files: if is_denylisted(filepath): self._tailers.pop(filepath, None) self._file_providers.pop(filepath, None) + self._pending_quarantined_offsets.pop(filepath, None) self.registry.remove(filepath) continue + tailer: JSONLTailer | None = None + tailer_snapshot: tuple[int, bytes] | None = None + source_inode = 0 + source_generation = 0 + read_accepted = False try: tailer = self._tailers.get(filepath) drain_buffer = False if tailer is not None and tailer.has_complete_buffered_line(): _registry_offset, registry_inode = self.registry.get(filepath) - inode_changed = registry_inode != 0 and registry_inode != tailer.get_inode() + current_inode = tailer.get_inode() + inode_changed = current_inode != 0 and ( + (registry_inode != 0 and registry_inode != current_inode) + or (tailer.observed_inode != 0 and tailer.observed_inode != current_inode) + ) if not inode_changed and not tailer.check_rewind(): drain_buffer = True elif tailer.rewound: @@ -1313,13 +1623,20 @@ def poll_once(self) -> int: if drain_buffer: read_start_offset = tailer.offset + tailer_snapshot = (tailer.offset, tailer._buffer) + source_inode = tailer.observed_inode + source_generation = self.registry.generation(filepath) new_lines = tailer.read_buffered_lines(max_lines=self.max_lines_per_file) else: - if self._skip_oversized_file(filepath): - continue tailer = self._ensure_tailer(filepath) read_start_offset = tailer.offset - new_lines = tailer.read_new_lines(max_lines=self.max_lines_per_file) + tailer_snapshot = (tailer.offset, tailer._buffer) + source_inode = tailer.observed_inode + source_generation = self.registry.generation(filepath) + new_lines = tailer.read_new_lines( + max_lines=self.max_lines_per_file, + max_bytes=self.max_read_bytes_per_file, + ) # Handle rewind detection (checkpoint restore) if tailer.rewound: @@ -1331,10 +1648,16 @@ def poll_once(self) -> int: tailer.get_inode(), ) tailer.rewound = False # Reset flag + source_inode = tailer.observed_inode + source_generation = self.registry.generation(filepath) normalized_lines = self._normalize_lines(filepath, new_lines) if new_lines else [] if normalized_lines: + for line in normalized_lines: + line["_source_inode"] = source_inode + line["_source_generation"] = source_generation self.indexer.add(normalized_lines) + read_accepted = True self._health_entries_seen += len(normalized_lines) total_new += len(normalized_lines) self._checkpoint_discarded_progress( @@ -1342,9 +1665,31 @@ def poll_once(self) -> int: read_start_offset, tailer.offset, normalized_lines, + source_inode, + source_generation, ) - except Exception: + if tailer.last_error is not None: + if not self._quarantine_failed_record( + filepath, + tailer, + source_inode, + source_generation, + ): + self._record_file_ingestion_failure(filepath, tailer.last_error) + else: + self._clear_file_ingestion_failure(filepath) + if tailer_snapshot is not None and ( + tailer.offset != tailer_snapshot[0] or tailer._buffer != tailer_snapshot[1] + ): + self.last_poll_made_progress = True + read_accepted = True + except Exception as error: + if tailer is not None and tailer_snapshot is not None and not read_accepted: + tailer.offset, tailer._buffer = tailer_snapshot + tailer.last_error = error + tailer.failed_record = None logger.exception("Poll file error: %s", filepath) + self._record_file_ingestion_failure(filepath, error) self.indexer.tick() if self.on_tick: diff --git a/tests/test_jsonl_watcher.py b/tests/test_jsonl_watcher.py index bb595524..d3d095f0 100644 --- a/tests/test_jsonl_watcher.py +++ b/tests/test_jsonl_watcher.py @@ -579,15 +579,50 @@ def test_partial_line_buffered(self, tmp_path): assert len(lines) == 1 assert lines[0]["partial"] == "value" - def test_corrupt_line_skipped(self, tmp_path): + def test_corrupt_line_stops_before_unparsed_bytes(self, tmp_path): f = tmp_path / "test.jsonl" f.write_text('{"good":"line"}\nnot json at all\n{"also":"good"}\n') tailer = JSONLTailer(str(f)) lines = tailer.read_new_lines() + first_line_end = len(b'{"good":"line"}\n') + + assert lines == [{"good": "line", "_line_end_offset": first_line_end}] + assert tailer.offset == first_line_end + assert tailer._buffer.startswith(b"not json at all\n") + assert tailer.last_error is not None + + def test_read_new_lines_limits_bytes_per_call(self, tmp_path): + f = tmp_path / "test.jsonl" + f.write_text(json.dumps({"role": "user", "content": "x" * 256}) + "\n") + tailer = JSONLTailer(str(f)) + + assert tailer.read_new_lines(max_bytes=64) == [] + assert tailer.offset == 0 + assert len(tailer._buffer) == 64 + + def test_read_new_lines_stops_reading_once_line_limit_is_buffered(self, tmp_path): + f = tmp_path / "test.jsonl" + record = json.dumps({"role": "user", "content": "x" * 2048}) + "\n" + f.write_text(record * 200) + tailer = JSONLTailer(str(f)) + + lines = tailer.read_new_lines(max_lines=2, max_bytes=1024 * 1024) + assert len(lines) == 2 - assert lines[0]["good"] == "line" - assert lines[1]["also"] == "good" + assert len(tailer._buffer) < 64 * 1024 + + def test_read_new_lines_bounds_an_oversized_incomplete_record(self, tmp_path): + f = tmp_path / "test.jsonl" + f.write_bytes(b'{"role":"user","content":"' + b"x" * 20_000) + tailer = JSONLTailer(str(f), max_record_bytes=4096) + + for _ in range(10): + assert tailer.read_new_lines(max_bytes=1024) == [] + + assert len(tailer._buffer) <= 4097 + assert type(tailer.last_error).__name__ == "OversizedJSONLRecordError" + assert tailer.offset == 0 def test_resume_from_offset(self, tmp_path): f = tmp_path / "test.jsonl" @@ -689,12 +724,12 @@ def test_total_flushed_counter(self): indexer.add([{"c": 3}, {"d": 4}]) assert indexer.total_flushed == 4 - def test_flush_callback_watermark_is_forwarded_to_offset_callback(self): + def test_flush_callback_watermark_is_forwarded_to_batch_callback(self): confirmed = [] indexer = BatchIndexer( on_flush=lambda items: {"/tmp/source.jsonl": items[-1]["_line_end_offset"]}, batch_size=2, - on_confirm_offsets=confirmed.append, + on_confirm_batch=lambda watermarks, _batch: confirmed.append(watermarks), ) indexer.add( @@ -1243,36 +1278,34 @@ def confirm_all(items): assert watcher.registry.get(str(rollout)) == (tailer.offset, rollout.stat().st_ino) assert tailer.offset < oversized_size - def test_poll_checkpoints_dropped_only_records_before_oversized_append(self, tmp_path, monkeypatch): + def test_poll_fully_indexes_file_larger_than_read_window(self, tmp_path, monkeypatch): sessions = tmp_path / "codex" / "sessions" sessions.mkdir(parents=True) rollout = sessions / "rollout.jsonl" - rollout.write_text(json.dumps({"type": "response_item", "payload": {"type": "function_call"}}) + "\n") - dropped_offset = rollout.stat().st_size + expected = [f"entry {index} " + "x" * 80 for index in range(4)] + rollout.write_text("".join(json.dumps({"role": "user", "content": content}) + "\n" for content in expected)) + assert rollout.stat().st_size > 128 monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "128") flushed = [] + def confirm_all(items): + flushed.extend(items) + return {item["_source_file"]: item["_line_end_offset"] for item in items} + watcher = JSONLWatcher( watch_roots=[WatchRoot("codex", sessions)], registry_path=tmp_path / "offsets.json", - on_flush=lambda items: flushed.extend(items), + on_flush=confirm_all, batch_size=1, ) - assert watcher.poll_once() == 0 - assert flushed == [] - assert watcher.registry.get(str(rollout)) == (dropped_offset, rollout.stat().st_ino) - - with rollout.open("a") as file_handle: - file_handle.write(json.dumps({"role": "user", "content": "x" * 256}) + "\n") - assert watcher.poll_once() == 0 - oversized_checkpoint = rollout.stat().st_size - assert watcher.registry.get(str(rollout))[0] == oversized_checkpoint + for _ in range(12): + watcher.poll_once() + if watcher.registry.get(str(rollout))[0] == rollout.stat().st_size: + break - with rollout.open("a") as file_handle: - file_handle.write(json.dumps({"role": "user", "content": "small append"}) + "\n") - assert watcher.poll_once() == 1 - assert flushed[0]["message"]["content"][0]["text"] == "small append" + assert [item["message"]["content"][0]["text"] for item in flushed] == expected + assert watcher.registry.get(str(rollout)) == (rollout.stat().st_size, rollout.stat().st_ino) def test_poll_does_not_checkpoint_dropped_tail_past_unconfirmed_record(self, tmp_path): sessions = tmp_path / "codex" / "sessions" @@ -1297,11 +1330,10 @@ def test_poll_does_not_checkpoint_dropped_tail_past_unconfirmed_record(self, tmp assert watcher.indexer.has_buffered_source(str(rollout)) assert watcher.registry.get(str(rollout)) == (0, 0) - def test_poll_skips_oversized_pending_file_with_warning_and_continues( + def test_poll_bounds_large_file_without_starving_healthy_file( self, tmp_path, monkeypatch, - caplog, ): sessions = tmp_path / "codex" / "sessions" sessions.mkdir(parents=True) @@ -1328,18 +1360,11 @@ def confirm_all(items): assert watcher.poll_once() == 1 assert [item["_source_file"] for item in flushed] == [str(healthy)] - assert watcher.registry.get(str(oversized)) == ( - oversized.stat().st_size, - oversized.stat().st_ino, - ) - assert any( - str(oversized) in record.getMessage() - and "pending_bytes=" in record.getMessage() - and "max_file_bytes=128" in record.getMessage() - for record in caplog.records - ) + assert watcher.registry.get(str(oversized)) == (0, 0) + assert watcher._tailers[str(oversized)].offset == 0 + assert len(watcher._tailers[str(oversized)]._buffer) == 128 - def test_poll_persists_oversized_checkpoint_immediately(self, tmp_path, monkeypatch): + def test_poll_never_checkpoints_past_unparsed_window(self, tmp_path, monkeypatch): sessions = tmp_path / "codex" / "sessions" sessions.mkdir(parents=True) oversized = sessions / "oversized.jsonl" @@ -1355,12 +1380,11 @@ def test_poll_persists_oversized_checkpoint_immediately(self, tmp_path, monkeypa ) assert watcher.poll_once() == 0 - assert OffsetRegistry(registry_path).get(str(oversized)) == ( - oversized.stat().st_size, - oversized.stat().st_ino, - ) + assert watcher._tailers[str(oversized)].offset == 0 + assert len(watcher._tailers[str(oversized)]._buffer) == 128 + assert OffsetRegistry(registry_path).get(str(oversized)) == (0, 0) - def test_poll_caps_oversized_replacement_from_start(self, tmp_path, monkeypatch): + def test_poll_indexes_large_inode_replacement_from_start(self, tmp_path, monkeypatch): sessions = tmp_path / "codex" / "sessions" sessions.mkdir(parents=True) rollout = sessions / "rollout.jsonl" @@ -1388,14 +1412,154 @@ def confirm_all(items): os.replace(replacement, rollout) assert rollout.stat().st_ino != original_inode - assert watcher.poll_once() == 0 - assert flushed == [] + for _ in range(8): + watcher.poll_once() + if flushed: + break + + assert [item["message"]["content"][0]["text"] for item in flushed] == ["y" * 256] assert watcher.registry.get(str(rollout)) == ( rollout.stat().st_size, rollout.stat().st_ino, ) - def test_poll_caps_oversized_same_inode_rewind_from_start(self, tmp_path, monkeypatch): + def test_poll_discards_unconfirmed_buffer_when_inode_is_replaced(self, tmp_path, monkeypatch): + sessions = tmp_path / "codex" / "sessions" + sessions.mkdir(parents=True) + rollout = sessions / "rollout.jsonl" + rollout.write_bytes(b'{"role":"user","content":"' + b"x" * 256) + monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "128") + flushed = [] + + def confirm_all(items): + flushed.extend(items) + return {item["_source_file"]: item["_line_end_offset"] for item in items} + + watcher = JSONLWatcher( + watch_roots=[WatchRoot("codex", sessions)], + registry_path=tmp_path / "offsets.json", + on_flush=confirm_all, + batch_size=1, + ) + watcher.poll_once() + original_inode = rollout.stat().st_ino + assert watcher.registry.get(str(rollout)) == (0, 0) + assert watcher._tailers[str(rollout)]._buffer + + replacement = sessions / "replacement.jsonl" + replacement.write_text(json.dumps({"role": "user", "content": "replacement"}) + "\n") + os.replace(replacement, rollout) + assert rollout.stat().st_ino != original_inode + + watcher.poll_once() + + assert [item["message"]["content"][0]["text"] for item in flushed] == ["replacement"] + assert watcher.registry.get(str(rollout)) == (rollout.stat().st_size, rollout.stat().st_ino) + + def test_old_inode_flush_watermark_cannot_advance_replacement_offset(self, tmp_path, monkeypatch): + sessions = tmp_path / "codex" / "sessions" + sessions.mkdir(parents=True) + rollout = sessions / "rollout.jsonl" + old_records = [f"old-{index}-" + "x" * 64 for index in range(20)] + rollout.write_text("".join(json.dumps({"role": "user", "content": content}) + "\n" for content in old_records)) + state = {"fail": True} + + def flush(items): + if state["fail"]: + raise RuntimeError("retain old-inode batch") + watermarks = {} + for item in items: + source = item["_source_file"] + watermarks[source] = max(watermarks.get(source, 0), item["_line_end_offset"]) + return watermarks + + def capture_alarm(code, message, context): + raise BrainLayerAlarm(code, message, context) + + monkeypatch.setattr("brainlayer.watcher.raise_alarm", capture_alarm) + watcher = JSONLWatcher( + watch_roots=[WatchRoot("codex", sessions)], + registry_path=tmp_path / "offsets.json", + on_flush=flush, + batch_size=20, + ) + watcher.poll_once() + assert watcher.registry.get(str(rollout)) == (0, 0) + assert len(watcher.indexer._buffer) == 20 + + replacement = sessions / "replacement.jsonl" + replacement.write_text( + "".join(json.dumps({"role": "user", "content": f"new-{index}"}) + "\n" for index in range(3)) + ) + os.replace(replacement, rollout) + replacement_size = rollout.stat().st_size + replacement_inode = rollout.stat().st_ino + watcher.poll_once() + + state["fail"] = False + watcher.indexer.flush() + + assert watcher.registry.get(str(rollout)) == (replacement_size, replacement_inode) + + with rollout.open("a") as file_handle: + file_handle.write(json.dumps({"role": "user", "content": "new-append"}) + "\n") + watcher.poll_once() + watcher.indexer.flush() + assert watcher.registry.get(str(rollout)) == (rollout.stat().st_size, replacement_inode) + + def test_replacement_between_read_and_confirmation_restarts_at_zero(self, tmp_path, monkeypatch): + sessions = tmp_path / "codex" / "sessions" + sessions.mkdir(parents=True) + rollout = sessions / "rollout.jsonl" + valid_line = (json.dumps({"role": "user", "content": "old valid record"}) + "\n").encode() + malformed_line = b'{"role":"user","content":}\n' + rollout.write_bytes(valid_line + malformed_line) + old_inode = rollout.stat().st_ino + registry_path = tmp_path / "offsets.json" + replacement_text = "replacement must be read from its first byte" + replacement = sessions / "replacement.tmp" + replacement.write_text(json.dumps({"role": "user", "content": replacement_text}) + "\n") + replacement_inode = replacement.stat().st_ino + flushed = [] + + def replace_during_flush(items): + flushed.extend(items) + os.replace(replacement, rollout) + return {item["_source_file"]: item["_line_end_offset"] for item in items} + + monkeypatch.setenv("BRAINLAYER_WATCHER_QUARANTINE_DIR", str(tmp_path / "quarantine")) + watcher = JSONLWatcher( + watch_roots=[WatchRoot("codex", sessions)], + registry_path=registry_path, + on_flush=replace_during_flush, + batch_size=1, + registry_flush_interval_s=3600, + ) + + assert watcher.poll_once() == 1 + assert rollout.stat().st_ino == replacement_inode + assert watcher.registry.get(str(rollout)) == (len(valid_line + malformed_line), old_inode) + assert watcher.registry.flush() is True + + replacement_items = [] + + def confirm_replacement(items): + replacement_items.extend(items) + return {item["_source_file"]: item["_line_end_offset"] for item in items} + + fresh_watcher = JSONLWatcher( + watch_roots=[WatchRoot("codex", sessions)], + registry_path=registry_path, + on_flush=confirm_replacement, + batch_size=1, + registry_flush_interval_s=3600, + ) + + assert fresh_watcher._ensure_tailer(str(rollout)).offset == 0 + assert fresh_watcher.poll_once() == 1 + assert [item["message"]["content"][0]["text"] for item in replacement_items] == [replacement_text] + + def test_poll_indexes_large_same_inode_rewind_from_start(self, tmp_path, monkeypatch): sessions = tmp_path / "codex" / "sessions" sessions.mkdir(parents=True) rollout = sessions / "rollout.jsonl" @@ -1406,10 +1570,14 @@ def test_poll_caps_oversized_same_inode_rewind_from_start(self, tmp_path, monkey flushed = [] rewinds = [] + def confirm_all(items): + flushed.extend(items) + return {item["_source_file"]: item["_line_end_offset"] for item in items} + watcher = JSONLWatcher( watch_roots=[WatchRoot("codex", sessions)], registry_path=tmp_path / "offsets.json", - on_flush=lambda items: flushed.extend(items), + on_flush=confirm_all, on_rewind=lambda *args: rewinds.append(args), batch_size=1, ) @@ -1420,8 +1588,12 @@ def test_poll_caps_oversized_same_inode_rewind_from_start(self, tmp_path, monkey file_handle.write(json.dumps({"role": "user", "content": "y" * 256}) + "\n") assert rollout.stat().st_ino == original_inode - assert watcher.poll_once() == 0 - assert flushed == [] + for _ in range(8): + watcher.poll_once() + if flushed: + break + + assert [item["message"]["content"][0]["text"] for item in flushed] == ["y" * 256] assert watcher.registry.get(str(rollout)) == ( rollout.stat().st_size, rollout.stat().st_ino, @@ -1489,34 +1661,248 @@ def confirm_first_only(items): assert watcher.poll_once() == 0 assert watcher.registry.get(str(rollout)) == (confirmed_offset, confirmed_inode) - def test_poll_forgets_oversized_files_that_disappear_or_become_denylisted(self, tmp_path, monkeypatch): + def test_file_processing_failure_raises_alarm_and_surfaces_in_health(self, tmp_path, monkeypatch): sessions = tmp_path / "codex" / "sessions" sessions.mkdir(parents=True) - disappeared = sessions / "disappeared.jsonl" - denylisted = sessions / "denylisted.jsonl" - for rollout in (disappeared, denylisted): - rollout.write_text(json.dumps({"role": "user", "content": "x" * 256}) + "\n") - monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "128") + rollout = sessions / "rollout.jsonl" + rollout.write_text(json.dumps({"role": "user", "content": "must not be checkpointed"}) + "\n") + health_path = tmp_path / "watcher-health.json" + alarms = [] watcher = JSONLWatcher( watch_roots=[WatchRoot("codex", sessions)], registry_path=tmp_path / "offsets.json", on_flush=lambda _items: None, + health_path=health_path, ) + def forced_failure(_filepath): + raise OSError("forced read failure") + + def capture_alarm(code, message, context): + alarms.append((code, message, context)) + raise BrainLayerAlarm(code, message, context) + + monkeypatch.setattr(watcher, "_ensure_tailer", forced_failure) + monkeypatch.setattr("brainlayer.watcher.raise_alarm", capture_alarm) assert watcher.poll_once() == 0 - assert watcher._oversized_files == {str(disappeared), str(denylisted)} + payload = json.loads(health_path.read_text()) - disappeared.unlink() - monkeypatch.setattr( - "brainlayer.watcher.is_denylisted", - lambda filepath: filepath == str(denylisted), + assert watcher.registry.get(str(rollout)) == (0, 0) + assert alarms[0][0] == "watcher_file_ingestion_failed" + assert alarms[0][2]["file_path"] == str(rollout) + assert payload["alerting"] is True + assert "file_ingestion_failure" in payload["alert_reasons"] + assert payload["file_ingestion_failure_count"] == 1 + assert payload["file_ingestion_failures"][0]["file_path"] == str(rollout) + + def test_normalization_failure_retries_without_crossing_failed_record(self, tmp_path, monkeypatch): + sessions = tmp_path / "codex" / "sessions" + sessions.mkdir(parents=True) + rollout = sessions / "rollout.jsonl" + expected = ["first", "second"] + rollout.write_text("".join(json.dumps({"role": "user", "content": content}) + "\n" for content in expected)) + flushed = [] + + def confirm_all(items): + flushed.extend(items) + return {item["_source_file"]: item["_line_end_offset"] for item in items} + + watcher = JSONLWatcher( + watch_roots=[WatchRoot("codex", sessions)], + registry_path=tmp_path / "offsets.json", + on_flush=confirm_all, + batch_size=1, + max_lines_per_file=1, ) + original_normalize = watcher._normalize_lines + attempts = 0 + + def fail_once(filepath, lines): + nonlocal attempts + attempts += 1 + if attempts == 1: + raise RuntimeError("forced normalization failure") + return original_normalize(filepath, lines) + + def capture_alarm(code, message, context): + raise BrainLayerAlarm(code, message, context) + + monkeypatch.setattr(watcher, "_normalize_lines", fail_once) + monkeypatch.setattr("brainlayer.watcher.raise_alarm", capture_alarm) assert watcher.poll_once() == 0 - assert watcher._oversized_files == set() + assert watcher.registry.get(str(rollout)) == (0, 0) + + watcher.poll_once() + watcher.poll_once() + + assert [item["message"]["content"][0]["text"] for item in flushed] == expected + assert watcher.registry.get(str(rollout))[0] == rollout.stat().st_size + + def test_malformed_record_is_quarantined_and_later_records_continue(self, tmp_path, monkeypatch): + sessions = tmp_path / "codex" / "sessions" + sessions.mkdir(parents=True) + rollout = sessions / "rollout.jsonl" + first = json.dumps({"role": "user", "content": "first"}) + "\n" + malformed = b"not json at all\n" + last = json.dumps({"role": "user", "content": "last"}) + "\n" + rollout.write_bytes(first.encode() + malformed + last.encode()) + health_path = tmp_path / "watcher-health.json" + quarantine_dir = tmp_path / "quarantine" + monkeypatch.setenv("BRAINLAYER_WATCHER_QUARANTINE_DIR", str(quarantine_dir)) + alarms = [] + flushed = [] + + def confirm_all(items): + flushed.extend(items) + return {item["_source_file"]: item["_line_end_offset"] for item in items} + + def capture_alarm(code, message, context): + alarms.append((code, message, context)) + raise BrainLayerAlarm(code, message, context) + + monkeypatch.setattr("brainlayer.watcher.raise_alarm", capture_alarm) + watcher = JSONLWatcher( + watch_roots=[WatchRoot("codex", sessions)], + registry_path=tmp_path / "offsets.json", + on_flush=confirm_all, + batch_size=1, + health_path=health_path, + ) + + for _ in range(3): + watcher.poll_once() + + payload = json.loads(health_path.read_text()) + quarantined = list(quarantine_dir.glob("watcher-parse-*.jsonl.bad")) + assert [item["message"]["content"][0]["text"] for item in flushed] == ["first", "last"] + assert watcher.registry.get(str(rollout))[0] == rollout.stat().st_size + assert len(quarantined) == 1 + assert quarantined[0].read_bytes() == malformed + assert len(alarms) == 1 + assert alarms[0][0] == "watcher_file_ingestion_failed" + assert alarms[0][2]["disposition"] == "quarantined" + assert payload["quarantined_record_count_total"] == 1 + assert payload["quarantined_records"][0]["file_path"] == str(rollout) + assert "quarantined_record" in payload["alert_reasons"] + + def test_quarantined_offset_waits_for_prior_indexable_record_confirmation(self, tmp_path, monkeypatch): + sessions = tmp_path / "codex" / "sessions" + sessions.mkdir(parents=True) + rollout = sessions / "rollout.jsonl" + first = json.dumps({"role": "user", "content": "first"}) + "\n" + malformed = b"not json at all\n" + rollout.write_bytes(first.encode() + malformed) + monkeypatch.setenv("BRAINLAYER_WATCHER_QUARANTINE_DIR", str(tmp_path / "quarantine")) + + def confirm_all(items): + return {item["_source_file"]: item["_line_end_offset"] for item in items} + + def capture_alarm(code, message, context): + raise BrainLayerAlarm(code, message, context) - def test_negative_watch_max_file_bytes_falls_back_to_default(self, tmp_path, monkeypatch, caplog): + monkeypatch.setattr("brainlayer.watcher.raise_alarm", capture_alarm) + + watcher = JSONLWatcher( + watch_roots=[WatchRoot("codex", sessions)], + registry_path=tmp_path / "offsets.json", + on_flush=confirm_all, + batch_size=10, + flush_interval_ms=360_000, + ) + + watcher.poll_once() + assert watcher.registry.get(str(rollout)) == (0, 0) + + watcher.indexer.flush() + + assert watcher.registry.get(str(rollout))[0] == rollout.stat().st_size + + def test_quarantine_write_failure_keeps_record_and_offset_unmodified(self, tmp_path, monkeypatch): + sessions = tmp_path / "codex" / "sessions" + sessions.mkdir(parents=True) + rollout = sessions / "rollout.jsonl" + rollout.write_bytes(b"not json at all\n") + invalid_quarantine_dir = tmp_path / "not-a-directory" + invalid_quarantine_dir.write_text("occupied") + monkeypatch.setenv("BRAINLAYER_WATCHER_QUARANTINE_DIR", str(invalid_quarantine_dir)) + alarms = [] + + def capture_alarm(code, message, context): + alarms.append((code, message, context)) + raise BrainLayerAlarm(code, message, context) + + monkeypatch.setattr("brainlayer.watcher.raise_alarm", capture_alarm) + watcher = JSONLWatcher( + watch_roots=[WatchRoot("codex", sessions)], + registry_path=tmp_path / "offsets.json", + on_flush=lambda _items: None, + ) + + watcher.poll_once() + + tailer = watcher._tailers[str(rollout)] + assert tailer.offset == 0 + assert tailer._buffer == b"not json at all\n" + assert watcher.registry.get(str(rollout)) == (0, 0) + assert len(alarms) == 1 + assert alarms[0][2]["error_type"] == "FileExistsError" + + def test_growing_blocked_record_emits_one_alarm(self, tmp_path, monkeypatch): + sessions = tmp_path / "codex" / "sessions" + sessions.mkdir(parents=True) + rollout = sessions / "rollout.jsonl" + rollout.write_bytes(b'{"role":"user","content":"' + b"x" * 64) + monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "16") + monkeypatch.setenv("BRAINLAYER_WATCH_MAX_RECORD_BYTES", "32") + alarms = [] + + def capture_alarm(code, message, context): + alarms.append((code, message, context)) + raise BrainLayerAlarm(code, message, context) + + monkeypatch.setattr("brainlayer.watcher.raise_alarm", capture_alarm) + watcher = JSONLWatcher( + watch_roots=[WatchRoot("codex", sessions)], + registry_path=tmp_path / "offsets.json", + on_flush=lambda _items: None, + ) + + for _ in range(4): + with rollout.open("ab") as file_handle: + file_handle.write(b"x") + watcher.poll_once() + + assert len(alarms) == 1 + assert watcher.registry.get(str(rollout)) == (0, 0) + assert len(watcher._tailers[str(rollout)]._buffer) <= 33 + + def test_health_caps_failure_details_and_reports_overflow(self, tmp_path): + health_path = tmp_path / "watcher-health.json" + watcher = JSONLWatcher( + watch_roots=[], + registry_path=tmp_path / "offsets.json", + on_flush=lambda _items: None, + health_path=health_path, + ) + watcher._file_ingestion_failures = { + f"/tmp/failure-{index}.jsonl": { + "file_path": f"/tmp/failure-{index}.jsonl", + "error": "forced", + "_fingerprint": ("OSError", "forced"), + } + for index in range(150) + } + + watcher._write_health_snapshot([]) + + payload = json.loads(health_path.read_text()) + assert payload["file_ingestion_failure_count"] == 150 + assert len(payload["file_ingestion_failures"]) == 100 + assert payload["file_ingestion_failures_overflow_count"] == 50 + + def test_negative_watch_read_window_falls_back_to_default(self, tmp_path, monkeypatch, caplog): monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "-1") watcher = JSONLWatcher( @@ -1525,10 +1911,10 @@ def test_negative_watch_max_file_bytes_falls_back_to_default(self, tmp_path, mon on_flush=lambda _items: None, ) - assert watcher.max_file_bytes == 100 * 1024 * 1024 + assert watcher.max_read_bytes_per_file == 100 * 1024 * 1024 assert any("BRAINLAYER_WATCH_MAX_FILE_BYTES='-1'" in record.getMessage() for record in caplog.records) - def test_invalid_watch_max_file_bytes_falls_back_to_default(self, tmp_path, monkeypatch, caplog): + def test_invalid_watch_read_window_falls_back_to_default(self, tmp_path, monkeypatch, caplog): monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "invalid") watcher = JSONLWatcher( @@ -1537,10 +1923,10 @@ def test_invalid_watch_max_file_bytes_falls_back_to_default(self, tmp_path, monk on_flush=lambda _items: None, ) - assert watcher.max_file_bytes == 100 * 1024 * 1024 + assert watcher.max_read_bytes_per_file == 100 * 1024 * 1024 assert any("BRAINLAYER_WATCH_MAX_FILE_BYTES='invalid'" in record.getMessage() for record in caplog.records) - def test_poll_ingests_small_append_after_oversized_checkpoint(self, tmp_path, monkeypatch): + def test_poll_ingests_append_after_large_file_is_fully_consumed(self, tmp_path, monkeypatch): sessions = tmp_path / "codex" / "sessions" sessions.mkdir(parents=True) rollout = sessions / "rollout.jsonl" @@ -1559,29 +1945,30 @@ def confirm_all(items): batch_size=1, ) - assert watcher.poll_once() == 0 + for _ in range(8): + watcher.poll_once() + if watcher.registry.get(str(rollout))[0] == rollout.stat().st_size: + break with rollout.open("a") as file_handle: file_handle.write(json.dumps({"role": "user", "content": "small append"}) + "\n") - assert watcher.poll_once() == 1 - assert [item["message"]["content"][0]["text"] for item in flushed] == ["small append"] + for _ in range(4): + watcher.poll_once() + if len(flushed) == 2: + break + assert [item["message"]["content"][0]["text"] for item in flushed] == ["x" * 256, "small append"] - def test_zero_watch_max_file_bytes_disables_checkpointing(self, tmp_path, monkeypatch): - sessions = tmp_path / "codex" / "sessions" - sessions.mkdir(parents=True) - rollout = sessions / "rollout.jsonl" - rollout.write_text(json.dumps({"role": "user", "content": "x" * 256}) + "\n") + def test_zero_watch_read_window_falls_back_to_default(self, tmp_path, monkeypatch, caplog): monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "0") watcher = JSONLWatcher( - watch_roots=[WatchRoot("codex", sessions)], + watch_roots=[], registry_path=tmp_path / "offsets.json", - on_flush=lambda items: {item["_source_file"]: item["_line_end_offset"] for item in items}, - batch_size=1, + on_flush=lambda _items: None, ) - assert watcher.poll_once() == 1 - assert watcher.registry.get(str(rollout))[0] == rollout.stat().st_size + assert watcher.max_read_bytes_per_file == 100 * 1024 * 1024 + assert any("BRAINLAYER_WATCH_MAX_FILE_BYTES='0'" in record.getMessage() for record in caplog.records) def test_codex_root_normalizes_role_content_entries(self, tmp_path): sessions = tmp_path / "codex" / "sessions" diff --git a/tests/test_watch_backfill_cli.py b/tests/test_watch_backfill_cli.py index bf5c7088..43c50a22 100644 --- a/tests/test_watch_backfill_cli.py +++ b/tests/test_watch_backfill_cli.py @@ -123,3 +123,45 @@ 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_keeps_polling_while_oversized_record_is_buffering(tmp_path, monkeypatch): + monkeypatch.delenv("BRAINLAYER_INGEST_DENYLIST", raising=False) + transcript = tmp_path / ".codex" / "sessions" / "2026" / "07" / "oversized.jsonl" + registry = tmp_path / "offsets.json" + queue_dir = tmp_path / "queue" + transcript.parent.mkdir(parents=True) + transcript.write_text( + json.dumps( + { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "x" * 512}], + "timestamp": "2026-07-31T21:00:00Z", + } + ) + + "\n" + ) + + result = CliRunner().invoke( + app, + [ + "watch-backfill", + "--home", + str(tmp_path), + "--registry", + str(registry), + "--max-cycles", + "10", + ], + env={ + "BRAINLAYER_QUEUE_DIR": str(queue_dir), + "BRAINLAYER_WATCH_MAX_FILE_BYTES": "128", + }, + ) + + assert result.exit_code == 0, result.output + assert "processed_entries=1" in result.output + assert len(list(queue_dir.glob("watcher-*.jsonl"))) == 1 + persisted = json.loads(registry.read_text()) + assert persisted[str(transcript)]["offset"] == transcript.stat().st_size