231 lines
7.8 KiB
Python
231 lines
7.8 KiB
Python
"""Append-only JSONL transcript whose records form a SHA-256 hash chain.
|
|
|
|
An agent transcript is evidence, and evidence that cannot be checked is just a
|
|
log. Every record carries the hash of the record before it, so the file is a
|
|
chain: rewrite one payload, drop one record, or swap two, and verification
|
|
breaks at the first affected record and names it. A single file digest cannot
|
|
do that -- it says "something changed", not "record 3 was rewritten".
|
|
|
|
WHAT THIS DOES NOT PROVE -- stated plainly, because a gate that overclaims is
|
|
worse than no gate.
|
|
|
|
A hash chain detects MODIFICATION and REORDERING of the records that are
|
|
present. It cannot detect that records are MISSING FROM THE END: delete the
|
|
last line and the remaining chain still verifies, because every surviving
|
|
record still points at a real predecessor. Completeness needs an anchor
|
|
outside the transcript -- a commit SHA, or a digest stored elsewhere.
|
|
verify_file therefore also returns the record count, so a caller holding an
|
|
external count can compare it.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
|
|
GENESIS = "0" * 64
|
|
|
|
KIND_USER = "user"
|
|
KIND_ASSISTANT = "assistant"
|
|
KIND_TOOL = "tool"
|
|
KIND_FINAL = "final"
|
|
_KINDS = (KIND_USER, KIND_ASSISTANT, KIND_TOOL, KIND_FINAL)
|
|
|
|
_FIELDS = ("seq", "kind", "payload", "prev", "hash")
|
|
|
|
|
|
class TranscriptError(Exception):
|
|
"""Raised when a transcript is malformed, or a record is rejected."""
|
|
|
|
|
|
def canonical_json(obj):
|
|
"""Serialise deterministically: sorted keys, no incidental whitespace.
|
|
|
|
Two runs of the same loop must produce byte-identical transcripts, so the
|
|
encoding has to be a function of the data and nothing else.
|
|
"""
|
|
return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
|
|
|
|
|
|
def _record_hash(seq, kind, payload, prev):
|
|
body = canonical_json({"seq": seq, "kind": kind, "payload": payload, "prev": prev})
|
|
return hashlib.sha256(body.encode("utf-8")).hexdigest()
|
|
|
|
|
|
class Transcript(object):
|
|
"""An in-memory record list that can mirror itself to a JSONL file."""
|
|
|
|
def __init__(self, path=None):
|
|
self.path = path
|
|
self.records = []
|
|
self._prev = GENESIS
|
|
if path and os.path.exists(path):
|
|
self.load(path)
|
|
|
|
# -- writing ---------------------------------------------------------
|
|
|
|
def append(self, kind, payload):
|
|
if kind not in _KINDS:
|
|
raise TranscriptError(
|
|
"unknown record kind %r (expected one of: %s)" % (kind, ", ".join(_KINDS))
|
|
)
|
|
if not isinstance(payload, (dict, list)):
|
|
raise TranscriptError(
|
|
"payload must be a dict or list, got %s" % type(payload).__name__
|
|
)
|
|
# Reject a payload that cannot be written back out: a transcript that
|
|
# cannot be serialised is not a transcript.
|
|
try:
|
|
canonical_json(payload)
|
|
except (TypeError, ValueError) as exc:
|
|
raise TranscriptError("payload is not JSON-serialisable: %s" % exc)
|
|
|
|
seq = len(self.records)
|
|
digest = _record_hash(seq, kind, payload, self._prev)
|
|
record = {
|
|
"seq": seq,
|
|
"kind": kind,
|
|
"payload": payload,
|
|
"prev": self._prev,
|
|
"hash": digest,
|
|
}
|
|
self.records.append(record)
|
|
self._prev = digest
|
|
if self.path:
|
|
self._append_line(record)
|
|
return record
|
|
|
|
def _append_line(self, record):
|
|
with open(self.path, "a", encoding="utf-8") as fh:
|
|
fh.write(canonical_json(record) + "\n")
|
|
|
|
# -- reading ---------------------------------------------------------
|
|
|
|
def load(self, path=None):
|
|
path = path or self.path
|
|
if not path:
|
|
raise TranscriptError("no path given")
|
|
with open(path, "rb") as fh:
|
|
raw = fh.read()
|
|
records = []
|
|
for lineno, line in enumerate(raw.decode("utf-8", "replace").splitlines(), 1):
|
|
if not line.strip():
|
|
continue
|
|
try:
|
|
records.append(json.loads(line))
|
|
except ValueError as exc:
|
|
raise TranscriptError("line %d is not JSON: %s" % (lineno, exc))
|
|
ok, violations = verify_records(records)
|
|
if not ok:
|
|
raise TranscriptError("transcript failed verification: %s" % (violations[0],))
|
|
self.records = records
|
|
self._prev = records[-1]["hash"] if records else GENESIS
|
|
self.path = path
|
|
return self
|
|
|
|
def save(self, path=None):
|
|
path = path or self.path
|
|
if not path:
|
|
raise TranscriptError("no path given")
|
|
data = self.bytes()
|
|
tmp = path + ".tmp"
|
|
with open(tmp, "wb") as fh:
|
|
fh.write(data)
|
|
os.replace(tmp, path)
|
|
self.path = path
|
|
return path
|
|
|
|
def bytes(self):
|
|
return "".join(canonical_json(r) + "\n" for r in self.records).encode("utf-8")
|
|
|
|
def __len__(self):
|
|
return len(self.records)
|
|
|
|
|
|
def verify_records(records):
|
|
"""Return ``(ok, violations)``.
|
|
|
|
``violations`` is a list of dicts, each naming the record index that broke
|
|
the chain and the reason. The scan stops at the FIRST break: past a
|
|
broken link every later record's ``prev`` is wrong too, so continuing would
|
|
report one fault as many. The first break is the one to fix.
|
|
"""
|
|
violations = []
|
|
prev = GENESIS
|
|
for i, raw in enumerate(records):
|
|
if not isinstance(raw, dict):
|
|
violations.append(
|
|
{"seq": i, "reason": "not_an_object", "detail": type(raw).__name__}
|
|
)
|
|
break
|
|
missing = [f for f in _FIELDS if f not in raw]
|
|
if missing:
|
|
violations.append({"seq": i, "reason": "missing_fields", "detail": missing})
|
|
break
|
|
if raw["seq"] != i:
|
|
violations.append({"seq": i, "reason": "seq_gap", "detail": raw["seq"]})
|
|
break
|
|
if raw["prev"] != prev:
|
|
violations.append(
|
|
{
|
|
"seq": i,
|
|
"reason": "prev_mismatch",
|
|
"detail": raw["prev"],
|
|
"expected": prev,
|
|
}
|
|
)
|
|
break
|
|
try:
|
|
expected = _record_hash(raw["seq"], raw["kind"], raw["payload"], raw["prev"])
|
|
except (TypeError, ValueError) as exc:
|
|
violations.append({"seq": i, "reason": "unserialisable_payload", "detail": str(exc)})
|
|
break
|
|
if expected != raw["hash"]:
|
|
violations.append(
|
|
{
|
|
"seq": i,
|
|
"reason": "hash_mismatch",
|
|
"detail": raw["hash"],
|
|
"expected": expected,
|
|
}
|
|
)
|
|
break
|
|
prev = raw["hash"]
|
|
return (not violations), violations
|
|
|
|
|
|
def verify_file(path):
|
|
"""Verify a transcript file on disk. Returns ``(ok, violations, count)``.
|
|
|
|
``count`` is the number of records read. Compare it against an external
|
|
expectation to detect truncation -- the chain alone cannot, see module doc.
|
|
"""
|
|
try:
|
|
with open(path, "rb") as fh:
|
|
raw = fh.read()
|
|
except OSError as exc:
|
|
return False, [{"seq": None, "reason": "unreadable", "detail": str(exc)}], 0
|
|
|
|
records = []
|
|
for lineno, line in enumerate(raw.decode("utf-8", "replace").splitlines(), 1):
|
|
if not line.strip():
|
|
continue
|
|
try:
|
|
records.append(json.loads(line))
|
|
except ValueError as exc:
|
|
return (
|
|
False,
|
|
[
|
|
{
|
|
"seq": len(records),
|
|
"reason": "bad_json",
|
|
"line": lineno,
|
|
"detail": str(exc),
|
|
}
|
|
],
|
|
len(records),
|
|
)
|
|
ok, violations = verify_records(records)
|
|
return ok, violations, len(records)
|