fix(watcher): stream oversized ingestion without loss - #631
Conversation
A worker-authored 'Verdict: ACCEPT' is not acceptance (plan Phase 7; collab constraint 4). These 1,047 lines were the implementer grading its own work and must not reach a PR. Review is performed by a fresh Claude seat, not the implementer. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
There was a problem hiding this comment.
Your trial has ended. Reactivate Greptile to resume code reviews.
📝 WalkthroughWalkthroughChangesJSONL ingestion
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant JSONLWatcher
participant JSONLTailer
participant BatchIndexer
participant OffsetRegistry
JSONLWatcher->>JSONLTailer: Read a bounded byte window
JSONLTailer-->>JSONLWatcher: Return parsed entries or retained failure
JSONLWatcher->>BatchIndexer: Process the source batch
BatchIndexer-->>JSONLWatcher: Return confirmed watermarks
JSONLWatcher->>OffsetRegistry: Advance matching inode and generation
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![macroscopeapp[bot]](https://avatars.githubusercontent.com/in/900172?s=60&v=4){kind=small}
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 31127db7f4
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| new_lines = tailer.read_new_lines( | ||
| max_lines=self.max_lines_per_file, | ||
| max_bytes=self.max_read_bytes_per_file, | ||
| ) |
There was a problem hiding this comment.
When a valid JSONL record exceeds BRAINLAYER_WATCH_MAX_FILE_BYTES but remains below the record ceiling (100–128 MB with the defaults), this bounded read buffers only the first window and poll_once() returns zero normalized lines. The watch-backfill loop in src/brainlayer/cli/__init__.py:3494-3498 interprets that zero as completion and exits with the offset still at zero, so the oversized record is never indexed even though another poll would continue it. Distinguish partial read progress from an idle watcher before terminating the backfill loop.
AGENTS.md reference: AGENTS.md:L5-L7
Useful? React with 👍 / 👎.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 8
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/brainlayer/watcher.py`:
- Around line 1378-1386: Update the watchdog alerting logic near the
`alert_reasons` construction to derive quarantine activity from recent entries
in `self._quarantined_records`, using the newest `observed_at` within the
current health window rather than lifetime
`self._quarantined_record_count_total`. Use that recent-activity flag for both
the `quarantined_record` reason and the `alerting` boolean, while retaining
`quarantined_record_count_total` solely as an informational counter.
- Around line 1228-1256: Update the quarantine handling around
_quarantine_failed_record to enforce retention for existing .bad files using an
age- or size-based prune, while preserving the current write and collision
behavior. When creating BRAINLAYER_WATCHER_QUARANTINE_DIR, apply a restrictive
owner-only directory mode and ensure pruning does not remove newly written or
unrelated files.
- Around line 1636-1642: Update the last_error attribute annotation in the
tailer class to BaseException | None so it accurately accepts the arbitrary
Exception assigned in the polling error handler. Leave the existing assignment
and runtime type checks unchanged.
- Around line 700-731: Update the buffer handling in the watcher read flow
around _partial_record_bytes and the read loop to track whether f.read()
appended any bytes. Only create or write back the combined buffer when data was
read; when the first or subsequent read returns empty, retain self._buffer
unchanged and avoid both full-buffer copies.
- Around line 1006-1018: Ensure watcher offset and quarantine state has
single-thread ownership or is protected by one shared lock: update the
poll_once/CLI flush path and BatchIndexer._do_flush callback path, including
_advance_confirmed_batch, OffsetRegistry updates, and watcher dictionaries such
as _file_ingestion_failures and _pending_quarantined_offsets. Apply the same
synchronization to every read-modify-write path so concurrent callers cannot
mutate these structures unsafely.
- Around line 1216-1222: Update _quarantine_failed_record to also accept
OversizedJSONLRecordError in its tailer.last_error type check, so complete
oversized records with a failed_record are quarantined and checkpointed through
the existing poll_once flow.
In `@tests/test_jsonl_watcher.py`:
- Around line 1738-1768: Add coverage alongside
test_quarantined_offset_waits_for_prior_indexable_record_confirmation for two
pending malformed ranges separated by a valid record, asserting sorting,
retention, and consecutive advancement behavior in _advance_quarantined_offsets;
add a separate rewind scenario with a pending quarantine range, asserting
_handle_rewind clears it and does not advance the offset over truncated or
replaced bytes.
- Line 624: Update the test assertion around tailer.last_error to import
OversizedJSONLRecordError from brainlayer.watcher and use
isinstance(tailer.last_error, OversizedJSONLRecordError) instead of comparing
the runtime type name string.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: a56c1c0f-1904-4b7c-befa-b3b113ce37ec
📒 Files selected for processing (2)
src/brainlayer/watcher.pytests/test_jsonl_watcher.py
📜 Review details
⏰ Context from checks skipped due to timeout. (5)
- GitHub Check: Cursor Bugbot
- GitHub Check: Macroscope - Correctness Check
- GitHub Check: test (3.13)
- GitHub Check: test (3.11)
- GitHub Check: test (3.12)
🧰 Additional context used
📓 Path-based instructions (5)
**/*.py
📄 CodeRabbit inference engine (AGENTS.md)
**/*.py: Flag risky DB or concurrency changes explicitly and do not hand-wave lock behavior
Enforce one-write-at-a-time concurrency constraint; reads are safe but brain_digest is write-heavy and must not run in parallel with other MCP work
Run pytest before claiming behavior changed safely; current test suite has 929 tests
Files:
tests/test_jsonl_watcher.pysrc/brainlayer/watcher.py
tests/**/*.py
📄 CodeRabbit inference engine (CLAUDE.md)
Tests must not refresh the production backup heartbeat log; use
BRAINLAYER_BACKUP_LOG_PATHand set provenance topytest.
Files:
tests/test_jsonl_watcher.py
src/**/*.py
📄 CodeRabbit inference engine (CLAUDE.md)
src/**/*.py: Usepaths.py:get_db_path()for all database path resolution; do not hard-code the BrainLayer database path.
Each worker must use its own database connection and retry onSQLITE_BUSY.
For bulk database operations, stop enrichment workers, checkpoint WAL before and after, drop and recreate FTS triggers around large deletes, batch deletes in 5–10K chunks, and checkpoint every three batches.
Use the canonical BrainLayer naming:BrainLayer (זיכרון)means “memory.”
Files:
src/brainlayer/watcher.py
src/brainlayer/**/*.py
📄 CodeRabbit inference engine (CLAUDE.md)
src/brainlayer/**/*.py: Preserveai_code,stack_trace, anduser_messageverbatim; skip noise, summarize build logs, and retain only structure for directory listings.
Use AST-aware tree-sitter chunking, never split stack traces, and mask large tool output.
Keep Groq as the primary enrichment backend, Gemini as fallback, and Ollama as the offline last resort; honorBRAINLAYER_ENRICH_BACKENDandBRAINLAYER_ENRICH_RATE.
Default searches must exclude lifecycle-managed chunks;include_archived=Truemay expose history.brain_supersedemust apply its personal-data safety gate, andbrain_archivemust soft-delete with a timestamp.
Files:
src/brainlayer/watcher.py
src/brainlayer/watcher*.py
📄 CodeRabbit inference engine (CLAUDE.md)
src/brainlayer/watcher*.py: The JSONL watcher must apply filters in order: entry-type whitelist, classification, minimum chunk length, and system-reminder stripping.
Persist watcher offsets, detect file rewinds, and soft-archive chunks reverted by checkpoint restoration.
Files:
src/brainlayer/watcher.py
🪛 ast-grep (0.45.0)
tests/test_jsonl_watcher.py
[info] 596-596: use jsonify instead of json.dumps for JSON output
Context: json.dumps({"role": "user", "content": "x" * 256})
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 605-605: use jsonify instead of json.dumps for JSON output
Context: json.dumps({"role": "user", "content": "x" * 2048})
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 729-729: Do not hardcode temporary file or directory names
Context: "/tmp/source.jsonl"
Note: [CWE-377] Insecure Temporary File.
(hardcoded-tmp-file)
[info] 1285-1285: use jsonify instead of json.dumps for JSON output
Context: json.dumps({"role": "user", "content": content})
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 1449-1449: use jsonify instead of json.dumps for JSON output
Context: json.dumps({"role": "user", "content": "replacement"})
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 1463-1463: use jsonify instead of json.dumps for JSON output
Context: json.dumps({"role": "user", "content": content})
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 1491-1491: use jsonify instead of json.dumps for JSON output
Context: json.dumps({"role": "user", "content": f"new-{index}"})
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 1504-1504: use jsonify instead of json.dumps for JSON output
Context: json.dumps({"role": "user", "content": "new-append"})
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 1615-1615: use jsonify instead of json.dumps for JSON output
Context: json.dumps({"role": "user", "content": "must not be checkpointed"})
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 1651-1651: use jsonify instead of json.dumps for JSON output
Context: json.dumps({"role": "user", "content": content})
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 1694-1694: use jsonify instead of json.dumps for JSON output
Context: json.dumps({"role": "user", "content": "first"})
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 1696-1696: use jsonify instead of json.dumps for JSON output
Context: json.dumps({"role": "user", "content": "last"})
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 1741-1741: use jsonify instead of json.dumps for JSON output
Context: json.dumps({"role": "user", "content": "first"})
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 1837-1837: Do not hardcode temporary file or directory names
Context: f"/tmp/failure-{index}.jsonl"
Note: [CWE-377] Insecure Temporary File.
(hardcoded-tmp-file)
[info] 1838-1838: Do not hardcode temporary file or directory names
Context: f"/tmp/failure-{index}.jsonl"
Note: [CWE-377] Insecure Temporary File.
(hardcoded-tmp-file)
[info] 1900-1900: use jsonify instead of json.dumps for JSON output
Context: json.dumps({"role": "user", "content": "small append"})
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
src/brainlayer/watcher.py
[warning] 696-696: File path is request-/variable-derived; validate and normalize to prevent path traversal.
Context: open(self.filepath, "rb")
Note: [CWE-22] Improper Limitation of a Pathname to a Restricted Directory ('Path Traversal').
(open-filename-from-request)
🔇 Additional comments (19)
src/brainlayer/watcher.py (11)
51-59: LGTM!Also applies to: 71-71, 82-99
364-367: LGTM!
634-657: LGTM!
744-807: LGTM!
882-883: LGTM!Also applies to: 929-945
1038-1081: LGTM!
1167-1214: LGTM!Also applies to: 1257-1293
1369-1377: LGTM!Also applies to: 1405-1409
1460-1480: LGTM!Also applies to: 1494-1494
1532-1542: LGTM!Also applies to: 1556-1578, 1592-1601, 1616-1621, 1630-1635
826-836: 🗄️ Data Integrity & IntegrationNo remaining
on_confirm_offsetscallers remain.tests/test_jsonl_watcher.py (8)
582-623: LGTM!Also applies to: 625-625
727-742: LGTM!
1281-1308: LGTM!
1333-1336: LGTM!Also applies to: 1363-1386
1415-1424: LGTM!Also applies to: 1426-1457, 1459-1508
1521-1528: LGTM!Also applies to: 1539-1544
1612-1736: LGTM!Also applies to: 1770-1851
1853-1919: LGTM!
| 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) |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick win
Avoid copying the whole buffer when no new bytes arrive.
Line 702 copies self._buffer into combined, and line 731 copies it back with bytes(combined). Both copies run even when f.read() returns no data on the first iteration.
A retained partial record can reach max_record_bytes (128 MiB by default). A blocked or slow-growing record therefore triggers two full-buffer copies on every poll for that file, at the poll interval. Track whether any bytes were appended and skip the write-back when nothing changed.
♻️ Proposed fix to skip the copy on an empty read
complete_lines = self._buffer.count(b"\n")
current_record_bytes = self._partial_record_bytes()
- combined = bytearray(self._buffer)
+ combined: bytearray | None = None
while remaining_bytes is None or remaining_bytes > 0:
@@
new_data = f.read(chunk_bytes)
if not new_data:
break
+ if combined is None:
+ combined = bytearray(self._buffer)
combined.extend(new_data)
@@
- self._buffer = bytes(combined)
+ if combined is not None:
+ self._buffer = bytes(combined)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| 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) | |
| complete_lines = self._buffer.count(b"\n") | |
| current_record_bytes = self._partial_record_bytes() | |
| combined: bytearray | None = None | |
| 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 | |
| if combined is None: | |
| combined = bytearray(self._buffer) | |
| 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 | |
| if combined is not None: | |
| self._buffer = bytes(combined) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/brainlayer/watcher.py` around lines 700 - 731, Update the buffer handling
in the watcher read flow around _partial_record_bytes and the read loop to track
whether f.read() appended any bytes. Only create or write back the combined
buffer when data was read; when the first or subsequent read returns empty,
retain self._buffer unchanged and avoid both full-buffer copies.
| 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}") |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Add a retention policy for the quarantine directory.
_quarantine_failed_record writes raw transcript bytes to BRAINLAYER_WATCHER_QUARANTINE_DIR and never removes them. _quarantined_records is capped in memory at _MAX_HEALTH_QUARANTINE_DETAILS, but the .bad files on disk grow without a bound and without an expiry.
The quarantined bytes are unredacted user transcript content, and the file name embeds the session id through Path(filepath).stem. This content sits outside the lifecycle-managed store, so archive and supersede paths never reach it.
Add an age-based or size-based prune for the quarantine directory, and restrict the directory mode when it is created.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/brainlayer/watcher.py` around lines 1228 - 1256, Update the quarantine
handling around _quarantine_failed_record to enforce retention for existing .bad
files using an age- or size-based prune, while preserving the current write and
collision behavior. When creating BRAINLAYER_WATCHER_QUARANTINE_DIR, apply a
restrictive owner-only directory mode and ensure pruning does not remove newly
written or unrelated files.
Source: Coding guidelines
| 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, | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
alerting stays true forever after one quarantined record.
self._quarantined_record_count_total is a lifetime counter and never resets. Lines 1378-1384 therefore force alerting: True and add quarantined_record to alert_reasons for the whole process lifetime, even after ingestion recovers.
_file_ingestion_failures is pruned each poll and clears correctly, so quarantine is the only permanently sticky reason. Downstream alerting cannot distinguish an active fault from one historical quarantine.
Gate the alert on recent quarantine activity, for example on the newest observed_at in self._quarantined_records within the current health window, and keep quarantined_record_count_total as an informational counter.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/brainlayer/watcher.py` around lines 1378 - 1386, Update the watchdog
alerting logic near the `alert_reasons` construction to derive quarantine
activity from recent entries in `self._quarantined_records`, using the newest
`observed_at` within the current health window rather than lifetime
`self._quarantined_record_count_total`. Use that recent-activity flag for both
the `quarantined_record` reason and the `alerting` boolean, while retaining
`quarantined_record_count_total` solely as an informational counter.
| 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) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Widen the last_error annotation to match this assignment.
Line 1639 assigns an arbitrary caught Exception to tailer.last_error. Line 655 declares the attribute as OSError | json.JSONDecodeError | UnicodeDecodeError | OversizedJSONLRecordError | None. A RuntimeError from _normalize_lines therefore violates the declared type.
Runtime behavior is safe, because _quarantine_failed_record re-checks the type with isinstance. Update the annotation on line 655 to BaseException | None, so type checkers stay accurate.
♻️ Proposed annotation fix at line 655
- self.last_error: OSError | json.JSONDecodeError | UnicodeDecodeError | OversizedJSONLRecordError | None = None
+ self.last_error: BaseException | None = None🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/brainlayer/watcher.py` around lines 1636 - 1642, Update the last_error
attribute annotation in the tailer class to BaseException | None so it
accurately accepts the arbitrary Exception assigned in the polling error
handler. Leave the existing assignment and runtime type checks unchanged.
| assert tailer.read_new_lines(max_bytes=1024) == [] | ||
|
|
||
| assert len(tailer._buffer) <= 4097 | ||
| assert type(tailer.last_error).__name__ == "OversizedJSONLRecordError" |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Assert the exception type directly.
The test compares type(tailer.last_error).__name__ to a string. A rename of OversizedJSONLRecordError then leaves the test passing against the wrong type, or failing with an unclear message. Import the class and use isinstance.
♻️ Proposed assertion change
- assert type(tailer.last_error).__name__ == "OversizedJSONLRecordError"
+ assert isinstance(tailer.last_error, OversizedJSONLRecordError)Add the symbol to the existing import block near line 24:
from brainlayer.watcher import OversizedJSONLRecordError🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/test_jsonl_watcher.py` at line 624, Update the test assertion around
tailer.last_error to import OversizedJSONLRecordError from brainlayer.watcher
and use isinstance(tailer.last_error, OversizedJSONLRecordError) instead of
comparing the runtime type name string.
| 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) | ||
|
|
||
| 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 |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick win
Add coverage for multiple pending quarantine ranges and for rewind clearing.
_advance_quarantined_offsets sorts a list of pending ranges and advances over consecutive ranges. This test exercises exactly one pending range, so the sort, the remaining retention branch, and the multi-range advance stay uncovered.
_pending_quarantined_offsets.pop(filepath, None) also runs in _ensure_tailer on inode change and in _handle_rewind. No test asserts that a pending range is dropped after a rewind, which would otherwise advance an offset over bytes that no longer exist.
Add two tests: one file with two malformed records separated by a valid record, and one file where a rewind occurs while a quarantined range is still pending.
🧰 Tools
🪛 ast-grep (0.45.0)
[info] 1741-1741: use jsonify instead of json.dumps for JSON output
Context: json.dumps({"role": "user", "content": "first"})
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/test_jsonl_watcher.py` around lines 1738 - 1768, Add coverage alongside
test_quarantined_offset_waits_for_prior_indexable_record_confirmation for two
pending malformed ranges separated by a valid record, asserting sorting,
retention, and consecutive advancement behavior in _advance_quarantined_offsets;
add a separate rewind scenario with a pending quarantine range, asserting
_handle_rewind clears it and does not advance the offset over truncated or
replaced bytes.
|
VERDICT: ITERATE |
Bugbot couldn't run - usage limit reachedBugbot is counted against Cursor usage for this user or team, and this run hit a usage or spend limit. A user or team admin can review and increase usage limits in the Cursor dashboard. (requestId: serverGenReqId_cb43ad0d-16c6-4678-a3c2-03c1f2478238) |
There was a problem hiding this comment.
Your trial has ended. Reactivate Greptile to resume code reviews.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 7e61455a06
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| 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() |
There was a problem hiding this comment.
When a watched path is atomically replaced after this constructor's stat but before read_new_lines() opens it, the bytes come from the replacement while _source_inode remains the old inode. The first flush is therefore accepted under the stale generation, and the next poll detects the inode mismatch, rewinds to zero, and queues the same replacement records again. Derive the source inode with fstat from the descriptor used for reading so the bytes and generation describe the same file.
AGENTS.md reference: AGENTS.md:L5-L7
Useful? React with 👍 / 👎.
Bugbot couldn't run - usage limit reachedBugbot is counted against Cursor usage for this user or team, and this run hit a usage or spend limit. A user or team admin can review and increase usage limits in the Cursor dashboard. (requestId: serverGenReqId_4429eaec-9056-42cf-ae2d-dc0cf7f9e658) |
There was a problem hiding this comment.
Your trial has ended. Reactivate Greptile to resume code reviews.
Bugbot couldn't run - usage limit reachedBugbot is counted against Cursor usage for this user or team, and this run hit a usage or spend limit. A user or team admin can review and increase usage limits in the Cursor dashboard. (requestId: serverGenReqId_6df053ff-2476-4fb4-bb59-ccb0084d4fd7) |
There was a problem hiding this comment.
Your trial has ended. Reactivate Greptile to resume code reviews.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 16eb0ec295
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if current_offset < start_offset: | ||
| remaining.append((start_offset, end_offset, pending_inode, pending_generation)) | ||
| continue |
There was a problem hiding this comment.
When one poll reads an indexable record, a provider record that normalization intentionally drops, and then a malformed record, delayed batch confirmation checkpoints only the indexable record. The quarantine starts after the dropped record, so this branch retains it because current_offset < start_offset, but the dropped range is no longer tracked and later idle polls cannot close the gap. The live watcher therefore leaves the registry permanently behind the tailer; after restart it reprocesses the dropped record and quarantines/alarms on the same malformed record again. Persist the discarded range or retry its checkpoint when the preceding batch confirms.
AGENTS.md reference: AGENTS.md:L5-L7
Useful? React with 👍 / 👎.
1 participant
What broke and how it was found
The watcher treated 100 MB as a correctness boundary: it advanced offsets to end-of-file without parsing oversized JSONL. Investigation found 17 abandoned files totaling 5.43 GB, including three created after the prior #613 fix.
Evidence
Review
This branch was pair-reviewed by a separate agent under the taste gate: smallest correct solution, behavioural tests, and no speculative machinery. The D1 review also mutation-tested the defect: reintroducing the bad offset advance killed 10 tests.
Operational note
Enrichment and drain are currently disabled with launchctl to prevent provenance-tag decay. Re-enable deliberately after the provenance-column ownership question is settled.
Note
High Risk
Changes core ingestion checkpointing, offset registry semantics, and failure handling for multi-GB JSONL streams; mistakes could cause data loss, stuck files, or incorrect resume positions.
Overview
Fixes JSONL ingestion so large or problematic files are no longer advanced to EOF without parsing.
JSONLTailerreads in bounded windows (max_bytes/max_lines, 64 KiB chunks) with a separateBRAINLAYER_WATCH_MAX_RECORD_BYTESceiling. Corrupt or oversized records stop parsing instead of being skipped; failed bytes stay in the buffer until quarantine or retry.JSONLWatcherremoves_skip_oversized_file. Offsets advance only over confirmed bytes, tagged with_source_inodeand_source_generationso stale flush watermarks cannot move a replaced file. Malformed complete lines can be durably quarantined, alarmed, and advanced in order after prior bytes confirm. Health exposes ingestion failures and quarantine counts;watch-backfillkeeps cycling whilelast_poll_made_progressis true even when a cycle returns zero indexed entries.Reviewed by Cursor Bugbot for commit 16eb0ec. Bugbot is set up for automated code reviews on this repo. Configure here.
Summary by CodeRabbit
Bug Fixes
Monitoring
Note
Fix JSONL watcher to stream oversized records incrementally without data loss
JSONLTailer.read_new_linesnow reads in 64 KiB chunks bounded by a per-poll byte window and line limit, buffering partial records instead of skipping or checkpointing past them.JSONLTailer.read_buffered_linesstops at the first malformed or oversized record, exposes the failed bytes infailed_record, and does not advance the offset past them.JSONLWatcher._quarantine_failed_record, which persists the bytes to a content-addressed quarantine file, emits an alarm, and schedules offset advancement so polling continues.OffsetRegistry.setnow accepts a generation parameter to prevent offset regressions across file rewinds and inode changes.JSONLWatcher.poll_oncesetslast_poll_made_progressso thewatch-backfillCLI loop continues polling while large records are buffering, stopping only when both item count and progress are zero.Macroscope summarized 16eb0ec.