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

Latest revision as of 04:37, 28 September 2026

"""tpc_volunteer 0.1.0 (DRAFT, pending the pause of landing round 8): the
VOLUNTEER channel's law of D126 as PMP-TPM 0.12.0 words it (VOLUNTEER
LAW, GOOD STANDING, the tuning options EARLY CALL LIMIT through
VOLUNTEER SHARE) with D155's FALCON answer - GOOD STANDING, the token
bucket, RETRY AFTER with full jitter, INTAKE's EWMA and the random early
throttle p_nw, the p_v epoch, the draw itself, and the VOLUNTEER SHARE
cap - as pure functions of recorded state and a supplied random value.

Reference implementation, Python standard library only; imports
tpc_probe (uniform, udiscrete) and tpc_beta (canon_float). DATA for
landing round 8 of branch tpm_project_closeout; not a declared TPC-PY
version and not TPX code. Every draw derives its uniform values from r
(32 lowercase hex) by the D97 digest of [tag, "1", r, counter]. Declared
parameters and tuning options are read from params under PMP-TPM
0.12.0's names: VOLUNTEER TIER; VOLUNTEER RATE (tokens per hour,
wall-clock); VOLUNTEER BURST; BACKOFF BASE and BACKOFF CAP (seconds,
wall-clock); INTAKE w_q; INTAKE MIN; INTAKE MAX; INTAKE MAX_P; P_V
BOUNDS; P_V ROTATION; EARLY CALL LIMIT; VOLUNTEER COOL-OFF; VOLUNTEER
SHARE; CONSISTENCY SHARE. Times are integer seconds of server time;
blocks are landed blocks. Every refused argument raises ValueError with
a short fixed message; render_*() give the canonical text the PHP
residence (tpc_volunteer.php) reproduces byte for byte.

good_standing(facts, params) -> dict                            (D126.1)
  "GOOD STANDING: certificate current, D(IS) matching, state GREEN, no
  reset in force, not SUSPENDED; VOLUNTEER TIER may narrow it to the
  TRUST TIER". facts: {certificate_current, dis_match, state, reset_in_
  force, suspended, in_tier}. GS1 The first failing clause, in that
  order, is the reason; VOLUNTEER TIER = TRUST TIER adds in_tier last.

bucket_refill(bucket, now, params) -> dict     (D126.1; RFC 2697 srTCM)
  "a token bucket on the call, tokens at VOLUNTEER RATE up to VOLUNTEER
  BURST". bucket: {tokens, updated}. BR1 tokens' = min(BURST, tokens +
  RATE x (now - updated) / 3600), updated' = now; now >= updated.
bucket_take(bucket) -> dict
  BT1 A token is taken when tokens >= 1; else the call has no token.

retry_after(attempt, r, params) -> float       (D126.1; Brooker 2015)
  "RETRY AFTER = random(0, min(BACKOFF CAP, BACKOFF BASE x 2^attempt)),
  attempt the consecutive NO WORK answers, reset by WORK". RA1 u from
  TPC-VOLUNTEER-RETRY counter 0; the bound is CAP once 2^attempt
  overflows a double (attempt >= 1024); seconds.

intake_update(intake, volunteer_blocks, landed_blocks, params) -> float
  "INTAKE the EWMA (w_q) of the share of the account's landed blocks in
  the last RATING PERIOD that were volunteer blocks" (D126.1; RED).
  IU1 share = volunteer / landed (0 when landed is 0, PD-SC-19);
      intake' = (1 - w_q) x intake + w_q x share.
p_nw(intake, params) -> float
  "p_nw 0 below INTAKE MIN, linear to INTAKE MAX_P at INTAKE MAX, 1
  above": 0 when intake <= MIN; (intake - MIN) / (MAX - MIN) x MAX_P
  when MIN < intake <= MAX; 1 when intake > MAX. RED's spacing
  correction is not used.

pv_step(state, r, params) -> dict                              (D126.1)
  "p_v is drawn per epoch from P_V BOUNDS with interior endpoints, the
  epoch ending with P_V ROTATION per volunteer call fleet-wide
  (Geometric)". state: {p_v: null or the value in force}. PV1 With p_v
  null, or when u0 < P_V ROTATION (TPC-PV-ROTATE 0), a new epoch starts
  on this call: p_v = lo + u1 x (hi - lo) (TPC-PV-DRAW 0). PV2 The step
  runs on every call TPX answers (PD-SC-20; alternative: on every call,
  refused ones included).

volunteer_share(window, g, v_max) -> dict                      (D126.1)
  "FOR weight from volunteer work beyond v_max of the account's FOR
  weight in the half-life window carries zero, the diversity quota's
  shape applied to the channel". window: [{volunteer: bool, weight}];
  g the match's base weight. VS1 q = clip((v_max x (W + g) - W_v) / g,
  0, 1), the same shape as diversity_quota (PD-SC-1).

volunteer_draw(state, call, r, params) -> dict     (D126.1; D155.1)
  "THE DRAW, per call, in order: no token or inside RETRY AFTER,
  REFUSED; u < p_nw(INTAKE), NO WORK; v < p_v, PROBE (canary or parrot
  by PARROT RATIO); a legitimate item available, that item; otherwise
  NO WORK"; "A too-early call is REFUSED, its own answer with RETRY
  AFTER, never NO WORK; ... past EARLY CALL LIMIT refusals in a RATING
  PERIOD the channel closes for VOLUNTEER COOL-OFF"; D155.1 "VOLUNTEER
  as a fourth answer kind FALCON, outside the p_v law". state: {bucket,
  retry_until (server time or null), attempt, refusals, period (the
  RATING PERIOD index the refusals belong to), closed_until (block or
  null), intake, p_v (pv_step's state)}; call: {now, block, period,
  has_item, has_falcon, has_own_record, ratio_k, standing (good_
  standing's facts)}.
  VD1 Not in GOOD STANDING: REFUSED (STANDING), not counted (it is not
      a too-early call). The channel closed (block < closed_until):
      REFUSED (CLOSED), not counted (PD-SC-21).
  VD2 The bucket is refilled to now. Inside RETRY AFTER (now < retry_
      until): REFUSED (EARLY); no token: REFUSED (TOKEN). A refused call
      is counted in its RATING PERIOD (a new period restarts the count);
      past EARLY CALL LIMIT the channel closes until block + VOLUNTEER
      COOL-OFF and the count restarts. Its RETRY AFTER is the jitter law
      at the current attempt, and retry_until becomes the later of the
      one in force and now + RETRY AFTER (PD-SC-22).
  VD3 A token is taken; pv_step runs. An assigned falcon in flight:
      FALCON, outside the p_v law, the attempt unchanged (PD-SC-23: the
      falcon is neither a WORK BLOCK nor a PROBE, so INTAKE's throttle
      does not apply to it).
  VD4 u < p_nw(intake) (TPC-VOLUNTEER-NW 0): NO WORK, attempt + 1.
  VD5 v < p_v (TPC-VOLUNTEER-PV 0): WORK, a PROBE; a parrot when w < k /
      (1 + k) (TPC-VOLUNTEER-KIND 0), a CONSISTENCY PARROT when x <
      CONSISTENCY SHARE (TPC-VOLUNTEER-KIND 1) and the account has an
      own record, else a CANARY; attempt 0.
  VD6 has_item: WORK, the ITEM; attempt 0. Else NO WORK, attempt + 1;
      when the queue is empty every WORK answer is therefore a probe.
  VD7 Every answer carries RETRY AFTER = retry_after(attempt', r) and
      retry_until' = now + RETRY AFTER; the record carries the answer,
      its kind, p_nw, p_v, intake and the attempt.
"""

import math
import os
import sys

_here = os.path.dirname(os.path.abspath(__file__))
for _p in (_here,):
    if _p not in sys.path:
        sys.path.insert(0, _p)
from tpc_beta import canon_float  # noqa: E402
from tpc_probe import uniform  # noqa: E402

VERSION = "0.1.0"
TIME_MAX = 253402300799
COUNT_MAX = 1000000000
STATES = ("GREEN", "YELLOW", "ORANGE", "SUSPENDED")
TIERS = ("ANY GREEN", "TRUST TIER")
_SKEYS = ("certificate_current", "dis_match", "state", "reset_in_force",
          "suspended", "in_tier")


# ---------------------------------------------------------------- common

def _is_int(x, lo, hi):
    return type(x) is int and lo <= x <= hi


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 _is_hex(s, n):
    if type(s) is not str or len(s) != n:
        return False
    for ch in s:
        if ch not in "0123456789abcdef":
            return False
    return True


def _pf(params, key, lo, hi, lo_open=False, hi_open=False):
    if type(params) is not dict or key not in params:
        raise ValueError("params:" + key)
    v = _num(params[key])
    if v is None:
        raise ValueError("params:" + key)
    if v < lo or v > hi or (lo_open and v == lo) or (hi_open and v == hi):
        raise ValueError("params:" + key)
    return v


def _pi(params, key, lo, hi):
    if type(params) is not dict or key not in params:
        raise ValueError("params:" + key)
    v = params[key]
    if not _is_int(v, lo, hi):
        raise ValueError("params:" + key)
    return v


def _clip(x, lo, hi):
    if x < lo:
        return lo
    if x > hi:
        return hi
    return x


def _f(x):
    return canon_float(x)


def _b(x):
    return "true" if x else "false"


# --------------------------------------------------------- good_standing

def good_standing(facts, params):
    if not _keys_are(facts, _SKEYS):
        raise ValueError("facts")
    for k in ("certificate_current", "dis_match", "reset_in_force", "suspended", "in_tier"):
        if type(facts[k]) is not bool:
            raise ValueError("facts")
    if facts["state"] not in STATES:
        raise ValueError("facts")
    if type(params) is not dict or params.get("VOLUNTEER TIER") not in TIERS:
        raise ValueError("params:VOLUNTEER TIER")
    if not facts["certificate_current"]:
        return {"ok": False, "reason": "CERTIFICATE"}
    if not facts["dis_match"]:
        return {"ok": False, "reason": "D(IS)"}
    if facts["state"] != "GREEN":
        return {"ok": False, "reason": "STATE"}
    if facts["reset_in_force"]:
        return {"ok": False, "reason": "RESET"}
    if facts["suspended"]:
        return {"ok": False, "reason": "SUSPENDED"}
    if params["VOLUNTEER TIER"] == "TRUST TIER" and not facts["in_tier"]:
        return {"ok": False, "reason": "TIER"}
    return {"ok": True, "reason": "GOOD STANDING"}


def render_standing(d):
    return _b(d["ok"]) + "|" + d["reason"]


# ---------------------------------------------------------------- bucket

def _bucket(bucket):
    if not _keys_are(bucket, ("tokens", "updated")):
        raise ValueError("bucket")
    t = _num(bucket["tokens"])
    if t is None or t < 0.0 or not _is_int(bucket["updated"], 0, TIME_MAX):
        raise ValueError("bucket")
    return t, bucket["updated"]


def bucket_refill(bucket, now, params):
    tokens, updated = _bucket(bucket)
    if not _is_int(now, 0, TIME_MAX) or now < updated:
        raise ValueError("now")
    rate = _pf(params, "VOLUNTEER RATE", 0.0, 1e9, True)
    burst = _pi(params, "VOLUNTEER BURST", 1, COUNT_MAX)
    t = tokens + rate * float(now - updated) / 3600.0
    if t > float(burst):
        t = float(burst)
    return {"tokens": t, "updated": now}


def bucket_take(bucket):
    tokens, updated = _bucket(bucket)
    if tokens >= 1.0:
        return {"ok": True, "bucket": {"tokens": tokens - 1.0, "updated": updated}}
    return {"ok": False, "bucket": {"tokens": tokens, "updated": updated}}


def render_bucket(b):
    return "%s|%d" % (_f(b["tokens"]), b["updated"])


def render_take(d):
    return _b(d["ok"]) + "|" + render_bucket(d["bucket"])


# ----------------------------------------------------------- retry_after

def retry_after(attempt, r, params):
    if not _is_int(attempt, 0, COUNT_MAX):
        raise ValueError("attempt")
    if not _is_hex(r, 32):
        raise ValueError("r")
    base = _pf(params, "BACKOFF BASE", 1.0, float(TIME_MAX))
    cap = _pf(params, "BACKOFF CAP", 1.0, float(TIME_MAX))
    if cap < base:
        raise ValueError("params:BACKOFF CAP")
    if attempt >= 1024:
        bound = cap
    else:
        bound = base * math.pow(2.0, float(attempt))
        if bound > cap:
            bound = cap
    return uniform("TPC-VOLUNTEER-RETRY", r, 0) * bound


# ---------------------------------------------------------------- intake

def intake_update(intake, volunteer_blocks, landed_blocks, params):
    i = _num(intake)
    if i is None or i < 0.0 or i > 1.0:
        raise ValueError("intake")
    if not _is_int(volunteer_blocks, 0, COUNT_MAX) or not _is_int(landed_blocks, 0, COUNT_MAX):
        raise ValueError("blocks")
    if volunteer_blocks > landed_blocks:
        raise ValueError("blocks")
    w = _pf(params, "INTAKE w_q", 0.0, 1.0)
    share = 0.0 if landed_blocks == 0 else float(volunteer_blocks) / float(landed_blocks)
    return (1.0 - w) * i + w * share


def p_nw(intake, params):
    i = _num(intake)
    if i is None or i < 0.0 or i > 1.0:
        raise ValueError("intake")
    lo = _pf(params, "INTAKE MIN", 0.0, 1.0)
    hi = _pf(params, "INTAKE MAX", 0.0, 1.0)
    mp = _pf(params, "INTAKE MAX_P", 0.0, 1.0)
    if hi <= lo:
        raise ValueError("params:INTAKE MAX")
    if i <= lo:
        return 0.0
    if i > hi:
        return 1.0
    return (i - lo) / (hi - lo) * mp


# ----------------------------------------------------------------- p_v

def pv_step(state, r, params):
    if not _keys_are(state, ("p_v",)):
        raise ValueError("state")
    pv = state["p_v"]
    if pv is not None:
        pv = _num(pv)
        if pv is None or pv < 0.0 or pv > 1.0:
            raise ValueError("state")
    if not _is_hex(r, 32):
        raise ValueError("r")
    if type(params) is not dict or "P_V BOUNDS" not in params:
        raise ValueError("params:P_V BOUNDS")
    b = params["P_V BOUNDS"]
    if type(b) is not list or len(b) != 2:
        raise ValueError("params:P_V BOUNDS")
    lo = _num(b[0])
    hi = _num(b[1])
    if lo is None or hi is None or lo <= 0.0 or hi >= 1.0 or lo > hi:
        raise ValueError("params:P_V BOUNDS")
    rot = _pf(params, "P_V ROTATION", 0.0, 1.0)
    started = False
    if pv is None or uniform("TPC-PV-ROTATE", r, 0) < rot:
        pv = lo + uniform("TPC-PV-DRAW", r, 0) * (hi - lo)
        started = True
    return {"p_v": pv, "started": started}


def render_pv(d):
    return _f(d["p_v"]) + "|" + _b(d["started"])


# ------------------------------------------------------- volunteer_share

def volunteer_share(window, g, v_max):
    if type(window) is not list:
        raise ValueError("window")
    gf = _num(g)
    if gf is None or gf <= 0.0:
        raise ValueError("g")
    vm = _num(v_max)
    if vm is None or vm <= 0.0 or vm >= 1.0:
        raise ValueError("v_max")
    w_all = 0.0
    w_v = 0.0
    for e in window:
        if not _keys_are(e, ("volunteer", "weight")) or type(e["volunteer"]) is not bool:
            raise ValueError("window")
        w = _num(e["weight"])
        if w is None or w < 0.0:
            raise ValueError("window")
        w_all = w_all + w
        if e["volunteer"]:
            w_v = w_v + w
    q = _clip((vm * (w_all + gf) - w_v) / gf, 0.0, 1.0)
    return {"q": q, "W": w_all, "W_v": w_v, "admitted": gf * q}


def render_share(d):
    return "%s|%s|%s|%s" % (_f(d["q"]), _f(d["W"]), _f(d["W_v"]), _f(d["admitted"]))


# -------------------------------------------------------- volunteer_draw

_VKEYS = ("bucket", "retry_until", "attempt", "refusals", "period",
          "closed_until", "intake", "p_v")
_CKEYS = ("now", "block", "period", "has_item", "has_falcon",
          "has_own_record", "ratio_k", "standing")


def _state_out(st, bucket, retry_until, attempt, refusals, period, closed, intake, pv):
    return {"bucket": bucket, "retry_until": retry_until, "attempt": attempt,
            "refusals": refusals, "period": period, "closed_until": closed,
            "intake": intake, "p_v": pv}


def volunteer_draw(state, call, r, params):
    if not _keys_are(state, _VKEYS) or not _keys_are(call, _CKEYS):
        raise ValueError("state")
    _bucket(state["bucket"])
    ru = state["retry_until"]
    if ru is not None:
        ru = _num(ru)
        if ru is None or ru < 0.0:
            raise ValueError("state")
    if not _is_int(state["attempt"], 0, COUNT_MAX) or not _is_int(state["refusals"], 0, COUNT_MAX):
        raise ValueError("state")
    if not _is_int(state["period"], 0, COUNT_MAX):
        raise ValueError("state")
    closed = state["closed_until"]
    if closed is not None and not _is_int(closed, 0, COUNT_MAX):
        raise ValueError("state")
    intake = _num(state["intake"])
    if intake is None or intake < 0.0 or intake > 1.0:
        raise ValueError("state")
    pvs = state["p_v"]
    if not _keys_are(pvs, ("p_v",)):
        raise ValueError("state")
    if not _is_int(call["now"], 0, TIME_MAX) or not _is_int(call["block"], 0, COUNT_MAX):
        raise ValueError("call")
    if not _is_int(call["period"], 0, COUNT_MAX) or call["period"] < state["period"]:
        raise ValueError("call")
    for k in ("has_item", "has_falcon", "has_own_record"):
        if type(call[k]) is not bool:
            raise ValueError("call")
    kf = _num(call["ratio_k"])
    if kf is None or kf <= 0.0:
        raise ValueError("call")
    if not _is_hex(r, 32):
        raise ValueError("r")
    limit = _pi(params, "EARLY CALL LIMIT", 0, COUNT_MAX)
    cool = _pi(params, "VOLUNTEER COOL-OFF", 1, COUNT_MAX)
    share = _pf(params, "CONSISTENCY SHARE", 0.0, 1.0)
    gs = good_standing(call["standing"], params)
    now = call["now"]
    block = call["block"]
    refusals = state["refusals"] if call["period"] == state["period"] else 0
    period = call["period"]
    attempt = state["attempt"]
    rec = {"answer": None, "kind": None, "reason": None, "retry_after": None,
           "p_nw": None, "p_v": None, "intake": intake, "attempt": attempt}

    def refused(reason, bucket, count):
        ra = retry_after(attempt, r, params)
        until = float(now) + ra
        if ru is not None and ru > until:
            until = ru
        nref = refusals
        nclosed = closed
        if count:
            nref = refusals + 1
            if nref > limit:
                nclosed = block + cool
                nref = 0
        rec.update(answer="REFUSED", reason=reason, retry_after=ra)
        return {"record": rec, "state": _state_out(state, bucket, until, attempt, nref,
                                                   period, nclosed, intake, pvs)}
    if not gs["ok"]:
        return refused("STANDING", state["bucket"], False)
    if closed is not None and block < closed:
        return refused("CLOSED", state["bucket"], False)
    bucket = bucket_refill(state["bucket"], now, params)
    if ru is not None and float(now) < ru:
        return refused("EARLY", bucket, True)
    take = bucket_take(bucket)
    if not take["ok"]:
        return refused("TOKEN", bucket, True)
    bucket = take["bucket"]
    pv = pv_step(pvs, r, params)
    pnw = p_nw(intake, params)
    rec["p_nw"] = pnw
    rec["p_v"] = pv["p_v"]
    if call["has_falcon"]:
        rec["answer"] = "FALCON"
        rec["kind"] = "FALCON"
    elif uniform("TPC-VOLUNTEER-NW", r, 0) < pnw:
        rec["answer"] = "NO WORK"
        rec["reason"] = "THROTTLE"
        attempt = attempt + 1
    elif uniform("TPC-VOLUNTEER-PV", r, 0) < pv["p_v"]:
        rec["answer"] = "WORK"
        if uniform("TPC-VOLUNTEER-KIND", r, 0) < kf / (1.0 + kf):
            if uniform("TPC-VOLUNTEER-KIND", r, 1) < share and call["has_own_record"]:
                rec["kind"] = "CONSISTENCY PARROT"
            else:
                rec["kind"] = "CORROBORATION PARROT"
        else:
            rec["kind"] = "CANARY"
        attempt = 0
    elif call["has_item"]:
        rec["answer"] = "WORK"
        rec["kind"] = "ITEM"
        attempt = 0
    else:
        rec["answer"] = "NO WORK"
        rec["reason"] = "EMPTY"
        attempt = attempt + 1
    ra = retry_after(attempt, r, params)
    rec["retry_after"] = ra
    rec["attempt"] = attempt
    return {"record": rec, "state": _state_out(state, bucket, float(now) + ra, attempt, refusals,
                                               period, closed, intake, {"p_v": pv["p_v"]})}


def render_draw(d):
    rec = d["record"]
    st = d["state"]
    lines = ["answer|%s|%s|%s|%s" % (rec["answer"], "-" if rec["kind"] is None else rec["kind"],
                                     "-" if rec["reason"] is None else rec["reason"],
                                     _f(rec["retry_after"]))]
    lines.append("draw|%s|%s|%s|%d" % ("-" if rec["p_nw"] is None else _f(rec["p_nw"]),
                                       "-" if rec["p_v"] is None else _f(rec["p_v"]),
                                       _f(rec["intake"]), rec["attempt"]))
    lines.append("state|%s|%s|%d|%d|%d|%s|%s" % (
        render_bucket(st["bucket"]), _f(st["retry_until"]), st["attempt"], st["refusals"],
        st["period"], "-" if st["closed_until"] is None else str(st["closed_until"]),
        "-" if st["p_v"]["p_v"] is None else _f(st["p_v"]["p_v"])))
    return "\n".join(lines)


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

FNS = ("standing", "refill", "take", "retry", "intake", "p_nw", "pv",
       "share", "draw")


def run_case(fn, args):
    """One corpus case: the render, or "ERROR:" and the message."""
    if fn not in FNS:
        raise ValueError("fn")
    try:
        p = args.get("params")
        if fn == "standing":
            return render_standing(good_standing(args["facts"], p))
        if fn == "refill":
            return render_bucket(bucket_refill(args["bucket"], args["now"], p))
        if fn == "take":
            return render_take(bucket_take(args["bucket"]))
        if fn == "retry":
            return _f(retry_after(args["attempt"], args["r"], p))
        if fn == "intake":
            return _f(intake_update(args["intake"], args["volunteer_blocks"],
                                    args["landed_blocks"], p))
        if fn == "p_nw":
            return _f(p_nw(args["intake"], p))
        if fn == "pv":
            return render_pv(pv_step(args["state"], args["r"], p))
        if fn == "share":
            return render_share(volunteer_share(args["window"], args["g"], args["v_max"]))
        return render_draw(volunteer_draw(args["state"], args["call"], args["r"], p))
    except ValueError as ex:
        return "ERROR:" + str(ex)