From 4a960b7fd9c68979cab6deb434ab3e1ee5e88d2e Mon Sep 17 00:00:00 2001 From: Shashank Shekhar Singh Date: Sat, 26 Sep 2026 14:37:33 +0530 Subject: [PATCH] server: the live stream's cursor describes what it read, not the file `frames()` polls the trace's size, rebuilds a snapshot when it changes, and stores that snapshot's `size` as "everything up to here has been sent": last_size = snapshot.size if snapshot.size else size `build_snapshot` took that number from `path.stat()` *after* reading the events, and the two can disagree in the direction that matters. The read stops at the last newline, and -- this being a live view -- the run is appending while it happens, so bytes can land between the read and the stat. The cursor then claims events the snapshot never saw, the next poll sees an unchanged file size and rebuilds nothing, and if the writer has stopped the file never changes size again. A finished run streams as a running one for as long as the page is open. Demonstrated against the old code rather than argued: with one append landing in that window, the cursor came back equal to the file's final size while `done` was still False -- the exact pair that leaves `frames()` with no reason to rebuild and nothing left to learn from. `TailRecorder.read_tail()` now returns the events *and* the byte count they came from; `read_events()` keeps its signature over it, since that is what every other caller uses. `build_snapshot` uses the count and reads only the mtime from the file's current state, which is safe to be too new: it makes a run look more recently active, delaying an idle verdict by one poll rather than stopping the stream. Nine tests, including the ordinary quiet-and-complete case -- where the cursor is the file size and the stream must still settle -- and the half-written final line, which is excluded from the count because bytes after the last newline cannot parse into an event yet. This is the `build_snapshot` read-then-stat race from the 2026-08 sweep. It was recorded as suspected and never demonstrated; it is demonstrated now. Verified: 194 tests across the live, server, trace and replay files, ruff clean, figure refreshed to 2,189. Co-Authored-By: Claude Opus 5 (1M context) --- docs/deep-dive.md | 2 +- grapharc/observe/trace.py | 25 +++- grapharc/server/live.py | 19 ++- tests/test_live_stream_cursor.py | 214 +++++++++++++++++++++++++++++++ 4 files changed, 251 insertions(+), 9 deletions(-) create mode 100644 tests/test_live_stream_cursor.py diff --git a/docs/deep-dive.md b/docs/deep-dive.md index b2a5be6..2e790f9 100644 --- a/docs/deep-dive.md +++ b/docs/deep-dive.md @@ -254,7 +254,7 @@ A stable system is not one that claims to have no edges — it is one whose edge - **`.env` and `grapharc.toml` follow the same discovery rule: the working directory, and nowhere else.** Neither searches parent directories — a run must not be governed by a file you did not know about, and must not be *billed* to one either. **This is a behaviour change:** the credential loader used to walk up to `/`, so a `.env` in an ancestor directory (a `$HOME` one on a shared box, a client project one above a demo checkout) was picked up silently. If you relied on that, move the file into the directory you run from, `export` the variable, or pass `env_file=` to name it explicitly. A real environment variable still beats any file. - **`grapharc run` has no budget unless you give it one.** Set any of `--max-tokens`, `--max-iterations`, `--max-seconds`, or `--max-concurrency`; without them each dimension is unlimited and the gate admits a topology of any worst-case cost. -**Verified this pass:** `pytest` → green, 2,180 selected and 13 deselected (the live ones); `ruff check .` clean; all eight `grapharc demo` stages green, plus the `trace` / `metrics` / `viz` / `replay` tour against a freshly recorded demo trace; the wheel builds and imports all submodules in a clean virtualenv with `[all]`, and `0.1.7` on PyPI is that wheel. The counts are a snapshot, not a property of the project — `pytest` re-derives them in one command, which is the only reason they are quoted, and `tests/test_deep_dive.py` fails this line rather than letting it drift. +**Verified this pass:** `pytest` → green, 2,189 selected and 13 deselected (the live ones); `ruff check .` clean; all eight `grapharc demo` stages green, plus the `trace` / `metrics` / `viz` / `replay` tour against a freshly recorded demo trace; the wheel builds and imports all submodules in a clean virtualenv with `[all]`, and `0.1.7` on PyPI is that wheel. The counts are a snapshot, not a property of the project — `pytest` re-derives them in one command, which is the only reason they are quoted, and `tests/test_deep_dive.py` fails this line rather than letting it drift. [ROADMAP.md](../ROADMAP.md) tracks what is built and what is not, item by item. diff --git a/grapharc/observe/trace.py b/grapharc/observe/trace.py index ae2a2bc..0507479 100644 --- a/grapharc/observe/trace.py +++ b/grapharc/observe/trace.py @@ -271,15 +271,32 @@ def record(self, event: TraceEvent) -> None: raise RuntimeError("TailRecorder is read-only") def read_events(self, run_id: str | None = None) -> list[TraceEvent]: + return self.read_tail(run_id)[0] + + def read_tail(self, run_id: str | None = None) -> tuple[list[TraceEvent], int]: + """The events, and **how many bytes they came from**. + + The byte count is not a detail a caller can re-derive with `stat()` + afterwards, which is why it is returned here. Two things make the + file's size at any later moment a different number: this read stops at + the last newline, so a half-written final line is excluded; and another + process may append between the read and the stat. + + A caller that uses a later `st_size` as its "everything up to here is + rendered" cursor therefore claims to have consumed bytes it never saw, + and will skip them for as long as the file stays that size — forever, + if the run has finished. `server.live` is that caller. + """ try: raw = self.path.read_bytes() except OSError: - return [] + return [], 0 cut = raw.rfind(b"\n") if cut < 0: - return [] + return [], 0 + consumed = cut + 1 events = [] - for line in raw[: cut + 1].splitlines(): + for line in raw[:consumed].splitlines(): if not line.strip(): continue try: @@ -288,7 +305,7 @@ def read_events(self, run_id: str | None = None) -> list[TraceEvent]: continue if run_id is None or event.run_id == run_id: events.append(event) - return events + return events, consumed def load_events( diff --git a/grapharc/server/live.py b/grapharc/server/live.py index 1d645c9..02a53fd 100644 --- a/grapharc/server/live.py +++ b/grapharc/server/live.py @@ -294,14 +294,25 @@ def build_snapshot(root: Path, rel: str, run_id: str | None) -> LiveSnapshot: """ path = resolve_trace(root, rel) recorder = TailRecorder(path) - events = recorder.read_events() + # `size` is the stream's "everything up to here is rendered" cursor, so it + # must describe *what was read*, not what the file is now. Those differ two + # ways: this read stops at the last newline, and the writer may append + # between the read and any later `stat()`. Taking it from `st_size` + # therefore claimed bytes the snapshot never saw, and `frames()` skips a + # file whose size has not changed since the last cursor — so those events + # were never sent, and never would be once the run stopped writing. A + # finished run then streamed as one still running, forever. + events, consumed = recorder.read_tail() if not events: return LiveSnapshot(trace=rel) + size = consumed try: - stat = path.stat() - size, quiet_for = stat.st_size, time() - stat.st_mtime + # Only the mtime comes from the file's current state. Erring *new* here + # is harmless: it makes the run look more recently active, which delays + # an idle verdict by one poll rather than stopping the stream. + quiet_for = time() - path.stat().st_mtime except OSError: - size, quiet_for = 0, float("inf") + quiet_for = float("inf") return compose_snapshot( rel, recorder, diff --git a/tests/test_live_stream_cursor.py b/tests/test_live_stream_cursor.py new file mode 100644 index 0000000..0cd51a6 --- /dev/null +++ b/tests/test_live_stream_cursor.py @@ -0,0 +1,214 @@ +"""`LiveSnapshot.size` is a cursor, and it has to describe what was read. + +`frames()` in `server.live` polls the trace file's size, rebuilds a snapshot +when it changes, and then stores the snapshot's own `size` as "everything up to +here has been sent": + + last_size = snapshot.size if snapshot.size else size + +So a `size` larger than the bytes the snapshot actually covers is not a +cosmetic error. It claims events that were never sent, and the next poll sees +an unchanged file size and rebuilds nothing. If the writer has stopped — the +run finished — the file never changes size again, the stream never rebuilds, +and a finished run is streamed as a running one for as long as the page is +open. + +`build_snapshot` used to take that number from `path.stat()` *after* reading +the events, which can exceed the bytes read two different ways: + + 1. `TailRecorder` cuts at the last newline, so a half-written final line is + excluded from the events but counted by `st_size`. + 2. The run is appending concurrently — that is the whole premise of a live + view — so bytes can land between the read and the stat. + +Only (2) loses whole events, because the bytes (1) leaves out cannot yet parse +into one. Both are fixed by the same change: the count comes from the read. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +from grapharc.observe.trace import TailRecorder +from grapharc.server.live import build_snapshot + + +def _event( + run_id: str = "r1", phase: str = "start", step: int = 1, *, terminal: bool = False +) -> str: + body: dict = { + "ts": "2026-01-01T00:00:00+00:00", + "run_id": run_id, + "graph": "g", + "node": "n", + "phase": phase, + "step": step, + } + if terminal: + # What `compose_snapshot` actually reads to decide a run is over; a bare + # `stop` phase on a node is not it. + body["state_delta"] = {"termination_reason": "completed"} + return json.dumps(body) + + +def _trace(tmp_path: Path, *lines: str, tail: str = "") -> Path: + path = tmp_path / "trace.jsonl" + path.write_text("".join(line + "\n" for line in lines) + tail, encoding="utf-8") + return path + + +# -- what the recorder reports ---------------------------------------------- + + +def test_the_recorder_reports_the_bytes_its_events_came_from(tmp_path): + path = _trace(tmp_path, _event(), _event(step=2)) + events, consumed = TailRecorder(path).read_tail() + + assert len(events) == 2 + assert consumed == path.stat().st_size + + +def test_a_half_written_final_line_is_excluded_from_the_count(tmp_path): + """The bytes after the last newline cannot parse into an event, so counting + them would claim an event that does not exist yet.""" + path = _trace(tmp_path, _event(), _event(step=2), tail='{"ts": "2026-01-0') + events, consumed = TailRecorder(path).read_tail() + + assert len(events) == 2 + assert consumed < path.stat().st_size + # And the count is exactly the complete prefix. + assert path.read_bytes()[:consumed].endswith(b"\n") + + +def test_a_file_with_no_complete_line_reads_as_nothing(tmp_path): + path = _trace(tmp_path, tail='{"ts": "2026-01-0') + + assert TailRecorder(path).read_tail() == ([], 0) + + +def test_read_events_still_returns_just_the_events(tmp_path): + """The old signature is what every other caller uses.""" + path = _trace(tmp_path, _event(), _event(step=2)) + + assert TailRecorder(path).read_events() == TailRecorder(path).read_tail()[0] + + +# -- the cursor the stream stores -------------------------------------------- + + +def test_the_snapshot_size_never_exceeds_what_it_read(tmp_path, monkeypatch): + """The wedge, demonstrated. + + A writer appending between the read and the stat is the ordinary case for a + live view, not an exotic one; it is made deterministic here by appending + from inside the read. With `size` taken from a later `stat()`, the snapshot + reported 2 events and a cursor past the third, and `frames()` would store + that cursor and skip the third event for as long as the file stayed that + size. + """ + path = _trace(tmp_path, _event(), _event(step=2)) + real_read_tail = TailRecorder.read_tail + + def read_then_append(self, run_id=None): + result = real_read_tail(self, run_id) + # The writer lands in the window. + with open(self.path, "a", encoding="utf-8") as handle: + handle.write(_event(phase="stop", step=3) + "\n") + return result + + monkeypatch.setattr(TailRecorder, "read_tail", read_then_append) + snapshot = build_snapshot(tmp_path, "trace.jsonl", None) + + assert len(snapshot.run_ids) == 1 + # The cursor describes the two events that were read, not the three bytes' + # worth now on disk. + assert snapshot.size < path.stat().st_size, ( + "size claims bytes the snapshot never read; frames() will skip them" + ) + + +def test_the_skipped_event_is_picked_up_on_the_next_read(tmp_path, monkeypatch): + """The point of the fix: the stream self-corrects. + + Because the cursor stopped short, the file's size now differs from it, so + the next poll rebuilds and the event arrives. That is the difference + between a one-poll delay and a stream that never recovers. + """ + path = _trace(tmp_path, _event(), _event(step=2)) + real_read_tail = TailRecorder.read_tail + appended = {"done": False} + + def read_then_append_once(self, run_id=None): + result = real_read_tail(self, run_id) + if not appended["done"]: + appended["done"] = True + with open(self.path, "a", encoding="utf-8") as handle: + handle.write(_event(phase="stop", step=3) + "\n") + return result + + monkeypatch.setattr(TailRecorder, "read_tail", read_then_append_once) + first = build_snapshot(tmp_path, "trace.jsonl", None) + + monkeypatch.setattr(TailRecorder, "read_tail", real_read_tail) + second = build_snapshot(tmp_path, "trace.jsonl", None) + + assert first.size < path.stat().st_size # cursor short, so the poll refires + assert second.size == path.stat().st_size + assert second.size > first.size + + +def test_a_quiet_complete_trace_reports_the_whole_file(tmp_path): + """The ordinary case must not regress: nothing is being appended, the last + line is complete, so the cursor is the file size and the stream settles.""" + path = _trace(tmp_path, _event(), _event(step=2), _event(phase="stop", step=3)) + snapshot = build_snapshot(tmp_path, "trace.jsonl", None) + + assert snapshot.size == path.stat().st_size + + +def test_an_empty_or_missing_trace_is_still_a_waiting_snapshot(tmp_path): + """The URL is posted before the run starts writing.""" + (tmp_path / "trace.jsonl").write_text("", encoding="utf-8") + + assert build_snapshot(tmp_path, "trace.jsonl", None).size == 0 + + +def test_a_finished_run_stops_reporting_itself_as_running(tmp_path, monkeypatch): + """The symptom a viewer actually sees, end to end. + + Reproduced against the old code: an append landing in the window left the + cursor equal to the file's final size while `done` was still False, so + `frames()` had no reason to rebuild and never learned the run had stopped. + The page showed a finished run as running for as long as it stayed open. + + With the cursor taken from the read, it falls short of the file, the next + poll rebuilds, and that rebuild sees the `stop` event. + """ + path = _trace(tmp_path, _event(), _event(step=2)) + real_read_tail = TailRecorder.read_tail + fired = {"n": 0} + + def read_then_append_once(self, run_id=None): + result = real_read_tail(self, run_id) + fired["n"] += 1 + if fired["n"] == 1: + with open(self.path, "a", encoding="utf-8") as handle: + handle.write(_event(phase="stop", step=3, terminal=True) + "\n") + return result + + monkeypatch.setattr(TailRecorder, "read_tail", read_then_append_once) + first = build_snapshot(tmp_path, "trace.jsonl", None) + final_size = path.stat().st_size + + # The old failure was exactly this pair: cursor == final size, done False. + assert not (first.size == final_size and not first.done), ( + "cursor matches the finished file while done is False: frames() will " + "never rebuild and the run streams as running forever" + ) + + monkeypatch.setattr(TailRecorder, "read_tail", real_read_tail) + second = build_snapshot(tmp_path, "trace.jsonl", None) + + assert second.done is True + assert second.size == final_size