KB Cluster Coherence-Judge (Phase 2.5) Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax.

Goal: Add a claude -p opus post-pass over the HDBSCAN clusters that drops incoherent grab-bags, prunes tangential members, and writes clean labels — so the /semantics explorer is trustworthy.

Architecture: Two new modules in /home/levander/kb-vectors/: judgeprompt.py (pure — build batch prompt + parse claude’s JSON) and judge.py (apply verdicts, recompute span/labels, write clusters.json with a raw backup). A kbclust judge subcommand runs it. Base HDBSCAN clustering is unchanged.

Tech Stack: Python (kb-vectors venv), claude -p --model opus (at /home/levander/.local/bin/claude, logged in as levander), the phase-2 cluster.py helpers.

Design: 2026-07-27-kb-cluster-judge-design

Global Constraints

• Code under /home/levander/kb-vectors/. Runs as levander, NOT root, no sudo. NO comments/docstrings/annotations (hard user rule). No git commit.
• Read-only over Qdrant + the KB docs. Only writes clusters.json and clusters.raw.json.
• claude: /home/levander/.local/bin/claude -p --model opus, prompt on stdin, timeout 300. Prefix runs with export PATH="$HOME/.local/bin:$PATH" or use the full path. —model opus explicit.
• FAIL-SAFE: if a batch’s claude call fails or its reply won’t parse, those clusters pass through UNCHANGED (never drop clusters on error). Snapshot clusters.raw.json before overwriting clusters.json.
• Reuse phase-2 cluster.py: source_of(manual_id), write_clusters(clusters, path), load_clusters(path), CLUSTERS_PATH (= /home/levander/kb-vectors/clusters.json). Cluster fields: id,label,members,size,source_count,source_manuals,sources,span; member = {manual_id,page_url,heading}.
• Serialized + thread-capped (OMP_NUM_THREADS=4 nice -n 10) — box is power-flaky + claude heavy; ~13 opus calls one-time.
• Run commands on the box: /usr/bin/ssh levander@telep-mainframe. On this Mac bare ssh is broken — always /usr/bin/ssh.

Test command: cd /home/levander/kb-vectors && venv/bin/python -m unittest <module> -v

---

Task 1: judgeprompt.py — batch prompt + reply parser

Files:

• Create: /home/levander/kb-vectors/judgeprompt.py
• Test: /home/levander/kb-vectors/test_judgeprompt.py

Interfaces:

• Produces:

  • build_batch_prompt(clusters) -> str — clusters = list of {id, members:[{manual_id,heading}...]}; returns the judge prompt (task + each cluster as CLUSTER <id>: then - [manual_id] heading lines)
  • parse_reply(text) -> list[dict] — extract the JSON array from claude stdout (tolerate code fences / surrounding text); return [{index, coherent, label, keep_headings}]; raise ValueError if no JSON array found

• Step 1: Write the failing test

Create /home/levander/kb-vectors/test_judgeprompt.py:

import unittest
 
from judgeprompt import build_batch_prompt, parse_reply
 
 
class TestBuildPrompt(unittest.TestCase):
    def _clusters(self):
        return [
            {"id": 0, "members": [
                {"manual_id": "vitara/workshop", "heading": "Window regulator"},
                {"manual_id": "chevy/geo", "heading": "Filling and painting"}]},
            {"id": 1, "members": [
                {"manual_id": "vitara/workshop", "heading": "Parking brake"}]},
        ]
 
    def test_lists_clusters_and_members(self):
        p = build_batch_prompt(self._clusters())
        self.assertIn("CLUSTER 0", p)
        self.assertIn("CLUSTER 1", p)
        self.assertIn("Window regulator", p)
        self.assertIn("[chevy/geo]", p)
 
    def test_asks_for_json(self):
        p = build_batch_prompt(self._clusters())
        low = p.lower()
        self.assertIn("json", low)
        self.assertIn("keep_headings", low)
        self.assertIn("coherent", low)
 
 
class TestParseReply(unittest.TestCase):
    def test_plain_json(self):
        r = parse_reply('[{"index":0,"coherent":false,"label":"x","keep_headings":[]}]')
        self.assertEqual(r[0]["coherent"], False)
 
    def test_code_fenced(self):
        r = parse_reply('```json\n[{"index":1,"coherent":true,"label":"Parking brake","keep_headings":["Parking brake"]}]\n```')
        self.assertEqual(r[0]["label"], "Parking brake")
 
    def test_surrounding_text(self):
        r = parse_reply('Here is the audit:\n[{"index":0,"coherent":true,"label":"y","keep_headings":["a"]}]\nDone.')
        self.assertEqual(len(r), 1)
 
    def test_junk_raises(self):
        with self.assertRaises(ValueError):
            parse_reply("no json here at all")
 
 
if __name__ == "__main__":
    unittest.main()

• Step 2: Run the test to verify it fails

/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/kb-vectors && venv/bin/python -m unittest test_judgeprompt -v'

Expected: FAIL with ModuleNotFoundError: No module named 'judgeprompt'

• Step 3: Write the implementation

Create /home/levander/kb-vectors/judgeprompt.py:

import json
import re
 
INSTRUCTIONS = """You are auditing automatically-formed clusters of car-repair manual sections. Each cluster is supposed to group sections covering THE SAME procedure/topic across manuals. Some clusters are incoherent grab-bags a density algorithm merged by vocabulary similarity.
 
For EACH cluster below, return a JSON object:
{"index": <cluster number>, "coherent": true|false, "label": "<clean concise topic label>", "keep_headings": ["<member headings that genuinely belong to the dominant topic>"]}
 
Rules:
- coherent=false if there is no single dominant topic (a grab-bag). Set keep_headings to [] when incoherent.
- When coherent=true, keep_headings lists only the member headings that belong to the dominant topic; drop tangential ones.
- label: a clean human topic name (fix OCR typos; do not invent facts).
- Output ONLY a JSON array of these objects, nothing else.
 
CLUSTERS:
"""
 
_ARRAY = re.compile(r"\[.*\]", re.DOTALL)
 
 
def build_batch_prompt(clusters):
    parts = [INSTRUCTIONS]
    for cluster in clusters:
        lines = ["CLUSTER %d:" % cluster["id"]]
        for member in cluster["members"]:
            lines.append("- [%s] %s" % (member["manual_id"], member["heading"]))
        parts.append("\n".join(lines))
    return "\n\n".join(parts)
 
 
def parse_reply(text):
    match = _ARRAY.search(text)
    if not match:
        raise ValueError("no JSON array in reply")
    return json.loads(match.group(0))

• Step 4: Run the test to verify it passes

/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/kb-vectors && venv/bin/python -m unittest test_judgeprompt -v'

Expected: OK — 6 tests pass.

---

Task 2: judge.py + kbclust judge — apply verdicts + real run

Files:

• Create: /home/levander/kb-vectors/judge.py
• Create: /home/levander/kb-vectors/test_judge.py
• Modify: /home/levander/kb-vectors/kbclust.py

Interfaces:

• Consumes: judgeprompt.build_batch_prompt, judgeprompt.parse_reply, cluster.source_of, cluster.write_clusters, cluster.load_clusters, cluster.CLUSTERS_PATH

• Produces:

  • apply_verdict(cluster, verdict) -> dict | None — refined cluster (recomputed source_manuals/sources/span/source_count/size + verdict label) or None (drop)
  • judge_all(clusters, run_claude, batch_size=15) -> tuple[list, dict] — run_claude is a prompt->text callable (injected for testing); returns (surviving_reassigned_sorted, stats)
  • main_judge(batch_size=15) — the real run (snapshot raw, judge, write)

• Step 1: Write the failing test

Create /home/levander/kb-vectors/test_judge.py:

import json
import unittest
 
from judge import apply_verdict, judge_all
 
 
def cl(cid, members):
    return {"id": cid, "label": "raw", "members": members,
            "size": len(members), "span": 1, "source_count": 1,
            "source_manuals": sorted(set(m["manual_id"] for m in members)),
            "sources": []}
 
 
def mem(manual_id, heading):
    return {"manual_id": manual_id, "page_url": manual_id + "/" + heading + "/", "heading": heading}
 
 
class TestApplyVerdict(unittest.TestCase):
    def test_incoherent_dropped(self):
        c = cl(0, [mem("a/x", "P"), mem("b/y", "Q")])
        self.assertIsNone(apply_verdict(c, {"coherent": False, "label": "grab", "keep_headings": []}))
 
    def test_refine_recomputes(self):
        c = cl(1, [mem("suzuki-vitara/workshop", "Parking brake"),
                   mem("chevy-tracker/geo", "Parking brake"),
                   mem("suzuki-vitara/workshop", "Master cylinder")])
        out = apply_verdict(c, {"coherent": True, "label": "Parking brake", "keep_headings": ["Parking brake"]})
        self.assertEqual(out["label"], "Parking brake")
        self.assertEqual(out["size"], 2)
        self.assertEqual(sorted(out["sources"]), ["chevy-tracker", "suzuki-vitara"])
        self.assertEqual(out["span"], 2)
 
    def test_kept_under_two_dropped(self):
        c = cl(2, [mem("a/x", "P"), mem("b/y", "Q")])
        self.assertIsNone(apply_verdict(c, {"coherent": True, "label": "l", "keep_headings": ["P"]}))
 
    def test_heading_trim_match(self):
        c = cl(3, [mem("a/x", " Brakes "), mem("b/y", "Brakes")])
        out = apply_verdict(c, {"coherent": True, "label": "Brakes", "keep_headings": ["Brakes"]})
        self.assertEqual(out["size"], 2)
 
 
class TestJudgeAll(unittest.TestCase):
    def test_applies_and_reassigns(self):
        clusters = [
            cl(0, [mem("a/x", "P"), mem("b/y", "Q")]),
            cl(1, [mem("suzuki-vitara/w", "Brake"), mem("chevy/g", "Brake")]),
        ]
        def fake_claude(prompt):
            return json.dumps([
                {"index": 0, "coherent": False, "label": "grab", "keep_headings": []},
                {"index": 1, "coherent": True, "label": "Brake", "keep_headings": ["Brake"]},
            ])
        out, stats = judge_all(clusters, fake_claude, batch_size=15)
        self.assertEqual(len(out), 1)
        self.assertEqual(out[0]["label"], "Brake")
        self.assertEqual(out[0]["id"], 0)
        self.assertEqual(stats["dropped"], 1)
 
    def test_failsafe_on_bad_reply(self):
        clusters = [cl(0, [mem("a/x", "P"), mem("b/y", "Q")])]
        def bad_claude(prompt):
            return "not json"
        out, stats = judge_all(clusters, bad_claude, batch_size=15)
        self.assertEqual(len(out), 1)
        self.assertEqual(stats["failed_batches"], 1)
 
 
if __name__ == "__main__":
    unittest.main()

• Step 2: Run the test to verify it fails

/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/kb-vectors && venv/bin/python -m unittest test_judge -v'

Expected: FAIL with ModuleNotFoundError: No module named 'judge'

• Step 3: Write judge.py

Create /home/levander/kb-vectors/judge.py:

import json
import os
import subprocess
 
from cluster import CLUSTERS_PATH, load_clusters, source_of, write_clusters
from judgeprompt import build_batch_prompt, parse_reply
 
CLAUDE = "/home/levander/.local/bin/claude"
RAW_PATH = "/home/levander/kb-vectors/clusters.raw.json"
 
 
def apply_verdict(cluster, verdict):
    if not verdict.get("coherent"):
        return None
    keep = set(h.strip() for h in verdict.get("keep_headings", []))
    members = [m for m in cluster["members"] if m["heading"].strip() in keep]
    if len(members) < 2:
        return None
    source_manuals = sorted(set(m["manual_id"] for m in members))
    sources = sorted(set(source_of(m["manual_id"]) for m in members))
    refined = dict(cluster)
    refined["members"] = members
    refined["label"] = verdict.get("label") or cluster["label"]
    refined["size"] = len(members)
    refined["source_manuals"] = source_manuals
    refined["sources"] = sources
    refined["span"] = len(sources)
    refined["source_count"] = len(sources)
    return refined
 
 
def _run_claude(prompt):
    result = subprocess.run(
        [CLAUDE, "-p", "--model", "opus"],
        input=prompt, capture_output=True, text=True, timeout=300,
    )
    if result.returncode != 0 or not result.stdout.strip():
        raise RuntimeError("claude judge failed: " + (result.stderr or "empty")[-300:])
    return result.stdout
 
 
def judge_all(clusters, run_claude, batch_size=15):
    stats = {"raw": len(clusters), "dropped": 0, "refined": 0, "failed_batches": 0}
    survivors = []
    for start in range(0, len(clusters), batch_size):
        batch = clusters[start:start + batch_size]
        try:
            verdicts = parse_reply(run_claude(build_batch_prompt(batch)))
            by_index = {v["index"]: v for v in verdicts}
        except Exception:
            stats["failed_batches"] += 1
            survivors.extend(batch)
            continue
        for cluster in batch:
            verdict = by_index.get(cluster["id"])
            if verdict is None:
                survivors.append(cluster)
                continue
            refined = apply_verdict(cluster, verdict)
            if refined is None:
                stats["dropped"] += 1
            else:
                if refined["size"] != cluster["size"] or refined["label"] != cluster["label"]:
                    stats["refined"] += 1
                survivors.append(refined)
    survivors.sort(key=lambda c: (c["span"], len(c["source_manuals"]), c["size"]), reverse=True)
    for position, cluster in enumerate(survivors):
        cluster["id"] = position
    stats["kept"] = len(survivors)
    return survivors, stats
 
 
def main_judge(batch_size=15):
    clusters = load_clusters()
    with open(RAW_PATH, "w", encoding="utf-8") as handle:
        json.dump({"clusters": clusters}, handle, ensure_ascii=False, indent=1)
    survivors, stats = judge_all(clusters, _run_claude, batch_size=batch_size)
    write_clusters(survivors)
    print("judge:", stats)

• Step 4: Run the test to verify it passes

/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/kb-vectors && venv/bin/python -m unittest test_judge -v'

Expected: OK — 6 tests pass.

• Step 5: Add the judge subcommand to kbclust.py

In /home/levander/kb-vectors/kbclust.py, add a judge subparser and dispatch. Add near the other sub.add_parser(...) lines:

    judge_parser = sub.add_parser("judge")
    judge_parser.add_argument("--batch", type=int, default=15)

And in the command dispatch (after the existing elif args.command == "show": block), add:

    elif args.command == "judge":
        import judge
        judge.main_judge(batch_size=args.batch)

Apply it:

/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/kb-vectors && python3 - <<PY
p="kbclust.py"
s=open(p,encoding="utf-8").read()
if "add_parser(\"judge\")" not in s:
    s=s.replace("    show_parser = sub.add_parser(\"show\")",
                "    judge_parser = sub.add_parser(\"judge\")\n    judge_parser.add_argument(\"--batch\", type=int, default=15)\n    show_parser = sub.add_parser(\"show\")")
    anchor="if __name__ =="
    disp="    elif args.command == \"judge\":\n        import judge\n        judge.main_judge(batch_size=args.batch)\n\n\n"
    s=s.replace("\n\nif __name__ ==", "\n\n"+disp+"if __name__ ==",1)
    open(p,"w",encoding="utf-8").write(s)
    print("kbclust judge added")
else:
    print("already present")
PY
venv/bin/python kbclust.py judge --help 2>&1 | head -3'

Expected: the judge subcommand help prints (argparse recognizes it).

• Step 6: REAL judge run on the live clusters (the proof)

Snapshot is automatic. Thread-capped, serialized (~13 opus calls — a few minutes).

/usr/bin/ssh levander@telep-mainframe 'export PATH="$HOME/.local/bin:$PATH"; cd /home/levander/kb-vectors && OMP_NUM_THREADS=4 nice -n 10 timeout 1200 venv/bin/python kbclust.py judge --batch 15'

Expected: judge: {raw: 194, dropped: N, refined: M, failed_batches: 0, kept: K} — a meaningful number dropped/refined, few or no failed batches.

• Step 7: Verify the grab-bag is gone + good clusters kept with clean labels

/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/kb-vectors && venv/bin/python -c "
import json
d=json.load(open(\"clusters.json\"))[\"clusters\"]
raw=json.load(open(\"clusters.raw.json\"))[\"clusters\"]
print(\"raw:\", len(raw), \"-> kept:\", len(d))
def has_windowreg_grabbag(cs):
    for c in cs:
        hs=[m[\"heading\"].lower() for m in c[\"members\"]]
        if any(\"window regulator\" in h for h in hs) and any(\"painting\" in h for h in hs):
            return c
    return None
print(\"grab-bag present now?:\", \"YES\" if has_windowreg_grabbag(d) else \"NO (fixed)\")
print(\"=== top 12 kept labels (clean?) ===\")
for c in d[:12]: print(\"  span=%d size=%d  %s\" % (c[\"span\"], c[\"size\"], c[\"label\"]))
"'
echo "=== explorer still 200 + shows clean labels ==="
curl -s -o /dev/null -w "semantics=%{http_code}\n" --max-time 15 "https://knowledgebase.taild4189d.ts.net/semantics"
curl -s --max-time 15 "https://knowledgebase.taild4189d.ts.net/semantics" | grep -ocE "forrás"