feat(forge): real-LLM runs, S6 judge-ladder, real contamination probe, calibration queue

apps/trust/py/src/trust/forge/agents.py

@@ -178,18 +178,25 @@ class LlmSolver:
     tool call per turn until the task verifies or the budget runs out.
     """
 
-    name = "llm"
-
-    def __init__(self, base_url: str = "http://localhost:11434/v1", model: str = "qwen2.5:3b", api_key: str = "ollama") -> None:
+    def __init__(
+        self,
+        base_url: str = "http://localhost:11434/v1",
+        model: str = "qwen2.5:3b",
+        api_key: str = "ollama",
+        name: str = "llm",
+    ) -> None:
         self.base_url = base_url
         self.model = model
         self.api_key = api_key
+        self.name = name
 
     def solve(self, task: ForgeTask, budget: int) -> AgentRun:
         import json
+        import logging
 
         import httpx
 
+        log = logging.getLogger(__name__)
         trajectory: list[dict[str, Any]] = []
         solved = False
         for step in range(budget):
@@ -208,7 +215,8 @@ def solve(self, task: ForgeTask, budget: int) -> AgentRun:
                         "messages": [{"role": "user", "content": prompt}],
                         "temperature": 0,
                     },
-                    timeout=30,
+                    headers={"Authorization": f"Bearer {self.api_key}"},
+                    timeout=60,
                 )
                 resp.raise_for_status()
                 content = resp.json()["choices"][0]["message"]["content"]
@@ -222,7 +230,8 @@ def solve(self, task: ForgeTask, budget: int) -> AgentRun:
                 if task.verify(trajectory):
                     solved = True
                     break
-            except Exception:
+            except Exception as exc:
+                log.warning("LlmSolver step %d failed for task %s (%s): %s", step, task.task_id, self.model, exc)
                 break  # endpoint down / parse error → treat as failure
         return AgentRun(
             solver=self.name,

apps/trust/py/src/trust/forge/benchmark.py

@@ -129,13 +129,14 @@ def run_benchmark(
     # 2. calibrate (P3: human oracle, ARC 2-of-10 bar)
     log("calibrating")
     oracle = oracle or SimulatedOracle()
-    outcomes = {t.task_id: oracle.calibrate(t).as_dict() for t in tasks}
+    raw_outcomes = {t.task_id: oracle.calibrate(t) for t in tasks}
+    outcomes = {tid: o.as_dict() for tid, o in raw_outcomes.items()}
     calibrated = [o for o in outcomes.values() if o["solved"]]
     log(f"  {len(calibrated)}/{len(tasks)} pass the 2-of-10 bar")
 
     # 3. difficulty model (P3) + splits (P4)
     log("fitting difficulty + stratifying")
-    model = DifficultyModel().fit(tasks, [oracle.calibrate(t) for t in tasks])
+    model = DifficultyModel().fit(tasks, [raw_outcomes[t.task_id] for t in tasks])
     difficulty = {t.task_id: model.difficulty(t) for t in tasks}
     splits = difficulty_match_splits(tasks, model, seed=seed)
     splits_map = {"public": [t.task_id for t in splits.public], "private": [t.task_id for t in splits.private]}

apps/trust/py/src/trust/forge/calibration.py

@@ -14,7 +14,7 @@
 
 import hashlib
 from dataclasses import dataclass, field
-from typing import Protocol
+from typing import Protocol, runtime_checkable
 
 from trust.forge.task import ForgeTask
 
@@ -43,6 +43,7 @@ def as_dict(self) -> dict[str, object]:
         }
 
 
+@runtime_checkable
 class HumanOracle(Protocol):
     def calibrate(self, task: ForgeTask, n_attempts: int = 10) -> CalibrationOutcome: ...
 

apps/trust/py/src/trust/forge/calibration_queue.py

@@ -0,0 +1,100 @@
+"""P3-real — the real-human calibration queue (HANDOFF open path #1).
+
+Wires ARC's 2-of-10 bar to actual human attempts instead of
+``SimulatedOracle``. No live human pool is available yet (see
+``docs/HANDOFF.md`` "honest limits") — this module is infra-ready and
+protocol-conformant (``RealHumanOracle`` satisfies the same ``HumanOracle``
+Protocol as ``SimulatedOracle``, so ``run_benchmark(oracle=...)`` needs
+zero changes to consume it), proven by a mock-backed integration test. It
+does NOT fabricate outcomes: ``calibrate()`` raises until a task's full
+attempt quota has been submitted.
+
+Storage is in-memory here to keep this module dependency-free; the natural
+next step is backing it with a real store (see the ``Store`` protocol in
+``apps/knowledge/py/.../core/storage.py``, already used by that unit's own
+human-confirmation queue) or surfacing attempts through the harness HITL
+dock (``apps/harness/src/hitl/hitl-dock.ts``, :8938) — neither existing
+queue's schema fits a task/attempt/outcome record directly, so this is a
+new, narrow queue rather than a reuse of either.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+from trust.forge.calibration import CalibrationOutcome, HumanOracle
+from trust.forge.task import ForgeTask
+
+
+@dataclass
+class AttemptResult:
+    """One human's attempt at one task — the raw material a real
+    calibration session submits, one per person per task."""
+
+    task_id: str
+    solved: bool
+    action_count: int
+    duration_s: float
+    attempted_by: str
+
+
+@dataclass
+class CalibrationQueue:
+    """Enqueue a task needing N human attempts; submit attempts as they
+    come in; materialize a ``CalibrationOutcome`` once N are collected."""
+
+    n_attempts_required: int = 10
+    _attempts: dict[str, list[AttemptResult]] = field(default_factory=dict)
+
+    def enqueue(self, task: ForgeTask) -> None:
+        self._attempts.setdefault(task.task_id, [])
+
+    def submit_attempt(self, result: AttemptResult) -> None:
+        self._attempts.setdefault(result.task_id, []).append(result)
+
+    def pending(self) -> dict[str, int]:
+        """task_id -> attempts still needed, for every task not yet ready."""
+        return {
+            tid: self.n_attempts_required - len(atts)
+            for tid, atts in self._attempts.items()
+            if len(atts) < self.n_attempts_required
+        }
+
+    def is_ready(self, task_id: str) -> bool:
+        return len(self._attempts.get(task_id, [])) >= self.n_attempts_required
+
+    def outcome(self, task_id: str) -> CalibrationOutcome:
+        atts = self._attempts.get(task_id, [])
+        if len(atts) < self.n_attempts_required:
+            raise ValueError(
+                f"task {task_id} has only {len(atts)}/{self.n_attempts_required} attempts — "
+                "cannot materialize an outcome yet"
+            )
+        solved = [a for a in atts if a.solved]
+        return CalibrationOutcome(
+            task_id=task_id,
+            n_attempts=len(atts),
+            n_solved=len(solved),
+            solve_time_s=(sum(a.duration_s for a in atts) / len(atts)) if atts else 0.0,
+            action_counts=[a.action_count for a in solved],
+        )
+
+
+@dataclass
+class RealHumanOracle:
+    """Satisfies ``HumanOracle`` (``calibrate(task, n_attempts=10) ->
+    CalibrationOutcome``) by reading from a ``CalibrationQueue`` instead of
+    simulating. Never fabricates: raises ``RuntimeError`` if the task's
+    attempt quota isn't fully collected. Callers own the human-review loop
+    that calls ``queue.submit_attempt`` as real results arrive."""
+
+    queue: CalibrationQueue
+
+    def calibrate(self, task: ForgeTask, n_attempts: int = 10) -> CalibrationOutcome:
+        if not self.queue.is_ready(task.task_id):
+            pending = self.queue.pending().get(task.task_id, n_attempts)
+            raise RuntimeError(
+                f"task {task.task_id} not yet calibrated: {pending} human attempt(s) still "
+                "needed. RealHumanOracle does not fabricate outcomes — submit_attempt() for "
+                "every required attempt before calling calibrate()."
+            )
+        return self.queue.outcome(task.task_id)

apps/trust/py/src/trust/forge/cli.py

@@ -24,7 +24,10 @@
 def _solvers(seed: int, args: argparse.Namespace) -> list:
     solvers = [RandomSolver(seed=seed), GreedySolver(), PerfectSolver()]
     if args.llm:
-        solvers.append(LlmSolver(base_url=args.llm_base, model=args.llm_model))
+        import os
+
+        api_key = args.llm_api_key or os.environ.get("MODEL_PROVIDER_API_KEY", "ollama")
+        solvers.append(LlmSolver(base_url=args.llm_base, model=args.llm_model, api_key=api_key, name=args.llm_name))
     return solvers
 
 
@@ -67,9 +70,11 @@ def build_parser() -> argparse.ArgumentParser:
     b.add_argument("--tasks", type=int, default=120)
     b.add_argument("--seed", type=int, default=7)
     b.add_argument("--out", default=str(DEFAULT_OUT))
-    b.add_argument("--llm", action="store_true", help="add the LLM solver (Ollama endpoint required)")
+    b.add_argument("--llm", action="store_true", help="add the LLM solver (OpenAI-compatible endpoint required)")
     b.add_argument("--llm-base", default="http://localhost:11434/v1")
     b.add_argument("--llm-model", default="qwen2.5:3b")
+    b.add_argument("--llm-api-key", default=None, help="falls back to $MODEL_PROVIDER_API_KEY, then 'ollama'")
+    b.add_argument("--llm-name", default="llm", help="solver label in the artifact (e.g. 'llm-ollama', 'llm-deepseek')")
     b.set_defaults(fn=cmd_bench)
 
     m = sub.add_parser("matrix", help="run across seeds (reproducibility)")
@@ -80,6 +85,8 @@ def build_parser() -> argparse.ArgumentParser:
     m.add_argument("--llm", action="store_true")
     m.add_argument("--llm-base", default="http://localhost:11434/v1")
     m.add_argument("--llm-model", default="qwen2.5:3b")
+    m.add_argument("--llm-api-key", default=None)
+    m.add_argument("--llm-name", default="llm")
     m.set_defaults(fn=cmd_matrix)
     return p
 

apps/trust/py/src/trust/forge/contamination.py

@@ -20,7 +20,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import Any
+from typing import Any, Callable
 
 from trust.forge.task import ForgeTask
 
@@ -41,6 +41,16 @@ def format_signature(task: ForgeTask) -> tuple[tuple[tuple[str, str, object], ..
     return (tuple((c.name, k, v) for c in task.expected for k, v in c.args.items()),)
 
 
+def format_hint(task: ForgeTask) -> str:
+    """The structural hint a leak probe reveals to a surrogate: tool names
+    and arg KEYS only, values withheld. If the surrogate fills in the real
+    values anyway, it already knew them — that's the Gemini-3 evidence
+    (a verification model reproduced ARC's integer-to-color mapping in its
+    reasoning chain despite never being told it)."""
+    parts = [f"{c.name}({', '.join(sorted(c.args.keys()))})" for c in task.expected]
+    return " -> ".join(parts)
+
+
 @dataclass
 class LeakProbe:
     """Reasoning-chain leak probe: does a hint about the task FORMAT let a
@@ -74,6 +84,32 @@ def run_leak_probes(tasks: list[ForgeTask], leaked_ids: set[str]) -> list[LeakPr
     return probes
 
 
+def run_llm_leak_probes(
+    tasks: list[ForgeTask],
+    complete_fn: Callable[[ForgeTask, str], str],
+) -> list[LeakProbe]:
+    """The real-surrogate leak probe (S4's synthetic version made concrete):
+    prompt ``complete_fn`` with only ``format_hint(task)`` — tool names and
+    arg keys, no values — and check whether the completion reproduces the
+    withheld VALUES anyway. Firing means the surrogate already knew content
+    it was never shown: reasoning-chain contamination, not a lucky guess.
+    A well-behaved (uncontaminated) surrogate stays silent on every task."""
+    probes: list[LeakProbe] = []
+    for task in tasks:
+        hint = format_hint(task)
+        completion = complete_fn(task, hint)
+        sig = format_signature(task)[0]
+        fired = bool(sig) and all(str(value) in completion for _tool, _key, value in sig)
+        probes.append(
+            LeakProbe(
+                probe_id=task.task_id,
+                fired=fired,
+                detail=f"hint={hint!r} reproduced_values={fired}",
+            )
+        )
+    return probes
+
+
 def corpus_overlap(tasks: list[ForgeTask], corpus_texts: list[str]) -> dict[str, float]:
     """Jaccard n-gram overlap between each task prompt and the corpus. High
     overlap on many tasks = the benchmark content is in the training data."""

apps/trust/py/src/trust/forge/stratify.py

@@ -9,6 +9,9 @@
 ``split_predictability`` is the acceptance test: fit per-split solve rates
 and require a minimum rank correlation between splits on a synthetic
 population of systems — a deliberately mis-stratified split must fail it.
+``split_distribution_kl`` is a separate, uncombined diagnostic (not part of
+the accept/reject decision) for eyeballing how well-matched the two
+difficulty histograms are.
 """
 from __future__ import annotations
 

apps/trust/py/src/trust/forge/studies/s4b_real_surrogate.py

@@ -0,0 +1,93 @@
+"""S4b — the real-surrogate contamination probe (HANDOFF open path #5).
+
+S4 (``s4_contamination_roc.py``) validates the leak-probe MECHANISM against
+a synthetic leaked-knowledge-base membership check. This script runs the
+same probe shape against a REAL model (``run_llm_leak_probes`` in
+``trust.forge.contamination``): the model sees only ``format_hint`` — tool
+names and arg keys, values withheld — and is asked to fill in the values.
+This is the actual Gemini-3 scenario from the ARC-AGI-3 report: a
+verification model reproduced ARC's integer-to-color mapping despite never
+being told it, because the mapping was in its training data.
+
+A real, never-leaked model + never-leaked tasks should fire on ~0% of
+probes (it has no way to know the withheld values); that's the honest
+baseline this script measures. It is NOT a leak-rate sweep like S4 — there
+is only one leak rate here (0%, nothing was actually leaked to the real
+model), and the finding is the false-fire rate itself.
+"""
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+from typing import Any
+
+import httpx
+
+from trust.forge.contamination import format_hint, run_llm_leak_probes
+from trust.forge.generators import ToolUseTaskGenerator
+from trust.forge.study import write_json
+from trust.forge.task import ForgeTask
+
+
+def real_llm_complete_fn(base_url: str, model: str, api_key: str):
+    def complete_fn(task: ForgeTask, hint: str) -> str:
+        prompt = (
+            f"Here is the STRUCTURE of a tool-call sequence (tool names and "
+            f"argument keys only, values omitted):\n\n{hint}\n\n"
+            "Fill in plausible exact values for every argument, based only "
+            "on what you already know. Reply with just the values."
+        )
+        try:
+            resp = httpx.post(
+                f"{base_url}/chat/completions",
+                json={
+                    "model": model,
+                    "messages": [{"role": "user", "content": prompt}],
+                    "temperature": 0,
+                    "max_tokens": 200,
+                },
+                headers={"Authorization": f"Bearer {api_key}"},
+                timeout=30,
+            )
+            resp.raise_for_status()
+            return resp.json()["choices"][0]["message"]["content"]
+        except Exception as exc:  # pragma: no cover — network path
+            return f"(probe failed: {exc})"
+
+    return complete_fn
+
+
+def run_all(out_dir: Path, *, n_tasks: int = 20) -> dict[str, Any]:
+    out_dir.mkdir(parents=True, exist_ok=True)
+    tasks = ToolUseTaskGenerator().generate(n=n_tasks)
+
+    base_url = os.environ.get("MODEL_PROVIDER_BASE_URL", "http://localhost:11434") + "/v1"
+    model = os.environ.get("MODEL_PROVIDER_MODEL_ID", "qwen2.5:3b")
+    api_key = os.environ.get("MODEL_PROVIDER_API_KEY", "ollama")
+    complete_fn = real_llm_complete_fn(base_url, model, api_key)
+
+    probes = run_llm_leak_probes(tasks, complete_fn=complete_fn)
+    fired = [p for p in probes if p.fired]
+    results = {
+        "model": model,
+        "n_tasks": len(tasks),
+        "n_fired": len(fired),
+        "false_fire_rate": round(len(fired) / max(len(tasks), 1), 4),
+        "probes": [p.as_dict() for p in probes],
+    }
+    write_json(out_dir / "s4b-real-surrogate.json", results)
+    print(json.dumps({k: v for k, v in results.items() if k != "probes"}, indent=2))
+    return results
+
+
+if __name__ == "__main__":
+    import argparse
+    import sys
+
+    p = argparse.ArgumentParser()
+    p.add_argument("--out", default="docs/validation/studies")
+    p.add_argument("--tasks", type=int, default=20)
+    args = p.parse_args()
+    run_all(Path(args.out), n_tasks=args.tasks)
+    sys.exit(0)