#!/usr/bin/env python3
"""drift.py: re-measure every access claim in the registry, with no credentials.

Every tool entry carries `agentAccess.noAccountNeeded` and `agentAccess.auth`,
and each was measured once, on the day it was filed, by the party that filed
it. Nobody measures them again. A server that was keyless in September can
require an account in October and the registry will go on saying it is free,
which is the one claim the registry actually sells.

This walks every `surfaces.mcp` in the registry, sends an unauthenticated MCP
`initialize`, and compares what the wire says to what the entry claims. It
writes nothing to the registry and files nothing: it produces a dated report.
Fixing an entry is the filer's job, never the merge gate's.

Protocol versions (wake 6b5e7fe7): an MCP server echoes back whichever
protocolVersion the request asks for, when it recognises it. So a version read
off a normal initialize is a property of the REQUEST. Ask for one the server
cannot possibly support (1999-01-01) and it substitutes the one it actually
prefers, which is a property of the SERVER. This probes with both.

Usage: python3 tools/drift.py [--clone /tmp/reg] [--json out.json]
"""

import sys as _sys

# tools/queue.py shadows the stdlib `queue`, which concurrent.futures imports,
# so drop this script's own directory from the import path before importing it.
_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 json
import os
import re
import subprocess
import sys
import urllib.error
import urllib.request

# corpus.py refuses to score an empty or a wrong corpus. Loaded by path
# because this directory is deliberately off sys.path (see the header).
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)

INIT = {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "initialize",
    "params": {
        "protocolVersion": "2025-06-18",
        "capabilities": {},
        "clientInfo": {"name": "lintel-drift", "version": "1"},
    },
}
# A version no server can know, so the answer is the server's own preference
# rather than a reflection of the question.
IMPOSSIBLE = "1999-01-01"
UA = "public-agents-lintel-drift"


def post(url, payload, timeout=25):
    """Returns (status, headers, body_text). Status None means no answer."""
    data = json.dumps(payload).encode()
    req = urllib.request.Request(
        url,
        data=data,
        method="POST",
        headers={
            "Content-Type": "application/json",
            "Accept": "application/json, text/event-stream",
            "User-Agent": UA,
        },
    )
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return r.status, dict(r.headers), r.read(65536).decode("utf-8", "replace")
    except urllib.error.HTTPError as e:
        return e.code, dict(e.headers), e.read(8192).decode("utf-8", "replace")
    except Exception as e:
        return None, {}, "%s: %s" % (type(e).__name__, e)


def mcp_result(body):
    """Pull the JSON-RPC result out of a plain or text/event-stream body."""
    for line in body.splitlines():
        line = line.strip()
        if line.startswith("data:"):
            line = line[5:].strip()
        if not line.startswith("{"):
            continue
        try:
            d = json.loads(line)
        except ValueError:
            continue
        if isinstance(d, dict) and ("result" in d or "error" in d):
            return d
    return None


def probe(url):
    """One endpoint, unauthenticated. Returns a dict of what the wire said."""
    out = {"url": url}
    status, headers, body = post(url, INIT)
    out["status"] = status
    out["www_authenticate"] = headers.get("www-authenticate") or headers.get(
        "WWW-Authenticate"
    )
    rpc = mcp_result(body) if status and 200 <= status < 300 else None

    if status is None:
        out["reading"] = "no answer"
        out["detail"] = body[:200]
    elif rpc and "result" in rpc:
        info = rpc["result"].get("serverInfo") or {}
        out["reading"] = "keyless"
        out["server"] = "%s %s" % (info.get("name", "?"), info.get("version", "?"))
        out["echoed"] = rpc["result"].get("protocolVersion")
        # The server's own preferred version, not the one we asked for.
        alt = dict(INIT)
        alt["params"] = dict(INIT["params"], protocolVersion=IMPOSSIBLE)
        s2, _, b2 = post(url, alt)
        r2 = mcp_result(b2) if s2 and 200 <= s2 < 300 else None
        if r2 and "result" in r2:
            out["preferred"] = r2["result"].get("protocolVersion")
            out["echoes"] = out["preferred"] != IMPOSSIBLE and out["echoed"] != out[
                "preferred"
            ]
    elif rpc and "error" in rpc:
        out["reading"] = "error"
        out["detail"] = str(rpc["error"])[:200]
    elif status in (401, 403):
        out["reading"] = "account-bound"
    elif status in (404, 405, 406, 415):
        out["reading"] = "not an endpoint"
    else:
        out["reading"] = "unclear"
        out["detail"] = body[:200]
    return out


def discloses(notes, status):
    """Do the entry's own agentAccess.notes name the status this sweep measured?

    This is the CONTROL, and it is the whole point of the function (issue #138,
    wake 8da7f882). What this sweep measures is ONE unauthenticated
    `initialize` against ONE surface; what `noAccountNeeded` claims is about the
    WHOLE ENTRY. Until 2026-09-22 nothing in the registry made those two come
    apart, so substituting one for the other looked free. Two entries made them
    come apart on the same night, from opposite directions:

      bitrefill  MCP is account-bound (401); the keyless path is a DIFFERENT
                 surface (x402 REST). The sweep read the wrong surface.
      firecrawl  MCP `initialize` answers 200 and every tool behind it refuses
                 with KEYLESS_ACCESS_NOT_AVAILABLE *inside* that 200. The sweep
                 read the cheap half of the right surface.

    Both entries had already written the measurement this sweep makes into
    their own notes. An entry that names the status the wire returned has
    accounted for it and is being MORE precise than this instrument; an entry
    that has genuinely drifted names the OLD status, not the new one, so the
    test comes out the other way exactly when it must:

      a keyless server that starts demanding an account -> wire 401, notes
      say 200 -> no match -> DRIFT, which is right.
      an account-bound server that opens up -> wire 200, notes say 401 ->
      no match -> DRIFT, which is right.

    WHAT THIS TEST CANNOT DO, stated because the first draft hid it. It reads
    a number out of prose, so a sentence that names a status in order to DENY
    it reads exactly like one that reports it. bitrefill's own notes are the
    example and they were the fixture that caught this: the first sentence
    mentioning 401 is "all 11 documented x402 routes answer 402, none 401 and
    none data", which is about a different surface and says the opposite, and
    the sentence that actually discloses is the next one. The first draft
    returned the first match and would have printed that denial as its reason
    -- a right verdict carrying a wrong reason, which is the defect I found in
    soft404.py one wake earlier and reproduced here the next day.

    So this returns EVERY sentence naming the status, never just the first,
    the row stays flagged in the report, and the verdict is "not counted as
    drift" rather than "cleared". A human reads the sentences. An instrument
    that reads prose is a reason to look, never a proof.
    """
    if not notes or status is None:
        return None
    token = str(status)
    hits = [s.strip() for s in re.split(r"(?<=[.;])\s+", notes)
            if re.search(r"(?<![0-9])%s(?![0-9])" % token, s)]
    return " | ".join(hits) if hits else None


def verdict(claim_free, reading, notes=None, status=None):
    """Does the wire agree with the entry, and if not, did the entry say so?

    A disagreement this sweep can see is a CANDIDATE, not a finding, because
    one request to one surface cannot settle an entry-level cell. It becomes
    DRIFT only when the entry has not already recorded what the wire says.
    """
    if reading == "not an endpoint":
        return "NOT AN ENDPOINT (surfaces.mcp is a documentation page)"
    if reading not in ("keyless", "account-bound"):
        return "UNRESOLVED (%s)" % reading
    agrees = claim_free if reading == "keyless" else not claim_free
    if agrees:
        return "AGREES"
    said = discloses(notes, status)
    if said:
        return "DISCLOSED: the entry's own notes record this reading: %s" % said
    return ("DRIFT: entry says an account is needed" if reading == "keyless"
            else "DRIFT: entry says no account is needed")


def selftest():
    """Falsify the verdict table with no network and no clone.

    Every rule here exists because getting it wrong is invisible in a report
    that only ever prints numbers. The two rows tonight's merges produced are
    both in here as fixtures, with their real notes text.
    """
    BITREFILL = ("Re-measured 2026-09-21, no credentials, no payment: all 11 "
                 "documented x402 routes answer 402, none 401 and none data. "
                 "MCP initialize and REST ping answer 401 (2026-09-11, re-read 09-18).")
    FIRECRAWL = ("Re-measured 2026-09-22 18:16Z, no credentials, same as "
                 "2026-09-09: REST /v2/scrape 403 (IP looks suspicious); keyless "
                 "MCP tools/call KEYLESS_ACCESS_NOT_AVAILABLE inside a 200.")
    ok = fail = 0

    def check(label, got, want):
        nonlocal ok, fail
        if got == want:
            ok += 1
            print("ok: %s" % label)
        else:
            fail += 1
            print("FAIL: %s\n    got  %r\n    want %r" % (label, got, want))

    def head(v):
        return v.split(":")[0] if ":" in v else v.split(" (")[0]

    # The plain agreements, which must not become disclosures.
    check("keyless wire + free claim agrees",
          head(verdict(True, "keyless", FIRECRAWL, 200)), "AGREES")
    check("account-bound wire + paid claim agrees",
          head(verdict(False, "account-bound", BITREFILL, 401)), "AGREES")

    # Tonight's two rows: disagreement the entry already recorded.
    check("bitrefill: 401 named in its own notes is disclosed, not drift",
          head(verdict(True, "account-bound", BITREFILL, 401)), "DISCLOSED")
    check("firecrawl: 200 named in its own notes is disclosed, not drift",
          head(verdict(False, "keyless", FIRECRAWL, 200)), "DISCLOSED")

    # THE CONTROL. Real drift must still be caught, in both directions, and
    # these are the fixtures that make the rule falsifiable rather than a way
    # of explaining every disagreement away.
    check("a keyless server that starts demanding an account is DRIFT",
          head(verdict(True, "account-bound",
                       "Measured 2026-09-01: initialize answers 200, keyless.", 401)),
          "DRIFT")
    check("an account-bound server that opens up is DRIFT",
          head(verdict(False, "keyless",
                       "Measured 2026-09-01: initialize answers 401, OAuth.", 200)),
          "DRIFT")
    check("no notes at all cannot disclose anything",
          head(verdict(True, "account-bound", None, 401)), "DRIFT")
    check("empty notes cannot disclose anything",
          head(verdict(True, "account-bound", "", 401)), "DRIFT")
    check("notes that name no status cannot disclose",
          head(verdict(True, "account-bound", "The vendor says it is free.", 401)),
          "DRIFT")

    # A status this sweep never measured must not be borrowed from a neighbour.
    check("notes naming 403 do not disclose a measured 401",
          head(verdict(True, "account-bound", "REST answers 403 to a cloud IP.", 401)),
          "DRIFT")

    # Substring accidents. "401" must not be found inside 1401 or 4010, and a
    # 200 must not be matched out of a byte count or a price.
    check("1401 does not contain a disclosing 401",
          head(verdict(True, "account-bound", "The page is 1401 bytes.", 401)), "DRIFT")
    check("4010 does not contain a disclosing 401",
          head(verdict(True, "account-bound", "Port 4010 is the mock.", 401)), "DRIFT")
    check("a 2000-byte body does not disclose a measured 200",
          head(verdict(False, "keyless", "The shell is 2000 bytes.", 200)), "DRIFT")

    # A status is not a verdict: statuses this sweep cannot read stay unread.
    check("not an endpoint is never drift",
          head(verdict(True, "not an endpoint", None, 404)), "NOT AN ENDPOINT")
    check("no answer is unresolved, never drift, and names why",
          verdict(True, "no answer", None, None), "UNRESOLVED (no answer)")
    check("an unclear status is unresolved, never drift, and names why",
          verdict(False, "unclear", None, 500), "UNRESOLVED (unclear)")
    check("a status of None cannot disclose", discloses(BITREFILL, None), None)

    # The disclosure must carry its evidence, not merely a flag.
    # The fixture that caught the first draft: bitrefill's notes name 401
    # TWICE, and the first mention is a denial about a different surface.
    said = discloses(BITREFILL, 401)
    check("the disclosing sentences are returned, not a boolean",
          said is not None and "MCP initialize" in said, True)
    check("EVERY sentence naming the status is returned, not just the first",
          said.count("401") == 2 and " | " in said, True)
    check("the denial is shown too, so a reader can see it is a denial",
          "none 401" in said, True)

    print("\n%d ok, %d failed" % (ok, fail))
    return 1 if fail else 0


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--selftest", action="store_true",
                    help="falsify the verdict table; no network, no clone")
    ap.add_argument("--clone", default="/tmp/reg")
    ap.add_argument("--json", default=None)
    ap.add_argument("--ref", default="origin/main",
                    help="the tree this sweep claims to have read; it must be "
                         "checked out at exactly this (wake fe04f262)")
    args = ap.parse_args()

    if args.selftest:
        return selftest()

    if not os.path.isdir(args.clone):
        subprocess.check_call(
            ["git", "clone", "-q",
             "https://github.com/PublicAgents/public-agents.git", args.clone]
        )
    # Wake 520aee0c. This sweep was the LAST one without the guards, and it is
    # the one whose number I quote most often ("0 of 31 disagree"). That day it
    # ran happily over a clone carrying four of my own uncommitted scratch
    # files, minutes after deployed.py and negatives.py had both refused the
    # very same tree. The result was right by luck: my edits were outside
    # registry/, which no caller of this number could have known. An instrument
    # is only as trustworthy as its worst silent corpus.
    corpus.require_clone(args.clone, "the drift sweep")
    head = corpus.require_ref(args.clone, args.ref, "the drift sweep")

    tools = []
    for f in sorted(glob.glob(os.path.join(args.clone, "registry/tools/*/tool.json"))):
        d = json.load(open(f))
        url = (d.get("surfaces") or {}).get("mcp")
        if not url:
            continue
        aa = d.get("agentAccess") or {}
        tools.append(
            {
                "slug": d["slug"],
                "url": url,
                "claim_free": aa.get("noAccountNeeded"),
                "auth": aa.get("auth"),
                "notes": aa.get("notes"),
                "updated": d.get("updated"),
            }
        )

    now = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    print("registry %s, %d tools with an mcp surface, measured %s\n" % (
        head, len(tools), now))

    with concurrent.futures.ThreadPoolExecutor(max_workers=6) as ex:
        results = list(ex.map(lambda t: probe(t["url"]), tools))

    rows = []
    for t, r in zip(tools, results):
        v = verdict(t["claim_free"], r["reading"], t.get("notes"), r.get("status"))
        row = dict(t, **r)
        row["verdict"] = v
        rows.append(row)
        flag = "  " if v.startswith("AGREES") else "!!"
        print("%s %-14s %-16s claim=%s/%-8s wire=%s %s" % (
            flag, t["slug"], v.split(":")[0], t["claim_free"], t["auth"],
            r["reading"], r.get("server", r.get("detail", "") or "")[:40]))

    echoers = [r for r in rows if r.get("echoes")]
    if echoers:
        print("\nprotocol versions (asked 2025-06-18, then an impossible one):")
        for r in echoers:
            print("   %-14s echoed %s, prefers %s" % (
                r["slug"], r.get("echoed"), r.get("preferred")))

    disclosed = [r for r in rows if r["verdict"].startswith("DISCLOSED")]
    if disclosed:
        print("\ndisclosed (the wire disagrees with the cell and the entry says so;")
        print("this sweep reads ONE initialize against ONE surface, and an entry-level")
        print("cell can be true of another surface or of a request this never makes):")
        for r in disclosed:
            print("   %-14s wire=%s, status %s" % (
                r["slug"], r["reading"], r.get("status")))
            print("      %s" % r["verdict"].split(": ", 2)[-1][:300])

    drift = [r for r in rows if r["verdict"].startswith("DRIFT")]
    print("\n%d of %d disagree with their entry and do not say so (drift)." % (
        len(drift), len(rows)))
    print("%d of %d disagree with this sweep and do say so (disclosed, not a finding)." % (
        len(disclosed), len(rows)))

    if args.json:
        json.dump({"head": head, "measured": now, "rows": rows},
                  open(args.json, "w"), indent=1)
    return 1 if drift else 0


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