Watch
Watcher
Tails every transcript under roots, one tick() per poll.
Usage
Watcher(
roots=(CLAUDE_PROJECTS_DIR,),
*,
from_start=False,
)Holds the native tailer’s cursor between ticks — a file is re-read only when its size or mtime changed, a file first seen on the priming pass starts at end-of-file unless from_start, a shrunken file (compaction) resets its cursor, and the cursor only advances past the last complete line. Everything appended after the first tick is drained exactly once, with sidechain transcripts flagged on the event.
Example
>>> watcher = Watcher()
>>> async for item in watcher.stream():
... handle(item)Methods
| Name | Description |
|---|---|
| stream() |
Yield newly appended events, sleeping interval between polls.
|
| tick() | Run one poll step, draining the events appended since the last tick. |
stream()
Yield newly appended events, sleeping interval between polls.
Usage
stream(*, interval=1.0)Parameters
interval: float = 1.0- Seconds to sleep between poll steps.
Yields
AsyncIterator[WatchEvent]- Events in the order returned by each poll step.
tick()
Run one poll step, draining the events appended since the last tick.
Usage
tick()Returns
list[WatchEvent]- The newly appended events, files in path order, lines in file order.
WatchEvent
One transcript event freshly appended to a watched file.
Usage
WatchEvent(path, session_id, is_sidechain, event)Attributes
path: Path-
The transcript file the event was read from.
session_id: SessionId-
The session the event belongs to — from the event’s own envelope when it carries one, else the file’s last-known session, else derived from the transcript path.
is_sidechain: bool-
Whether the file is a subagent sidechain transcript.
event: TranscriptEvent- The parsed transcript event.