# -*- coding: utf-8 -*-
"""
tpm_sandbox.py 0.2.0 --- a local UWC sandbox over a published TPC-PY
=====================================================================
NON-NORMATIVE. SANDBOX ONLY. NOT TPX. NOTHING HERE IS OF RECORD.
What it is: the thinnest runnable path through The Principia Machina. It plays
TPX's part on your own disk so that anyone, a person or an AI with a terminal,
can create a UWC and work its register before TPX exists. Every check it runs
is a function of TPC-PY, called unchanged; this file only sequences those
calls, stores the results and prints them. It runs only on a TPC-PY tree whose
digest it knows (KNOWN_TREES below) and says in every header whether that
version is declared or proposed.
Published on https://invarians.org as the page
"The Principia Machina/TPC-PY tools/tpm_sandbox.py"; how to use it:
"The Principia Machina/Hub/Outside AI".
Standard library only; Python 3.8 or later; Windows and Linux.
SET-UP: put this file in a folder that holds tpc/, the TPC-PY tree that
verify_tpc_py.py fetched and checked (or name another with --tpc DIR, before
the command). The UWCs it makes go in uwcs/ beside this file.
COMMANDS (on Windows use "py -3" in place of "python3")
python3 tpm_sandbox.py check DRAFT.json
the new-UWC gate's checks (TPC check_draft), nothing stored
python3 tpm_sandbox.py new DRAFT.json --prefix PFX [--initiator NAME]
runs the gate; on PASS writes the CANDIDATE certificate in uwcs/PFX/
python3 tpm_sandbox.py declare PFX --by NAME --confirm
THE HUMAN'S ACT (H-DECL's part), typed by a person, never by an AI.
Without --confirm it only prints what is being declared. With it: the
UWC is DECLARED, the register opens (MASTER at PFX-1, one NAMED_GAP per
IC2 or UNDER gap after it)
python3 tpm_sandbox.py push PFX RECORDS.json --block B [--account A]
[--kind WORK|EOM-0|REPRODUCTION]
lands a push: every record checked at PUSH, TPX's fields written,
checked again at LANDED, citations checked, promotions, EOM-0
settlement and SR3 screening run; all or nothing
python3 tpm_sandbox.py next-id PFX
python3 tpm_sandbox.py difference PFX VERIFICATION.json
computes a VERIFICATION's DIFFERENCE with TPC and prints the record
python3 tpm_sandbox.py compare PFX RID REPRODUCTION.json
compares a reproduction with a landed RESULT (lands nothing)
python3 tpm_sandbox.py withdraw PFX RID --account A
the author's own withdrawal (cause AUTHOR, WITHDRAWN only)
python3 tpm_sandbox.py status PFX
python3 tpm_sandbox.py show PFX RID
python3 tpm_sandbox.py verify PFX
the digest gate and the local chain, recomputed from the files
python3 tpm_sandbox.py report PFX
writes uwcs/PFX/REPORT.txt: everything about the UWC in one file
python3 tpm_sandbox.py example NAME
prints a worked example: draft (the spring draft from tpc/examples),
push_1, push_2, push_bad_id, verification_draft or reproduction
python3 tpm_sandbox.py selftest
runs the spring example end to end in a temporary folder
Every command prints one block from "=== TPM SANDBOX" to "=== END". Commands
on a UWC also save the block under uwcs/PFX/log/.
WHAT THE SANDBOX DOES NOT DO (by design; each is a TPX duty tested later):
wiki pages, the chain pages, leases and VOID lines, register volumes, the lock
and collaboration pages, CORROBORATION landing (use `compare`), strikes (SR3
contradictions are reported, never struck), holds, probes, canaries, falcons,
scores, the bibliography, ENVIRONMENT MODULES, the governor, accounts and
certification. Identifiers are taken in strict sequence from one mark; TPX's
own records (WITHDRAWAL, PROMOTION) take the next identifiers after a push's
records. These limits are not findings.
"""
import datetime
import hashlib
import json
import math
import os
import shutil
import sys
import tempfile
import uuid
sys.dont_write_bytecode = True # never write __pycache__ into the declared tree
SANDBOX_VERSION = "0.2.0"
HERE = os.path.dirname(os.path.abspath(__file__))
TPC_DIR = os.path.join(HERE, "tpc")
UWCS = os.path.join(HERE, "uwcs")
# The TPC-PY trees this sandbox runs on, by tree digest (TPC-PY page, "Versions").
# D_IS is tpc_digest d_is over the TPS version and D(TPS) with no other member.
KNOWN_TREES = {
"2998ac4c24dff83bd047a691081b7d27bb25c9d26512bffe519f5c2b6ba127e2": {
"TPC_PY": "0.2.0", "TPC_PY_STATUS": "declared 2026-09-30",
"TPS": "0.4.0", "TPS_STATUS": "declared 2026-09-30",
"D_TPS": "5a012a7378236044b88d543451bbe2e23a39ab48d8fec555937fad17d42a7e9c",
"D_IS": "2c22079d7395ed729f0e2468ee659a24830551dcb364516779c0f8364727a57e"},
"63ef0f2decb5b2c15c19274026b2f9a95160d0f54fee3f58e273088901f564f1": {
"TPC_PY": "0.3.0", "TPC_PY_STATUS": "PROPOSED, not declared on 2026-09-30",
"TPS": "0.5.0", "TPS_STATUS": "PROPOSED, not declared on 2026-09-30",
"D_TPS": "e45ef3a89f40c1335e2750dd7568e2be92b392717f90d23c917e0ee184688a0b",
"D_IS": "662a69d8a947ecb2c11caae2fc3f9de461d2e9b61e6a3fdaec02225b7935292e"},
}
PUSH_CLASSES = ("RESULT", "ASSUMPTION", "VERIFICATION", "ENTRY_STATUS")
KINDS = ("WORK", "EOM-0", "REPRODUCTION")
class Stop(Exception):
"""A refusal: the message is printed, nothing is written, exit 1."""
# ---- output ---------------------------------------------------------------------------
class Out:
def __init__(self, cmd, args):
self.lines = ["=== TPM SANDBOX %s | %s %s | %s ===" % (SANDBOX_VERSION, cmd, " ".join(args), now())]
def bound(self, b):
self.lines.append("(TPC-PY %s, %s; TPS %s, %s. Non-normative sandbox: nothing here is of record.)" % (
b["TPC_PY"], b["TPC_PY_STATUS"], b["TPS"], b["TPS_STATUS"]))
def __call__(self, s=""):
self.lines.extend(str(s).split("\n"))
def text(self, code):
return "\n".join(self.lines + ["=== END (exit %d) ===" % code]) + "\n"
def now():
return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def today():
return now()[:10]
def cj(v):
"""Canonical JSON for hashing and for small printed values."""
return json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
def sha256_bytes(b):
return hashlib.sha256(b).hexdigest()
def write_json(path, value):
tmp = path + ".tmp"
with open(tmp, "w", encoding="utf-8", newline="\n") as f:
f.write(json.dumps(value, indent=2, ensure_ascii=False) + "\n")
os.replace(tmp, path)
def read_json(path):
with open(path, "r", encoding="utf-8") as f:
return json.load(f, object_pairs_hook=_no_dupes)
# ---- the declared calculator -----------------------------------------------------------
def load_binding():
"""The versions this run is bound to, read from the tree digest of TPC_DIR."""
if not os.path.isdir(TPC_DIR):
raise Stop("TPC-PY not found at %s. Fetch it with verify_tpc_py.py --fetch into a folder named tpc "
"beside this file, or name the folder with --tpc DIR before the command." % TPC_DIR)
got = tree_digest()
b = KNOWN_TREES.get(got)
if b is None:
raise Stop("TPC-PY's tree digest here is %s, which this sandbox does not know (known: %s). Either "
"a file under %s was changed or added, or TPC-PY has a newer version than this sandbox; "
"fetch the tree again with verify_tpc_py.py, or use the sandbox published for that "
"version." % (got, ", ".join("%s (%s)" % (v["TPC_PY"], k[:16]) for k, v in
KNOWN_TREES.items()), TPC_DIR))
b = dict(b)
b["TPC_PY_TREE"] = got
return b
def tree_digest():
"""verify_tpc_py 0.3.0's tree digest over the files found under tpc/."""
fields = ["TPC-PY-TREE", "1"]
entries = []
for root, dirs, files in os.walk(TPC_DIR):
dirs[:] = [d for d in dirs if d != "__pycache__"]
for n in files:
if n.endswith(".pyc"):
continue
p = os.path.join(root, n)
rel = os.path.relpath(p, TPC_DIR).replace(os.sep, "/")
with open(p, "rb") as f:
entries.append((rel, sha256_bytes(f.read())))
h = hashlib.sha256()
for path, sha in sorted(entries):
fields += [path, sha]
for x in fields:
b = x.encode("utf-8")
h.update(str(len(b)).encode("ascii") + b":" + b)
return h.hexdigest()
def load_tpc(binding):
sys.path.insert(0, TPC_DIR)
import check_draft as CD # noqa
import tpc_zone as Z # noqa
import tpc_tests as K # noqa
import tpc_eom as E # noqa
import tpc_record as RC # noqa
import tpc_strike as S # noqa
import tpc_units as U # noqa
import tpc_probe as P # noqa
import tpc_digest as G # noqa
return dict(CD=CD, Z=Z, K=K, E=E, RC=RC, S=S, U=U, P=P, G=G)
# ---- the UWC folder ---------------------------------------------------------------------
def uwc_dir(prefix):
return os.path.join(UWCS, prefix)
def load_uwc(prefix):
d = uwc_dir(prefix)
if not os.path.isdir(d):
raise Stop("no UWC with prefix %s under %s" % (prefix, UWCS))
u = {"dir": d,
"draft": read_json(os.path.join(d, "draft.json")),
"cert": read_json(os.path.join(d, "certificate.json"))}
rp = os.path.join(d, "register.json")
u["register"] = read_json(rp) if os.path.exists(rp) else None
mp = os.path.join(d, "meta.json")
u["meta"] = read_json(mp) if os.path.exists(mp) else None
return u
def chain_lines(d):
p = os.path.join(d, "chain.jsonl")
if not os.path.exists(p):
return []
with open(p, "r", encoding="utf-8") as f:
return [json.loads(x) for x in f.read().splitlines() if x.strip()]
def chain_append(d, event, detail):
lines = chain_lines(d)
prev = lines[-1]["hash"] if lines else "GENESIS"
rp = os.path.join(d, "register.json")
reg_sha = sha256_bytes(open(rp, "rb").read()) if os.path.exists(rp) else ""
line = {"seq": len(lines) + 1, "time": now(), "event": event, "detail": detail,
"register_sha256": reg_sha, "prev": prev}
line["hash"] = sha256_bytes(cj(line).encode("utf-8"))
with open(os.path.join(d, "chain.jsonl"), "a", encoding="utf-8", newline="\n") as f:
f.write(cj(line) + "\n")
return line
def save_log(d, cmd, text):
ld = os.path.join(d, "log")
os.makedirs(ld, exist_ok=True)
n = len([x for x in os.listdir(ld) if x.endswith(".txt")]) + 1
with open(os.path.join(ld, "%04d_%s.txt" % (n, cmd)), "w", encoding="utf-8", newline="\n") as f:
f.write(text)
def rid(prefix, n):
return "%s-%d" % (prefix, n)
def rid_num(r):
return int(r.rsplit("-", 1)[1])
def keyed(register):
return [r for r in register if isinstance(r, dict)]
def problems_text(res):
"""A TPC ["REFUSED", problems] as lines."""
if res[0] == "REFUSED" and isinstance(res[1], list):
return [" %s %s: %s" % tuple(p) if isinstance(p, list) and len(p) == 3 else " %s" % (p,)
for p in res[1]]
return [" %s" % (res,)]
# ---- the gate ---------------------------------------------------------------------------
def gate(T, draft, out):
"""check_draft's check, then the parts the certificate needs."""
if "MODULES_USED" in draft:
out("DRAFT: MODULES_USED is not read by the sandbox at 0.1.0: no ENVIRONMENT MODULE is declared yet "
"(the interview's answer is NONE). Remove MODULES_USED and write the entries yourself.")
return None
ok = T["CD"].check(draft, out)
if not ok:
return None
Z, K, E = T["Z"], T["K"], T["E"]
fin = K.final_checks(draft)
ic2 = sorted(g.split("DOF ", 1)[1].rsplit(" is untied", 1)[0] for g in fin[2] if g.startswith("IC2: DOF "))
f = E.eom_findings(draft)
c = E.eom_class(draft)
params = draft["TMS"].get("PARAMETERS", {})
sc = params.get("STRUCTURE_CHECK", "ON")
under = sorted(fd[1] for fd in (f[1] if f[0] == "OK" else []) if fd[0] == "UNDER") if sc == "ON" else []
line = "STRUCTURE " + c[1] + ("" if len(c) == 2 else ": " + ", ".join(c[2]))
owed = list(T["CD"].OWED_INSPECTIONS)
findings = [fd[0] + ": " + fd[-1] for fd in (f[1] if f[0] == "OK" else [])] if sc == "ON" else []
return {"D_TPK": Z.d_tpk(draft["TPK"]), "D_TRM": Z.d_trm(draft["TrM"]), "D_TMS": Z.d_tms(draft["TMS"]),
"D_TFK": Z.d_tfk_of(draft), "IC2": ic2, "UNDER": under, "STRUCTURE_LINE": line,
"STRUCTURE_FINDINGS": findings, "INSPECTIONS_OWED": owed}
# ---- commands ---------------------------------------------------------------------------
def cmd_check(T, B, args, out):
if len(args) != 1:
raise Stop("usage: check DRAFT.json")
draft = T["CD"].load_draft(args[0])
g = gate(T, draft, out)
out("")
out("SANDBOX GATE: " + ("PASS" if g else "REFUSED"))
return 0 if g else 1
def cmd_new(T, B, args, out):
opts, pos = parse(args, {"--prefix": 1, "--initiator": 1})
if len(pos) != 1 or "--prefix" not in opts:
raise Stop("usage: new DRAFT.json --prefix PFX [--initiator NAME]")
prefix = opts["--prefix"]
if not T["RC"]._m(T["RC"]._IDENT, prefix) or "-" in prefix:
raise Stop("prefix %r is not an identifier ([A-Za-z_][A-Za-z0-9_]*)" % prefix)
if os.path.exists(uwc_dir(prefix)):
raise Stop("prefix %s is already used in this sandbox (uwcs/%s exists); prefixes are unique" % (prefix, prefix))
raw = open(pos[0], "rb").read()
draft = T["CD"].load_draft(pos[0])
g = gate(T, draft, out)
if not g:
out("")
out("SANDBOX GATE: REFUSED. No UWC was created. Fix the draft and run new again.")
return 1
cert = {"STATE": "CANDIDATE", "SANDBOX": "NOT OF RECORD: a local sandbox; no TPX, no chain page, no wiki",
"UUID": str(uuid.uuid4()), "PREFIX": prefix,
"D_TPK": g["D_TPK"], "D_TRM": g["D_TRM"], "D_TMS": g["D_TMS"], "D_TFK": g["D_TFK"],
"STRUCTURE_LINE": g["STRUCTURE_LINE"], "STRUCTURE_FINDINGS": g["STRUCTURE_FINDINGS"],
"IC2_GAPS": g["IC2"], "UNDER_GAPS": g["UNDER"],
"MODULES_USED": "NONE", "INTERVIEW": "NONE (no ENVIRONMENT MODULE is declared)",
"ACCESS_POLICY": "CLOSED (sandbox)", "AUDIT_TRIGGER": "ON REQUEST (sandbox)",
"BIBLIOGRAPHY_SETTING": "OFF", "INITIATOR": opts.get("--initiator", "unnamed"),
"CREATED": now(), "TPS": B["TPS"], "D_TPS": B["D_TPS"], "D_IS": B["D_IS"],
"TPC_PY": B["TPC_PY"], "TPC_PY_TREE": B["TPC_PY_TREE"],
"INSPECTIONS_OWED": g["INSPECTIONS_OWED"],
"CLOSED_WORLD_QUESTION": "Is every interaction not written in the TPK absent from the system modelled?"}
d = uwc_dir(prefix)
os.makedirs(os.path.join(d, "pushes"))
with open(os.path.join(d, "draft.json"), "wb") as f:
f.write(raw)
write_json(os.path.join(d, "certificate.json"), cert)
chain_append(d, "NEW", {"prefix": prefix, "uuid": cert["UUID"], "d_tfk": cert["D_TFK"],
"draft_sha256": sha256_bytes(raw)})
out("")
out("CANDIDATE CERTIFICATE written: uwcs/%s/certificate.json" % prefix)
out(" UUID %s PREFIX %s" % (cert["UUID"], prefix))
out(" D(TFK) %s" % cert["D_TFK"])
out(" %s" % cert["STRUCTURE_LINE"])
out(" named gaps at declaration: IC2 %s; UNDER %s" % (g["IC2"] or "none", g["UNDER"] or "none"))
out("NEXT: the human reads the certificate and declares it (only the human runs this):")
out(" py -3 tpm_sandbox.py declare %s --by <your name> --confirm" % prefix)
return 0
def cmd_declare(T, B, args, out):
opts, pos = parse(args, {"--by": 1, "--confirm": 0})
if len(pos) != 1 or "--by" not in opts:
raise Stop("usage: declare PFX --by NAME --confirm")
u = load_uwc(pos[0])
cert, d = u["cert"], u["dir"]
if cert["STATE"] != "CANDIDATE":
raise Stop("UWC %s is %s, not CANDIDATE" % (pos[0], cert["STATE"]))
draft = u["draft"]
if T["Z"].d_tfk_of(draft) != cert["D_TFK"]:
raise Stop("DIGEST GATE: draft.json no longer digests to the candidate's D(TFK); nothing declared")
out("DECLARING UWC %s (UUID %s)" % (cert["PREFIX"], cert["UUID"]))
out(" D(TFK) %s" % cert["D_TFK"])
out(" %s" % cert["STRUCTURE_LINE"])
for fd in cert["STRUCTURE_FINDINGS"]:
out(" finding: " + fd)
out(" inspections owed to the declaring authority:")
for i in cert["INSPECTIONS_OWED"]:
out(" - " + i)
out(" closed-world question: " + cert["CLOSED_WORLD_QUESTION"])
if "--confirm" not in opts:
out("")
out("NOT DECLARED: add --confirm to declare, after reading the lines above.")
return 1
stamp = now()
cert.update(STATE="DECLARED", DECLARED_BY=opts["--by"], DECLARED=stamp)
RC = T["RC"]
prefix = cert["PREFIX"]
master = {"ID": rid(prefix, 1), "CLASS": "MASTER", "GRADE": "NOT GRADED", "LANDED_BY": "TPX", "LANDED": stamp,
"TPS": B["TPS"], "UUID": cert["UUID"], "PREFIX": prefix,
"D_TPK": cert["D_TPK"], "D_TRM": cert["D_TRM"], "D_TMS": cert["D_TMS"], "D_TFK": cert["D_TFK"],
"CERTIFICATE": "SANDBOX certificate %s declared %s by %s" % (cert["UUID"], stamp[:10], opts["--by"]),
"STRUCTURE_LINE": cert["STRUCTURE_LINE"], "MODULES_USED": "NONE"}
dg = RC.record_digest(master)
if dg[0] != "OK":
raise Stop("MASTER not digestible:\n" + "\n".join(problems_text(dg)))
master["DIGEST"] = dg[1]
chk = RC.record_check(master, draft, "OFF", "LANDED")
if chk[0] != "OK":
raise Stop("MASTER refused at LANDED:\n" + "\n".join(problems_text(chk)))
trig = "the certificate %s, %s" % (cert["UUID"], cert["STRUCTURE_LINE"])
gaps = RC.gap_records(prefix, cert["IC2_GAPS"], cert["UNDER_GAPS"], trig, stamp, B["TPS"])
if gaps[0] != "OK":
raise Stop("gap records refused:\n" + "\n".join(problems_text(gaps)))
register = [master] + gaps[1]
meta = {"mark": 1 + len(gaps[1]), "pushes": 0, "records": {}, "setting": "OFF"}
for r in register:
meta["records"][r["ID"]] = {"push": 0, "block": "INSTANTIATION"}
write_json(os.path.join(d, "register.json"), register)
write_json(os.path.join(d, "meta.json"), meta)
write_json(os.path.join(d, "certificate.json"), cert)
chain_append(d, "DECLARE", {"by": opts["--by"], "master_digest": master["DIGEST"],
"gaps": [g["ID"] for g in gaps[1]]})
out("")
out("DECLARED. Register opened: %s (MASTER)%s" % (master["ID"], "".join(
", %s (NAMED_GAP %s %s)" % (g["ID"], g["SOURCE"], g["AGAINST"][0]) for g in gaps[1])))
if cert["UNDER_GAPS"]:
out("EOM-0 is OPEN: only pushes with --kind EOM-0 or --kind REPRODUCTION land until each UNDER gap settles.")
out("NEXT FREE IDENTIFIER: %s" % rid(prefix, meta["mark"] + 1))
return 0
def _declared(u):
if u["cert"]["STATE"] != "DECLARED" or u["register"] is None:
raise Stop("UWC %s is not DECLARED; the human declares it first" % u["cert"]["PREFIX"])
def digest_gate(T, u):
"""Before every write: the draft digests to the certificate's D(TFK) and the MASTER carries it."""
d = T["Z"].d_tfk_of(u["draft"])
m = u["register"][0]
if d != u["cert"]["D_TFK"] or m.get("D_TFK") != d:
raise Stop("DIGEST GATE FAILED: D(TFK) of draft.json is %s; certificate %s; MASTER %s. Writes halt."
% (d, u["cert"]["D_TFK"], m.get("D_TFK")))
def words_of(T, register):
w = T["RC"].status_words(register)
if w[0] != "OK":
raise Stop("register unreadable:\n" + "\n".join(problems_text(w)))
return w[1]
def load_push(T, path):
v = T["CD"]._load_json(path)
naming = []
if isinstance(v, dict):
extra = sorted(k for k in v if k not in ("RECORDS", "EOM0_SETTLEMENT"))
if extra or "RECORDS" not in v:
raise Stop("a push file is a JSON list of records, or an object with RECORDS "
"(and optionally EOM0_SETTLEMENT); found keys %s" % sorted(v))
naming = v.get("EOM0_SETTLEMENT", [])
v = v["RECORDS"]
if not isinstance(v, list) or not v or not all(isinstance(r, dict) for r in v):
raise Stop("a push holds a non-empty list of record objects")
return v, naming
def cmd_push(T, B, args, out):
opts, pos = parse(args, {"--block": 1, "--account": 1, "--kind": 1})
if len(pos) != 2 or "--block" not in opts:
raise Stop("usage: push PFX RECORDS.json --block B [--account A] [--kind WORK|EOM-0|REPRODUCTION]")
kind = opts.get("--kind", "WORK")
if kind not in KINDS:
raise Stop("--kind %s is not one of %s" % (kind, ", ".join(KINDS)))
account = opts.get("--account", "tester")
block = opts["--block"]
u = load_uwc(pos[0])
_declared(u)
digest_gate(T, u)
RC, S = T["RC"], T["S"]
draft, cert, meta = u["draft"], u["cert"], u["meta"]
prefix = cert["PREFIX"]
register = list(u["register"])
raw = open(pos[1], "rb").read()
records, naming = load_push(T, pos[1])
eo = RC.eom0_open(register)
if eo[0] == "OK" and eo[1] and kind not in ("EOM-0", "REPRODUCTION"):
raise Stop("EOM-0 OPEN: this UWC lands only blocks of kind EOM-0 or REPRODUCTION until its UNDER gaps "
"settle (TPS 3.3.16). Re-run with --kind EOM-0 if this push settles them.")
stamp = now()
push_no = meta["pushes"] + 1
mark = meta["mark"]
probs, landed, insp_notes = [], [], []
for i, rec in enumerate(records, 1):
tag = "record %d (%s)" % (i, rec.get("ID", "no ID"))
cls = rec.get("CLASS")
if cls == "CORROBORATION":
probs.append(tag + ": CORROBORATION is not landed by the sandbox at 0.1.0 (TPX writes OUTCOME, "
"DIFFERENCE, INDEPENDENCE and TARGET_STATUS after its own comparison); run compare")
continue
chk = RC.record_check(rec, draft, "OFF", "PUSH")
if chk[0] != "OK":
probs.append(tag + ": refused at PUSH")
probs.extend(problems_text(chk))
continue
if cls not in PUSH_CLASSES:
probs.append(tag + ": CLASS %s is not pushed" % cls)
continue
form_probs = []
if cls in ("RESULT", "ASSUMPTION"):
fc = RC.record_form_check(rec, draft) # parse, names, T4, D variable, ROLE (TPS 3.2.x)
if fc[0] != "OK":
form_probs += problems_text(fc)
if "DOMAIN_CONDITION" in rec:
dc = RC.domain_check(rec, draft) # T6 totality and dimensions of DOMAIN_CONDITION
if dc[0] != "OK":
form_probs += problems_text(dc)
if cls == "RESULT" and rec.get("REGIME") == "M":
bc = RC.budget_check(rec) # U_C against its BUDGET at the last written digit
if bc[0] == "REFUSED":
form_probs += problems_text(bc)
elif bc[0] in ("INSPECT", "CONVENTION"):
insp_notes.append("%s: BUDGET %s: %s" % (rec.get("ID"), bc[0], bc[1]))
if form_probs:
probs.append(tag + ": refused by the declared-form, domain or budget check")
probs.extend(form_probs)
continue
want = rid(prefix, mark + 1)
if rec["ID"] != want:
probs.append(tag + ": ID must be the next free identifier %s (identifiers are taken in sequence; "
"run next-id)" % want)
continue
mark += 1
L = dict(rec)
L["GRADE"] = RC.T["landgrade"][cls]
if RC.T["statuses"].get(cls):
L["STATUS"] = "UNMARKED"
L.update(BLOCK=block, AUTHOR=account, LANDED_BY=account, LANDED=stamp, TPS=B["TPS"], D_IS=B["D_IS"])
cc = RC.citation_check(L, register, [], False)
if cc[0] != "OK":
probs.append(tag + ": citation check refused")
probs.extend(problems_text(cc))
continue
if cls == "VERIFICATION":
tgt = next((r for r in keyed(register) if r["ID"] == L.get("TARGET")), None)
if tgt is None:
probs.append(tag + ": TARGET %s is not a landed record (a VERIFICATION follows its TARGET's push)"
% L.get("TARGET"))
continue
vo = RC.verification_order(meta["records"][tgt["ID"]]["push"], push_no, None)
if vo[0] not in ("OK", "INSPECT"):
probs.append(tag + ": order refused")
probs.extend(problems_text(vo))
continue
vd = RC.verification_difference(tgt, L, draft["TMS"]["PARAMETERS"]["K"])
if vd[0] != "OK":
probs.append(tag + ": DIFFERENCE refused (compute it with: difference %s FILE)" % prefix)
probs.extend(problems_text(vd) if vd[0] == "REFUSED" else [" " + str(vd)])
continue
dg = RC.record_digest(L)
if dg[0] != "OK":
probs.append(tag + ": not digestible")
probs.extend(problems_text(dg))
continue
L["DIGEST"] = dg[1]
chk2 = RC.record_check(L, draft, "OFF", "LANDED")
if chk2[0] != "OK":
probs.append(tag + ": refused at LANDED")
probs.extend(problems_text(chk2))
continue
register.append(L)
landed.append(L)
meta["records"][L["ID"]] = {"push": push_no, "block": block}
if probs:
out("PUSH REFUSED. Nothing landed; the register is unchanged. Problems:")
for p in probs:
out(" " + p)
out("NEXT FREE IDENTIFIER: %s" % rid(prefix, meta["mark"] + 1))
return 1
notes = ["INSPECTION " + n for n in insp_notes]
# supersessions: one WITHDRAWAL per superseded record, TPX-written, WITHDRAWN
for L in landed:
old = L.get("SUPERSEDES")
if not old:
continue
words = words_of(T, register)
sc = RC.withdrawal_scope(register, old, [])
if sc[0] != "OK":
raise Stop("withdrawal scope refused:\n" + "\n".join(problems_text(sc)))
cascade = [c for c in sc[1] if c[0] != L["ID"]]
act = {"cause": "SUPERSEDED", "withdrawn": old, "withdrawn_by": account, "status_set": "WITHDRAWN",
"prior_status": words[old], "superseded_by": L["ID"], "locus_gap": "", "still_stands": [],
"cascade": cascade, "tps_version": B["TPS"], "d_tms": cert["D_TMS"]}
mark += 1
wr = RC.withdrawal_record(rid(prefix, mark), act)
if wr[0] != "OK":
raise Stop("withdrawal record refused:\n" + "\n".join(problems_text(wr)))
register.append(wr[1])
meta["records"][wr[1][0]] = {"push": push_no, "block": block}
notes.append("WITHDRAWAL %s: %s SUPERSEDED by %s, WITHDRAWN; cascade %s"
% (wr[1][0], old, L["ID"], [c[0] for c in cascade] or "none"))
# EOM-0 settlement
settled = settle_eom0(T, draft, register, naming, stamp, notes)
# promotion
words = words_of(T, register)
pp = RC.promotion_pass(register, words, draft, [])
if pp[0] != "OK":
raise Stop("promotion pass refused:\n" + "\n".join(problems_text(pp)))
if pp[1]:
mark += 1
pr = RC.promotion_record(rid(prefix, mark), block, pp[1])
if pr[0] != "OK":
raise Stop("promotion record refused:\n" + "\n".join(problems_text(pr)))
register.append(pr[1])
meta["records"][pr[1][0]] = {"push": push_no, "block": block}
notes.append("PROMOTION %s: %s ESTABLISHED" % (pr[1][0], ", ".join(t[0] for t in pp[1])))
meta["mark"] = mark
meta["pushes"] = push_no
d = u["dir"]
with open(os.path.join(d, "pushes", "%04d_%s.json" % (push_no, safe(block))), "wb") as f:
f.write(raw)
write_json(os.path.join(d, "register.json"), register)
write_json(os.path.join(d, "meta.json"), meta)
chain_append(d, "PUSH", {"push": push_no, "block": block, "account": account, "kind": kind,
"landed": [L["ID"] for L in landed], "push_sha256": sha256_bytes(raw)})
out("PUSH %d LANDED (block %s, account %s, kind %s):" % (push_no, block, account, kind))
for L in landed:
out(" %s %s %s DIGEST %s" % (L["ID"], L["CLASS"], L["GRADE"], L["DIGEST"][:16]))
for n in notes:
out(" " + n)
screen(T, draft, register, out)
out("NEXT FREE IDENTIFIER: %s" % rid(prefix, mark + 1))
return 0
def safe(s):
return "".join(ch if ch.isalnum() or ch in "-_" else "_" for ch in s)[:40] or "block"
def settle_eom0(T, draft, register, naming, stamp, notes):
RC, E = T["RC"], T["E"]
eo = RC.eom0_open(register)
if eo[0] != "OK" or not eo[1]:
if naming:
raise Stop("EOM0_SETTLEMENT given but EOM-0 is not OPEN")
return False
words = words_of(T, register)
st = E.settlement_check(draft, keyed(register), words, naming)
if st[0] != "OK":
raise Stop("EOM-0 settlement refused:\n" + "\n".join(problems_text(st)))
res = st[1]
by = {}
for unk in res["BY_TEST"]:
by[unk] = sorted(res["USED"]) or ["-"]
for unk, r in res["BY_NAMING"]:
by.setdefault(unk, []).append(r)
for g in keyed(register):
if g["CLASS"] == "NAMED_GAP" and g["SOURCE"] == "UNDER" and g["STATE"] == "OPEN" and g["AGAINST"][0] in by:
ids = [x for x in by[g["AGAINST"][0]] if x != "-"]
gs = RC.gap_state("SETTLED", stamp[:10], sorted(set(ids)))
if gs[0] != "OK":
raise Stop("gap state refused:\n" + "\n".join(problems_text(gs)))
g["STATE"] = gs[1]
notes.append("NAMED_GAP %s (UNDER %s): %s" % (g["ID"], g["AGAINST"][0], gs[1]))
eo2 = RC.eom0_open(register)
if eo2[0] == "OK" and not eo2[1]:
notes.append("EOM-0 CLOSED")
return True
def screen(T, draft, register, out):
"""SR3 over every TrM entry: reported, never struck (sandbox)."""
S, U, RC = T["S"], T["U"], T["RC"]
words = words_of(T, register)
recs = keyed(register)
k = float(draft["TMS"]["PARAMETERS"]["K"])
byid = dict((r["ID"], r) for r in recs)
found = []
for e in draft["TrM"]["ENTRIES"]:
q = e["NAME"]
cs = S.comparison_set(recs, words, draft, q)
if cs[0] != "OK":
found.append("SR3 %s: not computed: %s" % (q, cs[1:]))
continue
if not cs[1]:
continue
groups = S.sr8_groups(recs, words, draft, q)
for a, b, mode in cs[1]:
if mode != "TEST":
found.append("SR3 %s: %s and %s by INSPECTION (%s)" % (q, a, b, mode))
continue
ra, rb = byid[a], byid[b]
def norm(r):
return U.normalize_measurement(r["VALUE"], r.get("U_C", "0") if r.get("REGIME") == "M" else "0",
r["UNIT"])
na, nb = norm(ra), norm(rb)
if na[0] != "OK" or nb[0] != "OK":
found.append("SR3 %s: %s and %s not normalized" % (q, a, b))
continue
sf = 1.0
if groups[0] == "OK":
for gm, ids in groups[1]:
if a in ids and b in ids and len(ids) >= 2:
ms = []
for x in ids:
nx = norm(byid[x])
ms.append({"x": nx[1], "u": nx[2]})
sf = S.sr8_scale_factor(ms)
if ra.get("REGIME") == "L" or rb.get("REGIME") == "L":
l_, m_ = (na, nb) if ra.get("REGIME") == "L" else (nb, na)
c = S.lm_contradicts(l_[1], m_[1], m_[2], k, sf)
else:
c = S.mm_contradicts(na[1], na[2], nb[1], nb[2], k, sf)
if c:
found.append("SR3 CONTRADICTION on %s: %s and %s (k %s, S %.4g). The sandbox reports it; TPX "
"would apply SR5 or SR7. Not struck." % (q, a, b, k, sf))
if found:
out("SCREENING:")
for f in found:
out(" " + f)
def cmd_next_id(T, B, args, out):
if len(args) != 1:
raise Stop("usage: next-id PFX")
u = load_uwc(args[0])
_declared(u)
out("NEXT FREE IDENTIFIER: %s" % rid(u["cert"]["PREFIX"], u["meta"]["mark"] + 1))
return 0
def cmd_difference(T, B, args, out):
if len(args) != 2:
raise Stop("usage: difference PFX VERIFICATION.json")
u = load_uwc(args[0])
_declared(u)
RC, U = T["RC"], T["U"]
v = T["CD"]._load_json(args[1])
if not isinstance(v, dict) or v.get("CLASS") != "VERIFICATION":
raise Stop("the file holds one VERIFICATION record object")
tgt = next((r for r in keyed(u["register"]) if r["ID"] == v.get("TARGET")), None)
if tgt is None:
raise Stop("TARGET %s is not a landed record" % v.get("TARGET"))
ext = v.get("EXTERNAL", {})
nt = U.normalize_measurement(tgt.get("VALUE"), tgt.get("U_C", "0") if tgt.get("REGIME") == "M" else "0",
tgt.get("UNIT"))
ne = U.normalize_measurement(ext.get("VALUE"), ext.get("U_C"), ext.get("UNIT"))
if nt[0] != "OK" or ne[0] != "OK":
raise Stop("values not normalized: target %s; external %s" % (nt, ne))
val = nt[1] - ne[1]
uu = math.sqrt(nt[2] * nt[2] + ne[2] * ne[2])
if uu == 0.0:
raise Stop("U is 0: both uncertainties are 0 (REC-95)")
v["DIFFERENCE"] = {"VALUE": RC._cd(val), "U": RC._cd(uu), "ZETA": RC._cd(val / uu)}
vd = RC.verification_difference(tgt, v, u["draft"]["TMS"]["PARAMETERS"]["K"])
if vd[0] != "OK":
out("TPC refused the verification:")
for p in problems_text(vd) if vd[0] == "REFUSED" else [str(vd)]:
out(p)
return 1
out("DIFFERENCE computed by TPC-PY (target minus external, coherent SI): %s" % cj(v["DIFFERENCE"]))
out("SR3 test between TARGET and EXTERNAL at k %s: %s" % (u["draft"]["TMS"]["PARAMETERS"]["K"],
"CONTRADICTS" if vd[2] else "does not contradict"))
out("The record with its DIFFERENCE, ready to push (after its TARGET's push):")
out(json.dumps(v, indent=2, ensure_ascii=False))
return 0
def cmd_compare(T, B, args, out):
if len(args) != 3:
raise Stop("usage: compare PFX RID REPRODUCTION.json")
u = load_uwc(args[0])
_declared(u)
RC, P, S, U = T["RC"], T["P"], T["S"], T["U"]
orig = next((r for r in keyed(u["register"]) if r["ID"] == args[1]), None)
if orig is None or orig.get("CLASS") != "RESULT":
raise Stop("%s is not a landed RESULT" % args[1])
rep = T["CD"]._load_json(args[2])
if not isinstance(rep, dict) or rep.get("CLASS") != "RESULT":
raise Stop("the reproduction is one RESULT record object")
if orig.get("REGIME") == "L":
lp = RC.l_finding_pair(orig, rep)
if lp[0] != "OK":
out("NOT COMPARED: " + " ".join(problems_text(lp) if lp[0] == "REFUSED" else [str(lp)]))
return 1
c = P.compare("L", lp[1], lp[2], None, None, None)
out("L COMPARISON over %s: %s" % (lp[1]["finding"]["type"], c["outcome"]))
out(" original %s" % c["original"])
out(" reproduction %s" % c["reproduction"])
else:
a = U.normalize_measurement(orig["VALUE"], orig["U_C"], orig["UNIT"])
b = U.normalize_measurement(rep.get("VALUE"), rep.get("U_C"), rep.get("UNIT"))
if a[0] != "OK" or b[0] != "OK" or a[3] != b[3]:
out("NOT COMPARED: %s / %s" % (a, b))
return 1
k = float(u["draft"]["TMS"]["PARAMETERS"]["K"])
stat = abs(a[1] - b[1]) / math.sqrt(a[2] ** 2 + b[2] ** 2)
c = S.mm_contradicts(a[1], a[2], b[1], b[2], k, 1.0)
out("M COMPARISON (SR3, S 1): %s; statistic %.6g against k %s"
% ("NON-CORROBORATION" if c else "MATCH", stat, k))
out("Nothing landed: the sandbox does not write CORROBORATION records at 0.1.0.")
return 0
def cmd_withdraw(T, B, args, out):
opts, pos = parse(args, {"--account": 1})
if len(pos) != 2 or "--account" not in opts:
raise Stop("usage: withdraw PFX RID --account A")
u = load_uwc(pos[0])
_declared(u)
digest_gate(T, u)
RC = T["RC"]
register, meta, cert = list(u["register"]), u["meta"], u["cert"]
target = next((r for r in keyed(register) if r["ID"] == pos[1]), None)
if target is None:
raise Stop("%s is not a keyed record of the register" % pos[1])
if target.get("AUTHOR") != opts["--account"]:
raise Stop("only the record's AUTHOR (%s) withdraws it (cause AUTHOR)" % target.get("AUTHOR"))
words = words_of(T, register)
if words[pos[1]] not in ("PROVISIONAL", "ESTABLISHED", "HELD"):
raise Stop("%s does not stand (%s)" % (pos[1], words[pos[1]]))
sc = RC.withdrawal_scope(register, pos[1], [])
if sc[0] != "OK":
raise Stop("\n".join(problems_text(sc)))
act = {"cause": "AUTHOR", "withdrawn": pos[1], "withdrawn_by": opts["--account"], "status_set": "WITHDRAWN",
"prior_status": words[pos[1]], "superseded_by": "", "locus_gap": "", "still_stands": [],
"cascade": sc[1], "tps_version": B["TPS"], "d_tms": cert["D_TMS"]}
meta["mark"] += 1
wr = RC.withdrawal_record(rid(cert["PREFIX"], meta["mark"]), act)
if wr[0] != "OK":
raise Stop("\n".join(problems_text(wr)))
register.append(wr[1])
meta["pushes"] += 1
meta["records"][wr[1][0]] = {"push": meta["pushes"], "block": "WITHDRAWAL"}
write_json(os.path.join(u["dir"], "register.json"), register)
write_json(os.path.join(u["dir"], "meta.json"), meta)
chain_append(u["dir"], "WITHDRAW", {"record": pos[1], "withdrawal": wr[1][0], "by": opts["--account"]})
out("WITHDRAWAL %s: %s WITHDRAWN (cause AUTHOR); cascade %s"
% (wr[1][0], pos[1], [c[0] for c in sc[1]] or "none"))
return 0
def cmd_status(T, B, args, out):
if len(args) != 1:
raise Stop("usage: status PFX")
u = load_uwc(args[0])
cert = u["cert"]
out("UWC %s UUID %s STATE %s" % (cert["PREFIX"], cert["UUID"], cert["STATE"]))
out(" D(TFK) %s" % cert["D_TFK"])
out(" %s" % cert["STRUCTURE_LINE"])
out(" TPS %s D(IS) %s TPC-PY %s" % (cert["TPS"], cert["D_IS"][:16], cert["TPC_PY"]))
if u["register"] is None:
out(" register: not opened (declare first)")
return 0
words = words_of(T, u["register"])
out(" register: %d records; next free identifier %s" % (len(u["register"]),
rid(cert["PREFIX"], u["meta"]["mark"] + 1)))
for r in u["register"]:
if isinstance(r, dict):
extra = ""
if r["CLASS"] == "NAMED_GAP":
extra = " %s %s: %s" % (r["SOURCE"], r["AGAINST"], r["STATE"])
elif r["CLASS"] == "VERIFICATION":
extra = " by %s: TARGET %s, DIFFERENCE %s" % (r.get("AUTHOR"), r.get("TARGET"), cj(r.get("DIFFERENCE")))
elif r["CLASS"] in PUSH_CLASSES:
extra = " by %s: %s" % (r.get("AUTHOR"), (r.get("STATEMENT") or r.get("PROPOSITION") or
r.get("ENTRY") or "")[:70])
out(" %-10s %-12s %-12s%s" % (r["ID"], r["CLASS"], words.get(r["ID"], "NOT GRADED"), extra))
else:
out(" %-10s %-12s %s" % (r[0], r[1], cj(r[2:5])[:90]))
eo = T["RC"].eom0_open(u["register"])
out(" EOM-0: %s" % ("OPEN" if eo[0] == "OK" and eo[1] else "not open"))
screen(T, u["draft"], u["register"], out)
return 0
def cmd_show(T, B, args, out):
if len(args) != 2:
raise Stop("usage: show PFX RID")
u = load_uwc(args[0])
_declared(u)
for r in u["register"]:
if (isinstance(r, dict) and r["ID"] == args[1]) or (isinstance(r, list) and r[0] == args[1]):
out(json.dumps(r, indent=2, ensure_ascii=False))
return 0
raise Stop("%s is not in the register" % args[1])
def cmd_verify(T, B, args, out):
if len(args) != 1:
raise Stop("usage: verify PFX")
u = load_uwc(args[0])
RC, Z = T["RC"], T["Z"]
bad = 0
d = Z.d_tfk_of(u["draft"])
ok = d == u["cert"]["D_TFK"]
out("draft D(TFK) %s: %s" % (d, "matches the certificate" if ok else "DIFFERS from the certificate"))
bad += not ok
lines = chain_lines(u["dir"])
prev = "GENESIS"
for ln in lines:
body = dict(ln)
h = body.pop("hash")
if body["prev"] != prev or sha256_bytes(cj(body).encode("utf-8")) != h:
out("chain line %d BROKEN" % ln["seq"])
bad += 1
prev = h
out("chain: %d lines%s" % (len(lines), "" if bad == 0 else " (see above)"))
if u["register"] is not None:
rp = os.path.join(u["dir"], "register.json")
last = next((ln for ln in reversed(lines) if ln["event"] in ("DECLARE", "PUSH", "WITHDRAW")), None)
if last and last["register_sha256"] != sha256_bytes(open(rp, "rb").read()):
out("register.json DIFFERS from the last chain line's hash (edited outside the sandbox?)")
bad += 1
# STATE LINEs change in place and are outside every digest; only chained pushes rewrite the file
if u["register"][0].get("D_TFK") != d:
out("MASTER D_TFK DIFFERS from the draft")
bad += 1
n = 0
for r in u["register"]:
if isinstance(r, dict):
c = RC.record_check(r, u["draft"], "OFF", "LANDED")
n += 1
if c[0] != "OK":
out("%s refused at LANDED:" % r["ID"])
for p in problems_text(c):
out(p)
bad += 1
out("records re-checked at LANDED, digests recomputed: %d" % n)
m = 0
for r in u["register"]:
if isinstance(r, list):
m += 1
if r[1] == "PROMOTION":
again = RC.promotion_record(r[0], r[2], r[3])
else:
keys = RC.ACT_KEYS
act = dict(zip(["cause", "withdrawn", "withdrawn_by", "status_set", "prior_status",
"superseded_by", "locus_gap", "still_stands", "cascade", "tps_version",
"d_tms"], r[2:13]))
again = RC.withdrawal_record(r[0], act) if sorted(act) == sorted(keys) else ["REFUSED", []]
if again[0] != "OK" or again[1] != r:
out("%s %s: its digest does not recompute" % (r[0], r[1]))
bad += 1
out("TPX-written list records recomputed: %d" % m)
out("VERIFY: %s" % ("PASS" if bad == 0 else "FAIL (%d)" % bad))
return 0 if bad == 0 else 1
def cmd_report(T, B, args, out):
if len(args) != 1:
raise Stop("usage: report PFX")
u = load_uwc(args[0])
d = u["dir"]
parts = ["TPM SANDBOX REPORT %s | UWC %s | %s" % (SANDBOX_VERSION, u["cert"]["PREFIX"], now()), "",
"== certificate.json ==", json.dumps(u["cert"], indent=2, ensure_ascii=False), "",
"== draft.json ==", open(os.path.join(d, "draft.json"), encoding="utf-8").read(), ""]
if u["register"] is not None:
parts += ["== register.json ==", json.dumps(u["register"], indent=2, ensure_ascii=False), ""]
parts += ["== chain.jsonl ==", open(os.path.join(d, "chain.jsonl"), encoding="utf-8").read()]
ld = os.path.join(d, "log")
if os.path.isdir(ld):
for n in sorted(os.listdir(ld)):
parts += ["== log/%s ==" % n, open(os.path.join(ld, n), encoding="utf-8").read()]
parts.append("END OF REPORT")
p = os.path.join(d, "REPORT.txt")
with open(p, "w", encoding="utf-8", newline="\n") as f:
f.write("\n".join(parts) + "\n")
out("REPORT written: uwcs/%s/REPORT.txt (%d bytes). Attach that file when you report back." %
(u["cert"]["PREFIX"], os.path.getsize(p)))
return 0
def cmd_selftest(T, B, args, out):
"""The spring example end to end, in a temporary folder."""
global UWCS
keep = UWCS
tmp = tempfile.mkdtemp(prefix="tpm_sandbox_selftest_")
UWCS = os.path.join(tmp, "uwcs")
ex = os.path.join(tmp, "examples")
os.makedirs(ex)
for name in EXAMPLES:
write_json(os.path.join(ex, name + ".json"), EXAMPLES[name])
shutil.copyfile(os.path.join(TPC_DIR, "examples", "spring_draft.json"), os.path.join(ex, "draft.json"))
steps = [("new", [os.path.join(ex, "draft.json"), "--prefix", "SPR", "--initiator", "selftest"], 0),
("push", ["SPR", os.path.join(ex, "push_1.json"), "--block", "B1"], 1),
("declare", ["SPR", "--by", "selftest", "--confirm"], 0),
("push", ["SPR", os.path.join(ex, "push_bad_id.json"), "--block", "B0"], 1),
("push", ["SPR", os.path.join(ex, "push_1.json"), "--block", "B1"], 0),
("difference", ["SPR", os.path.join(ex, "verification_draft.json")], 0),
("push", ["SPR", os.path.join(ex, "push_2.json"), "--block", "B2"], 0),
("compare", ["SPR", "SPR-3", os.path.join(ex, "reproduction.json")], 0),
("withdraw", ["SPR", "SPR-2", "--account", "tester"], 0),
("status", ["SPR"], 0),
("verify", ["SPR"], 0)]
fails = 0
try:
for cmd, a, want in steps:
sub = Out(cmd, a)
code = run_cmd(T, B, cmd, a, sub)
ok = code == want
fails += not ok
out("%-4s %s %s -> exit %d (want %d)" % ("ok" if ok else "FAIL", cmd, " ".join(
os.path.basename(x) for x in a), code, want))
if not ok:
out(sub.text(code))
reg = read_json(os.path.join(UWCS, "SPR", "register.json"))
words = words_of(T, reg)
expect = {"SPR-2": "WITHDRAWN", "SPR-3": "ESTABLISHED", "SPR-4": "WITHDRAWN", "SPR-6": "PROVISIONAL"}
for k, v in sorted(expect.items()):
ok = words.get(k) == v
fails += not ok
out("%-4s %s is %s (want %s)" % ("ok" if ok else "FAIL", k, words.get(k), v))
# tamper: edit a landed record by hand; verify must fail and push must refuse nothing silently
rp = os.path.join(UWCS, "SPR", "register.json")
reg[2]["STATEMENT"] = reg[2]["STATEMENT"] + " (edited by hand)"
write_json(rp, reg)
sub = Out("verify", ["SPR"])
code = run_cmd(T, B, "verify", ["SPR"], sub)
ok = code == 1
fails += not ok
out("%-4s a record edited by hand fails verify -> exit %d (want 1)" % ("ok" if ok else "FAIL", code))
finally:
UWCS = keep
shutil.rmtree(tmp, ignore_errors=True)
out("SELFTEST: %s" % ("PASS" if fails == 0 else "FAIL (%d)" % fails))
return 0 if fails == 0 else 1
# ---- worked examples ------------------------------------------------------------------
# The spring example's records, as the selftest pushes them. `example NAME` prints one.
EXAMPLES = {
"push_1": [
{
"ID": "SPR-2",
"CLASS": "ASSUMPTION",
"CITES": [],
"ASSUMPTIONS": [],
"PROPOSITION": "At t = 0 the mass is released at x = 0.02 m.",
"REFS": [
"x"
],
"SITUATION": "INITIAL",
"DOF": "x",
"VALUE": "0.02",
"UNIT": "m",
"AT": [
{
"DOF": "t",
"VALUE": "0",
"UNIT": "s"
}
]
},
{
"ID": "SPR-3",
"CLASS": "RESULT",
"CITES": [],
"ASSUMPTIONS": [],
"STATEMENT": "Dividing EOM_X by m, which A_mpos makes positive, gives dv/dt = -(k_s/m) x.",
"REGIME": "L",
"DERIVATION": {
"STEPS": 1,
"TEXT": "EOM_X: m D(v,t) = -k_s x. Divide both sides by m (m > 0 by A_mpos)."
},
"FALSIFIER": "A measured acceleration of the mass not proportional to x with ratio -k_s/m inside the domain.",
"FALSIFIER_CLASS": "F3",
"DOMAIN": "|x| < x_max",
"DOMAIN_CONDITION": "abs(x) < x_max",
"REFS": [
"v",
"t",
"k_s",
"m",
"x",
"x_max"
],
"EQUATION": "D(v, t) = -k_s / m * x",
"ENTRIES": [
"E_x",
"E_v"
]
},
{
"ID": "SPR-4",
"CLASS": "RESULT",
"CITES": [],
"ASSUMPTIONS": [
"SPR-2"
],
"STATEMENT": "At t = 0 the displacement is x = 0.02 m, by the initial value SPR-2.",
"REGIME": "L",
"DERIVATION": {
"STEPS": 1,
"TEXT": "Read x at t = 0 from SPR-2."
},
"FALSIFIER": "A measured displacement at release different from 0.02 m beyond its uncertainty.",
"FALSIFIER_CLASS": "F3",
"QUANTITY": "E_x",
"VALUE": "0.02",
"UNIT": "m",
"AT": [
{
"DOF": "t",
"VALUE": "0",
"UNIT": "s"
}
]
}
],
"push_2": [
{
"ID": "SPR-6",
"CLASS": "VERIFICATION",
"CITES": [],
"ASSUMPTIONS": [],
"REGIME": "M",
"TARGET": "SPR-4",
"EXTERNAL": {
"VALUE": "0.0199",
"U_C": "0.0002",
"UNIT": "m",
"QUANTITY": "E_x"
},
"SOURCE": "Bench measurement of the release displacement, lab notebook entry 2026-09-30 (example).",
"AT": [
{
"DOF": "t",
"VALUE": "0",
"UNIT": "s"
}
],
"DIFFERENCE": {
"VALUE": "1e-4",
"U": "2e-4",
"ZETA": "5e-1"
}
}
],
"push_bad_id": [
{
"ID": "SPR-9",
"CLASS": "ASSUMPTION",
"CITES": [],
"ASSUMPTIONS": [],
"PROPOSITION": "The spring is massless.",
"REFS": []
}
],
"reproduction": {
"CLASS": "RESULT",
"CITES": [],
"ASSUMPTIONS": [],
"STATEMENT": "An independent derivation: dv/dt = -(k_s/m) x.",
"REGIME": "L",
"DERIVATION": {
"STEPS": 1,
"TEXT": "From EOM_X, divide by m."
},
"FALSIFIER": "As the original.",
"FALSIFIER_CLASS": "F3",
"REFS": [
"v",
"t",
"k_s",
"m",
"x"
],
"EQUATION": "D(v,t) = -k_s/m*x"
},
"verification_draft": {
"ID": "SPR-6",
"CLASS": "VERIFICATION",
"CITES": [],
"ASSUMPTIONS": [],
"REGIME": "M",
"TARGET": "SPR-4",
"EXTERNAL": {
"VALUE": "0.0199",
"U_C": "0.0002",
"UNIT": "m",
"QUANTITY": "E_x"
},
"SOURCE": "Bench measurement of the release displacement, lab notebook entry 2026-09-30 (example).",
"AT": [
{
"DOF": "t",
"VALUE": "0",
"UNIT": "s"
}
]
}
}
def cmd_example(T, B, args, out):
if len(args) != 1:
raise Stop("usage: example NAME (draft, %s)" % ", ".join(sorted(EXAMPLES)))
name = args[0]
if name == "draft":
value = read_json(os.path.join(TPC_DIR, "examples", "spring_draft.json"))
elif name in EXAMPLES:
value = EXAMPLES[name]
else:
raise Stop("no example %r; the examples are draft, %s" % (name, ", ".join(sorted(EXAMPLES))))
out(json.dumps(value, indent=2, ensure_ascii=False))
return 0
def _no_dupes(pairs):
d = {}
for k, val in pairs:
if k in d:
raise ValueError("duplicate key %r" % k)
d[k] = val
return d
COMMANDS = {"check": cmd_check, "new": cmd_new, "declare": cmd_declare, "push": cmd_push,
"next-id": cmd_next_id, "difference": cmd_difference, "compare": cmd_compare,
"withdraw": cmd_withdraw, "status": cmd_status, "show": cmd_show, "verify": cmd_verify,
"report": cmd_report, "example": cmd_example, "selftest": cmd_selftest}
LOGGED = ("new", "declare", "push", "difference", "compare", "withdraw", "status", "verify", "next-id", "show")
def parse(args, spec):
opts, pos, i = {}, [], 0
while i < len(args):
a = args[i]
if a in spec:
if spec[a] == 1:
if i + 1 >= len(args):
raise Stop("%s needs a value" % a)
opts[a] = args[i + 1]
i += 2
else:
opts[a] = True
i += 1
elif a.startswith("--"):
raise Stop("unknown option %s" % a)
else:
pos.append(a)
i += 1
return opts, pos
def run_cmd(T, B, cmd, args, out):
try:
return COMMANDS[cmd](T, B, args, out)
except Stop as e:
out("REFUSED: %s" % e)
return 1
except (OSError, ValueError) as e:
out("REFUSED: %s: %s" % (type(e).__name__, e))
return 2
def main(argv):
global TPC_DIR
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(errors="backslashreplace")
argv = list(argv)
if len(argv) >= 3 and argv[1] == "--tpc":
TPC_DIR = os.path.abspath(argv[2])
del argv[1:3]
elif os.environ.get("TPM_TPC"):
TPC_DIR = os.path.abspath(os.environ["TPM_TPC"])
if len(argv) < 2 or argv[1] not in COMMANDS:
print(__doc__)
return 2
return dispatch(argv[1], argv[2:])
def dispatch(cmd, args, quiet=False):
out = Out(cmd, args)
try:
B = load_binding()
T = load_tpc(B)
out.bound(B)
except Stop as e:
out("REFUSED: %s" % e)
if quiet:
return 1, out.text(1)
print(out.text(1), end="")
return 1
code = run_cmd(T, B, cmd, args, out)
text = out.text(code)
if not quiet:
print(text, end="")
prefix = None
if cmd == "new":
try:
prefix = parse(args, {"--prefix": 1, "--initiator": 1})[0].get("--prefix")
except Stop:
prefix = None
elif cmd in LOGGED and args:
prefix = args[0]
if prefix and os.path.isdir(uwc_dir(prefix)):
save_log(uwc_dir(prefix), cmd, text)
return (code, text) if quiet else code
if __name__ == "__main__":
sys.exit(main(sys.argv))
# END OF FILE --- tpm_sandbox.py 0.2.0. If this line is absent, the file arrived truncated.
