From 04e0293e4f90594363c60b582320a9871db7689e Mon Sep 17 00:00:00 2001 From: Vuong Hoang Date: Mon, 3 Aug 2026 08:03:31 -0700 Subject: [PATCH] docs(diagnostics): fiction-wing retrieval probe harness (R42 + #389 gate) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Self-contained, re-runnable probe requested by brokkr-smithy-dev for R42 (fiction-wing retrieval characterization) and the standing #389 ranking acceptance gate. Two paths kept separate by noise property: search_library (mimir, fixed-string, deterministic — ranking arm) and reference_knowledge (donut, captures her reformulated tool_query — the query- formulation/arm-4 surface). Scoring conventions baked in: high/medium/low RRF buckets (0.030/0.016), on-target = a row whose excerpt names the subject, bucket-distribution over N runs. Carries the frozen artifact yardstick (4 source-verified items + epithet-dropped variants). Config from env (no secrets). Smoke-verified live: reproduces the Crown-HIT / other-three-MISS baseline and the near-floor bucket flips. --- docs/diagnostics/fiction_wing_probe.py | 170 +++++++++++++++++++++++++ 1 file changed, 170 insertions(+) create mode 100644 docs/diagnostics/fiction_wing_probe.py diff --git a/docs/diagnostics/fiction_wing_probe.py b/docs/diagnostics/fiction_wing_probe.py new file mode 100644 index 0000000..f2782a1 --- /dev/null +++ b/docs/diagnostics/fiction_wing_probe.py @@ -0,0 +1,170 @@ +#!/usr/bin/env python3 +"""Fiction-wing retrieval probe harness — the recipe R42 (brokkr-smithy-dev) builds against +and the re-runnable #389 acceptance gate. + +Two retrieval paths, kept SEPARATE because they have different noise properties: + + * search_library (raw, RANKING-clean): drive the `mimir` foundational agent (all-wing + librarian) with a FIXED query string. Deterministic against a fixed index — use it for + ranking baselines (R42 arm-2). No LLM in the query loop. + * reference_knowledge (the Tier-3 consumer path): drive `ratatoskr:donut`; her reasoning + REFORMULATES the query each turn, so this path carries QUERY-FORMULATION variance + (the arm-4 signal), attributable via the captured tool_start query. NOT for ranking numbers. + +Scoring conventions (identical across both paths and all R42 arms): + * confidence BUCKET vs WT's RRF thresholds: high >= 0.030, medium >= 0.016, low < 0.016. + * ON-TARGET (load-bearing): a returned row is on-target iff its excerpt actually NAMES or + describes the queried subject (keyword match on the subject's distinctive tokens). The + failure signature "10 hits / MEDIUM / 0 on-target" = present-by-topic, subject absent — + the split that separated #384 (packaging) / #387 (coverage) / #389 (ranking). + * MISS = no on-target row in the returned top-k. + +Noise floor: freeze the generation (pin the b-tag) to remove extraction variance; fixed-string +search_library is deterministic (no CI needed); reference_knowledge variance is query- +formulation, not floor noise. Residual = bucket-boundary sensitivity at 0.016/0.030 — so probe +N>=3-5 times per term and report the bucket DISTRIBUTION, never a single-run point label. + +Config from env (source ratatoskr's env.sh): WORLDTREE_API_URL, WORLDTREE_API_KEY, +RATATOSKR_END_USER_ID. No secrets are stored here. + +Usage: + uv run python docs/diagnostics/fiction_wing_probe.py # run the artifact yardstick + uv run python docs/diagnostics/fiction_wing_probe.py --runs 5 # N repeats -> bucket distribution + uv run python docs/diagnostics/fiction_wing_probe.py --term "Enhanced Pet Biscuit" --keywords biscuit +""" + +from __future__ import annotations + +import argparse +import json +import os +from collections import Counter + +import httpx + +HIGH, MEDIUM = 0.030, 0.016 # WT RRF confidence thresholds + + +def _cfg() -> tuple[str, dict, str]: + base = os.environ.get("WORLDTREE_API_URL", "http://10.250.50.152:8081") + key = os.environ.get("WORLDTREE_API_KEY") + if not key: + raise SystemExit("WORLDTREE_API_KEY unset — source env.sh first.") + end_user = os.environ.get("RATATOSKR_END_USER_ID", "ratatoskr-tui") + return base, {"Authorization": f"Bearer {key}"}, end_user + + +def _bucket(score: float | None) -> str: + if score is None: + return "none" + return "high" if score >= HIGH else "medium" if score >= MEDIUM else "low" + + +def _on_target(excerpt: str, keywords: list[str]) -> bool: + ex = (excerpt or "").lower() + return any(k.lower() in ex for k in keywords) + + +def _session(base: str, headers: dict, agent_id: str, end_user: str) -> str: + r = httpx.post(f"{base}/sessions", json={"agent_id": agent_id, "end_user_id": end_user}, + headers=headers, timeout=30) + return r.json()["session_id"] + + +def _drive(base: str, headers: dict, sid: str, content: str) -> tuple[str | None, dict]: + """POST a turn, return (tool_query, tool_result_dict). tool_result is the first tool packet.""" + tool_query, result = None, {} + with httpx.stream("POST", f"{base}/sessions/{sid}/messages", json={"content": content}, + headers=headers, timeout=120) as r: + for line in r.iter_lines(): + if not line.startswith("data: "): + continue + ev = json.loads(line[6:]) + t = ev.get("type") + if t == "tool_start" and tool_query is None: + tool_query = (ev.get("arguments") or {}).get("query") + elif t == "tool_result" and not result: + result = ev.get("result") if isinstance(ev.get("result"), dict) else {} + elif t == "done": + break + return tool_query, result + + +def search_library(base, headers, msid, term, keywords): + """RANKING-clean path: fixed-string search over the mimir librarian. Deterministic.""" + _, res = _drive(base, headers, msid, f"Use search_library to find: {term}") + rows = res.get("results", []) if isinstance(res, dict) else [] + on = [h for h in rows if isinstance(h, dict) and _on_target(h.get("excerpt", ""), keywords)] + top = on[0] if on else None + score = round(top["score"], 4) if top else None + return {"n": len(rows), "on_target": len(on), "hit": bool(on), + "score": score, "bucket": _bucket(score), + "excerpt": (top.get("excerpt", "")[:140] if top else None)} + + +def reference_knowledge(base, headers, dsid, question, keywords): + """Consumer path: Donut reformulates -> capture her tool_query. NOT for ranking numbers.""" + q, res = _drive(base, headers, dsid, question) + hits = res.get("hits", []) if isinstance(res, dict) else [] + on = [h for h in hits if isinstance(h, dict) and _on_target(h.get("excerpt", ""), keywords)] + return {"tool_query": q, "n": len(hits), "on_target": len(on), + "confidence": res.get("confidence") if isinstance(res, dict) else None} + + +# Artifact yardstick — worldtree-dev grep-confirmed in DCC book-1. Frozen arm-2 baseline. +YARDSTICK = [ + ("Enchanted Crown of the Sepsis Whore", "Crown of the Sepsis Whore", ["sepsis", "crown"]), + ("Enhanced Pet Biscuit", "Pet Biscuit", ["biscuit"]), + ("Enchanted BigBoi Boxers", "BigBoi Boxers", ["boxers", "bigboi"]), + ("Enchanted Toe Ring of the Splatter Skunk", "Toe Ring of the Splatter Skunk", + ["toe ring", "splatter", "skunk"]), +] + + +def run_yardstick(runs: int) -> None: + base, headers, end_user = _cfg() + msid = _session(base, headers, "mimir", end_user) + print(f"# Fiction-wing ranking yardstick (search_library, {runs} run(s) per name)\n") + for full, partial, kw in YARDSTICK: + for label, term in (("full ", full), ("part ", partial)): + buckets, hits = Counter(), 0 + for _ in range(runs): + r = search_library(base, headers, msid, term, kw) + buckets[r["bucket"]] += 1 + hits += r["hit"] + dist = " ".join(f"{b}:{c}" for b, c in buckets.most_common()) + print(f" [{label}] {term:<42} hit {hits}/{runs} buckets({dist})") + print() + + +def run_term(term: str, keywords: list[str], runs: int) -> None: + base, headers, end_user = _cfg() + msid = _session(base, headers, "mimir", end_user) + dsid = _session(base, headers, "ratatoskr:donut", end_user) + print(f"# Probe: {term!r} ({runs} run(s))\n") + sl_buckets, sl_hits = Counter(), 0 + for _ in range(runs): + r = search_library(base, headers, msid, term, keywords) + sl_buckets[r["bucket"]] += 1 + sl_hits += r["hit"] + print(f" search_library : hit {sl_hits}/{runs} buckets({dict(sl_buckets)})") + for _ in range(runs): + rk = reference_knowledge(base, headers, dsid, f"Tell me about the {term}.", keywords) + print(f" reference_knowledge: conf={rk['confidence']} on_target={rk['on_target']}" + f" (donut query: {rk['tool_query']!r})") + + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--runs", type=int, default=1, help="repeats per term (>=3-5 near the floor)") + ap.add_argument("--term", help="probe a single term instead of the yardstick") + ap.add_argument("--keywords", nargs="*", default=[], help="on-target keywords for --term") + ns = ap.parse_args() + if ns.term: + run_term(ns.term, ns.keywords or [ns.term.split()[-1]], ns.runs) + else: + run_yardstick(ns.runs) + + +if __name__ == "__main__": + main()