Merge remote-tracking branch 'origin/main' into feature/live-view-and-approvals

Author: Shashankss1205

grapharc/cli/main.py

@@ -48,7 +48,7 @@
 from grapharc.cli.output import EXIT_FAILED, EXIT_OK, emit, fail
 from grapharc.observe.metrics import summarize, to_mermaid
 from grapharc.observe.replay import ReplayError
-from grapharc.observe.trace import TraceRecorder
+from grapharc.observe.trace import TraceReadError, TraceRecorder
 
 EXAMPLES = (
     "stage0",
@@ -491,7 +491,13 @@ def _cmd_trace(args: argparse.Namespace) -> int:
     recorder = _existing_trace(args.path, command="trace", as_json=args.json)
     if isinstance(recorder, int):
         return recorder
-    events = recorder.read_events(args.run_id)
+    try:
+        events = recorder.read_events(args.run_id)
+    except TraceReadError as exc:
+        # A bad line — a truncated write, a hand edit — is the "unreadable
+        # trace" the exit-code contract names, not a traceback. Same for
+        # `metrics` and `viz` below: all three read this file.
+        return fail(str(exc), as_json=args.json, command="trace")
     payload: dict[str, Any] = {
         "ok": True,
         "command": "trace",
@@ -521,7 +527,10 @@ def _cmd_metrics(args: argparse.Namespace) -> int:
     recorder = _existing_trace(args.path, command="metrics", as_json=args.json)
     if isinstance(recorder, int):
         return recorder
-    metrics = summarize(recorder, args.run_id)
+    try:
+        metrics = summarize(recorder, args.run_id)
+    except TraceReadError as exc:
+        return fail(str(exc), as_json=args.json, command="metrics")
     if metrics is None:
         return fail(
             f"no events for run {args.run_id!r} in {args.path}",
@@ -544,6 +553,8 @@ def _cmd_viz(args: argparse.Namespace) -> int:
         return recorder
     try:
         mermaid = to_mermaid(recorder, args.run_id)
+    except TraceReadError as exc:
+        return fail(str(exc), as_json=args.json, command="viz")
     except ReplayError as exc:
         # Every other reading command answers this as a document; `viz` used to
         # let it out as a traceback with empty stdout, which breaks the CLI's own

grapharc/observe/__init__.py

@@ -40,7 +40,7 @@
     replay,
     replay_thread,
 )
-from grapharc.observe.trace import TraceEvent, TraceRecorder, load_events
+from grapharc.observe.trace import TraceEvent, TraceReadError, TraceRecorder, load_events
 
 __all__ = [
     "ListSpanExporter",
@@ -61,6 +61,7 @@
     "SpanExporter",
     "ThreadCost",
     "TraceEvent",
+    "TraceReadError",
     "TraceRecorder",
     "attribute",
     "attribute_thread",

grapharc/observe/trace.py

@@ -20,11 +20,29 @@
 from pathlib import Path
 from typing import Any
 
-from pydantic import BaseModel
+from pydantic import BaseModel, ValidationError
 
 _MAX_VALUE_CHARS = 2000
 
 
+class TraceReadError(Exception):
+    """A line in a trace file is not a `TraceEvent`.
+
+    Raised instead of skipped on purpose: a partially-read audit trail
+    presented as complete would be worse than a refusal. The message names the
+    file and the 1-based line so the one bad line — a process killed mid-write,
+    a hand edit — can be found without a pydantic field listing.
+    """
+
+    def __init__(self, path: Path, line_number: int, cause: Exception) -> None:
+        super().__init__(
+            f"unreadable trace file: {path}: line {line_number} is not a trace event"
+        )
+        self.path = path
+        self.line_number = line_number
+        self.cause = cause
+
+
 def _jsonable(value: Any) -> Any:
     """Best-effort conversion to something json.dumps accepts, truncating long text."""
     if isinstance(value, BaseModel):
@@ -186,10 +204,13 @@ def read_events(self, run_id: str | None = None) -> list[TraceEvent]:
             return []
         events = []
         with self.path.open(encoding="utf-8") as f:
-            for line in f:
+            for line_number, line in enumerate(f, start=1):
                 if not line.strip():
                     continue
-                ev = TraceEvent.model_validate_json(line)
+                try:
+                    ev = TraceEvent.model_validate_json(line)
+                except ValidationError as exc:
+                    raise TraceReadError(self.path, line_number, exc) from exc
                 if run_id is None or ev.run_id == run_id:
                     events.append(ev)
         return events
@@ -249,4 +270,4 @@ def load_events(
     return recorder.read_events(run_id)
 
 
-__all__ = ["TailRecorder", "TraceEvent", "TraceRecorder", "load_events"]
+__all__ = ["TailRecorder", "TraceEvent", "TraceReadError", "TraceRecorder", "load_events"]

grapharc/runtime/budget.py

@@ -380,8 +380,8 @@ def deadline_guard(meter: BudgetMeter, *, what: str) -> Iterator[None]:
     What this does *not* guarantee:
 
     - Mechanism 2 cannot interrupt a thread parked inside a C call: a
-      `time.sleep(60)` sleeps out its 60 seconds and raises on return. This is
-      not a fan-out-only weakness. Mechanism 1 needs `invoke()` to be on the
+      `time.sleep(60)` is not interrupted mid-call, but the guard still raises
+      on exit. This is not a fan-out-only weakness. Mechanism 1 needs `invoke()` to be on the
       process's main thread, so *any* run driven from a worker thread — every
       request handler in a threaded server, every `ThreadPoolExecutor` caller —
       falls back to mechanism 2 for the whole run, nodes and fan-out alike.
@@ -395,9 +395,10 @@ def deadline_guard(meter: BudgetMeter, *, what: str) -> Iterator[None]:
     - Like any asynchronous exception, the interrupt lands wherever the node
       happened to be: it is as safe as Ctrl-C, no safer.
 
-    Short of that last case the ceiling is honoured at the node boundary: if the
-    deadline passed and the node swallowed the exception, this guard raises on
-    exit, so the node's writes never reach state.
+    Short of the never-returns case the ceiling is honoured at the node
+    boundary: if the deadline passed — whether the node swallowed the exception
+    or no interrupt was ever delivered — this guard raises on exit, so the
+    node's writes never reach state.
     """
     remaining = meter.remaining_seconds()
     if remaining is None:
@@ -499,5 +500,11 @@ def disarm() -> None:
             disarm()
     except NodeDeadlineExceeded as exc:
         raise NodeDeadlineExceeded(detail()) from exc
-    if state["fired"]:
+
+    # Reached only when the node returned normally. It may have swallowed the
+    # interrupt, or the deadline may have passed without the timer firing —
+    # the timer thread needs the GIL, which a node inside a long C call
+    # withholds until it returns; either way its writes must not land.
+    left = meter.remaining_seconds()
+    if state["fired"] or (left is not None and left <= 0):
         raise NodeDeadlineExceeded(detail())

tests/test_budget_enforcement.py

@@ -542,6 +542,52 @@ def body():
     assert ran_for < 2.0, "the node asked for 5s and was not stopped"
 
 
+def test_an_overrun_is_refused_at_exit_even_if_the_timer_never_fired(monkeypatch):
+    """A node that holds the GIL through the deadline — a long C call, or plain
+    timer-scheduling latency — denies the timer thread its turn: `fire()` never
+    runs, the node returns normally, and an exit check that tests only
+    `state["fired"]` lets the overrun's writes land. The contract is the node
+    boundary, so the exit check itself must notice the spent deadline.
+
+    The timer is replaced with one that never fires, which makes this the
+    deterministic statement of that contract: asserting on whether a real
+    timer's async exception got delivered in time is a race (see
+    `_run_swallower` above), whereas the exit check runs unconditionally.
+    """
+
+    class NeverFires:
+        """`threading.Timer`'s surface as the guard uses it, minus the firing."""
+
+        def __init__(self, interval, function):
+            self.daemon = False
+
+        def start(self):
+            pass
+
+        def cancel(self):
+            pass
+
+    monkeypatch.setattr(threading, "Timer", NeverFires)
+    outcome: dict[str, object] = {}
+
+    def body():  # a worker thread uses mechanism 2, like any threaded server
+        meter = BudgetMeter(Budget(max_seconds=0.05))
+        try:
+            with deadline_guard(meter, what="node 'n'"):
+                time.sleep(0.2)  # outlast the deadline; nothing interrupts it
+            outcome["raised"] = None
+        except NodeDeadlineExceeded as exc:
+            outcome["raised"] = exc
+
+    worker = threading.Thread(target=body)
+    worker.start()
+    worker.join(timeout=10)
+    assert not worker.is_alive()
+    assert isinstance(outcome["raised"], NodeDeadlineExceeded), (
+        "the node overran, no interrupt fired, and the guard let its writes land"
+    )
+
+
 def test_a_node_that_finishes_in_time_is_left_alone():
     def brisk(state):
         time.sleep(0.05)

tests/test_cli.py

@@ -336,6 +336,43 @@ def test_viz_renders_the_executed_path(two_runs, capsys):
     assert "load" in payload["mermaid"]
 
 
+# Every reading command, with the arguments it needs beyond the path. `metrics`
+# and `viz` never reach the run id: the file refuses before any run is looked up.
+READERS = [["trace"], ["metrics", "r1"], ["viz", "r1"]]
+
+
+def _bad_trace(tmp_path) -> Path:
+    """A trace whose second line is not an event — a process killed mid-write."""
+    trace = TraceRecorder(tmp_path / "bad.jsonl")
+    trace.event(run_id="r1", graph="g", node="load", phase="start", step=1)
+    with trace.path.open("a", encoding="utf-8") as f:
+        f.write('{"not": "a trace event"}\n')
+    return trace.path
+
+
+@pytest.mark.parametrize("argv", READERS, ids=lambda argv: argv[0])
+def test_a_malformed_trace_is_a_report_not_a_traceback(argv, tmp_path, capsys):
+    """The contract in `output.py` names "an unreadable trace" as exit 2."""
+    bad = _bad_trace(tmp_path)
+    code, out, err = call([argv[0], str(bad), *argv[1:]], capsys)
+    assert code == 2
+    assert out == ""
+    assert f"error: unreadable trace file: {bad}: line 2 is not a trace event\n" == err
+
+
+@pytest.mark.parametrize("argv", READERS, ids=lambda argv: argv[0])
+def test_a_malformed_trace_fails_as_one_json_document(argv, tmp_path, capsys):
+    bad = _bad_trace(tmp_path)
+    code, payload, err = call_json([argv[0], str(bad), *argv[1:]], capsys)
+    assert code == 2
+    assert payload == {
+        "ok": False,
+        "command": argv[0],
+        "error": f"unreadable trace file: {bad}: line 2 is not a trace event",
+    }
+    assert err == ""
+
+
 # -- models -------------------------------------------------------------------