← Files MathboxARCHIVED FILE
skills/research-state/scripts/research_state.py
78.6 KB · Oct 3, 2026 · 06:33 UTC
#!/usr/bin/env python3
"""Local, append-only research evidence ledger. No network or code execution.
This checks bookkeeping, artifact freshness and dependency closure, not proofs.
Python 3.10+, standard library only. See ../references/ledger.md for the contract.
"""
from __future__ import annotations
import argparse
from collections import Counter
from contextlib import contextmanager, suppress
from datetime import datetime, timezone
import hashlib
import json
import os
from pathlib import Path
import re
import stat
import sys
import tempfile
SUPPORTING_STATUSES = {"proof-recorded", "source-recorded"}
RUN_OUTCOMES = {"succeeded", "failed", "blocked", "inconclusive", "abandoned"}
RECONCILIATION_DECISIONS = {
"continue",
"succeeded",
"failed",
"blocked",
"inconclusive",
"late-consistent",
"late-conflict",
"late-superseded",
"late-not-applicable",
}
TERMINAL_RECONCILIATIONS = {"succeeded", "failed", "blocked", "inconclusive"}
LATE_RECONCILIATIONS = {
"late-consistent", "late-conflict", "late-superseded", "late-not-applicable"
}
# Generated by the helper; deferred proposals may not supply them.
GENERATED_FIELDS = {"snapshot", "event_id", "created_at", "previous",
"manifest_inputs", "manifest_outputs"}
# Deferred packets create documents only; code and tool configuration stay under local authority.
DEFERRED_SUFFIXES = {".bib", ".md", ".tex", ".txt"}
# Hosts load these as agent instructions wherever they appear.
INSTRUCTION_FILES = {"agents.md", "agents.override.md", "claude.md", "claude.local.md",
"gemini.md", "skill.md"}
class LedgerError(ValueError):
pass
def require(condition, message):
if not condition:
raise LedgerError(message)
def canonical(value):
return json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
def json_text(value, label):
"""Reject what json.loads accepts but UTF-8 cannot store, such as lone surrogates."""
try:
canonical(value).encode("utf-8")
except (TypeError, ValueError) as exc:
raise LedgerError(f"{label} must be JSON with valid Unicode text: {exc}") from None
def digest(value):
return hashlib.sha256(canonical(value).encode()).hexdigest()
def file_hash(path):
h = hashlib.sha256()
with path.open("rb") as stream:
for block in iter(lambda: stream.read(1024 * 1024), b""):
h.update(block)
return h.hexdigest()
@contextmanager
def staged_copy(directory, data, mode=None):
"""A complete, fsynced temporary file in directory; removed unless renamed away."""
fd, temp = tempfile.mkstemp(prefix="pending-", dir=directory)
try:
with os.fdopen(fd, "wb") as stream:
stream.write(data)
stream.flush()
os.fsync(stream.fileno())
if mode is not None:
os.chmod(temp, mode)
yield temp
finally:
Path(temp).unlink(missing_ok=True)
def replace_file(path, data):
"""Atomically replace an existing file's bytes, keeping its permissions."""
with staged_copy(path.parent, data, stat.S_IMODE(path.stat().st_mode)) as temp:
os.replace(temp, path)
def write_new(path, data):
"""Create path with data, never replacing an existing entry; the umask sets its mode."""
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0), 0o666)
try:
with os.fdopen(fd, "wb") as stream:
stream.write(data)
stream.flush()
os.fsync(stream.fileno())
except BaseException:
path.unlink(missing_ok=True)
raise
def inside(root, name):
require(isinstance(name, str) and name.strip(), "path must be nonempty text")
path = Path(name)
require(not path.is_absolute() and ".." not in path.parts, "path must be project-relative")
current = root
for part in path.parts:
current = current / part
require(not current.is_symlink(), f"symlink not supported: {name}")
require(current.resolve().is_relative_to(root), f"path escapes project: {name}")
return current
def pinned_path(name):
"""Lexical spelling used to compare recorded pins, e.g. ./a//b.md -> a/b.md."""
return Path(name).as_posix()
def current_hash(root, name, overlays=None):
"""Path, stored spelling and SHA-256 of staged or on-disk bytes (None when absent)."""
path = inside(root, name)
key = path.relative_to(root).as_posix()
if overlays is not None and key in overlays:
return path, key, hashlib.sha256(overlays[key]).hexdigest()
return path, key, file_hash(path) if path.is_file() else None
def text_field(obj, key):
value = obj.get(key)
require(isinstance(value, str) and bool(value.strip()), f"{key} must be nonempty text")
return value
def strings(obj, key):
value = obj.get(key)
require(isinstance(value, list) and all(isinstance(x, str) and x.strip() for x in value),
f"{key} must be a list of nonempty strings")
require(len(value) == len(set(value)), f"duplicate {key}")
return value
def identifier(obj, key="id"):
value = text_field(obj, key)
require(bool(re.fullmatch(r"[A-Za-z][A-Za-z0-9_-]{0,79}", value)), f"invalid {key}")
return value
def artifact_record(artifact, label="artifact", locator=False):
require(isinstance(artifact, dict), f"{label} must be an object")
text_field(artifact, "path")
require(bool(re.fullmatch(r"[0-9a-f]{64}", artifact.get("sha256", ""))),
f"invalid {label} hash")
if locator:
text_field(artifact, "locator")
def base_pointer(state, payload):
base = text_field(payload, "base_event")
require(base in state["events"], "base_event must name an existing ledger event")
text_field(payload, "base_revision")
def route_closed(state, route):
return any(event["payload"]["route"] == route
for event in state["route_closures"].values())
def route_targets(route):
"""Claims a route can directly advance, with legacy routes targeting claim."""
return set(route.get("resolves", [route["claim"]]))
def route_in_scope(route, selected):
return route["claim"] in selected or bool(route_targets(route) & selected)
def program_closed(state, program):
return any(event["payload"]["program"] == program
for event in state["program_results"].values())
def closure(claims, start):
seen, todo = set(), [start]
while todo:
key = todo.pop()
require(key in claims, f"unknown claim: {key}")
if key not in seen:
seen.add(key)
todo.extend(claims[key]["dependencies"])
return seen
def topological(claims):
remaining = {key: set(c["dependencies"]) for key, c in claims.items()}
order = []
while remaining:
ready = sorted(key for key, deps in remaining.items() if not deps)
require(ready, "circular claim dependencies")
order.extend(ready)
for key in ready:
del remaining[key]
for deps in remaining.values():
deps.difference_update(ready)
return order
def dependency_map(claims):
"""Transitive dependencies of every claim, accumulated once in dependency order."""
result = {}
for key in topological(claims):
deps = set()
for parent in claims[key]["dependencies"]:
deps.add(parent)
deps |= result[parent]
result[key] = deps
return result
def snapshot(claims, claim):
return {key: claims[key]["revision"] for key in sorted(closure(claims, claim))}
def empty_state():
return {
"claims": {}, "evidence": {}, "reviews": {}, "routes": {}, "results": {},
"programs": {}, "program_observations": {}, "program_results": {},
"runs": {}, "run_observations": {}, "run_results": {},
"reconciliations": {}, "reconciled_results": set(), "route_closures": {},
"retracted": set(), "superseded": set(), "events": {},
}
def apply(state, event):
"""Validate an event against prior state. Mutates only an in-memory projection."""
require(isinstance(event, dict), "event must be an object")
text_field(event, "actor")
payload = event.get("payload")
require(isinstance(payload, dict), "payload must be an object")
p, kind, claims = payload, event.get("type"), state["claims"]
key = event["event_id"]
require(key not in state["events"], "duplicate event ID")
if kind == "claim":
cid = identifier(p)
for field in ("statement", "regime", "level"):
text_field(p, field)
strings(p, "hypotheses")
deps = strings(p, "dependencies")
require(all(d in claims for d in deps), "register dependency claims first")
if "statement_artifact" in p:
artifact_record(p["statement_artifact"], "statement artifact", locator=True)
if cid in claims:
text_field(p, "reason")
revision = claims.get(cid, {}).get("revision", 0) + 1
claims[cid] = dict(p, revision=revision, event_id=key)
topological(claims)
elif kind == "evidence":
cid = text_field(p, "claim")
require(cid in claims, "unknown evidence claim")
require(p.get("kind") in {"proof", "source", "computation", "counterexample"},
"unknown evidence kind")
text_field(p, "summary")
require(p.get("snapshot") == snapshot(claims, cid), "incorrect evidence revision snapshot")
artifacts = p.get("artifacts")
require(isinstance(artifacts, list), "artifacts must be a list")
for artifact in artifacts:
artifact_record(artifact)
if p["kind"] == "computation":
text_field(p, "assertion")
text_field(p, "bounds")
require(strings(p, "non_claims"), "computation needs non_claims")
if "manifest" in p:
artifact_record(p["manifest"], "computation manifest")
inputs, outputs = p.get("manifest_inputs"), p.get("manifest_outputs")
require(isinstance(inputs, list) and inputs,
"linked computation manifest needs pinned inputs")
require(isinstance(outputs, list) and outputs,
"linked computation manifest needs pinned outputs")
for artifact in inputs + outputs:
artifact_record(artifact, "manifest artifact")
require(not ({a["path"] for a in inputs} & {a["path"] for a in outputs}),
"computation inputs and outputs must be disjoint")
require(bool(artifacts) or p.get("kind") == "computation" and "manifest" in p,
"evidence needs durable artifacts")
if p["kind"] == "source":
for field in ("identifier", "version", "locator", "translation"):
text_field(p, field)
if p["kind"] == "counterexample":
text_field(p, "hypothesis_check")
if "supersedes" in p:
for old in strings(p, "supersedes"):
require(old in state["evidence"] and state["evidence"][old]["payload"]["claim"] == cid,
"supersedes must name prior evidence of this claim")
require(old not in state["superseded"] and old not in state["retracted"], "evidence is already inactive")
state["superseded"].add(old)
state["evidence"][key] = event
elif kind == "review":
target = p.get("evidence")
require(target in state["evidence"] and target not in state["retracted"] and target not in state["superseded"], "review needs active evidence")
require(p.get("outcome") in {"pass", "fail", "conditional"}, "invalid review outcome")
require(type(p.get("independent")) is bool, "independent must be a boolean")
if p["independent"]:
require(event["actor"] != state["evidence"][target]["actor"], "author cannot independently review own evidence")
text_field(p, "summary")
artifact = p.get("artifact")
require(isinstance(artifact, dict), "review needs a durable report")
text_field(artifact, "path")
require(bool(re.fullmatch(r"[0-9a-f]{64}", artifact.get("sha256", ""))), "invalid review hash")
state["reviews"][key] = event
elif kind == "retract":
target = text_field(p, "target")
require(target in state["evidence"] or target in state["reviews"], "retract an evidence or review event")
require(target not in state["retracted"], "already retracted")
text_field(p, "reason")
state["retracted"].add(target)
elif kind == "route":
rid = identifier(p)
require(rid not in state["routes"], "route IDs are immutable; open a new route")
require(p.get("claim") in claims, "unknown route claim")
for field in ("mechanism", "question", "discriminator", "success", "failure"):
text_field(p, field)
prerequisites = strings(p, "prerequisites")
require(all(d in claims for d in prerequisites), "unknown route prerequisite")
if "resolves" in p:
resolves = strings(p, "resolves")
require(resolves, "resolves must name at least one claim")
require(all(d in claims for d in resolves), "unknown resolved obligation")
require(set(resolves) <= closure(claims, p["claim"]),
"resolved obligations must belong to the owning claim's dependency closure")
require(not (set(prerequisites) & route_targets(p)),
"a route target cannot be its own prerequisite")
for field in ("gain", "cost"):
require(type(p.get(field)) is int and 1 <= p[field] <= 5, f"{field} must be an integer 1..5")
if "reopens" in p:
require(p["reopens"] in state["route_closures"],
"reopens must name a terminal route event")
prior = state["routes"][state["route_closures"][p["reopens"]]["payload"]["route"]]["payload"]
require(prior["claim"] == p["claim"]
and route_targets(prior) == route_targets(p)
and prior["mechanism"] == p["mechanism"],
"reopened route must retain owner, resolved obligations and mechanism")
text_field(p, "changed_input")
else:
for old in state["routes"].values():
prior = old["payload"]
require(not (prior["claim"] == p["claim"]
and prior["mechanism"] == p["mechanism"]),
"repeated mechanism: name reopens and changed_input")
state["routes"][rid] = event
elif kind == "route-result":
require(p.get("route") in state["routes"], "unknown route")
require(not route_closed(state, p["route"]), "route already closed")
require(p.get("outcome") in {"succeeded", "failed", "blocked", "inconclusive"}, "invalid route outcome")
text_field(p, "reason")
text_field(p, "next_question")
state["results"][key] = event
state["route_closures"][key] = event
elif kind == "program":
pid = identifier(p)
require(pid not in state["programs"], "program IDs are immutable")
require(p.get("goal") in claims, "unknown program goal")
text_field(p, "objective")
base_pointer(state, p)
state["programs"][pid] = event
elif kind == "program-observation":
pid = p.get("program")
require(pid in state["programs"], "unknown program")
require(not program_closed(state, pid), "program already closed")
require(p.get("state") in {"active", "waiting", "blocked", "unknown"},
"invalid program observation state")
text_field(p, "summary")
text_field(p, "observed_revision")
state["program_observations"][key] = event
elif kind == "program-result":
pid = p.get("program")
require(pid in state["programs"], "unknown program")
require(not program_closed(state, pid), "program already closed")
require(p.get("outcome") in {"completed", "blocked", "abandoned"},
"invalid program outcome")
active_runs = [run for run, item in state["runs"].items()
if item["payload"]["program"] == pid and
not any(result["payload"]["run"] == run
for result in state["run_results"].values())]
require(not active_runs, "close or abandon every program run first")
text_field(p, "reason")
text_field(p, "next_action")
text_field(p, "closed_revision")
state["program_results"][key] = event
elif kind == "route-run":
run = identifier(p)
require(run not in state["runs"], "run IDs are immutable")
rid, pid = p.get("route"), p.get("program")
require(rid in state["routes"], "unknown run route")
require(pid in state["programs"], "unknown run program")
require(not program_closed(state, pid), "cannot start a run in a closed program")
require(not route_closed(state, rid), "cannot start a run on a closed route")
goal = state["programs"][pid]["payload"]["goal"]
require(route_in_scope(state["routes"][rid]["payload"], closure(claims, goal)),
"run route is outside the program goal dependency closure")
base_pointer(state, p)
text_field(p, "executor")
require(strings(p, "work_scope"), "run needs a nonempty work_scope")
state["runs"][run] = event
elif kind == "run-observation":
run = p.get("run")
require(run in state["runs"], "unknown run")
require(not any(x["payload"]["run"] == run for x in state["run_results"].values()),
"run already closed")
require(p.get("state") in {"active", "waiting", "blocked", "unknown"},
"invalid run observation state")
text_field(p, "summary")
text_field(p, "observed_revision")
state["run_observations"][key] = event
elif kind == "run-result":
run = p.get("run")
require(run in state["runs"], "unknown run")
require(not any(x["payload"]["run"] == run for x in state["run_results"].values()),
"run already closed")
require(p.get("outcome") in RUN_OUTCOMES, "invalid run outcome")
text_field(p, "reason")
text_field(p, "next_question")
text_field(p, "result_revision")
artifacts = p.get("artifacts", [])
require(isinstance(artifacts, list), "run result artifacts must be a list")
for artifact in artifacts:
artifact_record(artifact, "run result artifact")
state["run_results"][key] = event
elif kind == "route-reconcile":
rid = p.get("route")
require(rid in state["routes"], "unknown reconciliation route")
results = strings(p, "results")
require(results, "reconciliation needs run results")
require(all(result in state["run_results"] for result in results),
"reconciliation results must name run-result events")
require(all(state["runs"][state["run_results"][result]["payload"]["run"]]["payload"]["route"] == rid
for result in results), "reconciliation result belongs to another route")
base_pointer(state, p)
require(all(
state["runs"][state["run_results"][result]["payload"]["run"]]["payload"]["base_event"]
== p["base_event"]
and state["runs"][state["run_results"][result]["payload"]["run"]]["payload"]["base_revision"]
== p["base_revision"]
for result in results
), "reconciliation results must share its declared base event and revision")
require(any(result not in state["reconciled_results"] for result in results),
"reconciliation must add at least one previously unreconciled result")
decision = p.get("decision")
require(decision in RECONCILIATION_DECISIONS, "invalid reconciliation decision")
text_field(p, "reason")
text_field(p, "next_question")
text_field(p, "current_revision")
conflicts = strings(p, "conflicts")
outcomes = {state["run_results"][result]["payload"]["outcome"]
for result in results}
require(len(outcomes) == 1 or conflicts,
"reconciliation of differing run outcomes must record conflicts")
require(decision != "late-conflict" or conflicts,
"late-conflict needs an explicit conflict")
if decision in LATE_RECONCILIATIONS:
require(route_closed(state, rid), "late reconciliation requires a closed route")
else:
require(not route_closed(state, rid), "route already closed; use a late disposition")
if decision in TERMINAL_RECONCILIATIONS:
state["route_closures"][key] = event
state["reconciled_results"].update(results)
state["reconciliations"][key] = event
else:
raise LedgerError(f"unknown event type: {kind}")
state["events"][key] = event
class Ledger:
def __init__(self, root):
self.root = Path(root).resolve()
require(self.root.is_dir(), "project root does not exist")
self.base = inside(self.root, ".mathbox")
self.events = inside(self.root, ".mathbox/events")
def initialize(self):
require(not self.base.exists(), ".mathbox already exists; initialization never overwrites")
self.base.mkdir()
self.events.mkdir()
config = {"schema_version": 1, "projection": {"semantics": "recorded-evidence-v1"}}
(self.base / "config.json").write_text(
json.dumps(config, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
)
def read(self):
config_path = inside(self.root, ".mathbox/config.json")
require(config_path.is_file(), "ledger not initialized; use init explicitly")
config = json.loads(config_path.read_text(encoding="utf-8"))
require(isinstance(config, dict) and type(config.get("schema_version")) is int and config["schema_version"] == 1,
"unsupported ledger schema")
projection = config.get("projection")
require(projection is None or
isinstance(projection, dict) and projection.get("semantics") == "recorded-evidence-v1",
"unsupported projection semantics")
require(self.events.is_dir(), "missing events directory")
state, previous = empty_state(), None
for number, path in enumerate(sorted(self.events.iterdir()), 1):
require(path.name == f"{number:06d}.json" and path.is_file() and not path.is_symlink(),
f"invalid or noncontiguous event: {path.name}")
event = json.loads(path.read_text(encoding="utf-8"))
require(isinstance(event, dict), f"invalid event: {path.name}")
recorded = event.get("sha256")
unsigned = {k: v for k, v in event.items() if k != "sha256"}
require(recorded == digest(unsigned), f"event checksum mismatch: {path.name}")
require(event.get("previous") == previous and event.get("event_id") == f"E{number:06d}",
f"broken event chain: {path.name}")
apply(state, event)
previous = recorded
return state
@contextmanager
def lock(self):
path = inside(self.root, ".mathbox/write.lock")
try:
path.mkdir()
except FileExistsError:
raise LedgerError("ledger writer active or stale write.lock; inspect before removing") from None
owned = path.lstat()
try:
yield
finally:
# Release only this writer's lock, not one recreated after a manual removal.
with suppress(FileNotFoundError):
current = path.lstat()
if (current.st_dev, current.st_ino) == (owned.st_dev, owned.st_ino):
path.rmdir()
def pin(self, artifact, overlays=None):
require(isinstance(artifact, dict), "artifact must be an object with path")
path, name, actual = current_hash(self.root, text_field(artifact, "path"), overlays)
require(not path.is_relative_to(self.base), "evidence must be outside the ledger itself")
require(actual is not None, f"missing artifact: {artifact['path']}")
if "sha256" in artifact:
require(artifact["sha256"] == actual, f"supplied artifact hash differs from current content: {name}")
# One stored spelling per file, so later path lookups compare like with like.
return dict(artifact, path=name, sha256=actual)
def pin_computation_manifest(self, artifact, claim, overlays=None):
"""Pin a manifest plus the exact input/output closure it declares.
This intentionally checks only ledger linkage and file hashes. The
computation-audit validator remains responsible for the manifest's
scientific and execution contract.
"""
pinned = self.pin(artifact, overlays)
path = inside(self.root, pinned["path"])
try:
raw = overlays.get(pinned["path"]) if overlays is not None else None
manifest = json.loads(raw.decode("utf-8") if raw is not None else
path.read_text(encoding="utf-8"))
except (OSError, ValueError, TypeError) as exc:
raise LedgerError(f"cannot read computation manifest: {exc}") from None
require(isinstance(manifest, dict), "computation manifest must be an object")
require(type(manifest.get("schema_version")) is int
and manifest["schema_version"] == 2,
"linked computation evidence requires a version 2 manifest")
require(manifest.get("claim_id") == claim,
"computation manifest claim_id must equal the ledger claim ID")
run = manifest.get("run")
require(isinstance(run, dict) and run.get("status") == "completed"
and run.get("exit_status") == 0,
"linked computation evidence needs a completed zero-exit run")
inputs, outputs = manifest.get("input_artifacts"), manifest.get("outputs")
require(isinstance(inputs, list) and inputs,
"linked computation manifest needs nonempty input_artifacts")
require(isinstance(outputs, list) and outputs,
"linked computation manifest needs nonempty outputs")
def declared(entry, role):
require(isinstance(entry, dict), f"manifest {role} must be an object")
checksum = entry.get("sha256_after") if role == "input" else entry.get("sha256")
require(bool(re.fullmatch(r"[0-9a-f]{64}", checksum or "")),
f"manifest {role} needs a current SHA-256")
if role == "input":
require(bool(re.fullmatch(r"[0-9a-f]{64}", entry.get("sha256") or "")),
"manifest input needs a pre-run SHA-256")
require(entry["sha256"] == entry["sha256_after"],
"completed computation evidence cannot have changed inputs")
return self.pin({"path": text_field(entry, "path"), "sha256": checksum}, overlays)
pinned_inputs = [declared(entry, "input") for entry in inputs]
pinned_outputs = [declared(entry, "output") for entry in outputs]
if "declared_results" in manifest:
declared_results = manifest["declared_results"]
require(isinstance(declared_results, list)
and all(isinstance(path, str) and path.strip() for path in declared_results)
and len(declared_results) == len(set(declared_results)),
"declared_results must be a duplicate-free path list")
result_outputs = {entry.get("path") for entry in outputs
if isinstance(entry, dict) and entry.get("kind") == "result"}
require(set(declared_results) == result_outputs,
"every declared result must be a result-kind output, with no extras")
paths = [entry["path"] for entry in pinned_inputs + pinned_outputs]
require(len(paths) == len(set(paths)), "manifest artifact paths must be unique")
return pinned, pinned_inputs, pinned_outputs
def prepare_event(self, proposal, state, overlays=None):
require(isinstance(proposal, dict) and set(proposal) == {"type", "actor", "payload"},
"proposal fields must be type, actor, payload")
event = json.loads(json.dumps(proposal))
require(isinstance(event["payload"], dict), "payload must be an object")
p = event["payload"]
if event["type"] == "claim" and "statement_artifact" in p:
require(isinstance(p["statement_artifact"], dict),
"statement artifact must be an object")
text_field(p["statement_artifact"], "locator")
p["statement_artifact"] = self.pin(p["statement_artifact"], overlays)
elif event["type"] == "evidence":
require(p.get("claim") in state["claims"], "unknown evidence claim")
require(not claim_contract_issues(self.root, state["claims"], p["claim"], overlays),
"claim statement contract changed; record a claim revision before new evidence")
require(isinstance(p.get("artifacts", []), list), "artifacts must be a list")
p["artifacts"] = [self.pin(a, overlays) for a in p.get("artifacts", [])]
if p.get("kind") == "computation" and "manifest" in p:
p["manifest"], p["manifest_inputs"], p["manifest_outputs"] = \
self.pin_computation_manifest(p["manifest"], p["claim"], overlays)
p["snapshot"] = snapshot(state["claims"], p["claim"])
elif event["type"] == "review":
p["artifact"] = self.pin(p.get("artifact"), overlays)
target = state["evidence"].get(p.get("evidence"))
require(target is not None, "unknown evidence to review")
require(not evidence_issues(self.root, state, target, overlays),
"cannot review stale evidence; record fresh evidence first")
elif event["type"] == "run-result":
require(isinstance(p.get("artifacts", []), list),
"run result artifacts must be a list")
p["artifacts"] = [self.pin(a, overlays) for a in p.get("artifacts", [])]
number = len(state["events"]) + 1
event.update(event_id=f"E{number:06d}",
created_at=datetime.now(timezone.utc).isoformat(),
previous=next(reversed(state["events"].values()))["sha256"] if state["events"] else None)
apply(state, event)
event["sha256"] = digest(event)
return event
def write_event(self, event):
# A complete file becomes visible atomically; writers serialize via the lock.
data = (json.dumps(event, indent=2, ensure_ascii=False) + "\n").encode("utf-8")
with staged_copy(self.base, data) as temp:
dest = self.events / f"{int(event['event_id'][1:]):06d}.json"
require(not dest.exists(), "event collision")
os.replace(temp, dest)
def prepare_many(self, proposals, state, overlays=None):
require(isinstance(proposals, list) and proposals, "batch must be a nonempty list")
json_text(proposals, "batch")
aliases, events = {}, []
def resolve(value):
if isinstance(value, dict) and set(value) == {"$event"}:
alias = value["$event"]
require(isinstance(alias, str) and alias in aliases,
f"unknown or forward batch event alias: {alias}")
return aliases[alias]
if isinstance(value, dict):
return {key: resolve(item) for key, item in value.items()}
if isinstance(value, list):
return [resolve(item) for item in value]
return value
for proposal in proposals:
require(isinstance(proposal, dict), "each batch proposal must be an object")
alias = proposal.get("alias")
require(alias is None or isinstance(alias, str) and
bool(re.fullmatch(r"[A-Za-z][A-Za-z0-9_-]{0,79}", alias)) and alias not in aliases,
"batch aliases must be unique identifiers")
clean = {key: value for key, value in proposal.items() if key != "alias"}
event = self.prepare_event(resolve(clean), state, overlays)
events.append(event)
if alias is not None:
aliases[alias] = event["event_id"]
return events
def append_many(self, events):
for event in events:
try:
self.write_event(event)
except (OSError, LedgerError) as exc:
raise LedgerError("batch append interrupted; a valid event prefix may remain; "
f"inspect the journal head before retrying: {exc}") from exc
def record_many(self, proposals, dry_run=False):
require(self.base.is_dir(), "ledger not initialized")
with self.lock():
events = self.prepare_many(proposals, self.read())
if not dry_run:
self.append_many(events)
return events
def deferred_path(self, name, label):
"""A path a deferred packet may name; hidden and agent instruction paths never qualify."""
path = inside(self.root, name)
relative = path.relative_to(self.root)
require(relative.parts, f"{label} must be below the project root")
# This also excludes .mathbox and case or trailing-dot aliases of it.
require(not any(part.startswith(".") for part in relative.parts),
f"{label} cannot contain hidden path components: {name}")
require(relative.name.casefold() not in INSTRUCTION_FILES,
f"{label} cannot be an agent instruction file: {name}")
return path, relative.as_posix()
def deferred_policy(self):
"""Artifact roots and index files that the local config opens to deferred packets."""
config = json.loads(inside(self.root, ".mathbox/config.json").read_text(encoding="utf-8"))
policy = config.get("deferred", {})
require(isinstance(policy, dict) and set(policy) <= {"artifact_roots", "index_files"},
"config deferred may contain only artifact_roots and index_files")
policy = {"artifact_roots": [], "index_files": [], **policy}
roots = [self.deferred_path(name, "deferred artifact root")[0]
for name in strings(policy, "artifact_roots")]
indexes = set()
for name in strings(policy, "index_files"):
path, key = self.deferred_path(name, "deferred index file")
require(path.suffix.casefold() in DEFERRED_SUFFIXES,
f"deferred index file must be a text document: {name}")
indexes.add(key)
return roots, indexes
def stage_deferred(self, packet, state):
"""Validate a packet's files against the local policy without writing anything."""
roots, indexes = self.deferred_policy()
artifacts = []
for item in packet["artifacts"]:
require(isinstance(item, dict) and set(item) == {"path", "content"},
"each deferred artifact needs path and content")
path, name = self.deferred_path(text_field(item, "path"), "deferred artifact")
require(roots, "deferred artifacts need deferred.artifact_roots in .mathbox/config.json")
require(any(path.is_relative_to(root) and path != root for root in roots),
f"deferred artifact is outside the configured artifact roots: {name}")
require(path.suffix.casefold() in DEFERRED_SUFFIXES,
f"deferred artifact must be a {', '.join(sorted(DEFERRED_SUFFIXES))} document: {name}")
require(isinstance(item["content"], str) and "\x00" not in item["content"],
"artifact content must be text without NUL")
require(all(name != other for _, other, _ in artifacts), f"duplicate artifact: {name}")
require(all(path not in other.parents and other not in path.parents
for other, _, _ in artifacts),
f"artifact path conflicts with another artifact: {name}")
require(not path.exists() and not path.is_symlink(),
f"artifact already exists: {name}")
require(all(not parent.exists() or parent.is_dir()
for parent in path.parents if parent.is_relative_to(self.root)),
f"artifact parent is not a directory: {name}")
artifacts.append((path, name, item["content"].encode("utf-8")))
request = packet["index_append"]
if request is None:
return artifacts, None
path, name = self.deferred_path(text_field(request, "path"), "index_append path")
require(name in indexes, f"index_append path is not a configured deferred index file: {name}")
require(path.is_file(), f"missing index: {name}")
old = path.read_bytes()
try:
current = old.decode("utf-8")
except UnicodeDecodeError as exc:
raise LedgerError(f"index is not UTF-8: {name}") from exc
tail, entry = request.get("expected_tail"), request.get("content")
require(not current or current.endswith("\n"), "index must end with a newline before appending")
# LF and CRLF compare equal: the inspected copy and a local checkout may differ.
last = current[current.rfind("\n", 0, len(current) - 1) + 1:]
require(isinstance(tail, str) and tail.replace("\r\n", "\n") == last.replace("\r\n", "\n"),
"index tail mismatch")
line = entry.removesuffix("\n").removesuffix("\r") if isinstance(entry, str) else ""
require(isinstance(entry, str) and entry.endswith("\n") and line.strip()
and not {"\n", "\r", "\x00"} & set(line),
"index content must be one nonempty newline-terminated entry")
new = old + (line + ("\r\n" if current.endswith("\r\n") else "\n")).encode("utf-8")
fresh = hashlib.sha256(old).hexdigest()
pinned = sorted({eid for eid, artifact in active_pins(state)
if pinned_path(artifact["path"]) == name and artifact["sha256"] == fresh})
require(not pinned, f"index_append would stale current pins of {name} "
f"({', '.join(pinned[:8])}); bind claims to a claim-scoped artifact instead")
return artifacts, (path, name, old, new)
def commit_deferred(self, artifacts, index, events):
"""Write staged files and events; undo the files if no event was written."""
created, made, index_written, recorded = [], [], False, 0
try:
for path, name, data in artifacts:
for parent in reversed(path.parents):
if parent.is_relative_to(self.root) and parent != self.root \
and not parent.exists():
parent.mkdir()
made.append(parent)
inside(self.root, name) # Reject a symlink swapped in since validation.
write_new(path, data)
created.append((path, data))
if index is not None:
index_path, index_name, old, new = index
inside(self.root, index_name)
require(index_path.read_bytes() == old, "index changed during ingest")
replace_file(index_path, new)
index_written = True
for event in events:
self.write_event(event)
recorded += 1
except (OSError, LedgerError) as exc:
if recorded:
raise LedgerError(f"deferred ingest interrupted after {recorded} of {len(events)} "
"events; its artifacts, index entry and that valid event prefix "
f"remain; inspect the journal head before retrying: {exc}") from exc
# Nothing refers to the new files yet, so remove those still holding our bytes.
kept = []
if index_written:
try:
if index_path.read_bytes() == new:
replace_file(index_path, old)
else:
kept.append(index_name)
except OSError:
kept.append(index_name)
for path, data in reversed(created):
try:
if path.read_bytes() == data:
path.unlink()
else:
kept.append(path.relative_to(self.root).as_posix())
except OSError:
kept.append(path.relative_to(self.root).as_posix())
for path in reversed(made):
with suppress(OSError):
path.rmdir()
cleanup = (f"left in place: {', '.join(kept)}; inspect before retrying" if kept else
"its files and index entry were removed")
raise LedgerError(f"deferred ingest recorded no events; {cleanup}: {exc}") from exc
def ingest(self, packet, dry_run=False):
"""Prevalidate a deferred packet, then commit it or undo its files."""
require(self.base.is_dir(), "ledger not initialized")
json_text(packet, "deferred packet")
require(isinstance(packet, dict) and set(packet) ==
{"format", "base", "artifacts", "index_append", "proposals"},
"deferred packet needs format, base, artifacts, index_append, proposals")
require(packet["format"] == "mathbox-deferred-v1", "unsupported deferred format")
base = packet["base"]
require(isinstance(base, dict) and set(base) == {"event_id", "event_sha256"},
"base needs event_id and event_sha256")
require(isinstance(packet["artifacts"], list), "artifacts must be a list")
index = packet["index_append"]
require(index is None or isinstance(index, dict) and
set(index) == {"path", "expected_tail", "content"},
"index_append must be null or have path, expected_tail, content")
def check_generated(value):
if isinstance(value, dict):
# An artifact reference may carry sha256; pin() then checks it.
generated = set(value) & GENERATED_FIELDS
if "sha256" in value and "path" not in value:
generated.add("sha256")
require(not generated, "deferred proposals must omit generated fields")
for child in value.values():
check_generated(child)
elif isinstance(value, list):
for child in value:
check_generated(child)
with self.lock():
state = self.read()
head = next(reversed(state["events"].values())) if state["events"] else None
require(base["event_id"] == (head["event_id"] if head else None) and
base["event_sha256"] == (head["sha256"] if head else None),
"stale deferred base: ledger head differs")
artifacts, index = self.stage_deferred(packet, state)
overlays = {name: data for _, name, data in artifacts}
if index is not None:
overlays[index[1]] = index[3]
check_generated(packet["proposals"])
events = self.prepare_many(packet["proposals"], state, overlays)
if not dry_run:
self.commit_deferred(artifacts, index, events)
return events
def record(self, proposal):
require(isinstance(proposal, dict) and set(proposal) == {"type", "actor", "payload"},
"proposal fields must be type, actor, payload")
return self.record_many([proposal])[0]
def artifact_issues(root, artifacts, overlays=None):
issues = []
for artifact in artifacts:
try:
if current_hash(root, artifact["path"], overlays)[2] != artifact["sha256"]:
issues.append(f"missing or changed artifact: {artifact['path']}")
except (OSError, LedgerError) as exc:
issues.append(str(exc))
return issues
def claim_contract_issues(root, claims, claim, overlays=None):
issues = []
for key in sorted(closure(claims, claim)):
artifact = claims[key].get("statement_artifact")
if artifact:
issues.extend(f"claim {key} statement contract: {issue}"
for issue in artifact_issues(root, [artifact], overlays))
return issues
def evidence_artifacts(payload):
artifacts = list(payload["artifacts"])
if payload.get("kind") == "computation" and "manifest" in payload:
artifacts.append(payload["manifest"])
artifacts.extend(payload["manifest_inputs"])
artifacts.extend(payload["manifest_outputs"])
return artifacts
def active_pins(state):
"""(event ID, artifact) for every pin a current claim, evidence, review or run result uses."""
inactive = state["retracted"] | state["superseded"]
for claim in state["claims"].values():
if "statement_artifact" in claim:
yield claim["event_id"], claim["statement_artifact"]
for eid, event in state["evidence"].items():
if eid not in inactive:
for artifact in evidence_artifacts(event["payload"]):
yield eid, artifact
for rid, event in state["reviews"].items():
if rid not in state["retracted"] and event["payload"]["evidence"] not in inactive:
yield rid, event["payload"]["artifact"]
for eid, event in state["run_results"].items():
for artifact in event["payload"].get("artifacts", []):
yield eid, artifact
def evidence_issues(root, state, event, overlays=None):
p = event["payload"]
issues = artifact_issues(root, evidence_artifacts(p), overlays)
issues.extend(claim_contract_issues(root, state["claims"], p["claim"], overlays))
if p["snapshot"] != snapshot(state["claims"], p["claim"]):
issues.append("claim or transitive dependency revision changed")
return issues
def execution_projection(root, state):
programs, runs, reconciliations, issues = {}, {}, [], []
for pid, event in state["programs"].items():
observations = [(eid, item) for eid, item in state["program_observations"].items()
if item["payload"]["program"] == pid]
terminals = [(eid, item) for eid, item in state["program_results"].items()
if item["payload"]["program"] == pid]
if terminals:
last_id, last = terminals[-1]
status = last["payload"]["outcome"]
terminal = dict(last["payload"], event_id=last_id,
created_at=last["created_at"])
elif observations:
last_id, last = observations[-1]
status, terminal = last["payload"]["state"], None
else:
last_id, last, status, terminal = event["event_id"], event, "active", None
programs[pid] = dict(event["payload"], event_id=event["event_id"], status=status,
last_observed={"event_id": last_id,
"created_at": last["created_at"]},
terminal=terminal)
for run, event in state["runs"].items():
observations = [(eid, item) for eid, item in state["run_observations"].items()
if item["payload"]["run"] == run]
terminals = [(eid, item) for eid, item in state["run_results"].items()
if item["payload"]["run"] == run]
result_issues = []
if terminals:
last_id, last = terminals[-1]
result_issues = artifact_issues(root, last["payload"].get("artifacts", []))
outcome = last["payload"]["outcome"]
status = "stale-result" if result_issues else outcome
terminal = dict(last["payload"], event_id=last_id,
created_at=last["created_at"])
issues.extend({"event": last_id, "issue": issue} for issue in result_issues)
elif observations:
last_id, last = observations[-1]
status, terminal = last["payload"]["state"], None
else:
last_id, last, status, terminal = event["event_id"], event, "active", None
route = state["routes"][event["payload"]["route"]]["payload"]
runs[run] = dict(event["payload"], event_id=event["event_id"], claim=route["claim"],
resolves=sorted(route_targets(route)),
status=status, last_observed={"event_id": last_id,
"created_at": last["created_at"]}, terminal=terminal,
issues=result_issues)
for eid, event in state["reconciliations"].items():
route = state["routes"][event["payload"]["route"]]["payload"]
reconciliations.append(dict(event["payload"], event_id=eid,
created_at=event["created_at"], claim=route["claim"],
resolves=sorted(route_targets(route))))
return {"programs": programs, "runs": runs,
"reconciliations": reconciliations, "issues": issues}
def project(root, state):
claims, evidence, reviews = state["claims"], {}, {}
for key, event in state["evidence"].items():
if key not in state["retracted"] and key not in state["superseded"]:
evidence[key] = dict(event["payload"], actor=event["actor"], issues=evidence_issues(root, state, event))
for key, event in state["reviews"].items():
if key not in state["retracted"] and event["payload"]["evidence"] in evidence:
reviews[key] = dict(event["payload"], event_id=key, actor=event["actor"],
created_at=event["created_at"],
issues=artifact_issues(root, [event["payload"]["artifact"]]))
resolved = SUPPORTING_STATUSES
# Losing a negative report cannot silently rehabilitate the challenged proof.
failed = {r["evidence"] for r in reviews.values() if r["outcome"] == "fail"}
uncertain = {r["evidence"] for r in reviews.values() if r["outcome"] == "conditional"}
result = {}
for key in topological(claims):
c = claims[key]
attached = {eid: e for eid, e in evidence.items() if e["claim"] == key}
live = {eid: e for eid, e in attached.items() if not e["issues"]}
attached_reviews = {rid: r for rid, r in reviews.items() if r["evidence"] in attached}
positive = {eid: e for eid, e in live.items() if e["kind"] in {"proof", "source"} and eid not in failed}
negative = {eid: e for eid, e in live.items() if e["kind"] == "counterexample" and eid not in failed}
computations = {eid: e for eid, e in live.items()
if e["kind"] == "computation" and eid not in failed}
blocked = [d for d in c["dependencies"] if result[d]["status"] not in resolved]
if positive and negative:
status = "disputed"
elif negative:
status = "counterexample-recorded" if any(eid not in uncertain for eid in negative) else "conditional"
elif positive:
unqualified = {eid: e for eid, e in positive.items() if eid not in uncertain}
status = "conditional" if blocked or not unqualified else ("proof-recorded" if any(e["kind"] == "proof" for e in unqualified.values()) else "source-recorded")
elif live and all(eid in failed for eid in live):
status = "incomplete"
elif computations:
status = "conditional" if blocked or not any(eid not in uncertain for eid in computations) else "computation-recorded"
elif attached and not live:
status = "stale"
else:
status = "conjectural"
independent = any(r["evidence"] in live and r["evidence"] not in uncertain
and r["evidence"] not in failed and r["independent"]
and r["outcome"] == "pass" and not r["issues"]
for r in reviews.values())
failed_evidence = {r["evidence"] for r in attached_reviews.values()
if r["outcome"] == "fail"}
passed_evidence = {r["evidence"] for r in attached_reviews.values()
if r["evidence"] in live and r["outcome"] == "pass"
and not r["issues"]}
has_conditional = any(r["outcome"] == "conditional"
for r in attached_reviews.values())
if failed_evidence & passed_evidence:
review_status = "conflicting-reviews-recorded"
elif failed_evidence:
review_status = "failed-review-recorded"
elif has_conditional:
review_status = "conditional-review-recorded"
elif independent:
review_status = "independent-pass-recorded"
else:
review_status = "no-independent-pass-recorded"
result[key] = dict(c, status=status, blocked_by=blocked,
review=review_status, reviews=attached_reviews,
evidence=attached)
issues = [{"event": key, "issue": issue} for key, e in evidence.items() for issue in e["issues"]]
issues += [{"event": key, "issue": issue} for key, r in reviews.items() for issue in r["issues"]]
for claim in claims.values():
if "statement_artifact" in claim:
issues += [{"event": claim["event_id"], "issue": issue}
for issue in artifact_issues(root, [claim["statement_artifact"]])]
executions = execution_projection(root, state)
issues += executions.pop("issues")
return {"claims": result, "route_context": {}, "issues": issues,
"events": len(state["events"]),
**executions}
def impact(claims, target):
require(target in claims, "unknown claim")
return sorted(key for key, deps in dependency_map(claims).items() if key != target and target in deps)
def next_routes(state, projection, goal=None):
claims = projection["claims"]
selected = closure(claims, goal) if goal else set(claims)
dependencies = dependency_map(claims)
closed = {r["payload"]["route"] for r in state["route_closures"].values()}
# The latest `continue` carries an open route's deferred or next step.
continuations = {}
for eid, event in state["reconciliations"].items():
p = event["payload"]
if p["decision"] == "continue":
continuations[p["route"]] = {"event_id": eid, "reason": p["reason"],
"next_question": p["next_question"]}
result = []
for rid, event in state["routes"].items():
p = event["payload"]
if rid in closed or not route_in_scope(p, selected):
continue
blocked = [key for key in p["prerequisites"]
if claims[key]["status"] not in SUPPORTING_STATUSES]
targets = route_targets(p)
reach = sum(1 for key in selected if key not in targets
and any(target in dependencies[key] for target in targets))
score = (p["gain"] + reach) / p["cost"]
result.append(dict(p, resolves=sorted(targets), ready=not blocked,
blocked_by=blocked, score=round(score, 3),
continuation=continuations.get(rid)))
return sorted(result, key=lambda r: (not r["ready"], -r["score"], r["id"]))
def closed_routes(state, goal=None):
selected = closure(state["claims"], goal) if goal else set(state["claims"])
result = []
for eid, event in state["route_closures"].items():
route = state["routes"][event["payload"]["route"]]
if route_in_scope(route["payload"], selected):
payload = event["payload"]
if event["type"] == "route-reconcile":
payload = {
"route": payload["route"], "outcome": payload["decision"],
"reason": payload["reason"], "next_question": payload["next_question"],
"run_results": payload["results"], "conflicts": payload["conflicts"],
}
result.append(dict(route["payload"], resolves=sorted(route_targets(route["payload"])),
event_id=route["event_id"],
result=dict(payload, event_id=eid)))
return result
def restrict(projection, state, selected):
"""Goal view: keep the selected claims and only the records reported under them."""
claims = {key: c for key, c in projection["claims"].items() if key in selected}
relevant_routes = [event["payload"] for event in state["routes"].values()
if route_in_scope(event["payload"], selected)]
context_ids = {route["claim"] for route in relevant_routes} - selected
prerequisite_roots = {key for route in relevant_routes
for key in route["prerequisites"]}
for key in prerequisite_roots:
context_ids |= closure(state["claims"], key)
context_ids -= selected
route_context = {key: c for key, c in projection["claims"].items()
if key in context_ids}
reported_claims = list(claims.values()) + list(route_context.values())
events = {c["event_id"] for c in reported_claims}
events |= {eid for c in reported_claims for eid in c["evidence"]}
events |= {rid for rid, r in state["reviews"].items() if r["payload"]["evidence"] in events}
runs = {key: value for key, value in projection["runs"].items()
if value["claim"] in selected or set(value["resolves"]) & selected}
programs = {key: value for key, value in projection["programs"].items()
if value["goal"] in selected
or any(run["program"] == key for run in runs.values())}
reconciliations = [value for value in projection["reconciliations"]
if value["claim"] in selected or set(value["resolves"]) & selected]
lifecycle_events = {value["event_id"] for value in programs.values()}
lifecycle_events |= {value["event_id"] for value in runs.values()}
lifecycle_events |= {value["event_id"] for value in reconciliations}
lifecycle_events |= {value["last_observed"]["event_id"] for value in programs.values()}
lifecycle_events |= {value["last_observed"]["event_id"] for value in runs.values()}
return dict(projection, claims=claims, route_context=route_context,
programs=programs, runs=runs,
reconciliations=reconciliations,
issues=[i for i in projection["issues"]
if i["event"] in events | lifecycle_events])
def run_attention(projection):
"""Runs a resuming agent must act on, with notes: live, stale, or unreconciled."""
reconciled = {result for item in projection.get("reconciliations", [])
for result in item["results"]}
attention = {}
for key, run in projection.get("runs", {}).items():
terminal = run["terminal"]
notes = [f"result {terminal['event_id']} unreconciled"] \
if terminal and terminal["event_id"] not in reconciled else []
if not terminal or notes or run["status"] == "stale-result":
attention[key] = notes
return dict(sorted(attention.items(), key=lambda item: (
projection["runs"][item[0]]["status"] != "stale-result", not item[1], item[0])))
def markdown(projection, routes=None, closed=None):
def cell(value):
return str(value).replace("|", "\\|").replace("\n", " ")
lines = ["# Research state", "", "Recorded evidence; this report does not certify mathematical correctness.", "",
"| Claim | Revision | Evidence status | Review | Blocked by |", "|---|---:|---|---|---|"]
for key, c in sorted(projection["claims"].items()):
lines.append(f"| {cell(key)} | {c['revision']} | {c['status']} | {c['review']} | {cell(', '.join(c['blocked_by']))} |")
def append_claim(key, c, heading="##"):
lines.extend(["", f"{heading} {key}", "", c["statement"], "",
f"Regime: {c['regime']}. Level: {c['level']}.",
"Hypotheses: " + ("; ".join(c["hypotheses"]) or "none recorded"),
f"Evidence status: {c['status']}. Review: {c['review']}. "
f"Blocked by: {', '.join(c['blocked_by']) or 'none'}."])
for eid, e in c["evidence"].items():
paths = list(dict.fromkeys(a["path"] for a in evidence_artifacts(e)))
lines.append(f"- {eid} ({e['kind']}): {e['summary']}; artifacts: " + ", ".join(paths))
for rid, review in c["reviews"].items():
independence = "independent" if review["independent"] else "non-independent"
lines.append(f"- Review {rid} ({review['outcome']}, {independence}) of "
f"{review['evidence']}: {review['summary']}; report: "
f"{review['artifact']['path']}")
for key, c in sorted(projection["claims"].items()):
append_claim(key, c)
if projection["route_context"]:
lines += ["", "## Route context", "",
"Claims needed to interpret relevant routes; these are not theorem dependencies of the goal."]
for key, c in sorted(projection["route_context"].items()):
append_claim(key, c, "###")
if projection["issues"]:
lines += ["", "## Stale records", ""] + [f"- {i['event']}: {i['issue']}" for i in projection["issues"]]
if projection["programs"]:
lines += ["", "## Program lifecycle", ""]
for p in projection["programs"].values():
lines.append(f"- {p['id']} ({p['status']}): goal {p['goal']}; base {p['base_event']} at {p['base_revision']}; last observed {p['last_observed']['event_id']}")
if projection["runs"]:
lines += ["", "## Route executions", ""]
attention = run_attention(projection)
for run in projection["runs"].values():
notes = "".join(f"; {note}" for note in attention.get(run["id"], []))
lines.append(f"- {run['id']} ({run['status']}): route {run['route']}; base {run['base_event']} at {run['base_revision']}; last observed {run['last_observed']['event_id']}{notes}")
if projection["reconciliations"]:
lines += ["", "## Reconciliations", ""]
for item in projection["reconciliations"]:
lines.append(f"- {item['event_id']}: route {item['route']}, {item['decision']}; run results {', '.join(item['results'])}; "
f"reason: {item['reason']}; next question: {item['next_question']}")
if routes is not None:
lines += ["", "## Candidate routes", "", "Scores order declared gain plus dependency reach per declared cost; they are not success probabilities."]
for r in routes:
lines.extend(["", f"- {r['id']} ({'ready' if r['ready'] else 'blocked'}, score {r['score']}): {r['question']}",
f" Resolves: {', '.join(sorted(route_targets(r)))}",
f" Decisive check: {r['discriminator']}"])
if r.get("continuation"):
c = r["continuation"]
lines.extend([f" Continuation {c['event_id']}: {c['next_question']}",
f" Continuation reason: {c['reason']}"])
if closed is not None:
lines += ["", "## Closed routes", ""]
for r in closed:
result = r["result"]
lines.extend([f"- {r['id']} (claim {r['claim']}, resolves {', '.join(sorted(route_targets(r)))}, route event {r['event_id']}): {r['mechanism']}",
f" Result {result['event_id']}: {result['outcome']}",
f" Reason / obstruction: {result['reason']}",
f" Next question: {result['next_question']}"])
return "\n".join(lines) + "\n"
def brief_markdown(projection, routes=None, closed=None):
def clip(text, limit=180):
return text if len(text) <= limit else text[:limit - 3] + "…"
claims = projection["claims"]
statuses = Counter(claim["status"] for claim in claims.values())
issues = projection["issues"]
lines = ["# Research state: brief view", "",
"Recorded evidence only; inspect the exact artifacts before a mathematical verdict.", "",
f"Events: {projection['events']}; selected claims: {len(claims)}; "
f"integrity/freshness issues: {len(issues)}.",
"Evidence states: " + (", ".join(f"{key} {count}" for key, count in sorted(statuses.items())) or "none") + "."]
candidates = sorted(claims.items(), key=lambda item: (item[1]["status"] in SUPPORTING_STATUSES, item[0]))
if candidates:
lines += ["", "## Claim sample", ""]
for key, claim in candidates[:10]:
blocked = claim["blocked_by"]
blockers = ", ".join(blocked[:6]) or "none"
if len(blocked) > 6:
blockers += f", … {len(blocked) - 6} more"
lines.append(f"- {key}: {claim['status']}; review {claim['review']}; "
f"blocked by {blockers}")
if len(candidates) > 10:
lines.append(f"- … {len(candidates) - 10} more claims; use --full or --json for details.")
active_reviews = [(key, rid, review) for key, claim in sorted(claims.items())
for rid, review in claim["reviews"].items()
if review["outcome"] != "pass" or review["issues"]]
if active_reviews:
lines += ["", f"## Review conditions ({len(active_reviews)})", ""]
for key, rid, review in active_reviews[:8]:
lines.append(f"- {rid} on {key}: {review['outcome']}; report {review['artifact']['path']}")
if len(active_reviews) > 8:
lines.append(f"- … {len(active_reviews) - 8} more review conditions; use --full or --json.")
if issues:
lines += ["", "## Issue sample", ""]
for issue in issues[:8]:
lines.append(f"- {issue['event']}: {clip(issue['issue'])}")
if len(issues) > 8:
lines.append(f"- … {len(issues) - 8} more issues; use --full or --json for all.")
programs, runs = projection.get("programs", {}), projection.get("runs", {})
reconciliations = projection.get("reconciliations", [])
attention = run_attention(projection)
if programs or runs or reconciliations:
def by_status(items):
counts = Counter(item["status"] for item in items.values())
detail = ", ".join(f"{key} {count}" for key, count in sorted(counts.items()))
return f"{len(items)} ({detail})" if items else "0"
lines += ["", "## Executions", "",
f"Programs: {by_status(programs)}; runs: {by_status(runs)}; "
f"reconciliations: {len(reconciliations)}; "
f"runs needing attention: {len(attention)}."]
live = [f"{key} ({program['status']})" for key, program in sorted(programs.items())
if program["terminal"] is None]
if live:
lines.append("Open programs: " + ", ".join(live[:8])
+ (f", … {len(live) - 8} more" if len(live) > 8 else "") + ".")
for key, notes in list(attention.items())[:8]:
run = runs[key]
lines.append(f"- {key}: " + "; ".join(
[run["status"], f"route {run['route']}", f"program {run['program']}", *notes]))
if len(attention) > 8:
lines.append(f"- … {len(attention) - 8} more runs needing attention; use --full or --json.")
route_runs = {}
for key in attention:
route_runs.setdefault(runs[key]["route"], []).append(f"{key} ({runs[key]['status']})")
if routes is not None:
lines += ["", f"## Open routes ({len(routes)})", ""]
for route in routes[:8]:
targets = sorted(route_targets(route))
resolves = ", ".join(targets[:6])
if len(targets) > 6:
resolves += f", … {len(targets) - 6} more"
active = route_runs.get(route["id"], [])
if active:
resolves += "; runs " + ", ".join(active[:3])
if len(active) > 3:
resolves += f", … {len(active) - 3} more"
lines.append(f"- {route['id']}: {'ready' if route['ready'] else 'blocked'}; "
f"resolves {resolves}")
if route.get("continuation"):
c = route["continuation"]
lines.append(f" Continue ({c['event_id']}): {clip(c['next_question'])}; "
f"reason: {clip(c['reason'])}")
if len(routes) > 8:
lines.append(f"- … {len(routes) - 8} more routes; use --full or --json for all.")
if closed is not None:
lines += ["", f"## Closed routes ({len(closed)})", ""]
for route in closed[-8:]:
result = route["result"]
lines.append(f"- {route['id']}: {result['outcome']} at {result['event_id']}")
if len(closed) > 8:
lines.append(f"- … {len(closed) - 8} earlier results; use --full or --json.")
if projection["route_context"]:
lines.append(f"Route-context claims: {len(projection['route_context'])}; use --full for their contracts.")
return "\n".join(lines) + "\n"
def pin_impact(state, path):
"""Active pins of one normalized path, found lexically without hashing artifacts."""
direct, evidence_ids, review_ids, result_ids = set(), set(), set(), set()
for eid, artifact in active_pins(state):
if pinned_path(artifact["path"]) != path:
continue
kind, p = state["events"][eid]["type"], state["events"][eid]["payload"]
if kind == "claim":
direct.add(p["id"])
elif kind == "evidence":
direct.add(p["claim"])
evidence_ids.add(eid)
elif kind == "review":
direct.add(state["evidence"][p["evidence"]]["payload"]["claim"])
review_ids.add(eid)
else:
# Run results are hash-checked too; a change makes the run stale-result.
result_ids.add(eid)
dependencies = dependency_map(state["claims"])
dependents = {key for key, closure_ids in dependencies.items()
if key not in direct and closure_ids & direct}
return {"path": path, "direct_claims": sorted(direct),
"dependent_claims": sorted(dependents),
"evidence_events": sorted(evidence_ids), "review_events": sorted(review_ids),
"run_result_events": sorted(result_ids),
"runs": sorted({state["run_results"][eid]["payload"]["run"] for eid in result_ids})}
def event_pins(event):
"""Artifacts an event pins, so a receipt can show what a dry-run would bind."""
p, kind = event["payload"], event["type"]
if kind == "claim":
return [p["statement_artifact"]] if "statement_artifact" in p else []
if kind == "evidence":
return evidence_artifacts(p)
if kind == "review":
return [p["artifact"]]
if kind == "run-result":
return p.get("artifacts", [])
return []
def receipt(event):
payload = event["payload"]
identity = next((payload[key] for key in
("id", "claim", "route", "evidence", "target", "run", "program")
if key in payload), None)
result = {"event_id": event["event_id"], "type": event["type"],
"subject": identity, "sha256": event["sha256"]}
pins = [{"path": item["path"], "sha256": item["sha256"]} for item in event_pins(event)]
if pins:
result["pins"] = pins[:8]
if len(pins) > 8:
result["pins_omitted"] = len(pins) - 8
return result
def batch_summary(events, dry_run):
return {"dry_run": dry_run, "count": len(events),
"first_event": events[0]["event_id"], "last_event": events[-1]["event_id"],
"types": dict(sorted(Counter(e["type"] for e in events).items())),
"sample": [receipt(e) for e in events[:8]],
"receipts_omitted": max(0, len(events) - 8)}
def unverified_pins(packet):
"""Existing files a deferred packet pins from local bytes without a supplied hash."""
def references(value):
if isinstance(value, dict):
if isinstance(value.get("path"), str) and "sha256" not in value:
yield pinned_path(value["path"])
for child in value.values():
yield from references(child)
elif isinstance(value, list):
for child in value:
yield from references(child)
created = {pinned_path(item["path"]) for item in packet["artifacts"]}
return sorted(set(references(packet["proposals"])) - created)
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--root", type=Path, default=Path.cwd())
parser.add_argument("--json", action="store_true")
commands = parser.add_subparsers(dest="command", required=True)
commands.add_parser("init", help="initialize a new ledger")
check = commands.add_parser("check", help="check freshness and integrity; brief by default")
check.add_argument("--summary", action="store_true")
check.add_argument("--full", action="store_true")
status = commands.add_parser("status", help="show recorded claim state; brief by default")
status.add_argument("--full", action="store_true")
record = commands.add_parser("record", help="append one proposal and print a receipt")
record.add_argument("proposal", type=Path)
batch = commands.add_parser("record-batch", help="prevalidate and append distinct proposals in one pass")
batch.add_argument("proposals", type=Path)
batch.add_argument("--dry-run", action="store_true")
ingest = commands.add_parser("ingest", help="validate and apply a deferred handoff packet")
ingest.add_argument("packet", help="JSON packet path, or - for stdin")
ingest.add_argument("--dry-run", action="store_true")
for name in ("next", "handoff"):
sub = commands.add_parser(name, help="show goal routes or a brief goal handoff")
sub.add_argument("--goal")
if name == "handoff":
sub.add_argument("--full", action="store_true")
sub = commands.add_parser("impact", help="list dependents of one claim")
sub.add_argument("claim")
sub = commands.add_parser("pin-impact", help="show claim and evidence pins of one project file")
sub.add_argument("path")
sub.add_argument("--full", action="store_true")
args = parser.parse_args(argv)
try:
ledger = Ledger(args.root)
if args.command == "init":
ledger.initialize()
print(json.dumps({"initialized": str(ledger.base)}))
return 0
if args.command == "record":
event = ledger.record(json.loads(args.proposal.read_text(encoding="utf-8")))
print(json.dumps(event if args.json else receipt(event), ensure_ascii=False))
return 0
if args.command == "record-batch":
events = ledger.record_many(json.loads(args.proposals.read_text(encoding="utf-8")),
dry_run=args.dry_run)
output = ({"dry_run": args.dry_run, "events": events} if args.json
else batch_summary(events, args.dry_run))
print(json.dumps(output, ensure_ascii=False))
return 0
if args.command == "ingest":
# Packets are UTF-8 whatever the locale; tolerate a byte-order mark.
raw = sys.stdin.buffer.read() if args.packet == "-" else Path(args.packet).read_bytes()
packet = json.loads(raw.decode("utf-8-sig"))
events = ledger.ingest(packet, dry_run=args.dry_run)
lists = {"artifacts": [pinned_path(a["path"]) for a in packet["artifacts"]],
"unverified_existing_pins": unverified_pins(packet)}
index = packet["index_append"]["path"] if packet["index_append"] else None
if args.json:
output = {"dry_run": args.dry_run, **lists, "index_append": index, "events": events}
else:
output = batch_summary(events, args.dry_run)
for key, value in lists.items():
output[key], output[key + "_omitted"] = value[:8], max(0, len(value) - 8)
output["index_append"] = index
print(json.dumps(output, ensure_ascii=False))
return 0
state = ledger.read()
# Dependency lookups need no artifact hashing, so they skip the projection.
if args.command == "impact":
output = {"claim": args.claim, "dependents": impact(state["claims"], args.claim)}
print(json.dumps(output, indent=2, ensure_ascii=False))
return 0
if args.command == "pin-impact":
path = inside(ledger.root, args.path).relative_to(ledger.root).as_posix()
output = pin_impact(state, path)
if args.json or args.full:
print(json.dumps(output, indent=2, ensure_ascii=False))
else:
lists = {key: value for key, value in output.items() if isinstance(value, list)}
print(json.dumps({"path": path,
"counts": {key: len(value) for key, value in lists.items()},
"samples": {key: value[:8] for key, value in lists.items()},
"detail_command": "--json pin-impact PATH"}, ensure_ascii=False))
return 0
projection = project(ledger.root, state)
routes = None
closed = None
if args.command in {"next", "handoff"}:
routes = next_routes(state, projection, args.goal)
if args.goal:
projection = restrict(projection, state, closure(state["claims"], args.goal))
if args.command == "handoff":
closed = closed_routes(state, args.goal)
output = {"state": projection, "routes": routes, "closed_routes": closed}
else:
output = routes
elif args.command == "check" and (args.summary or not (args.json or args.full)):
statuses = {}
reviews = {}
for claim in projection["claims"].values():
statuses[claim["status"]] = statuses.get(claim["status"], 0) + 1
reviews[claim["review"]] = reviews.get(claim["review"], 0) + 1
issues = projection["issues"]
shown = issues if args.full else issues[:8]
output = {"events": projection["events"], "claims": len(projection["claims"]),
"statuses": dict(sorted(statuses.items())),
"review_statuses": dict(sorted(reviews.items())),
"issue_count": len(issues), "issues": shown,
"issues_omitted": len(issues) - len(shown),
"detail_command": "--json check" if not args.full else None}
else:
output = projection
if args.json or args.command in {"next", "check"}:
print(json.dumps(output, indent=2, ensure_ascii=False))
elif getattr(args, "full", False):
print(markdown(projection, routes, closed), end="")
else:
print(brief_markdown(projection, routes, closed), end="")
return 1 if args.command == "check" and projection["issues"] else 0
except (LedgerError, OSError, ValueError, TypeError, KeyError) as exc:
print(f"research-state: {exc}", file=sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main())
SHA-256: 77c450aee94302de373431cf9cf5799c2a9309111058bd16acd0c324eaa6d7b9