The Principia Machina/TPC-PY/tpc identity.py
Appearance
"""tpc_identity 0.1.1 (D146.1, D156.1, D157.1 as D182.1 amends it, D158.1;
released at 0.1.0 with TPC-PY 0.1.0, 2026-09-28; 0.1.1, a PATCH of this
docstring only, proposed with TPC-PY 0.2.0: K-RESIDUALS PD3): the unit of
identity's pure functions: the PROMPT digest and TPM's canonical JSON,
the MODEL MAP lookup, the change class of a reported model, and the
IDENTITY RECORD's comparison.
Reference implementation, Python standard library only (the identity
speculation 0.1.0 sections 5 and 8; PMP-TPM 0.14.0 IDENTITY, PROMPT AND
VENDOR FAMILY; record D146.1 (a2), (c), D156.1, D157.1, D158.1, D182.1);
not TPX code. The code and the vectors are those of 0.1.0. Every
function is a pure function of its arguments. Every refused argument
raises ValueError with a short fixed message; a refused salt is never
echoed. render_*() give the canonical
text the PHP residence (tpc_identity.php) reproduces byte for byte.
Objects are JSON objects (dict), lists are JSON arrays (list); ints are
never bools. Text enters in NFC (D99's convention for every text a digest
covers); a string carrying a surrogate code point is refused.
canonical_tools(json_text) -> str (D157.1 (a))
"the tool definitions in TPM's canonical JSON" (the identity
speculation IM-4: "keys sorted, strings NFC, decimals canonical").
CJ1 PARSE. json_text is one RFC 8259 JSON text: one value (object,
array, string, number, true, false or null) with the six ASCII
whitespace characters SPACE, TAB, LF and CR around tokens and
nothing after it. Both residences carry their own parser so that no
library's leniency (NaN, Infinity, big numbers, lone surrogates,
trailing commas, comments) enters. Refused: ValueError("tools").
CJ2 STRINGS. Escapes \\" \\\\ \\/ \\b \\f \\n \\r \\t and \\uXXXX, a
high surrogate joined with its following low surrogate; a lone
surrogate, a raw control character below U+0020 or invalid UTF-8
is refused. Each string, a key included, is put in NFC.
CJ3 NUMBERS. The JSON number grammar; the canonical form is the zone
serialization's DECIMAL (zone schemas 5.6; tpc_falcon canon_
decimal): "0" for zero, else [-]M with "e"E when E is not 0, M an
integer without leading or trailing zeros; string arithmetic only,
so no residence rounds. An exponent of more than four digits is
refused (the DECIMAL grammar's bound; spec note PD-OC-6).
CJ4 OBJECTS. Members sorted by the code points of their NFC keys; two
members with one NFC key are refused (duplicate). Nesting deeper
than 64 levels is refused (PD-OC-6).
CJ5 OUTPUT. No whitespace; "," and ":" separators; strings quoted with
only \\" \\\\ and the controls escaped (\\b \\f \\n \\r \\t, else
\\u00XX lowercase); every other character raw UTF-8; true, false
and null as themselves.
prompt_digest(family, salt, template, tools) -> 64 hex (D157.1 (a), (f))
"SHA-256 under D97 over ["TPM-PROMPT", "1", VENDOR FAMILY, salt,
template, tools]: the salt 32 lowercase hexadecimal characters drawn
once per PROMPT; the template the SPONSOR ZONE in NFC, whole, no inner
boundary, at most SPONSOR ZONE CAP; the tools the live list of every
tool the model is offered, harness built-ins included, in TPM's
canonical JSON; the rendered PROMPT never hashed" (D157.1 (a) as D182.1
amends it: the boundary is no longer the sponsor's; the SPONSOR ZONE is
everything PRB places on the INSTRUCTION CHANNEL besides the GTP BUNDLE,
an implementation's text included; a built-in PRB cannot read is bound
by its harness, version and tool name; a change of a bound tool's text
is a PROMPT change).
PD1 family: a non-empty string, the VENDOR FAMILY key family_of gives
(PD-OC-5), taken in NFC. PD2 salt: exactly 32 characters of 0-9a-f,
else ValueError("salt") without echo. PD3 template: a string, put
in NFC: the SPONSOR ZONE whole (D182.1); its cap, SPONSOR ZONE
CAP, and the reading of the live tool list are PRB's, outside this
function. PD4 tools: the JSON text, canonical_tools applied; a
refusal is ValueError("tools"). PD5 the digest: SHA-256 over the D97 netstring
of the six fields, 64 lowercase hex.
family_of(model_string, model_map) -> dict (D156.1 (a)-(e))
model_map: the MODEL MAP's rows, a list of objects {name, family, open,
server, quantization}: name the model string as PRB relays it (one
row per spelling: a dated snapshot, a dateless alias, a server-
prefixed spelling; names unique after NFC); family the maker's family
(closed) or the maker lineage (open weights); open a boolean; server
and quantization the pinned serving provider and quantization of open
weights, both non-empty strings when open is true and both null when
it is false (D156.1 (b): the triple is pinned; (d): an unpinned
row cannot be approved). Anything else raises ValueError("model_map").
FO1 The row whose name equals model_string, both in NFC, exactly and
case-sensitively (PD-OC-7); none: status UNMAPPED (D147: a row the
MODEL MAP does not cover is UNMAPPED).
FO2 MAPPED carries family, open, server, quantization and the VENDOR
FAMILY KEY: the D97 netstring of ["CLOSED", family] or of ["OPEN",
family, server, quantization] (PD-OC-5), so that a change of server
or quantization of open weights is a change of family (D156.1 (b))
and the same closed model through another server is not (c).
change_class(certified_family, last_model, reported_model, lookup) -> dict
(D146.1 (a2))
certified_family: the certificate's VENDOR FAMILY KEY. last_model: the
model last recorded for the account (the DECLARED MODEL at admission,
then each RECORDED CHANGE). reported_model: the RELAYED MODEL of this
petition or push. lookup: {family: the VENDOR FAMILY KEY family_of
gives for reported_model, or "UNMAPPED"; cluster: the CLUSTER LOOKUP's
answer for it after substitution and succession, IN, CUSP, OUT,
FADING, UNLISTED or EXCLUDED, or null when TPX did not run it}.
CC1 lookup.family UNMAPPED: class REFUSED MODEL, detail UNMAPPED (the
row is held out of every class, D147; nothing lands).
CC2 lookup.family not the certified key: REFUSED FAMILY, whatever the
cluster says ("A change of family is never a MODEL CHANGE"; the
sponsor's path is a new agent).
CC3 reported_model equal to last_model (NFC, exact): SAME; the cluster
is not read.
CC4 Otherwise a MODEL CHANGE, read from lookup.cluster: IN, RECORDED
CHANGE; CUSP, CUSP PATH (the gate's cusp path, D146.1 (b)); OUT,
FADING, UNLISTED or EXCLUDED, REFUSED MODEL with that detail; null,
ValueError("lookup").
The tenure restart, the PROMPT EPOCH and the sponsor's notice are TPX's
acts on RECORDED CHANGE, outside this function.
identity_compare(declared, stated, relayed, inferred) -> dict (D158.1)
Four vendor-level values, each a string or null (ABSENT): the vendor of
the DECLARED MODEL, of the STATED MODEL, of the RELAYED MODEL and the
INFERRED VENDOR. "Compared at the vendor level only; within a family
nothing is compared": the reduction of each value to its vendor is the
caller's (spec note, UNCOMPUTED). IC1 The six pairs in the fixed order
declared-stated, declared-relayed, declared-inferred, stated-relayed,
stated-inferred, relayed-inferred are each AGREE (equal in NFC, exact,
PD-OC-7), DIFFER or ABSENT (a value missing). IC2 overall: DIFFER when
any pair differs; AGREE when none differs and at least one pair was
compared; ABSENT otherwise. The result is advisory: it feeds the
sponsor's notice and the counts, and no gate or adverse act reads it.
"""
import hashlib
import unicodedata
VERSION = "0.1.1"
PAIRS = (("declared", "stated"), ("declared", "relayed"),
("declared", "inferred"), ("stated", "relayed"),
("stated", "inferred"), ("relayed", "inferred"))
CLUSTER_CLASSES = ("IN", "CUSP", "OUT", "FADING", "UNLISTED", "EXCLUDED")
MAX_DEPTH = 64
MAX_EXP_DIGITS = 4
# ---------------------------------------------------------------- common
def _esc(s):
out = []
for ch in s:
o = ord(ch)
if ch == "|" or ch == "\\":
out.append("\\" + ch)
elif o < 0x20 or o == 0x7F:
out.append("\\x%02x" % o)
else:
out.append(ch)
return "".join(out)
def _netstring(fields):
return b"".join(b"%d:%s" % (len(b), b)
for b in (f.encode("utf-8") for f in fields))
def _digest_fields(fields):
return hashlib.sha256(_netstring(fields)).hexdigest()
def _is_text(s):
if type(s) is not str:
return False
for ch in s:
if 0xD800 <= ord(ch) <= 0xDFFF:
return False
return True
def _nfc(s, name):
if not _is_text(s):
raise ValueError(name)
return unicodedata.normalize("NFC", s)
def _keys_are(o, required, optional=()):
if type(o) is not dict:
return False
for k in required:
if k not in o:
return False
for k in o:
if k not in required and k not in optional:
return False
return True
# -------------------------------------------------------- canonical JSON
_WS = " \t\n\r"
_HEX = "0123456789abcdefABCDEF"
class _Parser(object):
def __init__(self, text):
self.t = text
self.i = 0
self.n = len(text)
def fail(self):
raise ValueError("tools")
def ws(self):
while self.i < self.n and self.t[self.i] in _WS:
self.i += 1
def value(self, depth):
if depth > MAX_DEPTH:
self.fail()
self.ws()
if self.i >= self.n:
self.fail()
c = self.t[self.i]
if c == "{":
return self.obj(depth)
if c == "[":
return self.arr(depth)
if c == '"':
return _quote(self.string())
if c == "-" or "0" <= c <= "9":
return self.number()
for lit in ("true", "false", "null"):
if self.t.startswith(lit, self.i):
self.i += len(lit)
return lit
self.fail()
def obj(self, depth):
self.i += 1
members = {}
self.ws()
if self.i < self.n and self.t[self.i] == "}":
self.i += 1
return "{}"
while True:
self.ws()
if self.i >= self.n or self.t[self.i] != '"':
self.fail()
k = self.string()
self.ws()
if self.i >= self.n or self.t[self.i] != ":":
self.fail()
self.i += 1
v = self.value(depth + 1)
if k in members:
self.fail()
members[k] = v
self.ws()
if self.i >= self.n:
self.fail()
if self.t[self.i] == ",":
self.i += 1
continue
if self.t[self.i] == "}":
self.i += 1
break
self.fail()
keys = sorted(members.keys())
return "{" + ",".join(_quote(k) + ":" + members[k]
for k in keys) + "}"
def arr(self, depth):
self.i += 1
items = []
self.ws()
if self.i < self.n and self.t[self.i] == "]":
self.i += 1
return "[]"
while True:
items.append(self.value(depth + 1))
self.ws()
if self.i >= self.n:
self.fail()
if self.t[self.i] == ",":
self.i += 1
continue
if self.t[self.i] == "]":
self.i += 1
break
self.fail()
return "[" + ",".join(items) + "]"
def hex4(self):
if self.i + 4 > self.n:
self.fail()
h = self.t[self.i:self.i + 4]
for ch in h:
if ch not in _HEX:
self.fail()
self.i += 4
return int(h, 16)
def string(self):
"""The string at a '"', decoded and put in NFC (unquoted)."""
self.i += 1
out = []
while True:
if self.i >= self.n:
self.fail()
c = self.t[self.i]
if c == '"':
self.i += 1
break
if c == "\\":
self.i += 1
if self.i >= self.n:
self.fail()
e = self.t[self.i]
self.i += 1
if e == '"' or e == "\\" or e == "/":
out.append(e)
elif e == "b":
out.append("\b")
elif e == "f":
out.append("\f")
elif e == "n":
out.append("\n")
elif e == "r":
out.append("\r")
elif e == "t":
out.append("\t")
elif e == "u":
cp = self.hex4()
if 0xD800 <= cp <= 0xDBFF:
if not self.t.startswith("\\u", self.i):
self.fail()
self.i += 2
lo = self.hex4()
if not (0xDC00 <= lo <= 0xDFFF):
self.fail()
cp = 0x10000 + ((cp - 0xD800) << 10) + (lo - 0xDC00)
elif 0xDC00 <= cp <= 0xDFFF:
self.fail()
out.append(chr(cp))
else:
self.fail()
elif ord(c) < 0x20:
self.fail()
elif 0xD800 <= ord(c) <= 0xDFFF:
self.fail()
else:
out.append(c)
self.i += 1
return unicodedata.normalize("NFC", "".join(out))
def number(self):
start = self.i
if self.t[self.i] == "-":
self.i += 1
if self.i >= self.n:
self.fail()
if self.t[self.i] == "0":
self.i += 1
elif "1" <= self.t[self.i] <= "9":
while self.i < self.n and "0" <= self.t[self.i] <= "9":
self.i += 1
else:
self.fail()
intpart = self.t[start:self.i].lstrip("-")
sign = "-" if self.t[start] == "-" else ""
frac = ""
if self.i < self.n and self.t[self.i] == ".":
self.i += 1
j = self.i
while self.i < self.n and "0" <= self.t[self.i] <= "9":
self.i += 1
if self.i == j:
self.fail()
frac = self.t[j:self.i]
e = 0
if self.i < self.n and self.t[self.i] in "eE":
self.i += 1
esign = 1
if self.i < self.n and self.t[self.i] in "+-":
if self.t[self.i] == "-":
esign = -1
self.i += 1
j = self.i
while self.i < self.n and "0" <= self.t[self.i] <= "9":
self.i += 1
if self.i == j or self.i - j > MAX_EXP_DIGITS:
self.fail()
e = esign * int(self.t[j:self.i])
e -= len(frac)
digits = (intpart + frac).lstrip("0")
if digits == "":
return "0"
trimmed = digits.rstrip("0")
e += len(digits) - len(trimmed)
return sign + trimmed + ("e%d" % e if e != 0 else "")
def _quote(s):
out = ['"']
for ch in s:
o = ord(ch)
if ch == '"':
out.append('\\"')
elif ch == "\\":
out.append("\\\\")
elif ch == "\b":
out.append("\\b")
elif ch == "\f":
out.append("\\f")
elif ch == "\n":
out.append("\\n")
elif ch == "\r":
out.append("\\r")
elif ch == "\t":
out.append("\\t")
elif o < 0x20:
out.append("\\u%04x" % o)
else:
out.append(ch)
out.append('"')
return "".join(out)
def canonical_tools(json_text):
if not _is_text(json_text):
raise ValueError("tools")
p = _Parser(json_text)
v = p.value(0)
p.ws()
if p.i != p.n:
raise ValueError("tools")
return v
# --------------------------------------------------------- prompt digest
def _is_salt(s):
if type(s) is not str or len(s) != 32:
return False
for ch in s:
if ch not in "0123456789abcdef":
return False
return True
def prompt_digest(family, salt, template, tools):
fam = _nfc(family, "family")
if fam == "":
raise ValueError("family")
if not _is_salt(salt):
raise ValueError("salt")
tmpl = _nfc(template, "template")
canon = canonical_tools(tools)
return _digest_fields(["TPM-PROMPT", "1", fam, salt, tmpl, canon])
# ------------------------------------------------------------- family_of
def _check_map(model_map):
if type(model_map) is not list:
raise ValueError("model_map")
names = set()
rows = []
for r in model_map:
if not _keys_are(r, ("name", "family", "open", "server",
"quantization")):
raise ValueError("model_map")
if not _is_text(r["name"]) or r["name"] == "":
raise ValueError("model_map")
if not _is_text(r["family"]) or r["family"] == "":
raise ValueError("model_map")
if type(r["open"]) is not bool:
raise ValueError("model_map")
if r["open"]:
for k in ("server", "quantization"):
if not _is_text(r[k]) or r[k] == "":
raise ValueError("model_map")
else:
if r["server"] is not None or r["quantization"] is not None:
raise ValueError("model_map")
name = unicodedata.normalize("NFC", r["name"])
if name in names:
raise ValueError("model_map")
names.add(name)
rows.append((name, r))
return rows
def family_key(family, open_weights, server, quantization):
"""The VENDOR FAMILY KEY (FO2)."""
if open_weights:
return _netstring(["OPEN", family, server, quantization]).decode(
"utf-8")
return _netstring(["CLOSED", family]).decode("utf-8")
def family_of(model_string, model_map):
ms = _nfc(model_string, "model")
if ms == "":
raise ValueError("model")
rows = _check_map(model_map)
for name, r in rows:
if name == ms:
fam = unicodedata.normalize("NFC", r["family"])
if r["open"]:
server = unicodedata.normalize("NFC", r["server"])
quant = unicodedata.normalize("NFC", r["quantization"])
else:
server = None
quant = None
return {"status": "MAPPED", "family": fam, "open": r["open"],
"server": server, "quantization": quant,
"key": family_key(fam, r["open"], server, quant)}
return {"status": "UNMAPPED"}
def render_family(d):
if d["status"] == "UNMAPPED":
return "UNMAPPED"
return "MAPPED|%s|%s|%s|%s|%s" % (
_esc(d["family"]), "true" if d["open"] else "false",
"-" if d["server"] is None else _esc(d["server"]),
"-" if d["quantization"] is None else _esc(d["quantization"]),
_esc(d["key"]))
# ---------------------------------------------------------- change_class
def change_class(certified_family, last_model, reported_model, lookup):
cert = _nfc(certified_family, "certified_family")
if cert == "":
raise ValueError("certified_family")
last = _nfc(last_model, "last_model")
if last == "":
raise ValueError("last_model")
rep = _nfc(reported_model, "reported_model")
if rep == "":
raise ValueError("reported_model")
if not _keys_are(lookup, ("family", "cluster")):
raise ValueError("lookup")
fam = lookup["family"]
if not _is_text(fam) or fam == "":
raise ValueError("lookup")
fam = unicodedata.normalize("NFC", fam)
cl = lookup["cluster"]
if cl is not None and cl not in CLUSTER_CLASSES:
raise ValueError("lookup")
if fam == "UNMAPPED":
return {"class": "REFUSED MODEL", "detail": "UNMAPPED"}
if fam != cert:
return {"class": "REFUSED FAMILY", "detail": "FAMILY"}
if rep == last:
return {"class": "SAME", "detail": "-"}
if cl is None:
raise ValueError("lookup")
if cl == "IN":
return {"class": "RECORDED CHANGE", "detail": "IN"}
if cl == "CUSP":
return {"class": "CUSP PATH", "detail": "CUSP"}
return {"class": "REFUSED MODEL", "detail": cl}
def render_class(d):
return d["class"] + "|" + d["detail"]
# ------------------------------------------------------ identity_compare
def identity_compare(declared, stated, relayed, inferred):
vals = {}
for name, v in (("declared", declared), ("stated", stated),
("relayed", relayed), ("inferred", inferred)):
if v is None:
vals[name] = None
else:
s = _nfc(v, name)
if s == "":
raise ValueError(name)
vals[name] = s
pairs = []
differ = 0
compared = 0
for a, b in PAIRS:
if vals[a] is None or vals[b] is None:
pairs.append((a, b, "ABSENT"))
elif vals[a] == vals[b]:
pairs.append((a, b, "AGREE"))
compared += 1
else:
pairs.append((a, b, "DIFFER"))
compared += 1
differ += 1
if differ > 0:
overall = "DIFFER"
elif compared > 0:
overall = "AGREE"
else:
overall = "ABSENT"
return {"overall": overall, "pairs": pairs, "differ": differ}
def render_compare(d):
lines = ["overall|%s|%d" % (d["overall"], d["differ"])]
for a, b, r in d["pairs"]:
lines.append("%s|%s|%s" % (a, b, r))
return "\n".join(lines)
# ---------------------------------------------------------------- corpus
FNS = ("tools", "digest", "family", "class", "compare")
def run_case(fn, args):
"""One corpus case: the render, or "ERROR:" and the message."""
if fn not in FNS:
raise ValueError("fn")
try:
if fn == "tools":
return canonical_tools(args["json"])
if fn == "digest":
return prompt_digest(args["family"], args["salt"],
args["template"], args["tools"])
if fn == "family":
return render_family(family_of(args["model"], args["map"]))
if fn == "class":
return render_class(change_class(
args["certified"], args["last"], args["reported"],
args["lookup"]))
return render_compare(identity_compare(
args["declared"], args["stated"], args["relayed"],
args["inferred"]))
except ValueError as ex:
return "ERROR:" + str(ex)