Jump to content

The Principia Machina/TPC-PY/tpc identity.py

From The Invariance Itself
"""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)