plexus pick: choose a tool for a request, or abstain

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+- New `plexus pick "REQUEST"` picks a tool for a plain-language request from what each manifest emits and consumes, with a probability for every tool, and abstains when no tool matches or the top probability is under `--threshold`. On a 51-request held-out set it picked the right tool 57% of the time, abstained on 10%, and abstained requests carried 2.7 times the error of picked ones. Details and the bar set before the run: `docs/PICK.md`.
 - The Claude plugin folder now carries the server code at `client-plugin/server/src/plexus/`, so an install of the folder alone starts. `python scripts/build_client_package.py --sync-vendored` rewrites that copy from `src/`, and a test fails when it drifts. The launcher no longer looks outside the plugin folder; a missing copy prints a one-line message and exits with status 1. A test keeps every code and text file under 256 KiB, images under 2 MB, and the folder at 512 files or fewer.
 - The skill and the MCP tool descriptions say "tool" where they said "organ".
 - The Claude plugin manifest adds the plugin directory listing fields: display name, keywords, homepage, repository, documentation, support, privacy and terms links, and a 1024 px icon. Plexus takes no launch settings, so the manifest declares no `userConfig`. Portable and Codex manifests are unchanged.

README.md

@@ -195,6 +195,8 @@ body cannot make it agree with itself. And the receipt carries a method version
 that has to match before anything else is compared, so a plan written by an older
 plexus is reported as failing rather than silently re-interpreted under new rules.
 
+Not sure which tool a request belongs to? `plexus pick "REQUEST"` returns a probability for each tool and abstains when none fits. Measured results are in [docs/PICK.md](docs/PICK.md).
+
 ## How a tool plugs in
 
 A manifest is plain JSON. A tool ships one and it joins the mesh. Drop

client-plugin/server/src/plexus/cli.py

@@ -4,6 +4,7 @@
   plexus wiring   [...]                      # capability -> (producer, consumer) pairs
   plexus plan --goal ORGAN [...]             # upstream pipeline that feeds ORGAN
   plexus route --from A --to B [...]         # capability path from A to B
+  plexus pick "REQUEST" [--threshold P]      # pick a tool for a request, or abstain
   plexus validate [...]                      # manifest problems (exit 1 if any)
 """
 from __future__ import annotations
@@ -71,6 +72,11 @@ def main(argv: "list[str] | None" = None) -> int:
     rp = sub.add_parser("route"); _add_source_flags(rp)
     rp.add_argument("--from", dest="src", required=True)
     rp.add_argument("--to", dest="dst", required=True)
+    kp = sub.add_parser("pick", help="pick a tool for a request, or abstain")
+    _add_source_flags(kp)
+    kp.add_argument("request", help="the request in plain language")
+    kp.add_argument("--threshold", type=float, default=0.23,
+                    help="abstain when the top tool's probability is under this")
     vp = sub.add_parser("verify"); _add_source_flags(vp)
     vp.add_argument("--plan", required=True,
                     help="a plan/route JSON file (from `plexus plan`/`route`) to re-derive "
@@ -106,6 +112,11 @@ def main(argv: "list[str] | None" = None) -> int:
     if args.cmd == "route":
         print(json.dumps(route(discover(mans), args.src, args.dst), indent=2))
         return 0
+    if args.cmd == "pick":
+        from .pick import pick
+        result = pick(discover(mans), args.request, threshold=args.threshold)
+        print(json.dumps(result, indent=2))
+        return 0
     if args.cmd == "verify":
         from .receipt import verify_plan
         with open(args.plan, encoding="utf-8") as f:

client-plugin/server/src/plexus/pick.py

@@ -0,0 +1,103 @@
+"""pick.py: choose a tool for a plain-language request, or abstain.
+
+Each tool in the mesh becomes a short document built from its manifest: its
+name, the title and summary of every capability it emits or consumes, and the
+words of each capability id. A request is scored against every tool with
+BM25, scores become probabilities with a softmax, and the top tool is picked
+only when its probability reaches ``threshold``. Otherwise the result is
+ABSTAIN: ``picked`` is None and ``abstained`` is True, so a person or a larger
+model decides. Deterministic: ties break by tool name.
+"""
+from __future__ import annotations
+
+import math
+import re
+
+from .mesh import Mesh
+
+SCHEMA = "plexus.pick/1"
+_TOKEN = re.compile(r"[a-z0-9]+")
+_STOP = frozenset(
+    "a an and are as at be by can do for from how i in into is it its me my of on "
+    "or our please so that the this to up us we what when which why will with you "
+    "your v1 v2 1 2".split()
+)
+
+
+def _stem(word: str) -> str:
+    for suffix in ("ing", "ies", "ed", "s"):
+        if word.endswith(suffix) and len(word) - len(suffix) >= 3:
+            return word[: -len(suffix)] + ("y" if suffix == "ies" else "")
+    return word
+
+
+def terms(text: str) -> list[str]:
+    """Lowercase word tokens, stop words dropped, light suffix stemming."""
+    return [_stem(t) for t in _TOKEN.findall(text.lower()) if t not in _STOP]
+
+
+def tool_document(manifest) -> list[str]:
+    parts = [manifest.organ.replace("-", " ")]
+    for port in list(manifest.emits) + list(manifest.consumes):
+        parts.extend([port.title, port.summary, port.capability])
+    return terms(" ".join(parts))
+
+
+def _bm25(docs: list[list[str]], query: list[str], k1: float = 1.2,
+          b: float = 0.75) -> list[float]:
+    n = len(docs)
+    avgdl = sum(len(d) for d in docs) / n if n else 0.0
+    df: dict[str, int] = {}
+    for d in docs:
+        for t in set(d):
+            df[t] = df.get(t, 0) + 1
+    out = []
+    for doc in docs:
+        total = 0.0
+        for t in set(query):
+            f = doc.count(t)
+            if f:
+                idf = math.log(1 + (n - df[t] + 0.5) / (df[t] + 0.5))
+                total += idf * f * (k1 + 1) / (f + k1 * (1 - b + b * len(doc) / (avgdl or 1)))
+        out.append(total)
+    return out
+
+
+def _softmax(scores: list[float], temperature: float) -> list[float]:
+    if not scores:
+        return []
+    top = max(scores)
+    exps = [math.exp((s - top) / temperature) for s in scores]
+    z = sum(exps)
+    return [e / z for e in exps]
+
+
+def rank_tools(mesh: Mesh, request: str, temperature: float = 1.0) -> list[dict]:
+    """Every tool with its raw score and probability, best first."""
+    organs = sorted(mesh.manifests)
+    raw = _bm25([tool_document(mesh.manifests[o]) for o in organs], terms(request))
+    probs = _softmax(raw, temperature)
+    rows = [{"tool": o, "score": round(r, 6), "probability": round(p, 6)}
+            for o, r, p in zip(organs, raw, probs)]
+    rows.sort(key=lambda row: (-row["probability"], row["tool"]))
+    return rows
+
+
+DEFAULT_THRESHOLD = 0.23   # the dev-half calibration on the labelled set, docs/PICK.md
+
+
+def pick(mesh: Mesh, request: str, *, threshold: float = DEFAULT_THRESHOLD,
+         temperature: float = 1.0) -> dict:
+    """Pick a tool for ``request`` or abstain. Returns a JSON-ready dict.
+
+    The default threshold is the value the dev-half calibration chose. A
+    threshold of 0.0 abstains only when no tool shares a word with the request.
+    """
+    if not 0.0 <= threshold <= 1.0 or temperature <= 0:
+        raise ValueError("threshold must be in [0, 1] and temperature > 0")
+    rows = rank_tools(mesh, request, temperature)
+    top = rows[0] if rows else None
+    abstained = top is None or top["score"] <= 0 or top["probability"] < threshold
+    return {"schema": SCHEMA, "request": request, "threshold": threshold,
+            "picked": None if abstained else top["tool"], "abstained": abstained,
+            "confidence": top["probability"] if top else 0.0, "candidates": rows}

client-plugin/server/src/plexus/pick_bench.py

@@ -0,0 +1,110 @@
+"""pick_bench.py: measure `plexus pick` on a labelled request set.
+
+Items split in half by a seeded hash of their id. The abstain threshold is
+calibrated on the dev half; numbers come from the test half. The baseline is
+the dev half's majority tool, always picked. A shuffled-label control checks
+that "abstained items carry more error" is not an artifact of the arithmetic.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+import random
+from collections import Counter
+from pathlib import Path
+
+from .mesh import discover
+from .pick import pick
+from .registry import builtin_manifests
+
+SEED = 20261003
+
+
+def split(items: list[dict], seed: int = SEED) -> tuple[list[dict], list[dict]]:
+    dev, test = [], []
+    for item in items:
+        digest = hashlib.sha256(f"{seed}:{item['id']}".encode()).digest()
+        (dev if digest[0] % 2 == 0 else test).append(item)
+    return dev, test
+
+
+def wilson(k: int, n: int, z: float = 1.96) -> list[float]:
+    if n == 0:
+        return [0.0, 0.0]
+    p, d = k / n, 1 + z * z / n
+    c = p + z * z / (2 * n)
+    h = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n))
+    return [round((c - h) / d, 3), round((c + h) / d, 3)]
+
+
+def _rate(k: int, n: int) -> float:
+    return round(k / n, 4) if n else 0.0
+
+
+def abstain_stats(rows: list[tuple]) -> dict:
+    """rows: (top-1 tool, picked tool or None, gold). Error is top-1 vs gold."""
+    dec = [r for r in rows if r[1] is not None]
+    abst = [r for r in rows if r[1] is None]
+    dec_rate = _rate(sum(1 for t, _, g in dec if t != g), len(dec))
+    abs_rate = _rate(sum(1 for t, _, g in abst if t != g), len(abst))
+    return {"decided": len(dec), "abstained": len(abst),
+            "abstain_rate": _rate(len(abst), len(rows)),
+            "decided_error_rate": dec_rate, "abstained_error_rate": abs_rate,
+            "error_ratio": round(abs_rate / dec_rate, 3) if dec_rate else None}
+
+
+def calibrate(mesh, items: list[dict], share: float = 0.10) -> float:
+    """Lowest threshold whose abstain rate on ``items`` is at least ``share``."""
+    tops = []
+    for item in items:
+        top = pick(mesh, item["text"])["candidates"][0]
+        tops.append(top["probability"] if top["score"] > 0 else -1.0)
+    need = math.ceil(share * len(tops))
+    forced = sum(1 for t in tops if t < 0)
+    if forced >= need:
+        return 0.0
+    ranked = sorted(t for t in tops if t >= 0)
+    return min(1.0, round(ranked[need - forced - 1] + 1e-6, 6))
+
+
+def run(items: list[dict], mesh=None, seed: int = SEED) -> dict:
+    mesh = mesh or discover(builtin_manifests())
+    dev, test = split(items, seed)
+    threshold = calibrate(mesh, dev)
+    rows = []
+    for item in test:
+        r = pick(mesh, item["text"], threshold=threshold)
+        rows.append((r["candidates"][0]["tool"], r["picked"], item["gold"]))
+    correct = sum(1 for t, _, g in rows if t == g)
+    majority = Counter(i["gold"] for i in dev).most_common(1)[0][0]
+    base = sum(1 for i in test if i["gold"] == majority)
+    stats = abstain_stats(rows)
+    golds = [g for _, _, g in rows]
+    random.Random(seed).shuffle(golds)
+    shuffled = abstain_stats([(t, p, g) for (t, p, _), g in zip(rows, golds)])
+    acc, base_acc = _rate(correct, len(rows)), _rate(base, len(test))
+    bar = {"P1_accuracy_baseline_plus_0_20": acc >= base_acc + 0.20,
+           "P2_abstain_under_15pct": stats["abstain_rate"] < 0.15,
+           "P3_error_ratio_at_least_2": (stats["error_ratio"] or 0.0) >= 2.0,
+           "control_fails_P3": (shuffled["error_ratio"] or 0.0) < 2.0}
+    return {"schema": "plexus.pick-bench/1", "seed": seed, "dev": len(dev),
+            "test": len(test), "threshold": threshold,
+            "top1_accuracy": acc, "top1_wilson": wilson(correct, len(rows)),
+            "baseline": {"tool": majority, "accuracy": base_acc}, **stats,
+            "shuffled_label_control": shuffled, "bar": bar}
+
+
+def main(argv: list[str] | None = None) -> int:
+    import sys
+    args = argv if argv is not None else sys.argv[1:]
+    if not args:
+        print("usage: python -m plexus.pick_bench LABELS.json", file=sys.stderr)
+        return 2
+    items = json.loads(Path(args[0]).read_text(encoding="utf-8"))["items"]
+    print(json.dumps(run(items), indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

docs/PICK.md

@@ -0,0 +1,52 @@
+# Pick a tool for a request, or abstain
+
+`plexus pick` reads a plain-language request and returns a probability for each tool in the mesh, scored against what each tool's manifest says it emits and consumes. When no tool matches, or the top probability is under a threshold, it abstains, so the request goes to a person or a larger model in place of a guessed tool. `plexus route` still answers the wiring question, how one tool's output reaches another.
+
+## Bar, set before the run
+
+This bar was committed before the picker was written and before its author saw the labelled set.
+
+**Labels.** 100 requests written by a separate labeller who saw only tool names, their capability titles and a one-line purpose, never code, each with one gold tool (`tests/fixtures/plexus_pick_labels.json`). The local routing-accuracy records hold 2 records, too few to use.
+
+**Split.** Items are split in half by a seeded hash of their id. The threshold is chosen on the dev half only: the lowest threshold whose dev abstain rate is at least 10%. All bar numbers come from the test half.
+
+**Baseline.** Plexus had no request picker. The baseline is the majority tool on the dev half, always picked.
+
+- **P1, accuracy.** Top-1 accuracy is at least the baseline's plus 0.20.
+- **P2, abstain rate.** Under 15% of test items abstain.
+- **P3, abstain carries error.** The top-1 error rate on abstained items is at least twice the error rate on decided items.
+- **Control.** With the gold labels shuffled, P3 must fail.
+- **Ship rule.** The picker ships as a new command whatever the result, with the numbers below.
+
+## Try it
+
+```bash
+plexus pick "verify a sealed wiki pinned to a commit"
+plexus pick --threshold 0.5 "check this memory still matches its source"
+python -m plexus.pick_bench tests/fixtures/plexus_pick_labels.json
+```
+
+Each tool is scored with BM25 against a document built from its manifest: its name, the title and summary of each capability, and the words of each capability id. A softmax turns the scores into probabilities. The default threshold, 0.23, is the calibrated value from the run below. A threshold of 0.0 abstains only when no tool shares a word with the request.
+
+## Results
+
+Run on 2026-10-03, seed 20261003. Dev half 49 items, test half 51. The calibration rule picked a threshold of 0.23.
+
+| Test half, 51 items | Value |
+|---|---|
+| Top-1 accuracy | 0.57 (29 of 51), Wilson 0.43 to 0.70 |
+| Baseline, dev majority tool (learn) | 0.02 |
+| Abstain rate | 0.10 (5 of 51) |
+| Error rate when picked | 0.37 |
+| Error rate when abstained | 1.00 |
+| Error ratio | 2.7 |
+| Shuffled-label control ratio | 1.15 |
+
+- **P1 passes.** 0.57 against 0.02. The baseline is weak because the dev majority tool appeared once in the test half; uniform guessing over 10 tools would score about 0.10, and the picker clears that too.
+- **P2 passes.** 10% of requests abstain.
+- **P3 passes.** Abstained requests err at 1.00 against 0.37 for picked ones.
+- **The control fails P3 as required.** Shuffled labels give 1.15.
+
+## Limits
+
+The labels come from one labeller, a Claude subagent that saw tool names, capability titles and a one-line purpose written for the task. 51 test items give wide intervals, and 4 in 10 picks are still wrong. Treat a pick as a suggestion to confirm. A tool whose manifest has sparse titles will rarely be picked.

src/plexus/cli.py

@@ -4,6 +4,7 @@
   plexus wiring   [...]                      # capability -> (producer, consumer) pairs
   plexus plan --goal ORGAN [...]             # upstream pipeline that feeds ORGAN
   plexus route --from A --to B [...]         # capability path from A to B
+  plexus pick "REQUEST" [--threshold P]      # pick a tool for a request, or abstain
   plexus validate [...]                      # manifest problems (exit 1 if any)
 """
 from __future__ import annotations
@@ -71,6 +72,11 @@ def main(argv: "list[str] | None" = None) -> int:
     rp = sub.add_parser("route"); _add_source_flags(rp)
     rp.add_argument("--from", dest="src", required=True)
     rp.add_argument("--to", dest="dst", required=True)
+    kp = sub.add_parser("pick", help="pick a tool for a request, or abstain")
+    _add_source_flags(kp)
+    kp.add_argument("request", help="the request in plain language")
+    kp.add_argument("--threshold", type=float, default=0.23,
+                    help="abstain when the top tool's probability is under this")
     vp = sub.add_parser("verify"); _add_source_flags(vp)
     vp.add_argument("--plan", required=True,
                     help="a plan/route JSON file (from `plexus plan`/`route`) to re-derive "
@@ -106,6 +112,11 @@ def main(argv: "list[str] | None" = None) -> int:
     if args.cmd == "route":
         print(json.dumps(route(discover(mans), args.src, args.dst), indent=2))
         return 0
+    if args.cmd == "pick":
+        from .pick import pick
+        result = pick(discover(mans), args.request, threshold=args.threshold)
+        print(json.dumps(result, indent=2))
+        return 0
     if args.cmd == "verify":
         from .receipt import verify_plan
         with open(args.plan, encoding="utf-8") as f: