From abb9f39ae393507555389cda2b26e707ee0d31af Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Mon, 20 Jul 2026 15:56:59 +0300 Subject: [PATCH 1/9] fix: prevent watcher ingestion starvation Precompute live-parent evidence during offset pruning and checkpoint oversized unread JSONL suffixes so one rollout cannot monopolize the serial watcher. Co-Authored-By: Claude Fable 5 --- src/brainlayer/watcher.py | 79 +++++++++++++++-- tests/test_jsonl_watcher.py | 163 ++++++++++++++++++++++++++++++++++++ 2 files changed, 236 insertions(+), 6 deletions(-) diff --git a/src/brainlayer/watcher.py b/src/brainlayer/watcher.py index 9332fa4b..a5ba5fd4 100644 --- a/src/brainlayer/watcher.py +++ b/src/brainlayer/watcher.py @@ -24,6 +24,7 @@ import tempfile import threading import time +from collections.abc import Set as AbstractSet from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path @@ -44,6 +45,32 @@ logger = logging.getLogger(__name__) +_WATCH_MAX_FILE_BYTES_ENV = "BRAINLAYER_WATCH_MAX_FILE_BYTES" +_DEFAULT_WATCH_MAX_FILE_BYTES = 100 * 1024 * 1024 + + +def _watch_max_file_bytes() -> int: + raw_value = os.environ.get(_WATCH_MAX_FILE_BYTES_ENV, str(_DEFAULT_WATCH_MAX_FILE_BYTES)) + try: + parsed_value = int(raw_value) + except ValueError: + logger.warning( + "Invalid %s=%r; using default %d", + _WATCH_MAX_FILE_BYTES_ENV, + raw_value, + _DEFAULT_WATCH_MAX_FILE_BYTES, + ) + return _DEFAULT_WATCH_MAX_FILE_BYTES + if parsed_value < 0: + logger.warning( + "Invalid %s=%r; using default %d", + _WATCH_MAX_FILE_BYTES_ENV, + raw_value, + _DEFAULT_WATCH_MAX_FILE_BYTES, + ) + return _DEFAULT_WATCH_MAX_FILE_BYTES + return parsed_value + @dataclass(frozen=True) class WatchRoot: @@ -499,7 +526,7 @@ def _has_unavailable_symlink_ancestor(candidate: Path, root: Path) -> bool: return False @staticmethod - def _has_live_parent_evidence(candidate: Path, live_files: list[Path]) -> bool: + def _has_live_parent_evidence(candidate: Path, live_parent_dirs: AbstractSet[Path]) -> bool: """Require a live transcript in the tracked file's containing directory.""" parent = candidate.parent try: @@ -507,7 +534,7 @@ def _has_live_parent_evidence(candidate: Path, live_files: list[Path]) -> bool: return False except OSError: return False - return any(live_file == parent or live_file.is_relative_to(parent) for live_file in live_files) + return parent in live_parent_dirs @property def last_prune_complete(self) -> bool: @@ -530,14 +557,13 @@ def prune_missing_files( live_files.append(candidate) except OSError: continue + live_parent_dirs = {parent for live_file in live_files for parent in live_file.parents} root_availability: dict[Path, bool] = {} for root in active_roots: candidate_root = Path(os.path.abspath(os.path.expanduser(str(root)))) try: - root_availability[candidate_root] = candidate_root.is_dir() and any( - live_file == candidate_root or live_file.is_relative_to(candidate_root) for live_file in live_files - ) + root_availability[candidate_root] = candidate_root.is_dir() and candidate_root in live_parent_dirs except OSError: root_availability[candidate_root] = False @@ -555,7 +581,7 @@ def prune_missing_files( if any(self._has_unavailable_symlink_ancestor(candidate, root) for root in most_specific_roots): self._last_prune_complete = False continue - if not self._has_live_parent_evidence(candidate, live_files): + if not self._has_live_parent_evidence(candidate, live_parent_dirs): self._last_prune_complete = False continue try: @@ -853,8 +879,10 @@ def __init__( self.poll_interval_s = poll_interval_s self.registry_flush_interval_s = registry_flush_interval_s self.max_lines_per_file = max(1, max_lines_per_file) + self.max_file_bytes = _watch_max_file_bytes() self._tailers: dict[str, JSONLTailer] = {} self._file_providers: dict[str, str] = {} + self._oversized_files: set[str] = set() self._stop = threading.Event() self._last_registry_flush = time.monotonic() self.health_path = Path(health_path).expanduser() if health_path else None @@ -1098,6 +1126,43 @@ def _ensure_tailer(self, filepath: str) -> JSONLTailer: self._tailers[filepath] = tailer return tailer + 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) + offset = tailer.offset if tailer else self.registry.get(filepath)[0] + 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) + 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 @@ -1128,6 +1193,8 @@ def poll_once(self) -> int: self.registry.remove(filepath) continue try: + if self._skip_oversized_file(filepath): + continue tailer = self._ensure_tailer(filepath) new_lines = tailer.read_new_lines(max_lines=self.max_lines_per_file) diff --git a/tests/test_jsonl_watcher.py b/tests/test_jsonl_watcher.py index 6d6be3f1..1742cb2c 100644 --- a/tests/test_jsonl_watcher.py +++ b/tests/test_jsonl_watcher.py @@ -136,6 +136,31 @@ def test_prune_missing_files_removes_only_deleted_paths_and_persists(self, tmp_p assert reloaded.get(str(existing)) == (100, 1) assert reloaded.get(str(deleted)) == (0, 0) + def test_prune_live_parent_evidence_uses_linear_ancestry_checks(self, monkeypatch, tmp_path): + tracked_count = 80 + registry = OffsetRegistry(tmp_path / "offsets.json") + live_files = [] + for index in range(tracked_count): + session_dir = tmp_path / f"session-{index}" + session_dir.mkdir() + live_file = session_dir / "live.jsonl" + live_file.write_text('{"id":"live"}\n') + live_files.append(live_file) + registry.set(str(session_dir / "deleted.jsonl"), 100, index + 1) + + real_is_relative_to = Path.is_relative_to + ancestry_checks = 0 + + def counting_is_relative_to(path, other): + nonlocal ancestry_checks + ancestry_checks += 1 + return real_is_relative_to(path, other) + + monkeypatch.setattr(Path, "is_relative_to", counting_is_relative_to) + + assert registry.prune_missing_files([tmp_path], live_files) == tracked_count + assert ancestry_checks <= tracked_count * 6 + def test_prune_flush_preserves_newer_offsets_from_concurrent_registry(self, tmp_path): registry_path = tmp_path / "offsets.json" existing = tmp_path / "existing.jsonl" @@ -1175,6 +1200,144 @@ def test_poll_once_limits_each_file_so_active_roots_do_not_starve(self, tmp_path assert watcher.poll_once() == 1 assert [item["_provider"] for item in flushed] == ["codex"] + def test_poll_skips_oversized_pending_file_with_warning_and_continues( + self, + tmp_path, + monkeypatch, + caplog, + ): + sessions = tmp_path / "codex" / "sessions" + sessions.mkdir(parents=True) + oversized = sessions / "oversized.jsonl" + healthy = sessions / "healthy.jsonl" + oversized.write_text(json.dumps({"role": "user", "content": "x" * 256}) + "\n") + healthy.write_text(json.dumps({"role": "user", "content": "healthy"}) + "\n") + os.utime(oversized, (2000, 2000)) + os.utime(healthy, (1000, 1000)) + 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, + ) + + 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 + ) + + def test_poll_persists_oversized_checkpoint_immediately(self, tmp_path, monkeypatch): + sessions = tmp_path / "codex" / "sessions" + sessions.mkdir(parents=True) + oversized = sessions / "oversized.jsonl" + oversized.write_text(json.dumps({"role": "user", "content": "x" * 256}) + "\n") + registry_path = tmp_path / "offsets.json" + monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "128") + + watcher = JSONLWatcher( + watch_roots=[WatchRoot("codex", sessions)], + registry_path=registry_path, + on_flush=lambda _items: None, + registry_flush_interval_s=3600, + ) + + assert watcher.poll_once() == 0 + assert OffsetRegistry(registry_path).get(str(oversized)) == ( + oversized.stat().st_size, + oversized.stat().st_ino, + ) + + def test_negative_watch_max_file_bytes_falls_back_to_default(self, tmp_path, monkeypatch, caplog): + monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "-1") + + watcher = JSONLWatcher( + watch_roots=[], + registry_path=tmp_path / "offsets.json", + on_flush=lambda _items: None, + ) + + assert watcher.max_file_bytes == 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): + monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "invalid") + + watcher = JSONLWatcher( + watch_roots=[], + registry_path=tmp_path / "offsets.json", + on_flush=lambda _items: None, + ) + + assert watcher.max_file_bytes == 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): + sessions = tmp_path / "codex" / "sessions" + sessions.mkdir(parents=True) + rollout = sessions / "rollout.jsonl" + rollout.write_text(json.dumps({"role": "user", "content": "x" * 256}) + "\n") + 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, + ) + + assert watcher.poll_once() == 0 + 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"] + + 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") + monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "0") + + watcher = JSONLWatcher( + watch_roots=[WatchRoot("codex", sessions)], + registry_path=tmp_path / "offsets.json", + on_flush=lambda items: { + item["_source_file"]: item["_line_end_offset"] for item in items + }, + batch_size=1, + ) + + assert watcher.poll_once() == 1 + assert watcher.registry.get(str(rollout))[0] == rollout.stat().st_size + def test_codex_root_normalizes_role_content_entries(self, tmp_path): sessions = tmp_path / "codex" / "sessions" sessions.mkdir(parents=True) From bf7e85e074eeb35496a057189dd642cb7ad7d46f Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Mon, 20 Jul 2026 16:28:08 +0300 Subject: [PATCH 2/9] fix: cap replaced watcher files from byte zero --- src/brainlayer/watcher.py | 8 +++--- tests/test_jsonl_watcher.py | 49 ++++++++++++++++++++++++++++--------- 2 files changed, 43 insertions(+), 14 deletions(-) diff --git a/src/brainlayer/watcher.py b/src/brainlayer/watcher.py index a5ba5fd4..8d4b8e47 100644 --- a/src/brainlayer/watcher.py +++ b/src/brainlayer/watcher.py @@ -1137,7 +1137,10 @@ def _skip_oversized_file(self, filepath: str) -> bool: return False tailer = self._tailers.get(filepath) - offset = tailer.offset if tailer else self.registry.get(filepath)[0] + registry_offset, registry_inode = self.registry.get(filepath) + offset = tailer.offset if tailer else registry_offset + if registry_inode != 0 and registry_inode != file_stat.st_ino: + offset = 0 pending_bytes = max(file_stat.st_size - offset, 0) if pending_bytes <= self.max_file_bytes: self._oversized_files.discard(filepath) @@ -1152,8 +1155,7 @@ def _skip_oversized_file(self, filepath: str) -> bool: ) 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", + "Oversized JSONL checkpointed and skipped: %s pending_bytes=%d max_file_bytes=%d offset=%d size=%d", filepath, pending_bytes, self.max_file_bytes, diff --git a/tests/test_jsonl_watcher.py b/tests/test_jsonl_watcher.py index 1742cb2c..778c6f4e 100644 --- a/tests/test_jsonl_watcher.py +++ b/tests/test_jsonl_watcher.py @@ -1263,6 +1263,41 @@ def test_poll_persists_oversized_checkpoint_immediately(self, tmp_path, monkeypa oversized.stat().st_ino, ) + def test_poll_caps_oversized_replacement_from_start(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" * 512}) + "\n") + original_size = rollout.stat().st_size + original_inode = rollout.stat().st_ino + 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.registry.set(str(rollout), original_size, original_inode) + watcher._tailers[str(rollout)] = JSONLTailer(str(rollout), offset=original_size) + + replacement = sessions / "replacement.tmp" + replacement.write_text(json.dumps({"role": "user", "content": "y" * 256}) + "\n") + os.replace(replacement, rollout) + assert rollout.stat().st_ino != original_inode + + assert watcher.poll_once() == 0 + assert flushed == [] + assert watcher.registry.get(str(rollout)) == ( + rollout.stat().st_size, + rollout.stat().st_ino, + ) + def test_negative_watch_max_file_bytes_falls_back_to_default(self, tmp_path, monkeypatch, caplog): monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "-1") @@ -1273,10 +1308,7 @@ def test_negative_watch_max_file_bytes_falls_back_to_default(self, tmp_path, mon ) assert watcher.max_file_bytes == 100 * 1024 * 1024 - assert any( - "BRAINLAYER_WATCH_MAX_FILE_BYTES='-1'" in record.getMessage() - for record in caplog.records - ) + 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): monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "invalid") @@ -1288,10 +1320,7 @@ def test_invalid_watch_max_file_bytes_falls_back_to_default(self, tmp_path, monk ) assert watcher.max_file_bytes == 100 * 1024 * 1024 - assert any( - "BRAINLAYER_WATCH_MAX_FILE_BYTES='invalid'" in record.getMessage() - for record in caplog.records - ) + 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): sessions = tmp_path / "codex" / "sessions" @@ -1329,9 +1358,7 @@ def test_zero_watch_max_file_bytes_disables_checkpointing(self, tmp_path, monkey watcher = JSONLWatcher( watch_roots=[WatchRoot("codex", sessions)], registry_path=tmp_path / "offsets.json", - on_flush=lambda items: { - item["_source_file"]: item["_line_end_offset"] for item in items - }, + on_flush=lambda items: {item["_source_file"]: item["_line_end_offset"] for item in items}, batch_size=1, ) From 704f598f202ebfe8b7636ad43caf3edd4402e121 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Mon, 20 Jul 2026 16:38:36 +0300 Subject: [PATCH 3/9] fix: reconcile oversized watcher paths --- src/brainlayer/watcher.py | 1 + tests/test_jsonl_watcher.py | 27 +++++++++++++++++++++++++++ 2 files changed, 28 insertions(+) diff --git a/src/brainlayer/watcher.py b/src/brainlayer/watcher.py index 8d4b8e47..798aa802 100644 --- a/src/brainlayer/watcher.py +++ b/src/brainlayer/watcher.py @@ -1173,6 +1173,7 @@ def poll_once(self) -> int: try: files = self._discover_jsonl_files() + self._oversized_files.intersection_update(filepath for filepath in files if not is_denylisted(filepath)) if not self._offset_prune_complete: pruned = self.registry.prune_missing_files( [root.resolved_path for root in self.watch_roots], diff --git a/tests/test_jsonl_watcher.py b/tests/test_jsonl_watcher.py index 778c6f4e..ab88ea4e 100644 --- a/tests/test_jsonl_watcher.py +++ b/tests/test_jsonl_watcher.py @@ -1298,6 +1298,33 @@ def confirm_all(items): rollout.stat().st_ino, ) + def test_poll_forgets_oversized_files_that_disappear_or_become_denylisted(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") + + watcher = JSONLWatcher( + watch_roots=[WatchRoot("codex", sessions)], + registry_path=tmp_path / "offsets.json", + on_flush=lambda _items: None, + ) + + assert watcher.poll_once() == 0 + assert watcher._oversized_files == {str(disappeared), str(denylisted)} + + disappeared.unlink() + monkeypatch.setattr( + "brainlayer.watcher.is_denylisted", + lambda filepath: filepath == str(denylisted), + ) + + assert watcher.poll_once() == 0 + assert watcher._oversized_files == set() + def test_negative_watch_max_file_bytes_falls_back_to_default(self, tmp_path, monkeypatch, caplog): monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "-1") From 794d5ce6d6856178088661cbfec7fa6487a1cbc3 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Mon, 20 Jul 2026 16:51:00 +0300 Subject: [PATCH 4/9] fix: preserve pending watcher entries at cap --- src/brainlayer/watcher.py | 21 +++++++++++++- tests/test_jsonl_watcher.py | 57 +++++++++++++++++++++++++++++++++++++ 2 files changed, 77 insertions(+), 1 deletion(-) diff --git a/src/brainlayer/watcher.py b/src/brainlayer/watcher.py index 798aa802..9fa3ca14 100644 --- a/src/brainlayer/watcher.py +++ b/src/brainlayer/watcher.py @@ -744,6 +744,11 @@ def flush(self): if self._buffer: self._do_flush() + def has_buffered_source(self, filepath: str) -> bool: + """Return whether unconfirmed buffered entries came from filepath.""" + with self._lock: + return any(item.get("_source_file") == filepath for item in self._buffer) + def _do_flush(self): """Internal flush — must be called with _lock held.""" batch = self._buffer @@ -1139,14 +1144,28 @@ def _skip_oversized_file(self, filepath: str) -> bool: tailer = self._tailers.get(filepath) registry_offset, registry_inode = self.registry.get(filepath) offset = tailer.offset if tailer else registry_offset - if registry_inode != 0 and registry_inode != file_stat.st_ino: + file_rewound = file_stat.st_size < offset + if (registry_inode != 0 and registry_inode != file_stat.st_ino) or file_rewound: offset = 0 pending_bytes = max(file_stat.st_size - offset, 0) if pending_bytes <= self.max_file_bytes: self._oversized_files.discard(filepath) return False + if self.indexer.has_buffered_source(filepath): + self.indexer.flush() + if self.indexer.has_buffered_source(filepath): + 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 + self._tailers.pop(filepath, None) + if file_rewound: + self.registry.mark_rewind(filepath, file_stat.st_ino) self.registry.set(filepath, file_stat.st_size, file_stat.st_ino) if not self.registry.flush(): logger.error( diff --git a/tests/test_jsonl_watcher.py b/tests/test_jsonl_watcher.py index ab88ea4e..d54fe0f4 100644 --- a/tests/test_jsonl_watcher.py +++ b/tests/test_jsonl_watcher.py @@ -1298,6 +1298,63 @@ def confirm_all(items): rollout.stat().st_ino, ) + def test_poll_caps_oversized_same_inode_rewind_from_start(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" * 512}) + "\n") + original_size = rollout.stat().st_size + original_inode = rollout.stat().st_ino + monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "128") + flushed = [] + + watcher = JSONLWatcher( + watch_roots=[WatchRoot("codex", sessions)], + registry_path=tmp_path / "offsets.json", + on_flush=lambda items: flushed.extend(items), + batch_size=1, + ) + watcher.registry.set(str(rollout), original_size, original_inode) + watcher._tailers[str(rollout)] = JSONLTailer(str(rollout), offset=original_size) + + with rollout.open("w") as file_handle: + 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 == [] + assert watcher.registry.get(str(rollout)) == ( + rollout.stat().st_size, + rollout.stat().st_ino, + ) + + def test_poll_does_not_checkpoint_past_retained_unconfirmed_entries(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": "pending"}) + "\n") + monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "128") + + def fail_flush(_items): + raise RuntimeError("write unavailable") + + watcher = JSONLWatcher( + watch_roots=[WatchRoot("codex", sessions)], + registry_path=tmp_path / "offsets.json", + on_flush=fail_flush, + batch_size=10, + flush_interval_ms=360000, + ) + + assert watcher.poll_once() == 1 + assert watcher.registry.get(str(rollout)) == (0, 0) + with rollout.open("a") as file_handle: + file_handle.write(json.dumps({"role": "user", "content": "x" * 256}) + "\n") + + assert watcher.poll_once() == 0 + assert watcher.registry.get(str(rollout)) == (0, 0) + assert len(watcher.indexer._buffer) == 1 + def test_poll_forgets_oversized_files_that_disappear_or_become_denylisted(self, tmp_path, monkeypatch): sessions = tmp_path / "codex" / "sessions" sessions.mkdir(parents=True) From 3c5f70a5bbe5417e7f812241a8aa9c74f9c8cfb8 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Mon, 20 Jul 2026 17:05:33 +0300 Subject: [PATCH 5/9] fix: archive oversized watcher rewinds --- src/brainlayer/watcher.py | 82 ++++++++++++++++++++++--------------- tests/test_jsonl_watcher.py | 3 ++ 2 files changed, 51 insertions(+), 34 deletions(-) diff --git a/src/brainlayer/watcher.py b/src/brainlayer/watcher.py index 9fa3ca14..836f1fc1 100644 --- a/src/brainlayer/watcher.py +++ b/src/brainlayer/watcher.py @@ -1131,6 +1131,44 @@ def _ensure_tailer(self, filepath: str) -> JSONLTailer: self._tailers[filepath] = tailer return tailer + def _handle_rewind( + self, + filepath: str, + old_offset: int, + new_offset: int, + inode: int, + ) -> None: + """Persist a rewind and notify archival consumers.""" + session_id = Path(filepath).stem + self.registry.mark_rewind(filepath, inode) + logger.warning( + "Checkpoint restore: %s (offset %d → %d)", + session_id, + old_offset, + new_offset, + ) + try: + from .telemetry import emit + + emit( + "brainlayer-watcher", + { + "_type": "rewind_detected", + "session_id": session_id, + "file_path": filepath, + "old_offset": old_offset, + "new_offset": new_offset, + }, + ) + except Exception: + pass + + if self.on_rewind: + try: + self.on_rewind(filepath, session_id, old_offset, new_offset) + 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) @@ -1144,6 +1182,7 @@ def _skip_oversized_file(self, filepath: str) -> bool: tailer = self._tailers.get(filepath) registry_offset, registry_inode = self.registry.get(filepath) offset = tailer.offset if tailer else registry_offset + rewind_old_offset = offset file_rewound = file_stat.st_size < offset if (registry_inode != 0 and registry_inode != file_stat.st_ino) or file_rewound: offset = 0 @@ -1165,7 +1204,12 @@ def _skip_oversized_file(self, filepath: str) -> bool: self._tailers.pop(filepath, None) if file_rewound: - self.registry.mark_rewind(filepath, file_stat.st_ino) + 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( @@ -1222,42 +1266,12 @@ def poll_once(self) -> int: # Handle rewind detection (checkpoint restore) if tailer.rewound: - session_id = Path(filepath).stem - self.registry.mark_rewind(filepath, tailer.get_inode()) - logger.warning( - "Checkpoint restore: %s (offset %d → %d)", - session_id, + self._handle_rewind( + filepath, tailer.rewind_old_offset, tailer.rewind_new_offset, + tailer.get_inode(), ) - try: - from .telemetry import emit - - emit( - "brainlayer-watcher", - { - "_type": "rewind_detected", - "session_id": session_id, - "file_path": filepath, - "old_offset": tailer.rewind_old_offset, - "new_offset": tailer.rewind_new_offset, - }, - ) - except Exception: - pass - - # Call rewind callback if set - if self.on_rewind: - try: - self.on_rewind( - filepath, - session_id, - tailer.rewind_old_offset, - tailer.rewind_new_offset, - ) - except Exception as e: - logger.error("Rewind callback failed: %s", e) - tailer.rewound = False # Reset flag if new_lines: diff --git a/tests/test_jsonl_watcher.py b/tests/test_jsonl_watcher.py index d54fe0f4..220dfa41 100644 --- a/tests/test_jsonl_watcher.py +++ b/tests/test_jsonl_watcher.py @@ -1307,11 +1307,13 @@ def test_poll_caps_oversized_same_inode_rewind_from_start(self, tmp_path, monkey original_inode = rollout.stat().st_ino monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "128") flushed = [] + rewinds = [] watcher = JSONLWatcher( watch_roots=[WatchRoot("codex", sessions)], registry_path=tmp_path / "offsets.json", on_flush=lambda items: flushed.extend(items), + on_rewind=lambda *args: rewinds.append(args), batch_size=1, ) watcher.registry.set(str(rollout), original_size, original_inode) @@ -1327,6 +1329,7 @@ def test_poll_caps_oversized_same_inode_rewind_from_start(self, tmp_path, monkey rollout.stat().st_size, rollout.stat().st_ino, ) + assert rewinds == [(str(rollout), "rollout", original_size, rollout.stat().st_size)] def test_poll_does_not_checkpoint_past_retained_unconfirmed_entries(self, tmp_path, monkeypatch): sessions = tmp_path / "codex" / "sessions" From 2001b074089715856eded7b4dee7efb6e78131b0 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Mon, 20 Jul 2026 17:20:37 +0300 Subject: [PATCH 6/9] fix: anchor watcher cap to confirmed offsets --- src/brainlayer/watcher.py | 21 ++++++++++++++------- tests/test_jsonl_watcher.py | 34 ++++++++++++++++++++++++++++++++++ 2 files changed, 48 insertions(+), 7 deletions(-) diff --git a/src/brainlayer/watcher.py b/src/brainlayer/watcher.py index 836f1fc1..a55e9b06 100644 --- a/src/brainlayer/watcher.py +++ b/src/brainlayer/watcher.py @@ -1181,19 +1181,21 @@ def _skip_oversized_file(self, filepath: str) -> bool: tailer = self._tailers.get(filepath) registry_offset, registry_inode = self.registry.get(filepath) - offset = tailer.offset if tailer else registry_offset - rewind_old_offset = offset - file_rewound = file_stat.st_size < offset - if (registry_inode != 0 and registry_inode != file_stat.st_ino) or file_rewound: - offset = 0 + 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 self.indexer.has_buffered_source(filepath): - self.indexer.flush() + 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", @@ -1201,6 +1203,11 @@ def _skip_oversized_file(self, filepath: str) -> bool: ) 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: diff --git a/tests/test_jsonl_watcher.py b/tests/test_jsonl_watcher.py index 220dfa41..19175eac 100644 --- a/tests/test_jsonl_watcher.py +++ b/tests/test_jsonl_watcher.py @@ -1358,6 +1358,40 @@ def fail_flush(_items): assert watcher.registry.get(str(rollout)) == (0, 0) assert len(watcher.indexer._buffer) == 1 + def test_poll_does_not_checkpoint_past_partially_confirmed_watermark(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": "first"}) + + "\n" + + json.dumps({"role": "user", "content": "second"}) + + "\n" + ) + monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "128") + + def confirm_first_only(items): + return {str(rollout): items[0]["_line_end_offset"]} + + watcher = JSONLWatcher( + watch_roots=[WatchRoot("codex", sessions)], + registry_path=tmp_path / "offsets.json", + on_flush=confirm_first_only, + batch_size=2, + flush_interval_ms=360000, + ) + + assert watcher.poll_once() == 2 + confirmed_offset, confirmed_inode = watcher.registry.get(str(rollout)) + assert confirmed_offset < watcher._tailers[str(rollout)].offset + assert watcher.indexer._buffer == [] + + with rollout.open("a") as file_handle: + file_handle.write(json.dumps({"role": "user", "content": "x" * 256}) + "\n") + + 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): sessions = tmp_path / "codex" / "sessions" sessions.mkdir(parents=True) From 7890d188f2b4ff1ebba63826d0e363359a6e00e3 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Mon, 20 Jul 2026 17:37:01 +0300 Subject: [PATCH 7/9] fix: drain buffered watcher lines before cap --- src/brainlayer/watcher.py | 35 ++++++++++++++++++++++++++---- tests/test_jsonl_watcher.py | 43 +++++++++++++++++++++++++++++++++++++ 2 files changed, 74 insertions(+), 4 deletions(-) diff --git a/src/brainlayer/watcher.py b/src/brainlayer/watcher.py index a55e9b06..c2ed2848 100644 --- a/src/brainlayer/watcher.py +++ b/src/brainlayer/watcher.py @@ -658,6 +658,14 @@ def read_new_lines(self, max_lines: int | None = None) -> list[dict]: if new_data: self._buffer += new_data + return self.read_buffered_lines(max_lines=max_lines) + + def has_complete_buffered_line(self) -> bool: + """Return whether an already-read complete record is waiting to be emitted.""" + return b"\n" in self._buffer + + def read_buffered_lines(self, max_lines: int | None = None) -> list[dict]: + """Parse complete buffered records without reading more bytes from disk.""" lines = [] while b"\n" in self._buffer: @@ -1266,10 +1274,29 @@ def poll_once(self) -> int: self.registry.remove(filepath) continue try: - if self._skip_oversized_file(filepath): - continue - tailer = self._ensure_tailer(filepath) - new_lines = tailer.read_new_lines(max_lines=self.max_lines_per_file) + 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() + if not inode_changed and not tailer.check_rewind(): + drain_buffer = True + elif tailer.rewound: + self._handle_rewind( + filepath, + tailer.rewind_old_offset, + tailer.rewind_new_offset, + tailer.get_inode(), + ) + tailer.rewound = False + + if drain_buffer: + 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) + new_lines = tailer.read_new_lines(max_lines=self.max_lines_per_file) # Handle rewind detection (checkpoint restore) if tailer.rewound: diff --git a/tests/test_jsonl_watcher.py b/tests/test_jsonl_watcher.py index 19175eac..31d392d8 100644 --- a/tests/test_jsonl_watcher.py +++ b/tests/test_jsonl_watcher.py @@ -1200,6 +1200,49 @@ def test_poll_once_limits_each_file_so_active_roots_do_not_starve(self, tmp_path assert watcher.poll_once() == 1 assert [item["_provider"] for item in flushed] == ["codex"] + def test_poll_drains_buffered_lines_before_checkpointing_oversized_append(self, tmp_path, monkeypatch): + sessions = tmp_path / "codex" / "sessions" + sessions.mkdir(parents=True) + rollout = sessions / "rollout.jsonl" + encoded_lines = [ + (json.dumps({"role": "user", "content": f"buffered line {idx} with enough content"}) + "\n").encode() + for idx in range(3) + ] + rollout.write_bytes(b"".join(encoded_lines)) + monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "256") + 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, + ) + + assert watcher.poll_once() == 1 + tailer = watcher._tailers[str(rollout)] + assert tailer.offset == len(encoded_lines[0]) + assert tailer._buffer == b"".join(encoded_lines[1:]) + + with rollout.open("ab") as file_handle: + file_handle.write(json.dumps({"role": "user", "content": "x" * 512}).encode() + b"\n") + oversized_size = rollout.stat().st_size + + assert watcher.poll_once() == 1 + assert [item["message"]["content"][0]["text"] for item in flushed] == [ + "buffered line 0 with enough content", + "buffered line 1 with enough content", + ] + assert tailer.offset == len(encoded_lines[0]) + len(encoded_lines[1]) + assert tailer._buffer == encoded_lines[2] + assert watcher.registry.get(str(rollout)) == (tailer.offset, rollout.stat().st_ino) + assert tailer.offset < oversized_size + def test_poll_skips_oversized_pending_file_with_warning_and_continues( self, tmp_path, From 9be818ad6524248818277f5a25d15009d8b696c2 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Mon, 20 Jul 2026 18:00:55 +0300 Subject: [PATCH 8/9] fix: checkpoint discarded watcher records --- src/brainlayer/watcher.py | 38 ++++++++++++++++++++++++-- tests/test_jsonl_watcher.py | 54 +++++++++++++++++++++++++++++++++++++ 2 files changed, 90 insertions(+), 2 deletions(-) diff --git a/src/brainlayer/watcher.py b/src/brainlayer/watcher.py index c2ed2848..838f106a 100644 --- a/src/brainlayer/watcher.py +++ b/src/brainlayer/watcher.py @@ -979,6 +979,31 @@ def _normalize_lines(self, filepath: str, new_lines: list[dict]) -> list[dict]: normalized.append(entry) return normalized + def _checkpoint_discarded_progress( + self, + filepath: str, + read_start_offset: int, + read_end_offset: int, + normalized_lines: list[dict], + ) -> None: + """Confirm intentionally discarded bytes without crossing indexable work.""" + required_confirmed_offset = max( + ( + line["_line_end_offset"] + for line in normalized_lines + if isinstance(line.get("_line_end_offset"), int) + ), + default=read_start_offset, + ) + if read_end_offset <= required_confirmed_offset: + return + if self.indexer.has_buffered_source(filepath): + return + confirmed_offset, _confirmed_inode = self.registry.get(filepath) + if confirmed_offset < required_confirmed_offset: + return + self._advance_confirmed_offsets({filepath: read_end_offset}) + def _max_offset_lag_bytes(self, files: list[str]) -> int: max_lag = 0 for filepath in files: @@ -1291,15 +1316,18 @@ def poll_once(self) -> int: tailer.rewound = False if drain_buffer: + read_start_offset = tailer.offset 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) # Handle rewind detection (checkpoint restore) if tailer.rewound: + read_start_offset = 0 self._handle_rewind( filepath, tailer.rewind_old_offset, @@ -1308,11 +1336,17 @@ def poll_once(self) -> int: ) tailer.rewound = False # Reset flag - if new_lines: - normalized_lines = self._normalize_lines(filepath, new_lines) + normalized_lines = self._normalize_lines(filepath, new_lines) if new_lines else [] + if normalized_lines: self.indexer.add(normalized_lines) self._health_entries_seen += len(normalized_lines) total_new += len(normalized_lines) + self._checkpoint_discarded_progress( + filepath, + read_start_offset, + tailer.offset, + normalized_lines, + ) except Exception: logger.exception("Poll file error: %s", filepath) diff --git a/tests/test_jsonl_watcher.py b/tests/test_jsonl_watcher.py index 31d392d8..bb595524 100644 --- a/tests/test_jsonl_watcher.py +++ b/tests/test_jsonl_watcher.py @@ -1243,6 +1243,60 @@ 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): + 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 + monkeypatch.setenv("BRAINLAYER_WATCH_MAX_FILE_BYTES", "128") + flushed = [] + + watcher = JSONLWatcher( + watch_roots=[WatchRoot("codex", sessions)], + registry_path=tmp_path / "offsets.json", + on_flush=lambda items: flushed.extend(items), + 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 + + 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" + + def test_poll_does_not_checkpoint_dropped_tail_past_unconfirmed_record(self, tmp_path): + sessions = tmp_path / "codex" / "sessions" + sessions.mkdir(parents=True) + rollout = sessions / "rollout.jsonl" + rollout.write_text( + json.dumps({"role": "user", "content": "indexable record"}) + + "\n" + + json.dumps({"type": "response_item", "payload": {"type": "function_call"}}) + + "\n" + ) + + watcher = JSONLWatcher( + watch_roots=[WatchRoot("codex", sessions)], + registry_path=tmp_path / "offsets.json", + on_flush=lambda _items: None, + batch_size=10, + flush_interval_ms=360000, + ) + + assert watcher.poll_once() == 1 + 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( self, tmp_path, From a4ed29d8debe52f1b2b10672ed5648b973ac7575 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Mon, 20 Jul 2026 18:09:54 +0300 Subject: [PATCH 9/9] style: format watcher watermark helper --- src/brainlayer/watcher.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/src/brainlayer/watcher.py b/src/brainlayer/watcher.py index 838f106a..dbeed8ef 100644 --- a/src/brainlayer/watcher.py +++ b/src/brainlayer/watcher.py @@ -988,11 +988,7 @@ def _checkpoint_discarded_progress( ) -> None: """Confirm intentionally discarded bytes without crossing indexable work.""" required_confirmed_offset = max( - ( - line["_line_end_offset"] - for line in normalized_lines - if isinstance(line.get("_line_end_offset"), int) - ), + (line["_line_end_offset"] for line in normalized_lines if isinstance(line.get("_line_end_offset"), int)), default=read_start_offset, ) if read_end_offset <= required_confirmed_offset: