felhom.eu/scripts/hub_copy_gate.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""hub_copy_gate — the hub's customer-facing copy comes under a guard (R-324).

THE SCOPE GAP THIS CLOSES. `retrieval_promise_gate.py` lives in `felhom-controller` and scans that
repo only. It was extended to Go string literals on 2026-08-12 (R-311) on the express ground that the
recovery screen's messages are Go strings in a handler and "none of it had ever been scanned" — and
the identical sentence was true one repo over the whole time. **The hub composes every customer
e-mail and renders the binding pages: the first sentences a customer ever reads, before they have
seen any box screen.** Nothing looked at them.

A hand scan on 2026-08-13 returned zero retrieval claims, so this was a SCOPE gap rather than a live
defect — which is the moment to close it, before someone writes the first hub-side promise assuming
the guard has them covered.

TWO CHECKS, and the difference is deliberate:

  1. RETIRED NAMES — banned outright, scanned across the WHOLE hub, not just declared surfaces. A
     name for a secret that a different secret now owns is never correct anywhere, so there is no
     allowlist and no "unless". Comments are stripped: prose explaining a rename is not the rename
     coming back, and the register rows that record these decisions quote the retired names by
     necessity.

  2. RETRIEVAL STEMS — registered, not banned, in the DECLARED customer-facing surfaces. The claim
     they carry is sometimes true and must stay sayable; what must not happen is a new one appearing
     where nobody was looking. Same contract as the controller gate: an occurrence is allowlisted
     WITH A REASON, and a stale allowlist entry is also a failure.

BOTH lists come from `customer_copy_vocab.py` — see that file for why they are not literals here.

DRIFT CHECK, and it is load-bearing. The controller gate still owns its own `STEMS` literal (the
session that wrote this could not touch felhom-controller). This gate reads that literal and FAILS if
it disagrees with the shared list, so the two cannot silently diverge in the meantime. An ABSENT
sibling clone is INCONCLUSIVE (exit 2), never a pass — the G-1 lesson: a gate that skips when its
sibling is missing runs in neither home.

Run:  python3 scripts/hub_copy_gate.py            (from the felhom.eu repo root)
      python3 scripts/hub_copy_gate.py --selftest (plant → convict → remove → pass)
"""
import json
import os
import re
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from customer_copy_vocab import (  # noqa: E402
    RETIRED_NAMES, RETRIEVAL_STEMS, RETRIEVAL_STEMS_EN)

_HERE = os.path.dirname(os.path.abspath(__file__))
_REPO = os.path.dirname(_HERE)
_HUB = os.path.join(_REPO, "hub")

# The hub files that carry HUNGARIAN CUSTOMER-FACING text — the surfaces where a retrieval claim
# could be made to a customer. Enumerated by scanning every non-test .go/.html under hub/internal for
# accented Hungarian outside comments, then keeping the ones a CUSTOMER (not the operator) reads.
#
# A declared file that is missing is a FAILURE, never a skip: the controller gate learned that when a
# renamed handler would have silently emptied its own scope.
CUSTOMER_SURFACES = [
    os.path.join("hub", "internal", "notify", "templates.go"),   # the mail wrappers + event copy
    os.path.join("hub", "internal", "web", "selfbind.go"),       # the binding pages
    os.path.join("hub", "internal", "api", "handler.go"),        # customer-visible event messages
    os.path.join("hub", "internal", "notify", "dispatcher.go"),  # the customer channel's own wording
    # ── THE BUNDLES (hub v0.118.0, R-558) ────────────────────────────────────────────────────────
    #
    # ⚠ THE SENTENCES MOVED, AND THIS IS THE LINE THAT KEPT THE GATE HONEST. Until v0.118.0 every
    # customer sentence was a Go literal in templates.go and selfbind.go; slice 3 moved all of them
    # into these two JSON files. Without adding them here, the four files above would still all
    # exist, the gate would still report "4 surfaces scanned", and it would be scanning NOTHING —
    # a green gate over an empty scope, which is the exact shape this project has found sixteen
    # times (AUDIT-gate-decoys-2026-09-01, "scope is a fact too").
    os.path.join("hub", "internal", "i18n", "locales", "hu.json"),
    os.path.join("hub", "internal", "i18n", "locales", "en.json"),
]

# (file basename, substring that identifies the occurrence) -> why it is allowed.
#
# It was empty until v0.118.0, when the English stems were added (R-558) and found exactly one
# occurrence. That it found one rather than none is the argument for having added them.
ALLOWLIST = {
    ("handler.go", "that history stays recoverable with the recovery code that sealed it"): (
        "OPERATOR-TIER, and verified so rather than assumed: the sentence is the message of "
        "`offsite_repo_key_changed`, which is listed in notify/dispatcher.go's operatorOnlyEvents "
        "with its own R-197 reasoning — a custody fact about escrow blobs that a customer can take "
        "no action on. It is operator-grade English naming host ids and retained-blob counts, and it "
        "reaches no customer channel. handler.go is a DECLARED customer surface because it also "
        "holds customer-visible event messages, so the file is right to be scanned and this "
        "occurrence is right to be registered. ⚠ If offsite_repo_key_changed is ever REMOVED from "
        "operatorOnlyEvents, this entry becomes false and the claim reaches households — the promise "
        "is one R-304 says the product cannot keep for a customer, because nothing in the product "
        "reads a retained key and a customer's correct old code is reported as wrong."),
}

def scannable_text(path):
    """The text a CUSTOMER could read, with everything else removed.

    For a bundle, that is the VALUES ONLY. The keys are identifiers — `mail.event.node_recovered`,
    `disaster_recovery_completed` — and scanning them convicts the machinery for the words it names
    things with. That is not a near miss either: keys accounted for a third of the first run's false
    convictions, and every one of them would have earned a permanent allowlist entry explaining that
    a key is not a sentence.
    """
    if path.endswith(".json"):
        with open(path, encoding="utf-8") as fh:
            return "\n".join(json.load(fh).values())
    return strip_comments(open(path, encoding="utf-8").read(), path)


GO_COMMENT = re.compile(r"//[^\n]*|/\*.*?\*/", re.S)
TPL_COMMENT = re.compile(r"\{\{/\*.*?\*/\}\}", re.S)
HTML_COMMENT = re.compile(r"<!--.*?-->", re.S)


def strip_comments(text, path):
    """Remove what never reaches a customer.

    Go `//` and `/* */`, Go-template `{{/* */}}`, and — unlike the controller gate — HTML `<!-- -->`
    too. The controller gate deliberately SCANS HTML comments because its templates ship to a browser
    where View Source is one keystroke. The hub's Hungarian lives in Go string literals that happen
    to contain HTML, and its operator templates are not customer copy, so an HTML comment here is a
    note to the next maintainer. Stated because it is a real difference between the two gates.
    """
    text = TPL_COMMENT.sub("", text)
    text = HTML_COMMENT.sub("", text)
    if path.endswith(".go"):
        text = GO_COMMENT.sub("", text)
    return text


def _iter_hub_sources():
    for root, _dirs, files in os.walk(os.path.join(_REPO, "hub", "internal")):
        for f in sorted(files):
            if f.endswith("_test.go") or not (f.endswith(".go") or f.endswith(".html")):
                continue
            yield os.path.join(root, f)


def scan_retired(extra_text=None):
    """Every retired name, everywhere in the hub. Returns a list of convictions."""
    convictions = []
    sources = [(p, open(p, encoding="utf-8").read()) for p in _iter_hub_sources()]
    if extra_text is not None:
        # The synthetic path MUST end in .go: strip_comments keys comment-stripping off the
        # extension, so a selftest source named anything else would be scanned WITH its comments —
        # which is how the first run of this selftest convicted its own step-3 control. The bug was
        # in the instrument, and the control is what found it.
        sources.append((os.path.join(_REPO, "hub", "internal", "selftest_planted.go"), extra_text))
    for path, raw in sources:
        text = strip_comments(raw, path)
        for name in RETIRED_NAMES:
            for m in re.finditer(re.escape(name), text):
                line = text[: m.start()].count("\n") + 1
                ctx = " ".join(text[max(0, m.start() - 70): m.end() + 70].split())[:150]
                convictions.append((os.path.relpath(path, _REPO), line, name, ctx))
    return convictions


def scan_stems():
    """Retrieval stems in the declared customer surfaces. Returns (convictions, seen_keys)."""
    convictions, seen = [], set()
    for rel in CUSTOMER_SURFACES:
        path = os.path.join(_REPO, rel)
        if not os.path.exists(path):
            raise SystemExit("hub-copy gate: declared customer surface is MISSING: %s" % rel)
        name = os.path.basename(path)
        text = scannable_text(path)
        # Both halves of the claim are scanned in every surface, deliberately rather than per-file:
        # an English sentence can be written into a Go literal and a Hungarian one into en.json, and
        # a gate that assumed each file holds one language would miss exactly the mistake that
        # matters.
        patterns = [(re.escape(st) + r"[a-záéíóöőúüű]*") for st in RETRIEVAL_STEMS]
        patterns += list(RETRIEVAL_STEMS_EN)  # already regexes; they carry their own modal
        for pattern in patterns:
            for m in re.finditer(pattern, text, re.I):
                hit = None
                for k in (k for k in ALLOWLIST if k[0] == name):
                    for om in re.finditer(re.escape(k[1]), text):
                        if om.start() <= m.start() and m.end() <= om.end():
                            hit = k
                            break
                    if hit:
                        break
                if hit:
                    seen.add(hit)
                else:
                    line = text[: m.start()].count("\n") + 1
                    ctx = " ".join(text[max(0, m.start() - 90): m.end() + 90].split())[:170]
                    convictions.append((name, line, m.group(0), ctx))
    return convictions, seen


def check_drift():
    """The controller gate's STEMS must equal the shared list. Returns (status, message).

    status: "ok" | "drift" | "inconclusive"
    """
    sibling = os.path.join(os.path.dirname(_REPO), "felhom-controller",
                           "controller", "scripts", "retrieval_promise_gate.py")
    if not os.path.exists(sibling):
        return "inconclusive", ("the felhom-controller clone is absent, so the shared stem list "
                               "could not be compared against the gate that also uses it (%s)" % sibling)
    text = open(sibling, encoding="utf-8").read()
    m = re.search(r"^STEMS\s*=\s*\[(.*?)\]", text, re.M | re.S)
    if not m:
        return "drift", "could not find a STEMS list in the controller gate — its shape changed"
    theirs = re.findall(r"[\"']([^\"']+)[\"']", m.group(1))
    if theirs != RETRIEVAL_STEMS:
        return "drift", ("the controller gate's STEMS have diverged from customer_copy_vocab.py\n"
                         "      controller : %r\n      shared     : %r" % (theirs, RETRIEVAL_STEMS))
    return "ok", "controller gate's STEMS match the shared list (%d stem(s))" % len(theirs)


def selftest():
    """Plant → convict → remove → pass. A guard never seen catching anything proves nothing."""
    print("hub-copy gate SELFTEST")
    baseline = scan_retired()
    if baseline:
        print("  FAIL: the tree is not clean before planting — %d conviction(s)" % len(baseline))
        for c in baseline:
            print("    %s:%d  [%s]" % (c[0], c[1], c[2]))
        return 1
    print("  1. clean tree                     : 0 conviction(s)  OK")

    planted = 'body := "Add meg a visszaállító jelszavadat a folytatáshoz."\n'
    convicted = scan_retired(extra_text=planted)
    if len(convicted) != 1 or convicted[0][2] != "isszaállító jelsz":
        print("  2. planted retired name           : NOT CONVICTED — the guard is inert")
        print("     got: %r" % (convicted,))
        return 1
    print("  2. planted „visszaállító jelszavadat”: CONVICTED (%s)  OK" % convicted[0][2])

    commented = '// the old name was „visszaállító jelszó" and is retired\n'
    if scan_retired(extra_text=commented):
        print("  3. same phrase inside a COMMENT   : convicted — comments must not be scanned")
        return 1
    print("  3. same phrase inside a comment   : not convicted  OK")

    if scan_retired():
        print("  4. planting removed               : still convicting — the scan is not deterministic")
        return 1
    print("  4. planting removed               : 0 conviction(s)  OK")
    print("hub-copy gate selftest OK — the guard has been watched catching, ignoring and releasing")
    return 0


def main():
    if "--selftest" in sys.argv:
        return selftest()

    retired = scan_retired()
    stems, seen = scan_stems()
    stale = [k for k in ALLOWLIST if k not in seen]
    drift_status, drift_msg = check_drift()

    for path, line, name, ctx in retired:
        print("  %s:%d RETIRED NAME in customer copy (%s):\n      …%s…" % (path, line, name, ctx))
        print("      reason it is retired: %s" % RETIRED_NAMES[name])
    for name, line, word, ctx in stems:
        print("  %s:%d unregistered retrieval claim (%s):\n      …%s…" % (name, line, word, ctx))
    for k in stale:
        print("  STALE ALLOWLIST ENTRY (no longer present): %s :: %r" % (k[0], k[1]))

    n_files = len(list(_iter_hub_sources()))
    if retired or stems or stale:
        print("\nHUB-COPY GATE FAILED: %d retired name(s), %d unregistered claim(s), %d stale."
              % (len(retired), len(stems), len(stale)))
        print("A retired name is never correct in customer copy — use the current name. If a new")
        print("retrieval claim is genuinely TRUE where it renders, make it conditional on what the")
        print("hub can actually see, then register it in ALLOWLIST with the reason.")
        return 1

    if drift_status == "drift":
        print("  DRIFT: %s" % drift_msg)
        print("\nHUB-COPY GATE FAILED: the shared vocabulary is no longer shared.")
        return 1

    print("hub-copy gate OK — %d hub file(s) scanned for %d retired name(s); %d customer surface(s) "
          "scanned for %d retrieval pattern(s) (%d hu + %d en), %d registered claim(s), none unregistered"
          % (n_files, len(RETIRED_NAMES), len(CUSTOMER_SURFACES),
             len(RETRIEVAL_STEMS) + len(RETRIEVAL_STEMS_EN),
             len(RETRIEVAL_STEMS), len(RETRIEVAL_STEMS_EN), len(ALLOWLIST)))
    if drift_status == "inconclusive":
        print("  ⚠ INCONCLUSIVE (exit 2): %s" % drift_msg)
        return 2
    print("  drift: %s" % drift_msg)
    print("  (BLIND SPOT: this checks the WORDS in the %d declared customer surfaces. It cannot"
          % len(CUSTOMER_SURFACES))
    print("   tell whether a true-looking sentence is wired to a predicate that is actually true —")
    print("   that is what render tests are for. And it does not read the operator's screens.)")
    return 0


if __name__ == "__main__":
    sys.exit(main())