#!/usr/bin/env python3
"""profiles.py: is each vendor sentence a profile quotes still on a page the profile links?

claims.py reads the quotations inside structured job claims and checks them
against the one `source` each claim names. The PROFILES carry most of this
registry's quotations (HubSpot's version 2 withdrew twenty from one profile;
ToolsGroup's three inventory sentences drifted off the vendor's page within
hours of filing) and nothing automated has ever read them.

A profile does not say which of its links carries which quotation, so this
asks the conservative question: is the quoted span on ANY page the entry
points at (every https link in the profile, the entry's homepage and docs,
every claim source)? A span found nowhere is a CANDIDATE, never a finding:
it needs the three-way test (archive, rendered browser, the words never
existed) before it is called anything.

Two kinds of quotation are verified against the registry instead of against a
page, because no page will ever carry them: the registry's own JSON and job
definitions (`registry:own-json-or-jobs`), and, since wake ec0e3ab2, the
MEASUREMENT fields of an evidence record filed about the same subject
(`registry:own-evidence`), which is where a server's own error text lives.
Only `surface`, `request` and `observed` are read there, never `finding`: see
`evidence_corpus` for the drift this would otherwise mask.

Quotation marks in a profile do many jobs besides quoting a vendor: a wire
body ("RBAC: access denied"), a header value, a field name, a tool name. So
only prose-shaped spans are scored: five words or more, mostly letters, not
inside code, no URL inside. Everything else is counted and skipped, and the
count is printed so the filter is visible.

    python3 tools/profiles.py [--clone /tmp/reg] [--ref origin/main] [--json out.json] [--only slug ...]

Written at wake c15fd999 (2026-09-18), designed at wake 9112e04d and left
unwritten then because it was the last forty minutes of a wake.
"""

import sys as _sys

_here = __file__.rsplit("/", 1)[0]
_sys.path[:] = [p for p in _sys.path if p not in ("", ".", _here)]

import argparse
import concurrent.futures
import datetime
import glob
import html
import json
import os
import re
import urllib.error
import urllib.request

import importlib.util as _ilu
_cspec = _ilu.spec_from_file_location("corpus", os.path.join(_here, "corpus.py"))
corpus = _ilu.module_from_spec(_cspec)
_cspec.loader.exec_module(corpus)

import importlib.util as _flilu
import os as _flos
# One disclosure for every sweep (tools/fetchlib.py). Loaded by path
# because this file drops its own directory from sys.path above, which is
# this directory's convention for importing a sibling.
_flspec = _flilu.spec_from_file_location(
    "fetchlib", _flos.path.join(_flos.path.dirname(_flos.path.abspath(__file__)), "fetchlib.py"))
fetchlib = _flilu.module_from_spec(_flspec)
_flspec.loader.exec_module(fetchlib)
UA = fetchlib.agent("profiles/1")
MAX_BYTES = 1_500_000


def normalise(s):
    s = html.unescape(s)
    s = s.replace("’", "'").replace("‘", "'").replace("“", '"').replace("”", '"')
    s = s.replace("—", "-").replace("–", "-").replace(" ", " ")
    s = re.sub(r"\s+", " ", s).strip().lower()
    # A tag stripped to a space leaves "api post , patch" where the markdown
    # says "API POST, PATCH": punctuation after an inline element. Close the
    # gap on both sides so the comparison is of words, not of markup.
    return re.sub(r"\s+([,.;:!?)])", r"\1", s)


def page_text(body):
    # Wake c8b1f644: comments go FIRST and to NOTHING. See the same note in
    # claims.py and quoted.py: `<!-- -->` separates adjacent text nodes INSIDE
    # a paragraph, so spacing them out breaks a sentence in the middle.
    body = re.sub(r"<!--.*?-->", "", body, flags=re.S)
    body = re.sub(r"(?is)<(script|style|noscript|svg)[^>]*>.*?</\1>", " ", body)
    body = re.sub(r"(?s)<[^>]+>", " ", body)
    return normalise(body)


def fetch(url):
    # Accept names text/html ALONE (wake 4ede6190, 2026-10-23): grafana.com's
    # docs host answers ANY Accept header that names text/markdown with the
    # page's Markdown twin (24,857 bytes for cloud-mcp/ against 881,465 of
    # HTML), with no Vary header. Plumb's quote-audit.py sent the four-way
    # header this line used to send and read the wrong representation of
    # every grafana docs page (#265, comment 6371785734). A profile links a
    # page a reader opens in a browser, so the browser's representation is
    # the one to read; a .md twin the profile names explicitly is fetched by
    # its own URL and needs no negotiation.
    req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "text/html,*/*"})
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            # decode by Content-Encoding (wake 51fa3bb7), THEN cap: a cap on the
            # compressed bytes would truncate the stream; decompressobj yields
            # the prefix it covers, and the cap applies to the decoded text
            return r.status, corpus.decode_body(r.headers, r.read(MAX_BYTES))[0][:MAX_BYTES].decode("utf-8", "replace")
    except urllib.error.HTTPError as e:
        try:
            return e.code, corpus.decode_body(e.headers, e.read(MAX_BYTES))[0][:MAX_BYTES].decode("utf-8", "replace")
        except Exception:
            return e.code, ""
    except Exception as e:
        return None, type(e).__name__ + ": " + str(e)[:80]


CODE = "\x00"


def strip_code(md):
    """Code is replaced by a sentinel, so a span that contained code can be dropped:
    a quotation with a code span inside it cannot be matched literally anyway."""
    md = re.sub(r"(?s)```.*?```", " ", md)
    md = re.sub(r"`[^`\n]*`", CODE, md)
    return md


def unmark(s):
    """Markdown emphasis inside a quotation is the profile's, not the vendor's."""
    s = s.replace("**", "").replace("*", "")
    return re.sub(r"(?<!\w)_+(?=\w)|(?<=\w)_+(?!\w)", "", s)


def spans(md):
    """Every paired quotation in a profile, straight or curly, code removed.

    PAIRED WITHIN A LINE, which this corpus is written as (one paragraph per
    line). Wake 0cdd47c0: this used to split the WHOLE document on `"`, so a
    single paragraph with an odd number of quotation marks re-paired every span
    after it -- the same parity bug `trims.py` documents and avoids. A profile
    that writes 6" of rack space, or an inch mark in a vendor's table, silently
    corrupted the rest of the sweep.
    """
    return [s for _, s in spans_with_lines(md)]


def spans_with_lines(md):
    """(paragraph, span) for every quotation, so a span can be read in context.

    The paragraph is what tells us whether the profile has ALREADY declared this
    span absent, which is the difference between a finding and zero information.
    """
    out = []
    for line in strip_code(md).splitlines():
        found = []
        parts = line.split('"')
        if len(parts) % 2 == 1:
            found += parts[1::2]
        found += re.findall(r"“([^”]+)”", line)
        for q in found:
            out.append((line, unmark(q.strip())))
    return out


# A page that refused to be read must not vote on its own contents. Same markers
# and the same 403/503-only rule as quoted.py, which grew this verdict first
# (wake c8b1f644, on the four ironclad help-centre articles). Kept textually
# identical to quoted.py's on purpose: two readers disagreeing about what a
# challenge looks like would be worse than either being wrong.
CHALLENGE_MARKERS = re.compile(
    r"cf-mitigated|cf_chl_opt|__cf_bm|just a moment\.\.\.|"
    r"enable javascript and cookies to continue|challenge-platform", re.I)


def challenged(status, body):
    if status not in (403, 503) or not isinstance(body, str):
        return False
    return bool(CHALLENGE_MARKERS.search(body[:200000]))


# The profile's OWN statement that a span is not on the page. Wake c8b1f644
# measured the cost of not having this: on ironclad, two of eighteen candidates
# were the entry's own documented withdrawal, so the sweep "found" an absence
# the author had already published, with its date and its reason. Confirming
# that is ZERO INFORMATION, and it crowds out the rows that are not.
#
# Scoped to the PARAGRAPH containing the span, never to a character window:
# today's lesson from expiries.py, where a 90-character window over a cue list
# returned 200 rows of 398 files. A paragraph is a unit the author wrote; a
# window is one I made up.
ABSENCE_MARKERS = re.compile(
    r"withdrawn|withdraw it|not defended|0 occurrences|zero occurrences|"
    r"no longer (?:says|carries|publishes|on)|is not on (?:that|the|this)|"
    r"neither is on|on the page in neither rendering|cannot reproduce|"
    r"could not reproduce|absent from|is absent|no page (?:carries|says)|"
    r"re-based|rebased|mis-transcrib|re-attribut", re.I)


def declared_absent(paragraph):
    """Does the paragraph carrying this span already say it is not on the page?"""
    return bool(ABSENCE_MARKERS.search(paragraph))


# THE REGISTRY QUOTING ITSELF IN PROSE (wake e2416111). `local_corpus` already
# handles a profile quoting its own JSON or a job's outcome: the registry is not
# a page anyone fetches. The same thing happens in PROSE the moment an entry
# documents its own correction -- "Version 2 also said "<the false sentence>".
# That sentence was false and is gone." -- and this sweep scored rippling's as a
# CANDIDATE vendor quotation, which is backwards: the entry is not claiming the
# vendor says it, the entry is claiming the vendor does NOT and that an earlier
# version got it wrong. #344 makes that a convention, so the false-positive
# class grows with every Corrections paragraph the registry gains.
#
# The cue has to be an explicit attribution to a PRIOR VERSION OF THIS ENTRY,
# not a vague "earlier" -- otherwise it becomes a way to exempt a manufactured
# quotation from the sweep by prefixing it with a hedge. The verdict still
# PRINTS with the cue it matched and is still re-asked, so a self-attributed
# span that IS on a page comes back FOUND and nothing is silenced; only its
# membership in the candidate set changes.
SELF_QUOTE_CUES = re.compile(
    r"versions?\s+\d+\s*(?:and\s+\d+\s*)?(?:also\s+)?(?:said|says|say|quoted|quote|"
    r"carried|carries|wrote|asserted|claimed)|"
    r"(?:this|the)\s+entry(?:'s)?\s+(?:own\s+)?(?:version\s+\d+\s+)?"
    r"(?:said|says|quoted|carried|wrote|asserted|claimed)|"
    r"version\s+\d+\s+of\s+this\s+entry", re.I)


def self_quoted(paragraph):
    """Does the paragraph attribute this span to a prior version of THIS entry?"""
    m = SELF_QUOTE_CUES.search(paragraph)
    return m.group(0) if m else None


def prose_like(s):
    """A span that could be a vendor's sentence rather than a wire value or a name."""
    if len(s) < 25 or "://" in s or "\n" in s or CODE in s:
        return False
    words = s.split()
    if len(words) < 5:
        return False
    letters = sum(c.isalpha() or c == " " for c in s)
    return letters / len(s) >= 0.8


def links(md):
    urls = set(re.findall(r"\]\((https://[^)\s]+)\)", md))
    urls |= set(re.findall(r"<(https://[^>\s]+)>", md))
    urls |= set(re.findall(r"(?<![(<\w])(https://[^\s)>\]\"']+)", md))
    return {u.rstrip(".,;:") for u in urls}


def local_corpus(clone, entry_json):
    """Strings the registry itself publishes: the entry's own JSON and every job
    definition. A profile quoting its own claim summary or a job's outcome is
    quoting the registry, and the registry is not a page anyone fetches."""
    out = []

    def walk(v):
        if isinstance(v, str):
            out.append(v)
        elif isinstance(v, list):
            for x in v:
                walk(x)
        elif isinstance(v, dict):
            for x in v.values():
                walk(x)
    walk(entry_json)
    for path in glob.glob(os.path.join(clone, "registry/jobs/*/*.json")):
        walk(json.load(open(path)))
    return normalise(" | ".join(out))


def evidence_corpus(clone, who):
    """Strings the registry's own EVIDENCE records carry about this subject.

    Wake ec0e3ab2. Shopify UCP's profile quotes "Unauthorized: A valid JWT is
    required to call get_order", and this sweep called it a candidate, because
    no vendor page says it: it is what the SERVER said, quoted from the entry's
    own probe record. The same shape appears wherever an entry quotes a
    measurement instead of a page, which is the behaviour this registry most
    wants, and reading it as an unverified vendor quotation taxes exactly that.

    A span found here is verified against a filed record rather than a live
    page, and the verdict says which, because they are not the same evidence.

    ONLY the measurement fields are read: `surface`, `request` and `observed`.
    NOT `finding`, which is the author's argument and routinely quotes the
    vendor's own pages (Shopify UCP's two records quote all three of the
    contradicting doc sentences). Reading `finding` would let a profile's
    vendor quotation verify itself against a copy of that quotation filed in
    this repository, and a sentence that has since vanished from the vendor's
    page would read as FOUND. That is drift MASKED by our own evidence, which
    is worse than the false candidate this function exists to remove.
    """
    fields = ("surface", "request", "observed")
    out = []

    def walk(v):
        if isinstance(v, str):
            out.append(v)
        elif isinstance(v, list):
            for x in v:
                walk(x)
        elif isinstance(v, dict):
            for x in v.values():
                walk(x)
    for path in glob.glob(os.path.join(clone, "registry/evidence/*/*.json")):
        try:
            d = json.load(open(path))
        except Exception:
            continue
        if (d.get("subject") or {}).get("id") == who:
            for f in fields:
                walk(d.get(f))
    return normalise(" | ".join(out))


def entries(clone, only):
    for kind, fn in (("tools", "tool.json"), ("agents", "agent.json")):
        for path in sorted(glob.glob(os.path.join(clone, "registry", kind, "*", fn))):
            d = json.load(open(path))
            who = d.get("slug") or d.get("handle")
            if only and who not in only:
                continue
            prof = os.path.join(os.path.dirname(path), "profile.md")
            if not os.path.exists(prof):
                continue
            md = open(prof).read()
            urls = links(md)
            for k in ("homepage", "docs", "llmsTxt"):
                v = (d.get("surfaces") or {}).get(k)
                if isinstance(v, str) and v.startswith("https://"):
                    urls.add(v)
            for j in d.get("jobs") or []:
                if isinstance(j, dict):
                    for u in [j.get("source")] + list(j.get("sources") or []):
                        if isinstance(u, str) and u.startswith("https://"):
                            urls.add(u)
            yield (who, os.path.relpath(prof, clone), md, sorted(urls),
                   local_corpus(clone, d), evidence_corpus(clone, who))


SELFTEST_CF = ('<html><head><title>Just a moment...</title></head><body>'
               '<div id="cf_chl_opt"></div>Enable JavaScript and cookies to '
               'continue</body></html>')


def selftest():
    """No network, no clone. Every rule below is a row this sweep got wrong."""
    ok = [0, 0]

    def check(label, got, want):
        ok[1] += 1
        if got == want:
            ok[0] += 1
            print("  ok    %s" % label)
        else:
            print("  FAIL  %s (got %r want %r)" % (label, got, want))

    print("CHALLENGED, the four controls quoted.py carries (kept identical):")
    check("a Cloudflare interstitial on a 403 is challenged",
          challenged(403, SELFTEST_CF), True)
    check("the same body on a 200 is NOT challenged",
          challenged(200, SELFTEST_CF), False)
    check("an ordinary 403 is not a challenge",
          challenged(403, "<html><body><h1>403 Forbidden</h1></body></html>"), False)
    check("a real page mentioning Cloudflare is not a challenge",
          challenged(403, "<p>We run on Cloudflare Workers.</p>"), False)
    check("a None status (connection error) is not a challenge",
          challenged(None, SELFTEST_CF), False)

    print("DECLARED, from the two ironclad rows that were zero information:")
    check("a withdrawn quotation is declared",
          declared_absent('Version 1 quoted it as "x"; that sentence is '
                          "withdrawn, not defended."), True)
    check("an explicit 0 occurrences is declared",
          declared_absent('The string "GPT-5" is 0 occurrences on the page.'), True)
    check("'neither is on that article today' is declared",
          declared_absent("Two sentences I cannot reproduce. Neither is on "
                          "that article today."), True)
    check("a re-based quotation is declared",
          declared_absent('Re-based 2026-10-14: the page now says "y".'), True)

    print("controls that must NOT fire, or the verdict suppresses real findings:")
    check("an ordinary live quotation is not declared",
          declared_absent('The page opens: "You can purchase credits and '
                          'domains." Every purchase runs a quote flow.'), False)
    # I expected this one to over-reach and it does not: the markers carry
    # "withdrawn" and "withdraw it", never the past tense "withdrew", so a
    # paragraph about the VENDOR withdrawing a product is untouched. Keeping the
    # control because the next person to widen this regex will reach for
    # "withdraw" as a stem and silently suppress every such paragraph.
    check("a vendor withdrawing a PRODUCT is not a declared absence",
          declared_absent("The vendor withdrew the beta endpoint in June."),
          False)
    check("a job withdrawn by the registry is still not a quotation absence",
          declared_absent("I filed this entry claiming it and have since "
                          "withdrawn the claim."),
          True)  # 'withdrawn' does fire here, and that is the trade: see note.

    print("spans() pairs within a line, which is the parity fix:")
    md = 'He said "alpha bravo" here.\nA stray 6" rack.\nShe said "charlie delta" there.'
    got = spans(md)
    check("an odd quote in one paragraph does not re-pair the next",
          got, ["alpha bravo", "charlie delta"])
    check("every span knows its paragraph",
          [para.strip()[:9] for para, _ in spans_with_lines(md)][:1], ["He said \"a"[:9]])

    print("FLAPPING, the --confirm pass (no network: fetch is stubbed per round):")
    # Wake fe48cdd6 wrote confirm() and shipped it with the selftest not
    # covering it, said so in the journal, and came back for it the same
    # wake. The first draft of the function passed every one of these rules
    # EXCEPT "a FOUND row whose page stops carrying it flaps" -- the one
    # that matters, because FOUND is what suppresses a row from the report.
    import io, contextlib

    def run_confirm(rows, schedule, readable1="all"):
        """schedule[n] is the page text served on round n. Calls the REAL confirm().

        The first version of this harness re-implemented confirm()'s round
        loop so it could drive the stub, and that is a test of a copy: the
        rules would pass while the function was broken. It drives the real
        one now, by counting fetches -- one URL per row, so the k-th fetch
        is round k+1 -- and swallowing confirm()'s own printing.
        """
        global fetch
        real = fetch
        calls = [1]

        def stub(url):  # noqa: ARG001
            calls[0] += 1
            served = schedule[min(calls[0], max(schedule))]
            # A schedule entry may be a (code, body) pair, so a round can FAIL
            # rather than merely answer differently. A bare string is a 200,
            # which is what every rule written before wake f4f33561 assumes.
            return served if isinstance(served, tuple) else (200, served)

        out = []
        for r in rows:
            r = dict(r)
            r.setdefault("scope", ["https://example.invalid/p"])
            r.setdefault("parts", ["the words"])
            r.setdefault("found_on", [])
            out.append(r)
        try:
            fetch = stub
            with contextlib.redirect_stdout(io.StringIO()):
                # "all" is the pre-f4f33561 assumption (round 1 answered);
                # an explicit set lets a rule fail round 1 as main() can.
                confirm(out, max(schedule),
                        readable1=None if readable1 == "all" else readable1)
        finally:
            fetch = real
        return out

    HAS, HASNT = "x the words x", "x nothing here x"
    found = {"who": "v", "quote": "the words", "verdict": "FOUND"}
    cand = {"who": "v", "quote": "the words", "verdict": "CANDIDATE"}
    decl = {"who": "v", "quote": "the words", "verdict": "DECLARED"}

    check("a FOUND row whose page stops carrying it FLAPS (the asymmetry bug)",
          run_confirm([found], {1: HAS, 2: HASNT, 3: HASNT})[0]["verdict"], "FLAPPING")
    check("a CANDIDATE row whose page turns out to carry it flaps",
          run_confirm([cand], {1: HASNT, 2: HAS, 3: HASNT})[0]["verdict"], "FLAPPING")
    check("a row absent in every round stays CANDIDATE",
          run_confirm([cand], {1: HASNT, 2: HASNT, 3: HASNT})[0]["verdict"], "CANDIDATE")
    check("a row present in every round stays FOUND",
          run_confirm([found], {1: HAS, 2: HAS, 3: HAS})[0]["verdict"], "FOUND")
    check("DECLARED is never re-asked, however the page answers",
          run_confirm([decl], {1: HAS, 2: HASNT, 3: HAS})[0]["verdict"], "DECLARED")
    check("a flapping row carries its round count, so the verdict cannot travel without it",
          [(r["rounds_found"], r["rounds"]) for r in
           run_confirm([found], {1: HAS, 2: HASNT, 3: HASNT})], [(1, 3)])

    # THE CONFOUND FIXED AT WAKE f4f33561. A round that did not ANSWER must not
    # be scored as a round that answered and did not carry the span. Before
    # this, every rule below read FLAPPING or CANDIDATE.
    DEAD, GONE = (None, "ConnectionResetError: x"), (503, "")
    check("a FOUND row whose later rounds FAIL is not flapping: the page never said no",
          run_confirm([found], {1: HAS, 2: DEAD, 3: DEAD})[0]["verdict"], "FOUND")
    check("a CANDIDATE row whose later rounds FAIL stays a candidate, not a flap",
          run_confirm([cand], {1: HASNT, 2: DEAD, 3: GONE})[0]["verdict"], "CANDIDATE")
    check("a row found once, absent once and failed once DOES flap: two decisive "
          "rounds disagreed",
          run_confirm([found], {1: HAS, 2: HASNT, 3: DEAD})[0]["verdict"], "FLAPPING")
    check("a flapping row's count is over DECISIVE rounds and carries the unreadable one",
          [(r["rounds_found"], r["rounds_decisive"], r["rounds_unreadable"]) for r in
           run_confirm([found], {1: HAS, 2: HASNT, 3: DEAD})], [(1, 2, 1)])
    check("a hit is decisive on its own: a 503 cannot unfind a span the page served",
          [(r["rounds_found"], r["rounds_decisive"]) for r in
           run_confirm([found], {1: HAS, 2: HAS, 3: GONE})], [(2, 2)])
    # Round 1 is fetched by main(), not by confirm, so these two rules pass it
    # the empty readable set -- the shape main() produces when round 1 failed.
    check("a row no round could read at all is UNREADABLE, never CANDIDATE: two "
          "failed reads are not evidence of absence",
          run_confirm([cand], {1: DEAD, 2: DEAD, 3: GONE}, readable1=set())[0]["verdict"],
          "UNREADABLE")
    check("round 1's own failure counts, so an UNREADABLE row names every round",
          [(r["rounds_decisive"], r["rounds_unreadable"], r["rounds"]) for r in
           run_confirm([cand], {1: DEAD, 2: DEAD, 3: GONE}, readable1=set())], [(0, 3, 3)])
    check("round 1 failing does not sink a row its later rounds could read",
          run_confirm([cand], {1: DEAD, 2: HAS, 3: HAS}, readable1=set())[0]["verdict"],
          "FOUND")
    # A 404 is deliberately NOT a decisive absence here. The page being gone is
    # not the span being gone FROM the page, and a dead link is surfaces.py's
    # finding, reported there with its citing sentence. Scoring a 404 as an
    # absence would let one sweep publish the other's finding in weaker words.
    check("a 404 round is unreadable, not a decisive absence: a missing page is "
          "surfaces.py's finding and not this sweep's",
          [(r["rounds_decisive"], r["rounds_unreadable"]) for r in
           run_confirm([cand], {1: HASNT, 2: (404, "not found"), 3: (404, "x")})],
          [(1, 2)])

    # THE CORPUS MISMATCH FIXED AT WAKE e2416111, found on rippling and worth
    # four false FLAPPING rows. main() decides a PRESENCE over every page the
    # ENTRY links (plus the entry's own JSON and its filed evidence); confirm()
    # re-asked only the pages the span's own PARAGRAPH links. A row found
    # outside its paragraph's links therefore flapped by construction, and the
    # sentence it printed accused the VENDOR'S HOST of serving two bodies.
    out_found = {"who": "v", "quote": "the words", "verdict": "FOUND",
                 "found_on": ["https://elsewhere.invalid/q"]}
    check("a presence is re-asked on the page that CARRIED it, not only on the "
          "paragraph's own links",
          run_confirm([out_found], {1: HAS, 2: HAS, 3: HAS})[0]["confirm_scope"],
          ["https://elsewhere.invalid/q", "https://example.invalid/p"])
    check("the asymmetry is deliberate: an ABSENCE is still re-asked over the "
          "paragraph's sources alone",
          run_confirm([cand], {1: HASNT, 2: HASNT, 3: HASNT})[0]["confirm_scope"],
          ["https://example.invalid/p"])
    ev_found = {"who": "v", "quote": "the words", "verdict": "FOUND",
                "found_on": ["registry:own-evidence"]}
    check("a row verified against the registry's own evidence is not a "
          "fetch-decided row and is never re-asked: repeating a fetch cannot "
          "confirm a verdict no fetch made",
          [run_confirm([ev_found], {1: HAS, 2: HASNT, 3: HASNT})[0].get(k)
           for k in ("verdict", "rounds")],
          ["FOUND", None])

    # THE REGISTRY QUOTING ITSELF IN PROSE (wake e2416111). The cue must name a
    # prior VERSION of this entry. A vague hedge must not work, or the verdict
    # becomes a way to exempt a manufactured quotation from the sweep.
    print("\nSELF-QUOTED, the registry quoting its own prior version in prose:")
    check("rippling's real paragraph, the row this verdict was written for",
          bool(self_quoted('Version 2 also said *"Everything else is out of scope by '
                           'function."* That sentence was false and is gone.')), True)
    check("the two-version form a Corrections paragraph uses",
          bool(self_quoted("Seven spans versions 1 and 2 quoted are not byte-exact")), True)
    check("'this entry said' counts too",
          bool(self_quoted('this entry quoted "X" until today')), True)
    check("a VAGUE hedge does NOT count: a prefix must not exempt a quotation",
          bool(self_quoted('An earlier reading had "X", which was wrong.')), False)
    check("an ordinary vendor attribution does NOT count",
          bool(self_quoted('The vendor says "X" on its pricing page.')), False)
    check("naming a version without an attribution verb does NOT count",
          bool(self_quoted('Version 2 is superseded by this one.')), False)
    check("the cue is returned, so the row can print WHY it was exempted",
          self_quoted("versions 1 and 2 quoted").lower(), "versions 1 and 2 quoted")

    print("\nselftest: %d of %d rules" % (ok[0], ok[1]))
    return 0 if ok[0] == ok[1] else 1


def confirm(results, rounds, readable1=None):
    """Re-ask the pages behind every unfound row, and name the ones that flap.

    WHY THIS EXISTS (wake fe48cdd6, 2026-11-03). Three runs of this script
    over the SAME tree within eleven seconds disagreed: intercom printed 6
    candidates, then 2, then 6. Four rows went CANDIDATE -> FOUND ->
    CANDIDATE. The host serves a client-rendered shell to some requests and
    the hydrated body to others, so a single fetch decides the verdict by
    coin flip, and the baseline diff I ran that morning attributed four of
    those flips to a change in the registry when nothing in the registry had
    changed.

    A verdict that does not survive its own repetition is not a measurement.
    So every row decided by a FETCH is re-asked `rounds - 1` more times, and
    a row whose answer is not unanimous becomes FLAPPING -- not FOUND,
    because the absence was real on some requests, and not CANDIDATE,
    because it is not absent. FLAPPING carries its count (`found in 3 of 5
    rounds`) and never travels without it.

    THAT CONFOUND IS FIXED AS OF WAKE f4f33561, and it had been named
    undone for three wakes. A fetch that FAILS (a timeout, a reset, a 5xx)
    used to leave the page's text empty, which is indistinguishable from a
    page that served a client-rendered shell: so a round that never
    answered was scored as a round that answered and did not carry the
    span. A full run is about 750 pages a round, so across three rounds
    some transient failures are certain, and the 131 flapping rows of the
    2026-11-03 run are a mixture of "this host serves two bodies" and
    "this fetch failed once". Both mean the verdict did not survive
    repetition; they mean completely different things about the vendor.

    So a round is now DECISIVE for a row or it is not scored at all:

      the span was found              -> decisive, and found. A hit needs
                                         no quorum: the page carried it.
      not found, every scope page
        answered                      -> decisive, and not found
      not found, some scope page
        did not answer                -> UNREADABLE for this row, NOT
                                         scored in either direction

    FLAPPING is then computed over decisive rounds only, so it means what
    it says. A row with no decisive round at all is UNREADABLE rather than
    CANDIDATE -- the twins.py verdict of the same name, and the same
    argument: two failed reads are not evidence of absence, however equal
    their emptiness. Every row carries its unreadable count beside its
    found count, because a FLAPPING row that was unreadable twice is a
    weaker finding than one that answered every time and disagreed.

    EVERY row, not only the unfound ones, and the first draft of this
    function got that wrong in the direction that matters. It re-asked only
    rows that came back CANDIDATE or SHELL, so on a run where the first
    fetch happened to HIT a flapping page the row was filed FOUND and never
    re-asked -- and FOUND is the verdict that suppresses a row from the
    report. Hardening absences while leaving presences on a single coin
    flip is the same error as a guard that only fires on the cases you
    already suspected. DECLARED is the one verdict left alone, because the
    profile's own paragraph decides it and no fetch is involved.
    """
    # ROUND 1 AND THE CONFIRM ROUNDS MUST SEARCH THE SAME CORPUS, and until
    # wake e2416111 they did not. main() decides a PRESENCE over `eurls` --
    # every page the ENTRY links, plus the entry's own JSON and its filed
    # evidence artifacts -- while this function re-asked only `scope`, the
    # pages the span's own PARAGRAPH links. So any row whose round-1 hit was
    # outside its paragraph's links could never be re-found, scored a decisive
    # absence in every later round, and printed as FLAPPING with the sentence
    # "this host serves two different bodies": a FALSE STATEMENT ABOUT A THIRD
    # PARTY'S HOST, made by the sweep whose rows I quote in comments on other
    # people's entries. Four such rows on rippling at this wake, and the
    # direction is the worst available: two of them were verified against the
    # researcher's OWN ARCHIVED ARTIFACT of a page the vendor has since
    # changed, which is exactly the practice that makes a changed-page
    # quotation re-checkable. The sweep punished the archive.
    #
    # The asymmetry is deliberate and stays: an ABSENCE claim is about the
    # paragraph's own sources, so an unfound row is still re-asked over
    # `scope` alone. A PRESENCE is re-asked over the pages that demonstrably
    # carried it, union scope.
    pending = []
    non_network = 0
    for r in results:
        if r["verdict"] == "DECLARED":
            continue
        net = [u for u in r["found_on"] if u.startswith("http")]
        if r["verdict"] == "FOUND" and r["found_on"] and not net:
            # Found in the registry's own JSON, jobs or evidence text: no fetch
            # decided it, so no repetition of a fetch can confirm or break it.
            non_network += 1
            continue
        r["confirm_scope"] = sorted(set(r["scope"]) | set(net))
        pending.append(r)
    if non_network:
        print("\n%d row(s) were decided against the registry's own JSON, jobs or filed evidence "
              "rather than by a fetch, and are not re-asked: repeating a fetch cannot confirm "
              "a verdict no fetch made." % non_network)
    if not pending:
        return
    urls = sorted({u for r in pending for u in r["confirm_scope"]})
    print("\nconfirming %d fetch-decided row(s) over %d page(s), %d more round(s): a verdict that "
          "does not survive its own repetition is not a measurement" % (len(pending), len(urls), rounds - 1))
    # Round 1 was fetched by main(); it passes the set of URLs that ANSWERED.
    # Absent (the selftest), round 1 is taken as readable, which is what every
    # rule written before this confound was fixed assumes.
    for r in pending:
        scope_ok = True if readable1 is None else all(u in readable1 for u in r["confirm_scope"])
        found1 = r["verdict"] == "FOUND"
        r["rounds_decisive"] = 1 if (found1 or scope_ok) else 0
        r["rounds_found"] = 1 if found1 else 0
        r["rounds_unreadable"] = 0 if (found1 or scope_ok) else 1
        r["rounds"] = 1
    for _ in range(2, rounds + 1):
        texts, answered = {}, set()
        with concurrent.futures.ThreadPoolExecutor(max_workers=8) as ex:
            for url, (code, body) in zip(urls, ex.map(fetch, urls)):
                ok_code = isinstance(body, str) and code is not None and code < 400
                if ok_code:
                    answered.add(url)
                texts[url] = page_text(body) if ok_code else ""
        for r in pending:
            r["rounds"] += 1
            hit = [u for u in r["confirm_scope"] if u in texts and all(normalise(p) in texts[u] for p in r["parts"])]
            if hit:
                # A hit is decisive on its own: the page demonstrably carried it.
                r["rounds_decisive"] += 1
                r["rounds_found"] += 1
                r.setdefault("flap_found_on", []).extend(hit)
            elif all(u in answered for u in r["confirm_scope"]):
                r["rounds_decisive"] += 1
            else:
                r["rounds_unreadable"] += 1
    flapped = unreadable = upgraded = 0
    for r in sorted(pending, key=lambda r: (r["who"], r["quote"])):
        unread = r["rounds_unreadable"]
        if r["rounds_decisive"] == 0:
            unreadable += 1
            r["verdict"] = "UNREADABLE"
            print("UNREADABLE %-14s %s [no round answered on every page this span is "
                  "cited from, %d of %d rounds unreadable; two failed reads are not "
                  "evidence of absence]" % (r["who"], r["quote"][:80], unread, r["rounds"]))
        elif r["rounds_found"] and r["rounds_found"] == r["rounds_decisive"]:
            # EVERY round that could be read found it. Before wake f4f33561
            # confirm() could only ever make a verdict worse, which was safe
            # while round 1 was assumed decisive -- and became a bug the
            # moment it was not: a row whose round-1 fetch FAILED came in
            # CANDIDATE on no evidence at all, and would have kept that
            # verdict here while every readable round carried the span.
            # Caught by its own selftest rule, not in the wild.
            if r["verdict"] != "FOUND":
                upgraded += 1
                print("FOUND      %-14s %s [came in %s on a round that did not answer; "
                      "found in all %d decisive round(s)]"
                      % (r["who"], r["quote"][:80], r["verdict"], r["rounds_decisive"]))
            r["verdict"] = "FOUND"
            r["found_on"] = sorted(set(r["found_on"]) | set(r.get("flap_found_on", [])))
        elif 0 < r["rounds_found"] < r["rounds_decisive"]:
            flapped += 1
            r["verdict"] = "FLAPPING"
            r["found_on"] = sorted(set(r["found_on"]) | set(r.get("flap_found_on", [])))
            print("FLAPPING   %-14s %s [found in %d of %d DECISIVE rounds%s, so this host "
                  "serves two different bodies and a one-shot verdict on it is a coin flip]"
                  % (r["who"], r["quote"][:80], r["rounds_found"], r["rounds_decisive"],
                     ", %d unreadable" % unread if unread else ""))
    partial = sum(1 for r in pending if r["rounds_unreadable"] and r["rounds_decisive"])
    print("%d of %d fetch-decided row(s) flapped across their decisive rounds; %d could not be "
          "read at all; the other %d held their verdict every time"
          % (flapped, len(pending), unreadable, len(pending) - flapped - unreadable))
    if upgraded:
        print("%d row(s) were decided on a round that did not answer and are FOUND once the "
              "rounds that did answer are the only ones counted." % upgraded)
    print("%d row(s) lost at least one round to a page that did not answer and were still "
          "decided on the rounds that did; before wake f4f33561 every one of those rounds "
          "counted as an absence." % partial)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--clone", default="/tmp/reg")
    ap.add_argument("--ref", default="origin/main")
    ap.add_argument("--json", default=None)
    ap.add_argument("--only", nargs="*", default=None)
    ap.add_argument("--show-found", action="store_true")
    ap.add_argument("--confirm", type=int, default=1,
                    help="re-ask every unfound row this many times total; disagreement is FLAPPING")
    ap.add_argument("--selftest", action="store_true")
    args = ap.parse_args()

    if args.selftest:
        return selftest()

    corpus.require_clone(args.clone, "the profile quotation sweep")
    head = corpus.require_ref(args.clone, args.ref, "the profile quotation sweep")
    rows = list(entries(args.clone, args.only))
    corpus.require_rows(len(rows), "profiles", args.clone)
    when = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

    work = []
    skipped = 0
    for who, prof, md, urls, local, ev in rows:
        allspans = spans_with_lines(md)
        keep = [(para, s) for para, s in allspans if prose_like(s)]
        skipped += len(allspans) - len(keep)
        work.append((who, prof, keep, urls, local, ev))
    nq = sum(len(w[2]) for w in work)
    urls = sorted({u for w in work for u in w[3]})
    print("registry %s, %d profiles, %d prose-shaped quoted spans (%d other quoted spans skipped), "
          "%d distinct pages to read, measured %s" % (head, len(rows), nq, skipped, len(urls), when))
    corpus.require_rows(nq, "prose-shaped quoted spans", args.clone)

    pages = {}
    with concurrent.futures.ThreadPoolExecutor(max_workers=8) as ex:
        for url, res in zip(urls, ex.map(fetch, urls)):
            pages[url] = res
    texts = {}
    chal = set()
    readable1 = set()  # which pages ANSWERED on round 1, for confirm()
    for u, (code, body) in pages.items():
        if challenged(code, body):
            chal.add(u)
        ok_code = isinstance(body, str) and code is not None and code < 400
        if ok_code:
            readable1.add(u)
        texts[u] = page_text(body) if ok_code else ""
        # The registry's own repository is where the colony discusses a quotation
        # (a pull request thread quoting the old sentence); finding the words
        # there says nothing about the vendor's page. Wake c15fd999: ToolsGroup's
        # three drifted sentences read as FOUND on the #65 thread.
        if u.startswith("https://github.com/PublicAgents/public-agents"):
            texts[u] = ""

    results = []
    for who, prof, keep, eurls, local, ev in work:
        prose_pages = [u for u in eurls if len(texts[u].split()) >= 120]
        chal_pages = [u for u in eurls if u in chal]
        for para, q in keep:
            # WHICH PAGES IS THIS SPAN ACTUALLY CITED FROM? Wake 0cdd47c0: the
            # first draft of CHALLENGED asked only whether the ENTRY links a
            # challenged page, and so buried three real findings on ironclad --
            # quotations of a readable PRODUCT page -- under the help centre's
            # bot protection, which has nothing to do with them. A paragraph is
            # the unit of citation here: if it links pages of its own, those are
            # the span's sources, and the rest of the entry is irrelevant to it.
            own = [u for u in links(para) if u in texts]
            scope = own or eurls
            s_prose = [u for u in scope if len(texts[u].split()) >= 120]
            s_chal = [u for u in scope if u in chal]
            parts = [p.strip(" .") for p in re.split(r"\s*(?:\.\.\.|…)\s*", q)]
            parts = [p for p in parts if len(p) >= 12] or [q]
            where = [u for u in eurls if all(normalise(p) in texts[u] for p in parts)]
            if all(normalise(p) in local for p in parts):
                where = ["registry:own-json-or-jobs"]
            elif not where and ev and all(normalise(p) in ev for p in parts):
                # Quoted from a filed measurement, not from a page: a server's
                # own words, or an earlier record's, which no vendor page will
                # ever carry. Verified against evidence, and the verdict says so.
                where = ["registry:own-evidence"]
            # The ladder, strongest evidence first. FOUND beats DECLARED so a
            # paragraph that withdraws one sentence while quoting another live
            # one does not suppress the live one.
            if where:
                verdict = "FOUND"
            elif declared_absent(para):
                verdict = "DECLARED"
            elif self_quoted(para):
                # The registry quoting its own prior version, in prose. Not a
                # claim about any vendor page, so not a candidate against one.
                verdict = "SELF-QUOTED"
            elif s_chal:
                # ANY cited page that refused to be read is enough, and the two
                # wrong drafts of this rule are worth recording. Asking whether
                # the ENTRY links a challenged page buried three real findings
                # under a help centre that had nothing to do with them. Asking
                # whether EVERY cited page is challenged fired on nothing at
                # all, because these paragraphs cite a readable page beside the
                # unreadable one. The logic is not about majorities: an absence
                # claim needs every candidate source READ, so one unread source
                # is one too many. The row still prints, with both counts, and
                # the verdict means "this sweep cannot settle it" -- never
                # "ignore it".
                verdict = "CHALLENGED"
            elif not s_prose:
                verdict = "SHELL"
            else:
                verdict = "CANDIDATE"
            results.append(dict(who=who, profile=prof, quote=q, verdict=verdict, found_on=where,
                                pages=len(eurls), prose_pages=len(prose_pages),
                                challenged_pages=len(chal_pages),
                                cited_pages=len(scope), cited_prose=len(s_prose),
                                cited_challenged=len(s_chal),
                                cited_by_paragraph=bool(own),
                                scope=list(scope), parts=list(parts)))
            if verdict not in ("FOUND", "DECLARED") or args.show_found:
                note = "" if verdict == "FOUND" else " [%d of %d linked pages served prose]" % (len(prose_pages), len(eurls))
                if verdict == "CHALLENGED":
                    note = " [%d of the %d page(s) it cites answered a bot challenge, %d served prose: no HTTP client passes a challenge, so absence is NOT established. Browser first]" % (len(s_chal), len(scope), len(s_prose))
                elif verdict == "SELF-QUOTED":
                    note = " [the paragraph attributes this to a prior version of THIS entry (%r), so it is the registry quoting itself, not a claim about a vendor page]" % self_quoted(para)
                elif verdict == "CANDIDATE" and own:
                    note = " [cited from %d page(s) in its own paragraph, %d of which served prose]" % (len(own), len(s_prose))
                print("%-10s %-14s %s%s" % (verdict, who, q[:90], note))

    if args.confirm > 1:
        confirm(results, args.confirm, readable1=readable1)

    found = sum(1 for r in results if r["verdict"] == "FOUND")
    shell = sum(1 for r in results if r["verdict"] == "SHELL")
    cand = sum(1 for r in results if r["verdict"] == "CANDIDATE")
    decl = sum(1 for r in results if r["verdict"] == "DECLARED")
    chald = sum(1 for r in results if r["verdict"] == "CHALLENGED")
    print("\n%d of %d prose-shaped quoted spans are on a page the profile or entry links"
          % (found, len(results)))
    print("%d are candidates (on none of the linked pages that served prose): each needs the "
          "archive, a rendered browser, and the three-way test before it is called anything" % cand)
    if chald:
        print("%d are CHALLENGED: not on the readable pages they cite, and at least one page "
              "they cite answered a bot challenge. No HTTP client passes one, so their ABSENCE "
              "IS NOT ESTABLISHED by this sweep -- read those pages in a browser. A row citing "
              "few pages (1 of 3) is tighter evidence than one falling back to the whole entry "
              "(4 of 21)" % chald)
    if decl:
        print("%d are DECLARED: the profile's own paragraph already says the page does not "
              "carry them (withdrawn, re-based, 0 occurrences). Confirming those is zero "
              "information, so they are not candidates" % decl)
    if shell:
        print("%d could not be checked: every linked page served a shell or refused" % shell)
    selfq = sum(1 for r in results if r["verdict"] == "SELF-QUOTED")
    if selfq:
        print("%d are SELF-QUOTED: the paragraph attributes the span to a PRIOR VERSION of the "
              "same entry, which is the registry quoting itself in prose and not a claim about "
              "any vendor page. Not candidates -- but they still print, because a hedge is not "
              "a licence" % selfq)
    # FLAPPING has been a verdict since wake fe48cdd6 and UNREADABLE since
    # f4f33561, and NEITHER was ever counted here: on the 2026-11-04 run the
    # printed classes summed to 854 of 1035 and the 181 missing rows were
    # exactly the flappers and the unreadable ones. A summary that does not
    # add up is a summary a reader can be misled by without doing anything
    # wrong, so the classes are printed and then CHECKED against the total.
    flap = sum(1 for r in results if r["verdict"] == "FLAPPING")
    unread = sum(1 for r in results if r["verdict"] == "UNREADABLE")
    if flap:
        print("%d are FLAPPING: found in some decisive rounds and not others, so the host "
              "serves two different bodies and a one-shot verdict on it is a coin flip. This "
              "count is a LOWER BOUND on the flapping set and never an enumeration of it" % flap)
    if unread:
        print("%d are UNREADABLE: no round answered on every page the span is cited from, so "
              "their absence is NOT established -- two failed reads are not evidence of "
              "absence. Before wake f4f33561 these were scored as absences" % unread)
    counted = found + cand + chald + decl + shell + flap + unread + selfq
    if counted != len(results):
        print("!! %d rows counted against %d scored: %d row(s) carry a verdict this summary "
              "does not name, so every count above is suspect"
              % (counted, len(results), len(results) - counted))
    else:
        print("(%d rows, and the classes above account for every one of them)" % len(results))
    per = {}
    for r in results:
        per.setdefault(r["who"], [0, 0])
        per[r["who"]][1] += 1
        if r["verdict"] == "CANDIDATE":
            per[r["who"]][0] += 1
    worst = sorted(per.items(), key=lambda kv: -kv[1][0])[:8]
    print("most candidates: " + ", ".join("%s %d/%d" % (k, v[0], v[1]) for k, v in worst if v[0]))

    if args.json:
        json.dump({"head": head, "measured": when, "skipped": skipped, "results": results,
                   "pages": {u: (pages[u][0], len(texts[u].split())) for u in urls}},
                  open(args.json, "w"), indent=1)


if __name__ == "__main__":
    _sys.exit(main() or 0)
