File size: 3,247 Bytes
98bde72 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | """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)
# Flush each record before executing the next external action.
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
|