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

Revision as of 04:37, 28 September 2026

# -*- coding: utf-8 -*-
"""
TPC-PY --- the hold-scope function (TPC step 2, increment "hold scope")
--- D111.1; charter CO-21; parrot handoff 0.5.0 4.2, 5.3, 13.3;
--- 0.2.0: the HOLD record's bibliography list (D171 (j), taken by
--- delegation 2026-09-26; tpm_biblio_resolution 0.1.0 section 6.14)
=======================================================================
0.2.0 (HOLD_FORM "2"): after a failed canary TPX re-runs D172's lookup on
every bibliography entry the account proposed or landed in the hold
window (tpc_biblio.biblio_hold_list) and lists in the HOLD record, in the
field BIBLIOGRAPHY ("biblio"), those whose re-run CHECK is not CONFIRMED;
entries are not HELD. hold_preimage and hold_record take the list as a
fourth argument (an empty list when the UWC's BIBLIOGRAPHY setting is OFF
or nothing is listed); the digest's field list gains, after the held
pairs, the list's count and its (bib_id, check) pairs; the HOLD record
gains "biblio" before "digest"; release_record reads form 2 (the hold
object must carry "biblio", whose digest is recomputed with it) and
carries the list unchanged into the RELEASE record, before its digest,
whose field list gains the count and pairs likewise. Every form-1 digest
moved once (fixtures only; no UWC exists).

SELECTED 2026-09-23 (D111.1, tpm_canary_lookback_pause 0.2.1 section 6):
after a failed canary TPX holds the account's recent work. HELD is a
value of the record's own status field; the prior status is kept and
restored on release; a HOLD record carries the cause and the list; a
RELEASE record restores the status after an independent, non-SAME-MODEL
REPRODUCTION matches. Spec: tpm_hold_scope 0.1.0. Standard library only;
Python 3.8+; Windows/Linux. Readings the sources leave open are marked
"taken by delegation 2026-09-24" in the spec note, section 6.

WHAT IS COMPUTED HERE (pure functions; TPX calls them, D111.1)
  hold_scope(history, lookback_passes)
      The held block set. history is the account's ordered blocks under
      the certificate in force: {"account", "admission_block", "blocks":
      [{"block_id", "canary": null | "PASS" | "FAIL"}, ...]}. The last
      block is the failing block. The scope reaches back to the block of
      the account's lookback_passes-th most recent passed canary,
      inclusive (LOOKBACK PASSES, declared 2), or, with fewer passes, to
      the admission block. Blocks before the admission block belong to a
      lapsed certificate: they are outside the scope and their passes do
      not anchor. A passed parrot is not an anchor (handoff 4.5); the
      caller records only canary outcomes in "canary".
      -> ["OK", [held block ids in order], anchor_kind, anchor_block,
          passes_counted] with anchor_kind "PASS" or "ADMISSION".
  held_closure(records, seeds)
      The HELD set of one register. seeds = {"account", "blocks"}: the
      failed account and the held blocks. Seeds are every record the
      account authored in a held block; every CORROBORATION it supplied
      (author, or "supplied_by" when TPX wrote it) in a held block; every
      record carried by a RETURN it vetted as AI-LEAD ("vetted_by", in
      the lead's block "vetted_in", the record's own block when absent)
      in a held block. Then every record whose citations or ASSUMPTIONS
      closure reaches a held record, transitively; a cycle does not loop.
      -> ["OK", [[id, prior_status], ...] sorted by id, [seed ids]].
      Collation: identifiers sort in Unicode code point order, which is
      UTF-8 byte order; both residences sort so.
  tpx_allocate(range)
      One identifier from the register's reserved range for TPX-written
      records (HOLD, RELEASE, a TPX-written corroboration): the leased-
      range form of TPS 5.7.5 held by TPX itself, so a write needs no
      mutex. -> ["OK", "<prefix>-<n>", [prefix, low, high, next]].
  hold_preimage(cause, seeds, held, biblio) / hold_record(hold_id, cause,
      seeds, held, biblio)
      The HOLD record: [hold_id, "HOLD", canary_id, account, snapshot,
      failing_block, anchor_kind, anchor_block, lookback_passes,
      held_blocks, seeds, [[id, prior_status], ...], [[bib_id, check],
      ...], digest]. biblio is the bibliography list of D171 (j): pairs
      of a BIB ID and its re-run CHECK value, none CONFIRMED, distinct,
      sorted by bib_id in code point order; [] when there is none. The
      digest is digest_fields (D97.1) over the tagged, counted, flat
      field list the preimage function shows; it excludes hold_id (the
      address).
  release_record(release_id, hold, reproduction, records)
      The RELEASE record. hold is the HOLD record's fields as an object;
      its digest is recomputed and a mismatch refused. reproduction is
      {"kind": "REPRODUCTION" | "LEDGER_MATCH", "id", "independence":
      {"account", "snapshot", "origin", "same_model"?}, "matched": [ids]}.
      Eligible when the independence names an account other than the held
      one, is not SAME-MODEL (flag, or the same snapshot), and its origin
      is REPRODUCTION BLOCK; a community-ledger MATCH counts on the same
      check (handoff 5.3, 13.3). The matched seeds leave the seed set,
      the closure is recomputed over the remaining seeds, and every held
      record no longer reached is released with its prior status; a
      matched record still reached by a remaining seed stays held.
      -> ["OK", [release_id, "RELEASE", hold_id, hold_digest, kind,
          reproduction_id, account, snapshot, [[id, restored_status], ...],
          [still held ids], [[bib_id, check], ...] (the HOLD's list,
          unchanged), digest]].

REFUSED, as ["REFUSED", reason], never raised; a reason that quotes input
escapes disallowed code points (zone schemas 6.3). A HOLD or RELEASE with
nothing to hold or release is refused, never written empty.

NOT HERE. Writing HELD into the records, the review list of older
same-snapshot work, the audit block on a mismatch or with no eligible
reproducer, and the hold-in-force check at release (a record listed by
another HOLD in force is restored no further than HELD) are TPX's (build
step 3). The status value a withdrawal sets on a struck record waits on
the record field schema (C-16) and is owed (spec note section 8).
"""

import tpc_digest as G
import tpc_zone as Z

TPC_HOLD_VERSION = "0.2.0"
HOLD_TAG = "TPC-HOLD"
RELEASE_TAG = "TPC-RELEASE"
HOLD_FORM = "2"                              # 0.2.0: the bibliography list (D171 (j))
LOOKBACK_PASSES_DECLARED = 2                 # PMP-TPM DECLARED PARAMETERS (D111.1)
CANARY_VALUES = ("PASS", "FAIL")
ANCHOR_KINDS = ("PASS", "ADMISSION")
ORIGIN_REPRODUCTION = "REPRODUCTION BLOCK"
REPRODUCTION_KINDS = ("REPRODUCTION", "LEDGER_MATCH")
RECORD_TEXT_FIELDS = ("block_id", "author", "kind", "status")
RECORD_LIST_FIELDS = ("cites", "assumptions")
RECORD_OPTIONAL_FIELDS = ("supplied_by", "vetted_by", "vetted_in")
HOLD_FIELDS = ("hold_id", "canary_id", "account", "snapshot", "failing_block", "anchor_kind",
               "anchor_block", "lookback_passes", "held_blocks", "seeds", "held", "biblio", "digest")
CHECK_CONFIRMED = "CONFIRMED"                # a listed entry's re-run CHECK is never CONFIRMED (D171 (j))


# ---- helpers -------------------------------------------------------------------
def _refuse(msg):
    return ["REFUSED", Z._show(msg)]


def _is_text(x):
    """A non-empty string with no disallowed code point (zone schemas 6)."""
    return isinstance(x, str) and x != "" and Z._first_bad(x) is None


def _is_int(x):
    return isinstance(x, int) and not isinstance(x, bool)


def _text_list(v):
    """The list when v is a list of texts, else None."""
    if not Z._list(v):
        return None
    for x in v:
        if not _is_text(x):
            return None
    return v


def _unique(v):
    return len(set(v)) == len(v)


# ---- hold_scope ------------------------------------------------------------------
def hold_scope(history, lookback_passes):
    if not _is_int(lookback_passes) or lookback_passes < 1:
        return _refuse("LOOKBACK PASSES is not an integer of at least 1")
    if not Z._obj(history):
        return _refuse("history is not an object")
    if not _is_text(history.get("account")):
        return _refuse("history: account is not a name")
    adm = history.get("admission_block")
    if not _is_text(adm):
        return _refuse("history: admission_block is not a name")
    blocks = history.get("blocks")
    if not Z._list(blocks) or not blocks:
        return _refuse("history: blocks is not a non-empty list")
    ids = []
    for b in blocks:
        if not Z._obj(b):
            return _refuse("history: block is not an object")
        bid = b.get("block_id")
        if not _is_text(bid):
            return _refuse("history: block_id is not a name")
        if bid in ids:
            return _refuse("history: duplicate block " + bid)
        c = b.get("canary")
        if c is not None and c not in CANARY_VALUES:
            return _refuse("history: block " + bid + ": canary is not null, PASS or FAIL")
        ids.append(bid)
    if adm not in ids:
        return _refuse("history: admission block " + adm + " is not in blocks")
    fi = len(blocks) - 1
    if blocks[fi].get("canary") != "FAIL":
        return _refuse("history: the last block is not the failing block")
    ai = ids.index(adm)
    passes, anchor = 0, None
    i = fi - 1
    while i >= ai:                                # the failing block excluded; the admission block included
        c = blocks[i].get("canary")
        if c == "FAIL":
            return _refuse("history: block " + ids[i] + " failed a canary inside the scope; an earlier failure is its own hold")
        if c == "PASS":
            passes += 1
            if passes == lookback_passes:
                anchor = i
                break
        i -= 1
    if anchor is None:
        kind, start = "ADMISSION", ai
    else:
        kind, start = "PASS", anchor
    return ["OK", ids[start:fi + 1], kind, ids[start], passes]


# ---- records -----------------------------------------------------------------------
def _read_records(records):
    """(by_id, None) or (None, reason)."""
    if not Z._list(records):
        return None, "records is not a list"
    by = {}
    for r in records:
        if not Z._obj(r):
            return None, "records: entry is not an object"
        rid = r.get("id")
        if not _is_text(rid):
            return None, "records: id is not a name"
        if rid in by:
            return None, "records: duplicate id " + rid
        for f in RECORD_TEXT_FIELDS:
            if not _is_text(r.get(f)):
                return None, "records: " + rid + ": " + f + " is not a name"
        for f in RECORD_LIST_FIELDS:
            if _text_list(r.get(f, [])) is None:
                return None, "records: " + rid + ": " + f + " is not a list of names"
        for f in RECORD_OPTIONAL_FIELDS:
            if f in r and not _is_text(r[f]):
                return None, "records: " + rid + ": " + f + " is not a name"
        by[rid] = r
    return by, None


def _seed_ids(by, account, blocks):
    bset = set(blocks)
    out = []
    for rid in sorted(by):
        r = by[rid]
        if r["author"] == account and r["block_id"] in bset:
            out.append(rid)
        elif r["kind"] == "CORROBORATION" and r.get("supplied_by", r["author"]) == account \
                and r["block_id"] in bset:
            out.append(rid)
        elif r.get("vetted_by") == account and r.get("vetted_in", r["block_id"]) in bset:
            out.append(rid)
    return out


def _propagate(by, seed_ids):
    """The seeds plus every record whose citations or ASSUMPTIONS closure
    reaches one, over reverse edges; a visited set guards cycles."""
    rev = dict((rid, []) for rid in by)
    for rid in sorted(by):
        for dep in list(by[rid].get("cites", [])) + list(by[rid].get("assumptions", [])):
            if dep in rev:
                rev[dep].append(rid)
    held = set()
    stack = list(seed_ids)
    while stack:
        cur = stack.pop()
        if cur in held:
            continue
        held.add(cur)
        stack.extend(rev[cur])
    return held


def held_closure(records, seeds):
    by, err = _read_records(records)
    if err is not None:
        return _refuse(err)
    if not Z._obj(seeds):
        return _refuse("seeds is not an object")
    account = seeds.get("account")
    if not _is_text(account):
        return _refuse("seeds: account is not a name")
    blocks = _text_list(seeds.get("blocks"))
    if blocks is None:
        return _refuse("seeds: blocks is not a list of names")
    sids = _seed_ids(by, account, blocks)
    held = _propagate(by, sids)
    pairs = [[rid, by[rid]["status"]] for rid in sorted(held)]
    return ["OK", pairs, sids]


# ---- the reserved range ----------------------------------------------------------------
def tpx_allocate(rng):
    if not Z._obj(rng):
        return _refuse("range is not an object")
    prefix = rng.get("prefix")
    if not Z._is_ident(prefix):
        return _refuse("range: prefix is not an identifier")
    low, high, nxt = rng.get("low"), rng.get("high"), rng.get("next")
    if not (_is_int(low) and _is_int(high) and low >= 1 and low <= high):
        return _refuse("range: low and high are not integers with 1 <= low <= high")
    if not _is_int(nxt) or nxt < low:
        return _refuse("range: next is not an integer at or above low")
    if nxt > high:
        return _refuse("range: the reserved range " + prefix + " " + str(low) + "-" + str(high) + " is exhausted")
    return ["OK", prefix + "-" + str(nxt), [prefix, low, high, nxt + 1]]


# ---- HOLD --------------------------------------------------------------------------------
def _read_pairs(held):
    """Sorted [[id, status], ...] or a reason string."""
    if not Z._list(held):
        return "held is not a list"
    out, ids = [], set()
    for p in held:
        if not Z._list(p) or len(p) != 2 or not _is_text(p[0]) or not _is_text(p[1]):
            return "held: entry is not [id, prior_status]"
        if p[0] in ids:
            return "held: duplicate id " + p[0]
        ids.add(p[0])
        out.append([p[0], p[1]])
    out.sort(key=lambda p: p[0])
    return out


def _read_cause(cause):
    """The validated cause, or a reason string."""
    if not Z._obj(cause):
        return "cause is not an object"
    for f in ("canary_id", "account", "snapshot", "failing_block", "anchor_kind", "anchor_block"):
        if not _is_text(cause.get(f)):
            return "cause: " + f + " is not a name"
    if cause["anchor_kind"] not in ANCHOR_KINDS:
        return "cause: anchor_kind is not PASS or ADMISSION"
    lb = cause.get("lookback_passes")
    if not _is_int(lb) or lb < 1:
        return "cause: lookback_passes is not an integer of at least 1"
    hb = _text_list(cause.get("held_blocks"))
    if hb is None or not hb or not _unique(hb):
        return "cause: held_blocks is not a non-empty list of distinct names"
    if hb[-1] != cause["failing_block"]:
        return "cause: failing_block is not the last held block"
    if hb[0] != cause["anchor_block"]:
        return "cause: anchor_block is not the first held block"
    return cause


def _read_biblio(biblio):
    """Sorted [[bib_id, check], ...] or a reason string (0.2.0, D171 (j))."""
    if not Z._list(biblio):
        return "biblio is not a list"
    out, ids = [], set()
    for p in biblio:
        if not Z._list(p) or len(p) != 2 or not _is_text(p[0]) or not _is_text(p[1]):
            return "biblio: entry is not [bib_id, check]"
        if p[1] == CHECK_CONFIRMED:
            return "biblio: entry " + p[0] + " is CONFIRMED; the list holds what is not"
        if p[0] in ids:
            return "biblio: duplicate entry " + p[0]
        ids.add(p[0])
        out.append([p[0], p[1]])
    out.sort(key=lambda p: p[0])
    return out


def _hold_fields(cause, seeds, pairs, biblio):
    hb = cause["held_blocks"]
    fields = [HOLD_TAG, HOLD_FORM, cause["canary_id"], cause["account"], cause["snapshot"],
              cause["failing_block"], cause["anchor_kind"], cause["anchor_block"],
              cause["lookback_passes"], len(hb)] + list(hb) + [len(seeds)] + list(seeds) + [len(pairs)]
    for p in pairs:
        fields += [p[0], p[1]]
    fields.append(len(biblio))
    for p in biblio:
        fields += [p[0], p[1]]
    return fields


def _hold_parts(cause, seeds, held, biblio):
    """(cause, sorted seeds, sorted pairs, sorted biblio) or a reason string."""
    c = _read_cause(cause)
    if isinstance(c, str):
        return c
    s = _text_list(seeds)
    if s is None or not _unique(s):
        return "seeds is not a list of distinct names"
    pairs = _read_pairs(held)
    if isinstance(pairs, str):
        return pairs
    ids = set(p[0] for p in pairs)
    for x in s:
        if x not in ids:
            return "seed " + x + " is not in the held list"
    b = _read_biblio(biblio)
    if isinstance(b, str):
        return b
    return c, sorted(s), pairs, b


def hold_preimage(cause, seeds, held, biblio):
    """The exact bytes the HOLD digest covers, as text (auditable)."""
    parts = _hold_parts(cause, seeds, held, biblio)
    if isinstance(parts, str):
        return _refuse(parts)
    c, s, pairs, b = parts
    return ["OK", G.preimage_fields(_hold_fields(c, s, pairs, b)).decode("utf-8")]


def hold_record(hold_id, cause, seeds, held, biblio):
    if not _is_text(hold_id):
        return _refuse("hold_id is not a name")
    parts = _hold_parts(cause, seeds, held, biblio)
    if isinstance(parts, str):
        return _refuse(parts)
    c, s, pairs, b = parts
    if not pairs:
        return _refuse("nothing to hold: the held list is empty")
    digest = G.digest_fields(_hold_fields(c, s, pairs, b))
    return ["OK", [hold_id, "HOLD", c["canary_id"], c["account"], c["snapshot"], c["failing_block"],
                   c["anchor_kind"], c["anchor_block"], c["lookback_passes"], list(c["held_blocks"]),
                   s, pairs, b, digest]]


# ---- RELEASE --------------------------------------------------------------------------------
def release_record(release_id, hold, reproduction, records):
    if not _is_text(release_id):
        return _refuse("release_id is not a name")
    if not Z._obj(hold):
        return _refuse("hold is not an object")
    for f in HOLD_FIELDS:
        if f not in hold:
            return _refuse("hold: missing field " + f)
    if not _is_text(hold["hold_id"]):
        return _refuse("hold: hold_id is not a name")
    parts = _hold_parts(hold, hold["seeds"], hold["held"], hold["biblio"])
    if isinstance(parts, str):
        return _refuse("hold: " + parts)
    c, seeds, pairs, biblio = parts
    if not pairs:
        return _refuse("hold: the held list is empty")
    if hold["digest"] != G.digest_fields(_hold_fields(c, seeds, pairs, biblio)):
        return _refuse("hold: digest does not match its fields")
    if not Z._obj(reproduction):
        return _refuse("reproduction is not an object")
    kind = reproduction.get("kind")
    if kind not in REPRODUCTION_KINDS:
        return _refuse("reproduction: kind is not REPRODUCTION or LEDGER_MATCH")
    if not _is_text(reproduction.get("id")):
        return _refuse("reproduction: id is not a name")
    ind = reproduction.get("independence")
    if not Z._obj(ind):
        return _refuse("reproduction: independence is not an object")
    for f in ("account", "snapshot", "origin"):
        if not _is_text(ind.get(f)):
            return _refuse("reproduction: independence " + f + " is not a name")
    if "same_model" in ind and not isinstance(ind["same_model"], bool):
        return _refuse("reproduction: independence same_model is not a boolean")
    matched = _text_list(reproduction.get("matched"))
    if matched is None or not _unique(matched):
        return _refuse("reproduction: matched is not a list of distinct names")
    # eligibility (D111.1; D42): not the failed account, not SAME-MODEL, a REPRODUCTION block
    if ind["account"] == c["account"]:
        return _refuse("reproduction by the held account is not independent")
    if ind.get("same_model") is True or ind["snapshot"] == c["snapshot"]:
        return _refuse("reproduction is SAME-MODEL")
    if ind["origin"] != ORIGIN_REPRODUCTION:
        return _refuse("reproduction origin is not REPRODUCTION BLOCK")
    held_ids = set(p[0] for p in pairs)
    for m in matched:
        if m not in held_ids:
            return _refuse("matched record " + m + " is not in the hold")
    by, err = _read_records(records)
    if err is not None:
        return _refuse(err)
    for rid in sorted(held_ids):
        if rid not in by:
            return _refuse("held record " + rid + " is not in records")
    remaining = [s for s in seeds if s not in matched]
    still = _propagate(by, remaining) & held_ids
    released = sorted(held_ids - still)
    if not released:
        return _refuse("no held record is released by this reproduction")
    restored = [[p[0], p[1]] for p in pairs if p[0] not in still]
    still_sorted = sorted(still)
    fields = [RELEASE_TAG, HOLD_FORM, hold["hold_id"], hold["digest"], kind, reproduction["id"],
              ind["account"], ind["snapshot"], len(restored)]
    for p in restored:
        fields += [p[0], p[1]]
    fields += [len(still_sorted)] + still_sorted + [len(biblio)]
    for p in biblio:
        fields += [p[0], p[1]]
    digest = G.digest_fields(fields)
    return ["OK", [release_id, "RELEASE", hold["hold_id"], hold["digest"], kind, reproduction["id"],
                   ind["account"], ind["snapshot"], restored, still_sorted, biblio, digest]]