feat: real-time JSONL file watcher prototype (R47 Phase 1) - #104
Conversation
Python polling-based watcher for ~/.claude/projects/ JSONL files. Validates the core design before porting to Swift DispatchSource. Components: - OffsetRegistry: persists file read offsets across restarts (atomic writes) - JSONLTailer: tail-follows JSONL files, buffers partial lines, skips corrupt - BatchIndexer: accumulates parsed lines, flushes on count or interval - JSONLWatcher: discovers .jsonl files, manages tailers, coordinates flush 30 tests covering offset persistence, incremental append, inode-aware restart, corrupt line handling, threading start/stop, multi-project. Co-Authored-By: Claude Opus 4.6 (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 free trial has ended. If you'd like to continue receiving code reviews, you can add a payment method here.
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: ASSERTIVE Plan: Pro Run ID: 📒 Files selected for processing (2)
📜 Recent review details⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (4)
🧰 Additional context used📓 Path-based instructions (4)**/*.py📄 CodeRabbit inference engine (AGENTS.md)
Files:
**/*test*.py📄 CodeRabbit inference engine (CLAUDE.md)
Files:
src/**/*.py📄 CodeRabbit inference engine (CLAUDE.md)
Files:
src/brainlayer/**/*.py📄 CodeRabbit inference engine (CLAUDE.md)
Files:
🧠 Learnings (9)📚 Learning: 2026-03-25T05:42:32.869ZApplied to files:
📚 Learning: 2026-03-14T02:20:54.656ZApplied to files:
📚 Learning: 2026-03-14T02:20:54.656ZApplied to files:
📚 Learning: 2026-03-14T02:20:54.656ZApplied to files:
📚 Learning: 2026-03-14T02:20:54.656ZApplied to files:
📚 Learning: 2026-03-25T05:42:32.868ZApplied to files:
📚 Learning: 2026-03-25T05:42:32.868ZApplied to files:
📚 Learning: 2026-03-25T05:42:32.868ZApplied to files:
📚 Learning: 2026-03-25T05:42:32.868ZApplied to files:
🔇 Additional comments (13)
📝 WalkthroughWalkthroughAdds Changes
Sequence Diagram(s)sequenceDiagram
participant Main as Main Thread
participant Watcher as JSONLWatcher
participant FS as Filesystem
participant Tailer as JSONLTailer
participant Indexer as BatchIndexer
participant Registry as OffsetRegistry
participant Callback as on_flush
Main->>Watcher: start()
Watcher->>Registry: load offsets
activate Watcher
loop Poll loop
Watcher->>FS: discover .jsonl files
FS-->>Watcher: file list
alt per file
Watcher->>Registry: get(filepath)
Registry-->>Watcher: offset, inode
Watcher->>Tailer: read_new_lines(offset)
Tailer->>FS: read bytes
FS-->>Tailer: bytes (lines + partial)
Tailer-->>Watcher: parsed dicts
Watcher->>Indexer: add(dicts + _source_file)
Watcher->>Registry: set(filepath, new_offset, inode)
end
Watcher->>Indexer: tick()
alt batch ready
Indexer-->>Callback: flush(batch)
Callback->>Callback: process
end
alt registry interval elapsed
Watcher->>Registry: flush()
Registry->>FS: atomic write (tmp -> rename)
end
end
Main->>Watcher: stop()
Watcher->>Indexer: flush()
Indexer-->>Callback: final flush
Watcher->>Registry: flush()
deactivate Watcher
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ 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}
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/brainlayer/watcher.py`:
- Around line 299-305: The registry is being updated immediately after
read_new_lines() using tailer.get_inode(), which may re-stat the path and
persist an offset/inode pair that wasn't durably flushed; change the logic so
you capture the inode from the same file descriptor used by
tailer.read_new_lines() (e.g., add or use a
tailer.get_inode_from_fd()/tailer.fd_inode or return inode with
read_new_lines()), call self.indexer.add(new_lines) and only call
self.registry.set(filepath, offset, inode) after the downstream flush/commit
succeeds; ensure the registry update is moved into the flush-success path and
mark this change as a risky DB/concurrency update so callers can handle
locking/transaction semantics appropriately.
- Around line 199-209: The _do_flush method currently clears self._buffer before
calling on_flush, which drops data on any exception; change it to keep the batch
until on_flush succeeds and only clear/reassign self._buffer after successful
flush and increment of self.total_flushed. On failure, re-queue the batch back
into the buffer (or prepend it) and implement retry logic for SQLITE_BUSY within
on_flush or the flush wrapper: detect sqlite3.OperationalError with "database is
locked"/SQLITE_BUSY, retry with exponential backoff a few times (each worker
must use its own DB connection), and only give up and log after retries fail so
transient DB locks do not lose data. Ensure _last_flush is updated only on
successful flush and that logger includes the exception details when finally
failing.
- Around line 179-197: The current implementation calls self._do_flush() (which
triggers on_flush callbacks) while holding self._lock, risking deadlocks and
blocking producers; change flush points in add/tick/flush so they swap out the
pending batch under self._lock into a local variable (e.g., pop/clear
self._buffer and capture the items and update _last_flush) then release
self._lock and invoke self._do_flush() or the on_flush callback using a
dedicated write coordinator/serial executor (or existing BatchIndexer
write-serializing mechanism) to ensure only one write/flush runs at a time and
that no arbitrary callback runs while the buffer mutex is held. Ensure
_do_flush/on_flush references and updates still occur consistently (use captured
batch and update _last_flush before releasing lock) and do not call on_flush
from inside the lock.
- Around line 38-39: The constructor for Watcher stores the registry path
without expanding '~', causing '~/...' to become a literal subdirectory; update
the __init__ in class Watcher so that self.path = Path(path).expanduser() (and
optionally .resolve() if you want absolute paths) to ensure tilde expansion when
storing the path.
- Around line 110-113: Before calling f.seek(self.offset + len(self._buffer))
check the current file size via os.stat(self.filepath).st_size and detect
in-place truncation: if self.offset + len(self._buffer) is greater than st_size,
adjust state by setting self.offset = st_size (or max(0, st_size -
len(self._buffer)) if you want to preserve trailing buffer content) and clear or
trim self._buffer accordingly so the subsequent open/seek/read will not seek
past EOF; then proceed with opening the file, seeking to the corrected position,
and reading new_data.
In `@tests/test_jsonl_watcher.py`:
- Around line 201-208: Add a new test (e.g., test_flush_error_with_restart) that
reproduces the regression: use a failing on_flush function (same as bad_flush in
test_flush_error_doesnt_crash) to force a flush exception, ensure the registry
is persisted/flushed, then construct a new watcher/indexer instance (create a
new BatchIndexer or JSONLWatcher instance that reads the same registry state)
and re-add or re-read the same JSONL line so it is observed again; finally
assert that after restart the line is processed (total_flushed or processed
counter increments) and no items are permanently skipped. Reference
BatchIndexer, on_flush, registry flush/persist, and the watcher/JSONL line
re-read when implementing.
🪄 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
Run ID: c8e0b1df-2899-4fb6-a56d-7381a0168383
📒 Files selected for processing (2)
src/brainlayer/watcher.pytests/test_jsonl_watcher.py
📜 Review details
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (4)
- 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 (4)
**/*.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
**/*.py: Usepaths.py:get_db_path()for database path resolution in all scripts and CLI instead of hardcoding paths
Implement retry logic onSQLITE_BUSYerrors with each worker using its own database connection
Never delete fromchunkstable while FTS trigger is active on large datasets
Files:
tests/test_jsonl_watcher.pysrc/brainlayer/watcher.py
**/*test*.py
📄 CodeRabbit inference engine (CLAUDE.md)
Run tests with
pytest
Files:
tests/test_jsonl_watcher.py
src/**/*.py
📄 CodeRabbit inference engine (CLAUDE.md)
Run linting and formatting with
ruff check src/ && ruff format src/on all source code
Files:
src/brainlayer/watcher.py
src/brainlayer/**/*.py
📄 CodeRabbit inference engine (CLAUDE.md)
Use Typer CLI framework in
src/brainlayer/for command-line interface implementation
Files:
src/brainlayer/watcher.py
| try: | ||
| with open(self.filepath, "rb") as f: | ||
| f.seek(self.offset + len(self._buffer)) | ||
| new_data = f.read() |
There was a problem hiding this comment.
Handle in-place truncation before seeking.
If a producer truncates the file without changing the inode, self.offset + len(self._buffer) can stay past EOF forever. Line 112 will then keep seeking beyond the new end and all later appends to that file are missed.
Possible fix
try:
with open(self.filepath, "rb") as f:
+ if self.offset + len(self._buffer) > os.fstat(f.fileno()).st_size:
+ self.offset = 0
+ self._buffer = b""
f.seek(self.offset + len(self._buffer))
new_data = f.read()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/brainlayer/watcher.py` around lines 110 - 113, Before calling
f.seek(self.offset + len(self._buffer)) check the current file size via
os.stat(self.filepath).st_size and detect in-place truncation: if self.offset +
len(self._buffer) is greater than st_size, adjust state by setting self.offset =
st_size (or max(0, st_size - len(self._buffer)) if you want to preserve trailing
buffer content) and clear or trim self._buffer accordingly so the subsequent
open/seek/read will not seek past EOF; then proceed with opening the file,
seeking to the corrected position, and reading new_data.
| with self._lock: | ||
| self._buffer.extend(items) | ||
| if len(self._buffer) >= self.batch_size: | ||
| self._do_flush() | ||
|
|
||
| def tick(self): | ||
| """Check if flush interval has elapsed. Call periodically.""" | ||
| with self._lock: | ||
| if not self._buffer: | ||
| return | ||
| elapsed_ms = (time.monotonic() - self._last_flush) * 1000 | ||
| if elapsed_ms >= self.flush_interval_ms: | ||
| self._do_flush() | ||
|
|
||
| def flush(self): | ||
| """Force flush remaining items.""" | ||
| with self._lock: | ||
| if self._buffer: | ||
| self._do_flush() |
There was a problem hiding this comment.
Don't invoke on_flush while _lock is held.
Lines 179-197 can enter _do_flush() from inside the buffer mutex, and _do_flush() executes arbitrary callback code. A slow or reentrant flush path can deadlock BatchIndexer, and it blocks every producer behind the write. Swap the batch under _lock, then run the callback under a dedicated/shared write coordinator instead.
As per coding guidelines, "Flag risky DB or concurrency changes explicitly and do not hand-wave lock behavior" and "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".
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/brainlayer/watcher.py` around lines 179 - 197, The current implementation
calls self._do_flush() (which triggers on_flush callbacks) while holding
self._lock, risking deadlocks and blocking producers; change flush points in
add/tick/flush so they swap out the pending batch under self._lock into a local
variable (e.g., pop/clear self._buffer and capture the items and update
_last_flush) then release self._lock and invoke self._do_flush() or the on_flush
callback using a dedicated write coordinator/serial executor (or existing
BatchIndexer write-serializing mechanism) to ensure only one write/flush runs at
a time and that no arbitrary callback runs while the buffer mutex is held.
Ensure _do_flush/on_flush references and updates still occur consistently (use
captured batch and update _last_flush before releasing lock) and do not call
on_flush from inside the lock.
| new_lines = tailer.read_new_lines() | ||
| if new_lines: | ||
| # Tag each line with source metadata | ||
| for line in new_lines: | ||
| line.setdefault("_source_file", filepath) | ||
| self.indexer.add(new_lines) | ||
| self.registry.set(filepath, tailer.offset, tailer.get_inode()) |
There was a problem hiding this comment.
Persist offsets/inodes only after the same bytes are durably flushed.
Line 305 checkpoints tailer.offset immediately and gets the inode by re-statting the path after the read. If the batch later fails, or the file is replaced between read_new_lines() and get_inode(), the registry can store an offset/inode pair that never made it downstream and future polls/restarts will skip records. Capture the inode from the same file descriptor used for the read, and move the registry update into the flush-success path.
As per coding guidelines, "Flag risky DB or concurrency changes explicitly and do not hand-wave lock behavior".
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/brainlayer/watcher.py` around lines 299 - 305, The registry is being
updated immediately after read_new_lines() using tailer.get_inode(), which may
re-stat the path and persist an offset/inode pair that wasn't durably flushed;
change the logic so you capture the inode from the same file descriptor used by
tailer.read_new_lines() (e.g., add or use a
tailer.get_inode_from_fd()/tailer.fd_inode or return inode with
read_new_lines()), call self.indexer.add(new_lines) and only call
self.registry.set(filepath, offset, inode) after the downstream flush/commit
succeeds; ensure the registry update is moved into the flush-success path and
mark this change as a risky DB/concurrency update so callers can handle
locking/transaction semantics appropriately.
| def test_flush_error_doesnt_crash(self): | ||
| def bad_flush(items): | ||
| raise RuntimeError("flush failed") | ||
|
|
||
| indexer = BatchIndexer(on_flush=bad_flush, batch_size=1) | ||
| # Should not raise | ||
| indexer.add([{"a": 1}]) | ||
| assert indexer.total_flushed == 0 |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Add a regression that combines flush failure with restart.
These tests cover on_flush exceptions and restart semantics separately, but not the combination that guards against permanent skips. Please add a case where on_flush raises, the registry is flushed, a new watcher instance is created, and the same JSONL line must be observed again.
Also applies to: 302-331
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@tests/test_jsonl_watcher.py` around lines 201 - 208, Add a new test (e.g.,
test_flush_error_with_restart) that reproduces the regression: use a failing
on_flush function (same as bad_flush in test_flush_error_doesnt_crash) to force
a flush exception, ensure the registry is persisted/flushed, then construct a
new watcher/indexer instance (create a new BatchIndexer or JSONLWatcher instance
that reads the same registry state) and re-add or re-read the same JSONL line so
it is observed again; finally assert that after restart the line is processed
(total_flushed or processed counter increments) and no items are permanently
skipped. Reference BatchIndexer, on_flush, registry flush/persist, and the
watcher/JSONL line re-read when implementing.
- Expand ~ in registry_path (was only expanded for watch_dir) - Retain buffer on flush failure instead of dropping data - Per-directory OSError handling in file discovery - Test verifies buffer retention on flush error Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
| def _do_flush(self): | ||
| """Internal flush — must be called with _lock held.""" | ||
| batch = self._buffer |
There was a problem hiding this comment.
🟢 Low brainlayer/watcher.py:199
In _do_flush, batch = self._buffer assigns a reference to the same list object, not a copy. If on_flush mutates batch in-place (e.g., batch.clear()) then raises, the items are lost from _buffer even though the code claims they are "retaining in buffer". Consider copying the batch with batch = self._buffer.copy() before the callback.
def _do_flush(self):
"""Internal flush — must be called with _lock held."""
- batch = self._buffer
+ batch = self._buffer.copy()
self._last_flush = time.monotonic()🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file src/brainlayer/watcher.py around lines 199-201:
In `_do_flush`, `batch = self._buffer` assigns a reference to the same list object, not a copy. If `on_flush` mutates `batch` in-place (e.g., `batch.clear()`) then raises, the items are lost from `_buffer` even though the code claims they are "retaining in buffer". Consider copying the batch with `batch = self._buffer.copy()` before the callback.
Evidence trail:
src/brainlayer/watcher.py lines 199-208 (REVIEWED_COMMIT): Line 199 `batch = self._buffer` creates a reference (not a copy). Line 202 calls `self.on_flush(batch)`. Lines 206-207 show except block logging 'retaining in buffer'. Python semantics: assignment creates a reference to the same mutable list object, so in-place mutations via `batch` would affect `self._buffer`.
Wire the JSONLWatcher (PR #104) into BrainLayer's indexing pipeline: - watcher_bridge.py: processes raw JSONL entries through classify → chunk → INSERT OR IGNORE (deferred embedding, FTS5-searchable immediately) - `brainlayer watch` CLI command with --poll, --batch-size, --flush-ms - launchd/com.brainlayer.watch.plist: persistent LaunchAgent with KeepAlive - 8 integration tests: insert, noise skip, dedup, FTS5 search, full pipeline Replaces 30-min batch indexing with ~1s polling for real-time ingestion. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Wire the JSONLWatcher (PR #104) into BrainLayer's indexing pipeline: - watcher_bridge.py: processes raw JSONL entries through classify → chunk → INSERT OR IGNORE (deferred embedding, FTS5-searchable immediately) - `brainlayer watch` CLI command with --poll, --batch-size, --flush-ms - launchd/com.brainlayer.watch.plist: persistent LaunchAgent with KeepAlive - 8 integration tests: insert, noise skip, dedup, FTS5 search, full pipeline Replaces 30-min batch indexing with ~1s polling for real-time ingestion. Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1 participant
Summary
~/.local/share/brainlayer/offsets.json, survives restarts, detects file replacement via inode comparison.jsonlfiles under~/.claude/projects/, manages tailer lifecycle, coordinates batch flush and offset persistenceArchitecture
This is the Python polling prototype (R47 research). Validates the core design:
on_flushcallback for downstream indexingProduction version will use Swift DispatchSource kqueue in BrainBar for sub-1ms latency vs ~1s polling.
Test plan
pytest tests/test_jsonl_watcher.py)ruff check+ruff format)~/.claude/projects/directoryon_flushto BrainLayer's chunk insertion pipeline🤖 Generated with Claude Code
Note
Add real-time JSONL file watcher with persistent offset tracking (R47 Phase 1)
JSONLWatcherthat polls a projects directory tree for.jsonlfiles, tails new records incrementally, and dispatches them in batches via a caller-provided flush callback.OffsetRegistrypersists per-file byte offsets and inodes to disk using atomic writes, enabling resumption across restarts and detection of file replacement via inode change.JSONLTailerhandles partial writes and malformed lines gracefully, returning only well-formed dict records since the last read position.BatchIndexerbuffers records and flushes either when a size threshold is reached or a time interval elapses; on flush errors, the buffer is retained for retry.tests/test_jsonl_watcher.pycovering registry persistence, tailing, batching, and watcher polling lifecycle.Macroscope summarized 417df23.
Summary by CodeRabbit
New Features
Tests