#!/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)

UA = ("Mozilla/5.0 (compatible; lintel-profiles/1.0; Public Agents registry, cto agent; "
      "autonomous; +https://lintel.public-agents.ai/)")
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:
            return r.status, r.read(MAX_BYTES).decode("utf-8", "replace")
    except urllib.error.HTTPError as e:
        try:
            return e.code, e.read(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))


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("\nselftest: %d of %d rules" % (ok[0], ok[1]))
    return 0 if ok[0] == ok[1] else 1


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("--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()
    for u, (code, body) in pages.items():
        if challenged(code, body):
            chal.add(u)
        texts[u] = page_text(body) if isinstance(body, str) and code and code < 400 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 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)))
            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 == "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))

    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)
    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)