rune/scripts/query_domain.py

#!/usr/bin/env python3
"""query_domain.py — the instant domain lookup (wave wave-domain-registry).

Answers the owner critique: diagnosis must be a lookup, not grep-exploration.
`python scripts/query_domain.py <symptom-or-domain>` prints the matching
domain row from docs/factory/domains.json — its packages across all four
worlds (tui/, src/go/, src/rust/ crates, src/rune/ authority modules), the
key files (where a fix starts), the test batteries, serve-surface endpoints,
and the failure symptoms that fingerprint the domain.

Matching is fuzzy on purpose: domain names, aliases, summaries, symptoms,
and package basenames all score. Exit 0 with the best hit; exit 1 with the
closest matches when nothing clears the bar (a miss is loud, never blank).
"""
from __future__ import annotations

import argparse
import difflib
import json
import re
import sys
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
REGISTRY = REPO / "docs/factory/domains.json"

TOKEN_RE = re.compile(r"[a-z0-9]+")
SCORE_HIT = 6.0          # a query token matching a high-signal field
SCORE_PACKAGE = 3.0      # package basename hit
SCORE_RATIO = 4.0        # fuzzy phrase ratio on name/symptom strings


def _tokens(text: str) -> set[str]:
    return set(TOKEN_RE.findall(text.lower()))


def _domain_text(row: dict) -> str:
    parts = [row["domain"], " ".join(row.get("aliases", [])),
             row.get("summary", "")]
    parts.extend(row.get("symptoms", []))
    parts.extend(str(kf.get("path", "")) for kf in row.get("key_files", []))
    return " ".join(parts)


def _package_names(row: dict) -> list[str]:
    names: list[str] = []
    for pkg_list in row.get("packages", {}).values():
        for p in pkg_list:
            names.extend(TOKEN_RE.findall(Path(p).name.lower()))
            names.extend(TOKEN_RE.findall(Path(p).parent.name.lower()))
    return names


def score_row(row: dict, query: str) -> float:
    q = " ".join(query.lower().split())
    q_tokens = _tokens(q)
    if not q_tokens:
        return 0.0

    name = row["domain"].lower()
    aliases = [a.lower() for a in row.get("aliases", [])]

    score = 0.0
    # exact / substring dominance
    if q == name or q in aliases:
        return 1000.0
    if q in name or any(q in a for a in aliases):
        score += 100.0
    if name in q:
        score += 60.0

    high_signal = _tokens(_domain_text(row))
    score += SCORE_HIT * len(q_tokens & high_signal)

    pkg_tokens = set(_package_names(row))
    score += SCORE_PACKAGE * len(q_tokens & pkg_tokens)

    # fuzzy phrase ratio keeps typos and phrasings honest
    candidates = [name] + aliases + [s.lower() for s in row.get("symptoms", [])]
    best = max(difflib.SequenceMatcher(None, q, c).ratio() for c in candidates)
    score += SCORE_RATIO * best

    # strong symptom phrases should dominate over incidental token overlap
    for s in row.get("symptoms", []):
        s_low = s.lower()
        if q in s_low or s_low in q:
            score += 50.0
            break

    return score


def _format_row(row: dict) -> str:
    lines = [f"domain: {row['domain']}"]
    if row.get("aliases"):
        lines.append(f"  aliases: {', '.join(row['aliases'])}")
    lines.append(f"  summary: {row['summary']}")
    pkgs = row.get("packages", {})
    for world in ("tui", "src_go", "src_rust", "src_rune"):
        if pkgs.get(world):
            lines.append(f"  {world}: {' '.join(pkgs[world])}")
    if row.get("serve_endpoints"):
        lines.append(f"  serve-surface: {' '.join(row['serve_endpoints'])}")
    lines.append("  fix starts at:")
    for kf in row["key_files"][:4]:
        lines.append(f"    - {kf['path']}  ({kf['why']})")
    lines.append("  test batteries:")
    for tb in row["test_batteries"][:6]:
        lines.append(f"    - {tb}")
    lines.append("  failure symptoms:")
    for s in row["symptoms"][:5]:
        lines.append(f"    - {s}")
    if row.get("scaffold_contract"):
        lines.append(f"  scaffold contract: {row['scaffold_contract']}")
    if row.get("retirement_row"):
        lines.append(f"  retirement-map row: {row['retirement_row']}")
    return "\n".join(lines)


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="domain registry lookup")
    ap.add_argument("query", nargs="+", help="symptom or domain name")
    ap.add_argument("--json", action="store_true", dest="as_json",
                    help="emit the matched row(s) as JSON")
    ap.add_argument("--all", action="store_true",
                    help="list every domain name and exit 0")
    args = ap.parse_args(argv)

    if not REGISTRY.is_file():
        print(f"RED: {REGISTRY} missing — the domain registry IS the map",
              file=sys.stderr)
        return 1
    doc = json.loads(REGISTRY.read_text())
    rows = doc["domains"]

    if args.all:
        for row in rows:
            print(row["domain"])
        return 0

    query = " ".join(args.query)
    scored = sorted(
        ((score_row(r, query), r) for r in rows),
        key=lambda pair: pair[0], reverse=True)
    top_score, top = scored[0]

    # A hit must actually relate to the query: some token overlap or a
    # strong fuzzy ratio — never a free win off pure noise.
    q_tokens = _tokens(query)
    relates = (top_score >= 9.0
               and (q_tokens & (_tokens(_domain_text(top))
                                | set(_package_names(top)))))
    if relates:
        if args.as_json:
            print(json.dumps(top, indent=2))
        else:
            print(_format_row(top))
        if len(scored) > 1 and scored[1][0] >= 0.5 * top_score:
            other = scored[1][1]["domain"]
            print(f"(also plausible: {other} — re-query it by name)")
        return 0

    print(f"no domain matched {query!r}. closest matches:")
    for score, row in scored[:3]:
        print(f"  {row['domain']}  (score {score:.1f})")
    print("try: python scripts/query_domain.py --all")
    return 1


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