feat: best-of-N + LLM-as-judge detector + untell-ceiling (free-ceiling moves)

README.md

@@ -82,8 +82,10 @@ score the text and lock your facts. Zero dependencies (lite tier).
 git clone https://github.com/ssamba1/untell && cd untell
 pip install -e ".[full]"                          # real detector ensemble on CPU
 untell-loop "Your AI-sounding paragraph here."    # rewrite until it passes
+untell-loop "text" --best-of 3                     # draw 3 rewrites/round, keep the best valid one
 untell-score "text" --tier full --threshold 0.3   # just score it
 untell-verify --file draft.txt                    # honest pass/fail per detector
+untell-ceiling                                     # measure the loop's evasion vs the local detectors
 ```
 
 <details>

eval/ceiling.py

@@ -0,0 +1,175 @@
+"""Measure untell's inference-only evasion ceiling against the LOCAL detector ensemble.
+
+The literature has no data point for what the training-free closed loop actually achieves: only the
+~1% one-shot-style floor and the ~97% RL-trained ceiling (see docs/free-ceiling-report.md). This
+script produces that missing number. It scores a corpus of AI text, runs the untell loop on each,
+and reports the before/after flagged rate plus per-detector mean P(AI).
+
+Without a rewriter configured it reports the BASELINE (pre-rewrite detection) only, which is always
+runnable and still useful. With a rewriter (an API key, or one passed to ``measure_ceiling``) it
+reports the full before/after delta — the actual inference-only ceiling on the local tier.
+
+    untell-ceiling                       # built-in sample, baseline (or full delta if a key is set)
+    untell-ceiling --file corpus.txt     # paragraphs separated by blank lines
+    untell-ceiling --tier full --best-of 3 --json
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+
+# Run-as-file support: put the package parent on sys.path when executed directly.
+if __package__ in (None, ""):
+    import sys as _sys
+    from pathlib import Path as _Path
+
+    for _p in _Path(__file__).resolve().parents:
+        if (_p / "untell" / "__init__.py").exists():
+            _sys.path.insert(0, str(_p))
+            break
+
+from untell.scripts.run import untell_text
+from untell.scripts.score import DEFAULT_THRESHOLD, score_text
+
+# A few formulaic AI paragraphs (no locked facts needed; this measures detector movement).
+_SAMPLE = [
+    "Furthermore, artificial intelligence has fundamentally transformed numerous industries in recent "
+    "years. Moreover, organizations increasingly leverage these technologies to optimize operational "
+    "efficiency and drive innovation. Overall, the transformative impact continues to expand across "
+    "various sectors.",
+    "In today's rapidly evolving digital landscape, cybersecurity has become paramount. It is important "
+    "to note that organizations must navigate the complexities of an ever-changing threat environment. "
+    "Ultimately, a robust and comprehensive security posture is essential for success.",
+    "Climate change represents one of the most pressing challenges of our time. Notably, rising global "
+    "temperatures underscore the urgent need for action. By fostering collaboration and harnessing "
+    "innovative solutions, society can pave the way toward a more sustainable future.",
+]
+
+
+def _numeric(score: dict) -> dict:
+    return {
+        k: v
+        for k, v in score.get("detectors", {}).items()
+        if isinstance(v, (int, float)) and not k.endswith("__error")
+    }
+
+
+def _mean(xs: list[float]) -> float | None:
+    return round(sum(xs) / len(xs), 4) if xs else None
+
+
+def measure_ceiling(
+    texts: list[str],
+    tier: str = "full",
+    threshold: float = DEFAULT_THRESHOLD,
+    max_iters: int = 5,
+    rewriter=None,
+    best_of: int = 1,
+) -> dict:
+    """Score each text, run the loop, and aggregate the before/after detector movement."""
+    pre_max: list[float] = []
+    post_max: list[float] = []
+    per_pre: dict[str, list[float]] = {}
+    per_post: dict[str, list[float]] = {}
+    rewrote = 0
+
+    for t in texts:
+        pre = score_text(t, tier=tier, threshold=threshold)
+        pre_max.append(pre["max"])
+        for k, v in _numeric(pre).items():
+            per_pre.setdefault(k, []).append(v)
+
+        res = untell_text(
+            t, tier=tier, threshold=threshold, max_iters=max_iters, rewriter=rewriter, best_of=best_of
+        )
+        if "error" not in res and "post" in res:
+            post = res["post"]
+            post_max.append(post["max"])
+            for k, v in _numeric(post).items():
+                per_post.setdefault(k, []).append(v)
+            rewrote += 1
+
+    def flagged_rate(scores: list[float]) -> float | None:
+        return round(sum(1 for s in scores if s >= threshold) / len(scores), 4) if scores else None
+
+    return {
+        "n": len(texts),
+        "tier": tier,
+        "threshold": threshold,
+        "max_iters": max_iters,
+        "best_of": best_of,
+        "rewrote": rewrote,
+        "rewriter_available": rewrote > 0,
+        "pre_flagged_rate": flagged_rate(pre_max),
+        "post_flagged_rate": flagged_rate(post_max),
+        "pre_mean_max": _mean(pre_max),
+        "post_mean_max": _mean(post_max),
+        "per_detector_pre": {k: _mean(v) for k, v in per_pre.items()},
+        "per_detector_post": {k: _mean(v) for k, v in per_post.items()} or None,
+    }
+
+
+def _render(r: dict) -> str:
+    lines = [
+        f"untell inference-only ceiling — tier={r['tier']} threshold={r['threshold']} "
+        f"best_of={r['best_of']} n={r['n']}",
+        "",
+        f"  pre  flagged rate: {r['pre_flagged_rate']}   mean max P(AI): {r['pre_mean_max']}",
+    ]
+    if r["rewriter_available"]:
+        lines.append(
+            f"  post flagged rate: {r['post_flagged_rate']}   mean max P(AI): {r['post_mean_max']}   "
+            f"(rewrote {r['rewrote']}/{r['n']})"
+        )
+        lines.append("")
+        lines.append("  per-detector mean P(AI)  before -> after:")
+        for k, before in sorted(r["per_detector_pre"].items()):
+            after = (r["per_detector_post"] or {}).get(k)
+            lines.append(f"    {k:24} {before} -> {after}")
+    else:
+        lines.append("")
+        lines.append(
+            "  No rewriter configured (no ANTHROPIC_API_KEY / OPENAI_API_KEY, and not in the skill) "
+            "— showing BASELINE detection only. Set a key, or run inside the /untell skill where "
+            "Claude is the rewriter, to measure the after-rewrite ceiling."
+        )
+    return "\n".join(lines)
+
+
+def _read_corpus(path: str) -> list[str]:
+    with open(path, encoding="utf-8") as fh:
+        raw = fh.read()
+    blocks = [b.strip() for b in raw.split("\n\n")]
+    return [b for b in blocks if b]
+
+
+def main(argv: list[str] | None = None) -> int:
+    from untell.scripts.io_utils import configure_utf8_io
+
+    configure_utf8_io()
+    parser = argparse.ArgumentParser(prog="untell-ceiling", description=__doc__)
+    parser.add_argument("--file", "-f", help="corpus file (paragraphs separated by blank lines)")
+    parser.add_argument("--tier", default="full", choices=["lite", "full", "heavy", "commercial"])
+    parser.add_argument("--threshold", "-t", type=float, default=DEFAULT_THRESHOLD)
+    parser.add_argument("--max-iters", type=int, default=5)
+    parser.add_argument("--best-of", type=int, default=1)
+    parser.add_argument("--json", action="store_true")
+    args = parser.parse_args(argv)
+
+    from untell._env import load_env
+
+    load_env()
+    texts = _read_corpus(args.file) if args.file else _SAMPLE
+    if not texts:
+        print(json.dumps({"error": "empty corpus"}))
+        return 2
+    result = measure_ceiling(
+        texts, tier=args.tier, threshold=args.threshold, max_iters=args.max_iters, best_of=args.best_of
+    )
+    print(json.dumps(result, ensure_ascii=True, indent=2) if args.json else _render(result))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

pyproject.toml

@@ -92,6 +92,7 @@ untell-verify = "untell.scripts.verify:main"
 untell-prove = "eval.prove:main"
 untell-sentences = "untell.scripts.sentences:main"
 untell-mcp = "untell.mcp_server:main"
+untell-ceiling = "eval.ceiling:main"
 untell-distill = "training.distill:main"
 untell-surrogate = "training.surrogate:main"
 

tests/test_ceiling.py

@@ -0,0 +1,44 @@
+"""Inference-only ceiling measurement tests — offline (baseline + stub-rewriter delta)."""
+
+from __future__ import annotations
+
+import re
+
+from eval.ceiling import _SAMPLE, main, measure_ceiling
+
+
+def test_baseline_without_rewriter():
+    # No rewriter and no API key => baseline (pre) only; post is None but the run still succeeds.
+    r = measure_ceiling(_SAMPLE[:2], tier="lite", max_iters=2, rewriter=None)
+    assert r["n"] == 2
+    assert r["rewriter_available"] is False
+    assert r["pre_flagged_rate"] is not None
+    assert r["post_flagged_rate"] is None
+    assert r["pre_mean_max"] is not None
+
+
+def test_full_delta_with_stub_rewriter():
+    class _RW:
+        name = "stub"
+
+        def available(self):
+            return True
+
+        def rewrite(self, text, score_result, threshold=0.30):
+            sentinels = re.findall(r"⟦HZ\d{4}⟧", text)
+            return "Plain, short, human line. " + " ".join(sentinels)
+
+    r = measure_ceiling(_SAMPLE[:2], tier="lite", threshold=0.30, max_iters=2, rewriter=_RW())
+    assert r["rewrote"] == 2
+    assert r["rewriter_available"] is True
+    assert r["post_flagged_rate"] is not None
+    assert r["pre_mean_max"] is not None and r["post_mean_max"] is not None
+    assert isinstance(r["per_detector_pre"], dict) and r["per_detector_pre"]
+
+
+def test_cli_smoke(capsys):
+    rc = main(["--tier", "lite", "--max-iters", "2"])
+    assert rc == 0
+    out = capsys.readouterr().out
+    assert "ceiling" in out.lower()
+    assert "flagged rate" in out.lower()

tests/test_llm_judge.py

@@ -0,0 +1,43 @@
+"""LLM-as-judge detector tests — offline (no key => unavailable; completion mocked otherwise)."""
+
+from __future__ import annotations
+
+from untell.detectors.llm_judge import LLMJudgeDetector
+
+
+def test_unavailable_without_key(monkeypatch):
+    monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
+    monkeypatch.delenv("OPENAI_API_KEY", raising=False)
+    d = LLMJudgeDetector()
+    assert d.available() is False
+    assert d.score("some text") is None  # unavailable => no signal (excluded from the ensemble)
+
+
+def test_score_parses_number(monkeypatch):
+    d = LLMJudgeDetector()
+    monkeypatch.setattr(d, "available", lambda: True)
+    monkeypatch.setattr(d, "_complete", lambda prompt: "0.82")
+    assert d.score("text") == 0.82
+
+
+def test_score_handles_percentage(monkeypatch):
+    d = LLMJudgeDetector()
+    monkeypatch.setattr(d, "available", lambda: True)
+    monkeypatch.setattr(d, "_complete", lambda prompt: "I'd rate this 73")
+    assert d.score("text") == 0.73  # a percentage answer is normalized to [0,1]
+
+
+def test_empty_and_unparseable_return_none(monkeypatch):
+    d = LLMJudgeDetector()
+    monkeypatch.setattr(d, "available", lambda: True)
+    monkeypatch.setattr(d, "_complete", lambda prompt: "0.5")
+    assert d.score("   ") is None  # empty input => no signal
+    monkeypatch.setattr(d, "_complete", lambda prompt: "no idea")
+    assert d.score("real text") is None  # no number in the reply => None, not a crash
+
+
+def test_registered_in_commercial_tier():
+    from untell.detectors.base import all_detectors
+
+    names = {d.name for d in all_detectors()}
+    assert "llm_judge" in names

tests/test_run.py

@@ -232,3 +232,30 @@ def test_loop_rejects_sentinel_dropping_rewrite(monkeypatch):
     # ...and rejected every time, so the locked facts survive into the final output.
     assert "Smith (2020)" in res["final"]
     assert "47%" in res["final"]
+
+
+def test_best_of_n_draws_multiple_candidates_and_keeps_facts(monkeypatch):
+    import untell.scripts.run as run_mod
+
+    calls = {"n": 0}
+
+    class _MultiRW:
+        name = "multi"
+
+        def available(self):
+            return True
+
+        def rewrite(self, text, score_result, threshold=0.30):
+            import re
+
+            calls["n"] += 1
+            sentinels = re.findall(r"⟦HZ\d{4}⟧", text)
+            tail = (" " + " ".join(sentinels)) if sentinels else ""
+            return f"It shifted, and people noticed. Variant {calls['n']}.{tail}"
+
+    monkeypatch.setattr(run_mod, "get_rewriter", lambda prefer=None: _MultiRW())
+    # threshold=0.0 forces a rewrite; best_of=3 => exactly three candidates drawn in the one iteration.
+    res = untell_text(AI, tier="lite", threshold=0.0, max_iters=1, best_of=3)
+    assert "error" not in res
+    assert calls["n"] == 3
+    assert "Smith (2020)" in res["final"] and "47%" in res["final"]  # facts survive best-of selection

untell/detectors/base.py

@@ -55,6 +55,7 @@ def all_detectors() -> list[Detector]:
     from .commercial import commercial_detectors
     from .fast_detectgpt import FastDetectGPTDetector
     from .hc3_roberta import HC3RobertaDetector
+    from .llm_judge import LLMJudgeDetector
     from .mage import MageDetector
     from .perplexity_burstiness import PerplexityBurstinessDetector
     from .radar import RadarDetector
@@ -66,8 +67,9 @@ def all_detectors() -> list[Detector]:
         HC3RobertaDetector(),
         MageDetector(),
         FastDetectGPTDetector(),
-        RadarDetector(),  # opt-in (HUMANIZE_ENABLE_RADAR=1); robust-to-paraphrase, non-commercial
+        RadarDetector(),  # opt-in (UNTELL_ENABLE_RADAR=1); robust-to-paraphrase, non-commercial
         BinocularsDetector(),
+        LLMJudgeDetector(),  # commercial tier: the frontier LLM as a detector (key-gated); strong free signal
         *commercial_detectors(),
     ]