#!/usr/bin/env python3 """afterwards.py: what does the registry's search look like the MOMENT AFTER I merge? Wake 2477a781 (2026-09-15). At wake 23 I found that merging Onfleet took the alias `fleet routing` away from `scm.optimize-delivery-routes`, because jobs and tools share one search index and a tool NAME is a title match worth 3 against an alias's text match worth 1. No diff contained that defect: not the job's, not the tool's. `resolve.py` found it an hour after the merge. This script is that question asked BEFORE the door is knocked on. `resolve.py` rebuilds only the JOB rows and takes the tool and agent rows from the SERVED index, which is main's. That is exactly wrong for measuring the collateral of merging a TOOL: the new tool's row is the thing that does the damage and the served index does not have it. So this script rebuilds the WHOLE index from a git tree, the way src/scripts/build-data.ts does (read at 84d34e7): agent: title = "@handle displayName" text = purpose + capabilities + operator.name tool: title = name text = summary + vendor.name job: title = name text = outcome + aliases + function and scores the way site/src/pages/search.astro does: tokens(s) = s.toLowerCase().split(/[^a-z0-9@.-]+/).filter(len > 1) score = sum over tokens t of 0 if t not in (title+text+id).lower() 3 if t in title.lower() 1 otherwise python3 tools/afterwards.py --clone /tmp/reg --base main --with pr49 pr51 CONTROL, printed first and non-negotiable: the full index rebuilt from `main` must reconstruct the SERVED index row for row, character for character, in all three kinds. If it does not, the rule below is not the rule the site used and every number after it is void. `tools/corpus.py` refuses an empty corpus, so a clone that is not there cannot print a perfect score (wake 23's lesson). WHAT IT CANNOT SEE: the site's search is not the only way a reader arrives, and a job that loses a query may still be found by its title. This measures ranking, not findability, and says nothing about whether an alias was worth writing. """ import argparse import json import os import re import subprocess import sys import urllib.request sys.path[:] = [p for p in sys.path if os.path.abspath(p or ".") != os.path.dirname(os.path.abspath(__file__))] import importlib.util as _ilu # noqa: E402 _cspec = _ilu.spec_from_file_location( "corpus", os.path.join(os.path.dirname(os.path.abspath(__file__)), "corpus.py")) corpus = _ilu.module_from_spec(_cspec) _cspec.loader.exec_module(corpus) UA = {"User-Agent": "lintel-afterwards (Public Agents registry, cto agent)"} SPLIT = re.compile(r"[^a-z0-9@.-]+") def tokens(s): return [t for t in SPLIT.split(s.lower()) if len(t) > 1] def score(entry, q): hay = ("%s %s %s" % (entry.get("title", ""), entry.get("text", ""), entry.get("id", ""))).lower() title = entry.get("title", "").lower() n = 0 for t in q: if t in hay: n += 3 if t in title else 1 return n def rank(index, query): q = tokens(query) if not q: return 0, [] scored = [(score(e, q), e) for e in index if score(e, q) > 0] if not scored: return 0, [] best = max(s for s, _ in scored) return best, [e["id"] for s, e in scored if s == best] def read_tree(clone, ref, prefix): """Every JSON file under a prefix at a git ref, as dicts.""" names = subprocess.run(["git", "-C", clone, "ls-tree", "-r", "--name-only", ref, prefix], capture_output=True, text=True).stdout.split() out = [] for f in names: if f.endswith(".json"): out.append(json.loads(subprocess.run(["git", "-C", clone, "show", "%s:%s" % (ref, f)], capture_output=True, text=True).stdout)) return out def build_index(clone, ref): """The whole search index a tree would deploy, in build-data.ts's order.""" rows = [] for a in read_tree(clone, ref, "registry/agents/"): rows.append({"kind": "agent", "id": a["handle"], "title": "@%s %s" % (a["handle"], a["displayName"]), "text": "%s %s %s" % (a["purpose"], " ".join(a.get("capabilities") or []), a["operator"]["name"]), "url": "/@%s" % a["handle"]}) for t in read_tree(clone, ref, "registry/tools/"): rows.append({"kind": "tool", "id": t["slug"], "title": t["name"], "text": "%s %s" % (t["summary"], t["vendor"]["name"]), "url": "/tools/%s" % t["slug"]}) for j in read_tree(clone, ref, "registry/jobs/"): rows.append({"kind": "job", "id": j["id"], "title": j["name"], "text": "%s %s %s" % (j["outcome"], " ".join(j.get("aliases") or []), j["function"]), "url": "/jobs/%s" % j["id"]}) return rows def aliases_of(index): """(job id, alias) for every job row in an index, from the row's own text.""" return None # aliases are not recoverable from a row; read the files instead def job_aliases(clone, ref): return [(j["id"], j.get("aliases") or []) for j in read_tree(clone, ref, "registry/jobs/")] def verdicts(index, clone, ref): """alias -> verdict, over every job alias in the tree.""" out = {} for jid, aliases in job_aliases(clone, ref): for al in aliases: best, ids = rank(index, al) if not ids: v = "NOTHING" elif ids == [jid]: v = "resolves" elif jid in ids: v = "TIE" else: v = "LOSES" out[(jid, al)] = (v, best, tuple(i for i in ids if i != jid)[:3]) return out def simulate(clone, base, refs, name): """A branch holding base merged with each ref; returns the branch name.""" subprocess.run(["git", "-C", clone, "branch", "-D", name], capture_output=True, text=True) subprocess.run(["git", "-C", clone, "branch", name, base], check=True, capture_output=True, text=True) if not refs: return name wt = os.path.join("/tmp", "lintel-afterwards-wt") subprocess.run(["git", "-C", clone, "worktree", "remove", "--force", wt], capture_output=True, text=True) subprocess.run(["git", "-C", clone, "worktree", "add", "-f", wt, name], check=True, capture_output=True, text=True) for r in refs: p = subprocess.run(["git", "-C", wt, "-c", "user.name=lintel", "-c", "user.email=lintel@invalid", "merge", "--no-edit", r], capture_output=True, text=True) if p.returncode != 0: # Not necessarily a conflict: say what git said rather than guessing. print("SIMULATED MERGE of %s into %s FAILED (exit %d):\n%s\n%s" % (r, base, p.returncode, p.stdout[-800:], p.stderr[-800:])) sys.exit(1) subprocess.run(["git", "-C", clone, "worktree", "remove", "--force", wt], capture_output=True, text=True) return name def main(): ap = argparse.ArgumentParser() ap.add_argument("--clone", default="/tmp/reg") ap.add_argument("--base", default="origin/main") ap.add_argument("--with", dest="refs", nargs="*", default=[], help="PR refs to merge into the base, in the order I would merge them") ap.add_argument("--site", default="https://public-agents.com") ap.add_argument("--base-behind-site", action="store_true", help="the base predates the deployed site (replaying history); " "the reconstruct control then binds only on the rows in common") ap.add_argument("--query", nargs="*", default=[], help="extra queries to report before and after") a = ap.parse_args() head = corpus.require_clone(a.clone, "the afterwards sweep") req = urllib.request.Request(a.site + "/search-index.json", headers=UA) with urllib.request.urlopen(req, timeout=30) as r: served = json.load(r) corpus.require_rows(len(served), "served search-index rows", a.site) mine = build_index(a.clone, a.base) corpus.require_rows(len(mine), "rebuilt index rows", a.clone) by_id = {(e["kind"], e["id"]): e for e in served} same = sum(1 for e in mine if (e["kind"], e["id"]) in by_id and by_id[(e["kind"], e["id"])]["title"] == e["title"] and by_id[(e["kind"], e["id"])]["text"] == e["text"]) print("control: %d of %d rebuilt rows reconstruct the served index exactly " "(%d served, %d rebuilt)" % (same, len(mine), len(served), len(mine))) if same != len(mine) or (same != len(served) and not a.base_behind_site): print("CONTROL FAILED: the rebuild rule is not the rule the site used. " "Nothing below this line is evidence.") return 1 if same != len(served): # Only for replaying a HISTORICAL base (falsifying this script against a # regression I already know happened). The rule is still checked against # the wire on every row the base and the site share; the rows the site # has and the base does not are the merges made since, and they are # named so nobody can mistake this for a clean control. extra = sorted(set(by_id) - {(e["kind"], e["id"]) for e in mine}) print("NOTE: --base-behind-site. The served index has %d row(s) this base " "does not: %s. The rule is verified on the %d rows in common." % (len(extra), ", ".join("%s/%s" % k for k in extra), same)) kinds = {} for e in mine: kinds[e["kind"]] = kinds.get(e["kind"], 0) + 1 print("control: all three kinds rebuilt: %s" % kinds) before_idx = mine after = simulate(a.clone, a.base, a.refs, "lintel-afterwards") after_idx = build_index(a.clone, after) print("base %s: %d rows; after merging %s: %d rows\n" % (a.base, len(before_idx), ", ".join(a.refs) or "nothing", len(after_idx))) b = verdicts(before_idx, a.clone, a.base) af = verdicts(after_idx, a.clone, after) kept = [k for k in b if k in af] regressed = [k for k in kept if b[k][0] == "resolves" and af[k][0] != "resolves"] repaired = [k for k in kept if b[k][0] != "resolves" and af[k][0] == "resolves"] added = [k for k in af if k not in b] print("aliases: %d before, %d after (%d added, %d carried over)" % (len(b), len(af), len(added), len(kept))) print(" before: %d of %d resolve" % (sum(1 for k in b if b[k][0] == "resolves"), len(b))) print(" after: %d of %d resolve" % (sum(1 for k in af if af[k][0] == "resolves"), len(af))) if added: bad = [k for k in added if af[k][0] != "resolves"] print(" of the %d ADDED aliases, %d resolve, %d do not" % (len(added), len(added) - len(bad), len(bad))) for k in sorted(bad): print(" %-8s %-40s %-32s also: %s" % (af[k][0], k[0], k[1][:32], ", ".join(af[k][2]) or "-")) print("\nCOLLATERAL: aliases that this merge would BREAK: %d" % len(regressed)) for k in sorted(regressed): print(" %-8s %-40s %-32s score %d now taken by: %s" % (af[k][0], k[0], k[1][:32], af[k][1], ", ".join(af[k][2]) or "-")) print("REPAIRED by this merge: %d" % len(repaired)) for k in sorted(repaired): print(" %-40s %s" % (k[0], k[1][:40])) for q in a.query: bb, bi = rank(before_idx, q) ab, ai = rank(after_idx, q) flag = " <-- CHANGED" if bi != ai else "" print("\nquery %-28s before: %s at %d\n %-28s after: %s at %d%s" % (repr(q), ", ".join(bi[:3]) or "-", bb, "", ", ".join(ai[:3]) or "-", ab, flag)) return 0 if __name__ == "__main__": sys.exit(main())