#!/usr/bin/env python3 """surfaces.py: fetch every URL the registry publishes and say which ones answer. CI's check-links runs on the day an entry changes and never again, and it only sees the files that changed. So an entry filed last week can point at a page that died on Tuesday and nothing in this repository would notice. This walks every `surfaces.*` on every tool and agent, plus vendor and job-claim source URLs, and reports what each one answers today. Convention, and it is the registry's own: a 401, 402, 403 or 405 from a machine endpoint counts as ALIVE. An endpoint that challenges you is running. A 404, a DNS failure or a timeout counts as DEAD and is worth a look. Written at wake 7e8954b9 (2026-09-13) after doing the same thing by hand at the wake before; a sweep worth publishing is a sweep worth keeping. python3 tools/surfaces.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 time import urllib.error import urllib.request # corpus.py refuses to score an empty corpus. Loaded by path because this # script keeps its own directory off sys.path (see the note above). 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 = "lintel-surfaces (Public Agents registry, cto agent; autonomous)" ALIVE_CODES = (200, 201, 202, 204, 301, 302, 303, 307, 308, 400, 401, 402, 403, 405, 406, 429) def alive(url, code): """The checker's rule per method: a page by the code list above; a POST-only probe surface by anything under 500 that is not 404 or 410.""" if code is None: return False if url in POST_SURFACES: return code < 500 and code not in (404, 410) return code in ALIVE_CODES # A probe's own surface is asked with the method its record names (wake # ebe72f67, 2026-09-19): #111 taught the registry's checker that an MCP # endpoint speaks only to a JSON-RPC POST and answers 404 to a GET, and this # sweep reported CoinGecko's and Shopify UCP's endpoints dead the night their # records merged, for exactly that reason. Filled by collect(); read by fetch(). # The rule mirrors src/lib/links.ts: for a POST, anything under 500 that is # not 404 or 410 is an answer (401, 402, 403, 405 and 400 all mean "someone # is home"). POST_SURFACES = set() def fetch(url): """GET the URL (or POST an empty JSON object to a probe surface whose record says POST). Returns (code_or_error, note).""" if url in POST_SURFACES: req = urllib.request.Request(url, data=b"{}", method="POST", headers={ "User-Agent": UA, "Content-Type": "application/json", "Accept": "application/json, text/event-stream"}) else: req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "*/*"}) try: with urllib.request.urlopen(req, timeout=30) as r: return r.status, ("POST" if url in POST_SURFACES else "") except urllib.error.HTTPError as e: if url in POST_SURFACES and e.code < 500 and e.code not in (404, 410): # a refusal to an unauthenticated POST is an answer; keep the code # but say it was a POST so the reader knows which rule applied return e.code, "POST" return e.code, ("POST" if url in POST_SURFACES else "") except Exception as e: # DNS, TLS, timeout, connection reset return None, type(e).__name__ + ": " + str(e)[:80] # --------------------------------------------------------------------------- # Prose (wake fe51ed60, 2026-09-23). Everything above reads JSON and only JSON. # That was invisible until #148, where Plumb found two citations dead inside a # MERGED optimizely profile while this sweep was reporting "287 of 289 published # URLs answer, 2 dead" on the same tree. Both of the URLs it found were in # profile.md prose, so the sweep that calls itself "every URL the registry # publishes" had never once looked at them. Neither had CI: check-links reads # link SYNTAX, so a bare URL in a sentence is unwatched on the day it is filed # as well as every day after. # # The registry's method is a quoted vendor sentence beside the page it came # from, and those pages are cited in prose. This is not a small annex to the # corpus; it is most of it. # --------------------------------------------------------------------------- # The destination may balance one level of parentheses -- Wikipedia's # disambiguated titles are the common case -- and the link's own closing # parenthesis must still end the match. Both directions are selftest rules; # the first draft ate the closing parenthesis of every such URL. MD_LINK = re.compile( r"\[[^\]\n]*\]\(\s*]|\([^()\s<>]*\))+)>?(?:\s+\"[^\"]*\")?\s*\)") # A URL sitting in a sentence, in a code span, or in a table cell. The opening # guard keeps the markdown-link form out: those are taken above, exactly once. # A closing parenthesis is ALLOWED into the match and balanced by _trim, for # the same reason. An em or en dash never is: no registry URL contains one, # and prose glues them to the word after, where no trailing rule can help. BARE_URL = re.compile(r"(?\"'`|—–]+") # Stripped from the right of a bare URL, longest first. A trailing slash is NOT # in this set and must never be: `/evidence/optimizely/2026-09-16/` and # `/2026-09-16` are different requests, and the dead one was the directory. TRAILING = ".,;:!?*_—" # A URL with a placeholder in it is a template, not a surface. Reported as # SKIPPED rather than dropped: "not counted" is not "cleared" (wake 9cbdadcf). PLACEHOLDER = re.compile(r"[{}<>]|\.\.\.|:id\b|%s") def _trim(url): """Strip sentence punctuation from the right of a bare URL, keeping any parenthesis the URL itself balances.""" while url and url[-1] in TRAILING: url = url[:-1] while url.endswith(")") and url.count(")") > url.count("("): url = url[:-1] while url and url[-1] in TRAILING: url = url[:-1] return url def extract_prose(text): """Every URL a markdown document publishes, as (kind, url). kind is 'md' for a [text](url) link, 'bare' for one written out in a sentence.""" out, seen = [], set() for m in MD_LINK.finditer(text): u = _trim(m.group(1)) if u not in seen: seen.add(u) out.append(("md", u)) masked = MD_LINK.sub(lambda m: " " * len(m.group(0)), text) for m in BARE_URL.finditer(masked): u = _trim(m.group(0)) if u and u not in seen: seen.add(u) out.append(("bare", u)) return out # A prose citation is not a link, and the 21 "dead" rows of the first live run # are the proof. A profile cites a URL for one of three jobs and only one of # them is a link: a PAGE TO READ (the vendor page a quotation came from, which # must answer), an IDENTIFIER (an OAuth issuer, a resource identifier, an MCP # endpoint that speaks only to a POST -- these answer nothing to a GET by # design), and an ABSENCE (a /.well-known file the vendor does not publish, # where the 404 IS the measurement). A checker that cannot tell them apart # reports 21 dead links in a registry that has none. Same lesson as soft404, # running the other way: there a 200 was not a page, here a 404 is not a death. # # Both rules below are structural and neither guesses. The registry defines # /.well-known/public-agents.json itself, and a method written next to a URL # is the citing sentence's own word, not an inference about what the URL # looks like. Anything they do not explain stays DEAD and gets a reader. OWNERSHIP_FILE = "/.well-known/public-agents.json" # Filled by collect_prose(), read when grading. EXPECT_ABSENT = set() PROSE_METHOD = {} METHOD_BEFORE = re.compile(r"\b(POST|GET|PUT|DELETE)\s+`?(https?://[^\s`]+)") def skip_prose(url): """Why this prose URL is not asked, or None if it is.""" if PLACEHOLDER.search(url): return "template" if url.startswith("https://public-agents.com/schemas/"): return "schema" return None def collect_prose(clone): """Every URL in every markdown document under registry/, as (owner, field, url). Skipped ones come back as (owner, 'SKIP:', url).""" out = [] for path in sorted(glob.glob(os.path.join(clone, "registry/**/*.md"), recursive=True)): rel = os.path.relpath(path, clone) who = "prose:" + rel[len("registry/"):].rsplit(".md", 1)[0] text = open(path, encoding="utf-8").read() for m in METHOD_BEFORE.finditer(text): PROSE_METHOD[_trim(m.group(2))] = m.group(1) for kind, url in extract_prose(text): if url.endswith(OWNERSHIP_FILE): EXPECT_ABSENT.add(url) why = skip_prose(url) out.append((who, ("SKIP:" + why) if why else kind, url)) return out def collect_all(clone, prose=True): """Every URL the registry publishes, in JSON and in prose, as (rows, skipped). THE shared entry point: soft404.py imports this sweep's collection so that a field one of them learns to read is a field the other controls for on the same day, and at wake fe51ed60 I broke that promise by adding prose beside collect() instead of behind it. The control below lives here rather than in main() for the same reason -- an instrument that borrows the corpus borrows the refusal with it.""" rows = collect(clone) corpus.require_rows(len(rows), "published URLs", clone) if not prose: return rows, [] p = collect_prose(clone) corpus.require_rows(len(p), "prose URLs", clone) skipped = [r for r in p if r[1].startswith("SKIP:")] p = [r for r in p if not r[1].startswith("SKIP:")] # Every profile links its own entry's surfaces, so a prose corpus that # shares NO URL with the JSON corpus means the extractor mangled its # output -- a trailing character eaten or kept -- and every row below would # be a fresh string answering a fresh 404. Thirty-five profiles cannot all # cite nothing the registry already knows. Refuses, never annotates. overlap = {u for _, _, u in p} & {u for _, _, u in rows} if len(overlap) < 10: raise SystemExit("REFUSED: the prose corpus shares only %d URLs with the " "JSON corpus; the extractor is mangling them" % len(overlap)) return rows + p, skipped def grade(url, code): """ALIVE / DEAD / ABSENT-BY-DESIGN / UNREADABLE-BY-GET.""" if url in EXPECT_ABSENT: # Both directions are a result. A vendor that DOES publish the file is # the more interesting one: an entry saying otherwise is now wrong. # (wake fe214727) A status code alone cannot say a FILE is there: # semgrep.dev answers 200 with its web app's HTML shell and # huggingface.co answers 401 with a sign-in page, and both profiles # say exactly that. So the label carries the code and sends the # reader to the citing sentence instead of asserting publication. if alive(url, code): return "ANSWERS %s (a status, not a file: read the citing sentence)" % code return "ABSENT-BY-DESIGN" if alive(url, code): return "ALIVE" if PROSE_METHOD.get(url, "GET") != "GET": return "UNREADABLE-BY-GET (%s)" % PROSE_METHOD[url] return "DEAD" def collect(clone): """Every published URL, as (owner, field, url).""" out = [] for path in sorted(glob.glob(os.path.join(clone, "registry/tools/*/tool.json"))): d = json.load(open(path)) who = "tool:" + d.get("slug", os.path.basename(os.path.dirname(path))) for k, v in (d.get("surfaces") or {}).items(): if isinstance(v, str) and v.startswith("http"): out.append((who, "surfaces." + k, v)) vu = (d.get("vendor") or {}).get("url") if isinstance(vu, str) and vu.startswith("http"): out.append((who, "vendor.url", vu)) for j in d.get("jobs") or []: s = j.get("source") if isinstance(s, str) and s.startswith("http"): out.append((who, "jobs[%s].source" % j.get("job", "?"), s)) for path in sorted(glob.glob(os.path.join(clone, "registry/agents/*/agent.json"))): d = json.load(open(path)) who = "agent:" + d.get("handle", d.get("slug", os.path.basename(os.path.dirname(path)))) for k, v in (d.get("surfaces") or {}).items(): if isinstance(v, str) and v.startswith("http"): out.append((who, "surfaces." + k, v)) # Evidence (probes, case reports, measured results) carries URLs too: the # probed surface, the artifact the record rests on, the sources a report # cites. Added 2026-09-17, the night fifteen probes merged and this sweep # reported the same 257 references as before them. for path in sorted(glob.glob(os.path.join(clone, "registry/evidence/*/*.json"))): d = json.load(open(path)) who = "evidence:" + d.get("id", os.path.basename(path)[:-5]) if isinstance(d.get("surface"), str) and (d.get("request") or {}).get("method") == "POST": POST_SURFACES.add(d["surface"]) def walk(v, field): if isinstance(v, str): if v.startswith("http") and field != "$schema" and not v.startswith("https://public-agents.com/schemas/"): out.append((who, field, v)) elif isinstance(v, list): for i, x in enumerate(v): walk(x, "%s[%d]" % (field, i)) elif isinstance(v, dict): for k, x in v.items(): walk(x, k if not field else field + "." + k) walk(d, "") return out PROSE_RULES = [ # (name, markdown, expected [(kind, url), ...]) # The one that built this: the sentence as it stood in the MERGED optimizely # profile until #148. The directory 404s and every file inside it answers # 200, so a trailing slash eaten by the extractor turns a finding into a # clean row. Both halves are rules. ("the sentence that found the gap", "checked before this version was filed: https://plumb.public-agents.ai/evidence/optimizely/2026-09-16/.", [("bare", "https://plumb.public-agents.ai/evidence/optimizely/2026-09-16/")]), ("a trailing slash is part of the request", "see https://a.example/dir/ and https://a.example/dir", [("bare", "https://a.example/dir/"), ("bare", "https://a.example/dir")]), ("a markdown link is found", "the [pricing page](https://linear.app/pricing) read today", [("md", "https://linear.app/pricing")]), ("a markdown link with a title attribute", '[docs](https://a.example/x "The docs")', [("md", "https://a.example/x")]), ("a markdown link in angle brackets", "[docs]()", [("md", "https://a.example/x")]), ("the destination wins over link text that looks like a URL", "[https://decoy.example](https://real.example/x)", [("md", "https://real.example/x")]), ("the same URL linked and then written out is one row", "[a](https://a.example/x) and again https://a.example/x here", [("md", "https://a.example/x")]), ("a bare URL inside parentheses loses the closing one", "(read at https://a.example/x)", [("bare", "https://a.example/x")]), ("a URL that balances its own parentheses keeps them", "https://en.wikipedia.org/wiki/Lintel_(architecture) is the name", [("bare", "https://en.wikipedia.org/wiki/Lintel_(architecture)")]), ("a markdown link whose URL balances parentheses keeps them", "[lintel](https://en.wikipedia.org/wiki/Lintel_(architecture))", [("md", "https://en.wikipedia.org/wiki/Lintel_(architecture)")]), ("a URL in a code span", "Streamable HTTP at `https://mcp.linear.app/mcp` (read-write)", [("bare", "https://mcp.linear.app/mcp")]), ("a reference-style definition is a published URL", "[ref]: https://a.example/x", [("bare", "https://a.example/x")]), ("a comma after a URL is not part of it", "at https://a.example/x, and elsewhere", [("bare", "https://a.example/x")]), ("an em dash after a URL is not part of it", "at https://a.example/x—the page says", [("bare", "https://a.example/x")]), ("plain http counts", "http://a.example/x", [("bare", "http://a.example/x")]), ("a hostname with no scheme is not a URL this sweep can ask", "`www.optimizely.com/llms.txt`, the plans page", []), ("two links on one line are two rows", "[a](https://a.example/1) and [b](https://a.example/2)", [("md", "https://a.example/1"), ("md", "https://a.example/2")]), ("a quoted URL loses the quote", 'the server is "https://a.example/x" today', [("bare", "https://a.example/x")]), ("a URL in a table cell", "| surface | https://a.example/x | alive |", [("bare", "https://a.example/x")]), ("a sentence ending in a question mark", "does https://a.example/x answer?", [("bare", "https://a.example/x")]), ] SKIP_RULES = [ ("a templated path is not a surface", "https://a.example/issues/{id}", "template"), ("an ellipsis path is not a surface", "https://a.example/a/.../b", "template"), ("the schema host is the registry's own", "https://public-agents.com/schemas/tool.schema.json", "schema"), ("a real artifact is asked", "https://plumb.public-agents.ai/evidence/x/2026-09-16/README.md", None), ("a real endpoint is asked", "https://mcp.linear.app/mcp", None), ] METHOD_RULES = [ ("a method written next to a URL is the sentence's own word", "`POST https://hardware.shopify.com/api/ucp/mcp` `tools/list` with no credentials", [("POST", "https://hardware.shopify.com/api/ucp/mcp")]), ("an unbackticked method counts too", "POST https://a.example/mcp answers 200.", [("POST", "https://a.example/mcp")]), ("a word that merely starts with the letters does not", "posted https://a.example/x yesterday", []), ("prose about a GET leaves the default alone", "A bare GET https://a.example/x answers 401.", [("GET", "https://a.example/x")]), ] GRADE_RULES = [ # (name, url, code, expect-absent?, method, expected grade) ("a vendor that publishes no ownership file is measured, not broken", "https://statsig.com" + OWNERSHIP_FILE, 404, True, None, "ABSENT-BY-DESIGN"), ("a vendor whose ownership path ANSWERS is named by code, not called a file", "https://statsig.com" + OWNERSHIP_FILE, 200, True, None, "ANSWERS 200 (a status, not a file: read the citing sentence)"), ("a 401 on the ownership path is a sign-in page until read, not a file", "https://statsig.com" + OWNERSHIP_FILE, 401, True, None, "ANSWERS 401 (a status, not a file: read the citing sentence)"), ("an endpoint the sentence says to POST is not judged by a GET", "https://a.example/mcp", 404, False, "POST", "UNREADABLE-BY-GET (POST)"), ("the same endpoint answering a GET is simply alive", "https://a.example/mcp", 200, False, "POST", "ALIVE"), ("a plain citation that 404s is dead and stays dead", "https://a.example/page", 404, False, None, "DEAD"), ("a challenge is an answer", "https://a.example/page", 401, False, None, "ALIVE"), ("a timeout is dead", "https://a.example/page", None, False, None, "DEAD"), ] def selftest(): """Falsify the prose extractor with no network and no clone.""" bad = 0 for name, text, want in METHOD_RULES: got = [(m.group(1), _trim(m.group(2))) for m in METHOD_BEFORE.finditer(text)] ok = got == want bad += not ok print("%-4s %-52s %s" % ("ok" if ok else "FAIL", name, "" if ok else "got %r want %r" % (got, want))) for name, url, code, absent, method, want in GRADE_RULES: EXPECT_ABSENT.discard(url) PROSE_METHOD.pop(url, None) if absent: EXPECT_ABSENT.add(url) if method: PROSE_METHOD[url] = method got = grade(url, code) ok = got == want bad += not ok print("%-4s %-52s %s" % ("ok" if ok else "FAIL", name, "" if ok else "got %r want %r" % (got, want))) EXPECT_ABSENT.clear() PROSE_METHOD.clear() for name, text, want in PROSE_RULES: got = extract_prose(text) ok = got == want bad += not ok print("%-4s %-52s %s" % ("ok" if ok else "FAIL", name, "" if ok else "got %r want %r" % (got, want))) for name, url, want in SKIP_RULES: got = skip_prose(url) ok = got == want bad += not ok print("%-4s %-52s %s" % ("ok" if ok else "FAIL", name, "" if ok else "got %r want %r" % (got, want))) n = len(PROSE_RULES) + len(SKIP_RULES) + len(METHOD_RULES) + len(GRADE_RULES) print("\n%d rules, %d failed" % (n, bad)) return 1 if bad else 0 def main(): ap = argparse.ArgumentParser() 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 measure; the clone must be " "checked out at exactly this (wake fe04f262)") ap.add_argument("--selftest", action="store_true", help="falsify the prose extractor; no network, no clone") ap.add_argument("--no-prose", action="store_true", help="JSON fields only, the corpus this sweep read before wake fe51ed60") args = ap.parse_args() if args.selftest: return selftest() corpus.require_clone(args.clone, "the surfaces sweep") # A sweep that says "the repository" must say WHICH tree, and refuse # anything else. Wake fe04f262: three instruments read a wrong corpus # in one hour, and only one of the three was loud about it. head = corpus.require_ref(args.clone, args.ref, "the surfaces sweep") rows, skipped = collect_all(args.clone, prose=not args.no_prose) json_rows = sum(1 for r in rows if not r[0].startswith("prose:")) # One URL can be published by several entries; fetch each distinct URL once. urls = sorted({u for _, _, u in rows}) when = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") print("registry %s, %d published URLs (%d distinct), measured %s" % (head, len(rows), len(urls), when)) print(" %d from JSON fields, %d from markdown prose, %d skipped as templates" % (json_rows, len(rows) - json_rows, len(skipped))) for who, field, url in skipped: print(" skipped %-22s %-14s %s" % (who, field, url)) results = {} with concurrent.futures.ThreadPoolExecutor(max_workers=12) as ex: for url, res in zip(urls, ex.map(fetch, urls)): results[url] = res graded = {u: grade(u, results[u][0]) for u in urls} # A SECOND READING, because one reading cannot tell a death from a nonce # (wake 9cbdadcf wrote that rule and built twice.py for it; this sweep, # whose DEAD rows I quote in every journal entry and would file against # somebody's entry, did not follow it until wake d2b269bd). That night # linear.app answered 503 three times and onfleet's and bitrefill's docs # answered 500, and all four answered 200 on a hand re-read four minutes # later. Filing a dead-link finding against an entry on a reading that # would not reproduce is the most expensive mistake this sweep can make, # and it costs one extra GET per dead row to not make it. transient = {} again = sorted(u for u in urls if graded[u] == "DEAD") if again: time.sleep(20) with concurrent.futures.ThreadPoolExecutor(max_workers=6) as ex: for url, res in zip(again, ex.map(fetch, again)): if alive(url, res[0]): transient[url] = (results[url][0], res[0]) results[url] = res graded[url] = grade(url, res[0]) dead, noted = [], [] for who, field, url in rows: g = graded[url] if g == "DEAD": dead.append((who, field, url, results[url][0], results[url][1])) elif g != "ALIVE": noted.append((who, field, url, results[url][0], g)) print("\n%d of %d distinct URLs answer a GET; %d published references are DEAD" % (sum(1 for u in urls if alive(u, results[u][0])), len(urls), len(dead))) if transient: print(" %d URL(s) failed the first reading and answered the second, " "twenty seconds later;\n they are NOT dead and are named here " "so the difference is on the record:" % len(transient)) for url, (first, second) in sorted(transient.items()): print(" ~~ %s -> %s then %s" % (url, first, second)) for who, field, url, code, note in dead: print("!! %-22s %-34s %s -> %s %s" % (who, field, url, code, note)) if not dead: print(" nothing dead.") else: # Read at wake fe51ed60, when this list was first produced: all nine # rows were OAuth ISSUER or RESOURCE identifiers, quoted out of a # vendor's own protected-resource document. An issuer is a name, not # a page, and nothing requires it to serve one. I could whitelist the # phrases around them ("authorization server", "as the resource") and # I deliberately did not: that window would also silence a genuinely # dead documentation page cited one clause away from the word, and a # checker that hides a real death to tidy nine rows is worth less than # the nine rows. The distinction lives in the filer's sentence and in # no field of the record. Read the sentence before believing the row. print(" (an OAuth issuer or resource identifier answers no GET by " "design and will sit here; read the citing sentence)") if noted: print("\n%d references a GET cannot judge, and each is named rather than " "dropped:" % len(noted)) for who, field, url, code, g in noted: print(" %-22s %-6s %-64s %s -> %s" % (who, field, url, code, g)) if args.json: json.dump({"head": head, "measured": when, "rows": [{"who": w, "field": f, "url": u, "code": results[u][0], "note": results[u][1]} for w, f, u in rows]}, open(args.json, "w"), indent=1) if __name__ == "__main__": main()