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 os
import re
import sys

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

_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"),   # every customer e-mail + 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
]

# (file basename, substring that identifies the occurrence) -> why it is allowed.
# Empty today, and that is a measurement rather than an oversight: the hub makes no retrieval promise.
ALLOWLIST = {}

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 = strip_comments(open(path, encoding="utf-8").read(), path)
        for stem in RETRIEVAL_STEMS:
            for m in re.finditer(re.escape(stem) + r"[a-záéíóöőúüű]*", text):
                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 stem(s), %d registered claim(s), none unregistered"
          % (n_files, len(RETIRED_NAMES), len(CUSTOMER_SURFACES), len(RETRIEVAL_STEMS), 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 four declared customer surfaces. It cannot")
    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())