The Principia Machina/TPC-PY/tpc sponsor.py: Difference between revisions

Latest revision as of 04:37, 28 September 2026

"""tpc_sponsor 0.1.0 (DRAFT, pending D160): the sponsor's CONDUCT EVENTS,
the user evidence E_u with its interval and n_u, and the WHITEWASH count.

Reference implementation, Python standard library only; imports tpc_beta
for the Beta layer. DATA for landing round 8 of branch
tpm_project_closeout (the identity speculation 0.1.0 SM-1 to SM-3 and
section 8; PMP-TPM 0.12.0 AI CERTIFICATION CR12 as 0.12.0 words it, the
blend retired by D161.1; record D160.1 (b)-(e); TUNING OPTIONS g_C, h_C,
SUCCESSION WINDOW); not a declared TPC-PY version and not TPX code. Every
function is a pure function of its arguments. Every refused argument
raises ValueError with a short fixed message. render_*() give the
canonical text the PHP residence (tpc_sponsor.php) reproduces byte for
byte; floats through canon_float (the tree's %.12g). Ints are never
bools. Times are integer seconds UTC; ages and block counts are integers
of landed blocks.

conduct_resolve(event, facts) -> dict                   (D160.1 (c), (e))
  event: {kind}, kind one of CE-F1, CE-F2, CE-A1, CE-A2, CE-A3, CE-A4,
    CE-A5. facts: the counted facts or the audit's finding the kind
    needs (below); anything else raises ValueError("facts"). Result
    {resolution: FOR, AGAINST, NEUTRAL or PENDING; reason}. "Each is a
    counted fact or an audit finding; none resolves AGAINST on a
    statistic"; "a PENDING item updates nothing". The 100-block windows
    are the rule's own figures (D160.1 (c)), not parameters.
  CR-F1 PROACTIVE SUCCESSION. facts {named: the sponsor named the agent
     it succeeds or the change it anticipates; predecessor: null or
     {state, pending} at the admission; blocks: landed by the new agent
     since admission; first_violation_block: null or the 1-based block of
     its first confirmed violation; period_taken: another CE-F1 of the
     sponsor already stands in this RATING PERIOD of its blocks}.
     Not named, a predecessor not GREEN or with a PENDING item, or the
     period taken ("at most one CE-F1 per RATING PERIOD"): NEUTRAL. A
     confirmed violation at a block up to and including 100: NEUTRAL.
     blocks at least 100 with none: FOR. Otherwise PENDING.
  CR-F2 DECLARED CHANGE KEPT CLEAN. facts {declared: declared on TPX's
     form before use; blocks: landed after the change; first_violation_
     block; first_canary_failure_block}. Not declared: NEUTRAL (an
     undeclared change is CE-A1's matter). A canary failure or a
     confirmed violation at a block up to 100: NEUTRAL. blocks at least
     100 with neither: FOR. Otherwise PENDING.
  CR-A1, CR-A2, CR-A3. facts {audit: OPEN, CONFIRMED or DISMISSED}, the
     audit block's state. CONFIRMED: AGAINST; OPEN: PENDING; DISMISSED:
     NEUTRAL. No statistic enters (the exclusion of D160).
  CR-A4 WHITEWASH. facts {admission, exit: times; state, pending: the
     other agent's at its retirement or lapse; window: SUCCESSION WINDOW
     in seconds}. "A new agent admitted within SUCCESSION WINDOW of the
     retirement or lapse of another agent of the sponsor that was then
     not GREEN or had a PENDING item": |admission - exit| <= window
     (either side; the window closed on both ends: PD-OC-9) and (state
     not GREEN or pending > 0): AGAINST, "counted at the admission";
     otherwise NEUTRAL; never PENDING.
  CR-A5 REMOVAL FOR CAUSE. facts {removed: H-DECL's removal declared;
     cause: CANARY, TRIP, OTHER or null}. removed with cause CANARY or
     TRIP: AGAINST (the SPONSOR CUT is applied at once by TPX, outside
     this function); removed with cause OTHER: NEUTRAL (not for cause);
     not removed: PENDING (the decision item open).

user_evidence(events, ages, g, h, a0, b0, half_life) -> dict
                                                        (D160.1 (b), (d))
  events: the resolved events of both parts, each {kind, resolution,
    weight?}: a CONDUCT EVENT (kind CE-F1 .. CE-A5, resolution FOR for an
    F kind, AGAINST for an A kind, or NEUTRAL) weighted by g[kind] or
    h[kind]; or an agent event of the pooled part (kind AGENT, resolution
    FOR or AGAINST, weight the agent-level weight CR11 gives it, g_F x q
    or h(standing), the DIVERSITY QUOTA already applied by the caller:
    "the diversity quota acts on the pooled part only"). A PENDING event,
    a resolution against its kind, or a weight on a conduct event raises
    ValueError("events"). ages: parallel, integers of the sponsor's
    agents' landed blocks since each event resolved. g: {CE-F1, CE-F2};
    h: {CE-A1 .. CE-A5}; a0, b0 > 0; half_life > 0.
  UE1 Each event's weight is its base weight x 2^(-age / half_life), pow
      of both residences on the same doubles.
  UE2 F_u is the sum of the FOR weights and A_u of the AGAINST weights,
      in list order; a NEUTRAL event adds nothing.
  UE3 n_u counts the FOR and AGAINST events (PD-OC-10: a NEUTRAL event
      is resolved but carries no evidence and is not counted).
  UE4 E_u = (a0 + F_u) / (a0 + b0 + F_u + A_u) (tpc_beta.beta_mean);
      the 95 per cent equal-tailed interval of Beta(a0 + F_u, b0 + A_u)
      (tpc_beta.beta_interval). T_idle's scaling of F_u and A_u toward
      the prior is applied by TPX to the sums before this function or by
      recomputing from them (both are returned).

whitewash_count(admissions, retirements, states, window) -> dict (CE-A4)
  admissions: [{agent, time}] the sponsor's new agents; retirements:
    [{agent, time}] the retirement or lapse of its agents; states:
    {agent: {state, pending}} at each retirement (every retired agent
    present, else ValueError("states")); window: SUCCESSION WINDOW in
    seconds. WC1 An admission counts once when some retirement of ANOTHER
    agent lies within the window of it (CR-A4's test) and that agent was
    then not GREEN or had a PENDING item; the first such retirement in
    list order is named. WC2 count is the number of admissions that
    count.
"""

import math
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from tpc_beta import beta_interval, beta_mean, canon_float  # noqa: E402

VERSION = "0.1.0"
TIME_MAX = 253402300799
COUNT_MAX = 1000000000
WINDOW_BLOCKS = 100
FOR_KINDS = ("CE-F1", "CE-F2")
AGAINST_KINDS = ("CE-A1", "CE-A2", "CE-A3", "CE-A4", "CE-A5")
KINDS = FOR_KINDS + AGAINST_KINDS
STATES = ("GREEN", "YELLOW", "ORANGE", "SUSPENDED")
AUDITS = ("OPEN", "CONFIRMED", "DISMISSED")
CAUSES = ("CANARY", "TRIP", "OTHER")


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 _is_int(x, lo, hi):
    return type(x) is int and lo <= x <= hi


def _is_name(x):
    return type(x) is str and x != ""


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


def _num(x):
    if type(x) is bool or not isinstance(x, (int, float)):
        return None
    x = float(x)
    if not math.isfinite(x):
        return None
    return x


def _opt_block(x):
    return x is None or _is_int(x, 1, COUNT_MAX)


def _state_ok(o):
    return (_keys_are(o, ("state", "pending")) and o["state"] in STATES
            and _is_int(o["pending"], 0, COUNT_MAX))


# ------------------------------------------------------- conduct_resolve

def conduct_resolve(event, facts):
    if not _keys_are(event, ("kind",)) or event["kind"] not in KINDS:
        raise ValueError("event")
    kind = event["kind"]
    if kind == "CE-F1":
        if not _keys_are(facts, ("named", "predecessor", "blocks",
                                 "first_violation_block", "period_taken")):
            raise ValueError("facts")
        if (type(facts["named"]) is not bool
                or type(facts["period_taken"]) is not bool
                or not _is_int(facts["blocks"], 0, COUNT_MAX)
                or not _opt_block(facts["first_violation_block"])):
            raise ValueError("facts")
        pred = facts["predecessor"]
        if pred is not None and not _state_ok(pred):
            raise ValueError("facts")
        if not facts["named"]:
            return {"resolution": "NEUTRAL", "reason": "NOT NAMED"}
        if pred is not None and (pred["state"] != "GREEN"
                                 or pred["pending"] > 0):
            return {"resolution": "NEUTRAL", "reason": "PREDECESSOR"}
        if facts["period_taken"]:
            return {"resolution": "NEUTRAL", "reason": "PERIOD"}
        v = facts["first_violation_block"]
        if v is not None and v <= WINDOW_BLOCKS:
            return {"resolution": "NEUTRAL", "reason": "VIOLATION"}
        if facts["blocks"] >= WINDOW_BLOCKS:
            return {"resolution": "FOR", "reason": "CLEAN"}
        return {"resolution": "PENDING", "reason": "WINDOW"}
    if kind == "CE-F2":
        if not _keys_are(facts, ("declared", "blocks",
                                 "first_violation_block",
                                 "first_canary_failure_block")):
            raise ValueError("facts")
        if (type(facts["declared"]) is not bool
                or not _is_int(facts["blocks"], 0, COUNT_MAX)
                or not _opt_block(facts["first_violation_block"])
                or not _opt_block(facts["first_canary_failure_block"])):
            raise ValueError("facts")
        if not facts["declared"]:
            return {"resolution": "NEUTRAL", "reason": "NOT DECLARED"}
        c = facts["first_canary_failure_block"]
        if c is not None and c <= WINDOW_BLOCKS:
            return {"resolution": "NEUTRAL", "reason": "CANARY"}
        v = facts["first_violation_block"]
        if v is not None and v <= WINDOW_BLOCKS:
            return {"resolution": "NEUTRAL", "reason": "VIOLATION"}
        if facts["blocks"] >= WINDOW_BLOCKS:
            return {"resolution": "FOR", "reason": "CLEAN"}
        return {"resolution": "PENDING", "reason": "WINDOW"}
    if kind in ("CE-A1", "CE-A2", "CE-A3"):
        if not _keys_are(facts, ("audit",)) or facts["audit"] not in AUDITS:
            raise ValueError("facts")
        a = facts["audit"]
        if a == "CONFIRMED":
            return {"resolution": "AGAINST", "reason": "AUDIT"}
        if a == "OPEN":
            return {"resolution": "PENDING", "reason": "AUDIT"}
        return {"resolution": "NEUTRAL", "reason": "DISMISSED"}
    if kind == "CE-A4":
        if not _keys_are(facts, ("admission", "exit", "state", "pending",
                                 "window")):
            raise ValueError("facts")
        if (not _is_int(facts["admission"], 0, TIME_MAX)
                or not _is_int(facts["exit"], 0, TIME_MAX)
                or facts["state"] not in STATES
                or not _is_int(facts["pending"], 0, COUNT_MAX)
                or not _is_int(facts["window"], 0, TIME_MAX)):
            raise ValueError("facts")
        gap = facts["admission"] - facts["exit"]
        if gap < 0:
            gap = -gap
        if gap > facts["window"]:
            return {"resolution": "NEUTRAL", "reason": "OUTSIDE WINDOW"}
        if facts["state"] == "GREEN" and facts["pending"] == 0:
            return {"resolution": "NEUTRAL", "reason": "GREEN"}
        return {"resolution": "AGAINST", "reason": "WHITEWASH"}
    if not _keys_are(facts, ("removed", "cause")):
        raise ValueError("facts")
    if type(facts["removed"]) is not bool or (
            facts["cause"] is not None and facts["cause"] not in CAUSES):
        raise ValueError("facts")
    if not facts["removed"]:
        return {"resolution": "PENDING", "reason": "DECISION ITEM"}
    if facts["cause"] in ("CANARY", "TRIP"):
        return {"resolution": "AGAINST", "reason": facts["cause"]}
    return {"resolution": "NEUTRAL", "reason": "NOT FOR CAUSE"}


def render_resolve(d):
    return d["resolution"] + "|" + d["reason"]


# --------------------------------------------------------- user_evidence

def _check_weights(t, kinds):
    if not _keys_are(t, kinds):
        raise ValueError("weights")
    out = {}
    for k in kinds:
        v = _num(t[k])
        if v is None or v < 0.0:
            raise ValueError("weights")
        out[k] = v
    return out


def user_evidence(events, ages, g, h, a0, b0, half_life):
    if type(events) is not list or type(ages) is not list:
        raise ValueError("events")
    if len(events) != len(ages):
        raise ValueError("events")
    gw = _check_weights(g, FOR_KINDS)
    hw = _check_weights(h, AGAINST_KINDS)
    a0f = _num(a0)
    b0f = _num(b0)
    hl = _num(half_life)
    if a0f is None or b0f is None or a0f <= 0.0 or b0f <= 0.0:
        raise ValueError("prior")
    if hl is None or hl <= 0.0:
        raise ValueError("half_life")
    weights = []
    for e, age in zip(events, ages):
        if not _keys_are(e, ("kind", "resolution"), ("weight",)):
            raise ValueError("events")
        if not _is_int(age, 0, COUNT_MAX):
            raise ValueError("ages")
        kind = e["kind"]
        res = e["resolution"]
        if kind == "AGENT":
            if res not in ("FOR", "AGAINST") or "weight" not in e:
                raise ValueError("events")
            w = _num(e["weight"])
            if w is None or w < 0.0:
                raise ValueError("events")
        elif kind in FOR_KINDS:
            if res not in ("FOR", "NEUTRAL") or "weight" in e:
                raise ValueError("events")
            w = gw[kind]
        elif kind in AGAINST_KINDS:
            if res not in ("AGAINST", "NEUTRAL") or "weight" in e:
                raise ValueError("events")
            w = hw[kind]
        else:
            raise ValueError("events")
        weights.append((res, w * math.pow(2.0, -float(age) / hl)))
    f = 0.0
    a = 0.0
    n = 0
    for res, w in weights:
        if res == "FOR":
            f = f + w
            n += 1
        elif res == "AGAINST":
            a = a + w
            n += 1
    e_u = beta_mean(a0f + f, b0f + a)
    lo, hi = beta_interval(a0f + f, b0f + a, 0.95)
    return {"F": f, "A": a, "n": n, "E": e_u, "lo": lo, "hi": hi}


def render_evidence(d):
    return "%s|%s|%d|%s|%s|%s" % (canon_float(d["F"]), canon_float(d["A"]),
                                  d["n"], canon_float(d["E"]),
                                  canon_float(d["lo"]), canon_float(d["hi"]))


# ------------------------------------------------------- whitewash_count

def whitewash_count(admissions, retirements, states, window):
    if type(admissions) is not list or type(retirements) is not list:
        raise ValueError("lists")
    for lst in (admissions, retirements):
        for x in lst:
            if not _keys_are(x, ("agent", "time")) or not _is_name(x["agent"]) \
                    or not _is_int(x["time"], 0, TIME_MAX):
                raise ValueError("lists")
    if type(states) is not dict:
        raise ValueError("states")
    for k, v in states.items():
        if not _is_name(k) or not _state_ok(v):
            raise ValueError("states")
    for r in retirements:
        if r["agent"] not in states:
            raise ValueError("states")
    if not _is_int(window, 0, TIME_MAX):
        raise ValueError("window")
    matches = []
    for adm in admissions:
        for r in retirements:
            if r["agent"] == adm["agent"]:
                continue
            gap = adm["time"] - r["time"]
            if gap < 0:
                gap = -gap
            if gap > window:
                continue
            st = states[r["agent"]]
            if st["state"] == "GREEN" and st["pending"] == 0:
                continue
            matches.append((adm["agent"], r["agent"]))
            break
    return {"count": len(matches), "matches": matches}


def render_whitewash(d):
    lines = ["count|%d" % d["count"]]
    for a, r in d["matches"]:
        lines.append("match|%s|%s" % (_esc(a), _esc(r)))
    return "\n".join(lines)


# ---------------------------------------------------------------- corpus

FNS = ("resolve", "evidence", "whitewash")


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 == "resolve":
            return render_resolve(conduct_resolve(args["event"],
                                                  args["facts"]))
        if fn == "evidence":
            return render_evidence(user_evidence(
                args["events"], args["ages"], args["g"], args["h"],
                args["a0"], args["b0"], args["half_life"]))
        return render_whitewash(whitewash_count(
            args["admissions"], args["retirements"], args["states"],
            args["window"]))
    except ValueError as ex:
        return "ERROR:" + str(ex)