| """Append-only event records with deterministic replay and artifact checks.""" |
| from __future__ import annotations |
| import json |
| import os |
| from datetime import datetime, timezone |
| from pathlib import Path |
| from typing import Any |
| from .schema import canonical, digest |
|
|
|
|
| class Trace: |
| def __init__(self, path: str | Path): |
| self.path = Path(path) |
| self.path.parent.mkdir(parents=True, exist_ok=True) |
| self.events = self.read(self.path) if self.path.exists() else [] |
|
|
| @staticmethod |
| def read(path: Path) -> list[dict]: |
| events = [] |
| previous = "0" * 64 |
| for line_no, line in enumerate(path.read_text().splitlines(), 1): |
| event = json.loads(line) |
| checksum = event.pop("hash") |
| if event["previous"] != previous or event["index"] != len(events) or digest(event) != checksum: |
| raise ValueError(f"trace integrity failure at line {line_no}") |
| event["hash"] = checksum |
| previous = checksum |
| events.append(event) |
| return events |
|
|
| def append(self, kind: str, payload: Any) -> dict: |
| event = {"index": len(self.events), "previous": self.events[-1]["hash"] if self.events else "0" * 64, |
| "time": datetime.now(timezone.utc).isoformat(), "kind": kind, "payload": payload} |
| event["hash"] = digest(event) |
| |
| with self.path.open("a", encoding="utf-8") as f: |
| f.write(canonical(event) + "\n") |
| f.flush() |
| os.fsync(f.fileno()) |
| self.events.append(event) |
| return event |
|
|
|
|
| def replay(path: str | Path) -> dict: |
| """Reconstruct scientific state without invoking tools or language models.""" |
| state = {"spec": None, "candidates": {}, "measurements": [], "evidence": {}, |
| "spent": {}, "plan": {}, "artifacts": {}, "stopped": False, "errors": [], "controller_feedback": [], "tool_messages": []} |
| for event in Trace.read(Path(path)): |
| p, kind = event["payload"], event["kind"] |
| if kind == "initialize": |
| if state["spec"] is not None: |
| raise ValueError("duplicate initialization") |
| state["spec"] = p |
| elif kind == "reserve": |
| for key, value in p["cost"].items(): |
| state["spent"][key] = state["spent"].get(key, 0) + value |
| elif kind == "result": |
| from .schema import Candidate |
| for c in p["candidates"]: |
| state["candidates"][Candidate.model_validate(c).id] = c |
| for e in p["evidence"]: |
| state["evidence"][e["id"]] = e |
| state["measurements"].extend(p["measurements"]) |
| state["artifacts"].update(p["artifacts"]) |
| if p.get("message"):state["tool_messages"].append(p["message"]) |
| elif kind == "revise": |
| state["plan"].update(p) |
| elif kind == "error": |
| state["errors"].append(p) |
| elif kind == "feedback": |
| state["controller_feedback"].append(p["error"]) |
| elif kind == "stop": |
| state["stopped"] = True |
| if state["spec"] is None: |
| raise ValueError("trace has no initial specification") |
| return state |
|
|
