feat(evidence_probe): port to FULL Go — dichotomy wave

docs/dichotomy-rebase-honest-audit.md

@@ -2,14 +2,14 @@
 
 _Regenerate with `python3 scripts/dichotomy_audit.py --write`. A module is FULL only when 80%+ of its Python functions have name-preserving counterparts in the Go/Rust implementation — directories without the functions are facades and count as NOT TOUCHED._
 
-**Function-level completion: 28%** (686/2480 runtime functions matched)
+**Function-level completion: 32%** (838/2627 runtime functions matched)
 
 | Status | Modules | Lines | Meaning |
 |---|---|---|---|
-| FULL | 15 | 4264 | 80%+ of functions ported |
-| PARTIAL | 83 | 41656 | 20–80% — real logic exists, gaps named below |
-| NOT TOUCHED | 119 | 26856 | no name-preserving implementation |
-| Total runtime | 217 | 72776 | |
+| FULL | 25 | 8218 | 80%+ of functions ported |
+| PARTIAL | 89 | 44507 | 20–80% — real logic exists, gaps named below |
+| NOT TOUCHED | 113 | 22686 | no name-preserving implementation |
+| Total runtime | 227 | 75411 | |
 
 ## NOT TOUCHED — the port queue, in priority order (largest first)
 
@@ -17,50 +17,36 @@ _Regenerate with `python3 scripts/dichotomy_audit.py --write`. A module is FULL
 |---|---|---|---|
 | rune_tui | 4170 | 159 | TUI-LAST |
 | curator | 1838 | 43 | GO |
-| abstraction | 656 | 17 | GO |
 | canary_audit_ledger | 541 | 2 | TEST-INFRA |
 | tui_validator | 514 | 13 | TUI-LAST |
-| construction_pattern | 496 | 12 | RUST |
-| model_routing | 432 | 13 | GO |
 | worthiness | 411 | 13 | unclassified |
-| token_optimizer | 410 | 22 | RUST |
-| behavior | 408 | 9 | RUST |
-| capability_ledger | 405 | 18 | RUST |
 | chat_bubble | 401 | 12 | TUI-LAST |
-| prompt_sync | 379 | 14 | RUST |
 | drift-ledger | 355 | 14 | unclassified |
-| programme_continuation | 349 | 11 | RUST |
+| pc_loader | 337 | 22 | unclassified |
 | canary_rune_serve | 329 | 13 | TEST-INFRA |
 | canary_p1_07 | 324 | 10 | TEST-INFRA |
-| message_queue | 316 | 13 | GO |
-| drift_observability | 315 | 8 | RUST |
-| build_basis | 314 | 12 | RUST |
 | canary_p0_07 | 310 | 10 | TEST-INFRA |
-| attention_capsule | 296 | 7 | RUST |
 | canary_p1_01 | 294 | 12 | TEST-INFRA |
-| drift_cache | 292 | 15 | RUST |
 | outcome_ledger | 286 | 7 | RUST |
 | time_awareness | 286 | 14 | RUST |
 | canary_p1_06 | 280 | 7 | TEST-INFRA |
-| auto_context | 273 | 8 | GO |
 | canary_p1_03 | 271 | 9 | TEST-INFRA |
 | canary_gitea_automation | 263 | 3 | TEST-INFRA |
 | canary_p1_02 | 253 | 10 | TEST-INFRA |
 | construction_patterns_builtin | 249 | 1 | RUST |
 | canary_full_system | 248 | 2 | TEST-INFRA |
-| toolchain | 236 | 5 | RUST |
 | task_gate | 235 | 8 | RUST |
 | canary_native | 231 | 3 | TEST-INFRA |
 | lexicon_evolution | 229 | 7 | unclassified |
 | openai_compat_transport | 227 | 9 | GO |
 | shard_manager | 223 | 9 | RUST |
 | spine_language | 218 | 5 | RUST |
-| evidence_probe | 213 | 9 | GO |
 | canary_p0_04 | 207 | 8 | TEST-INFRA |
 | cc_loader | 205 | 28 | RUST |
 | forge_client | 203 | 18 | GO |
 | canary_context_broker | 199 | 10 | TEST-INFRA |
 | canary_p0_05 | 199 | 13 | TEST-INFRA |
+| cl_loader | 196 | 20 | unclassified |
 | testing_mode | 196 | 12 | RUST |
 | grading | 193 | 3 | unclassified |
 | canary_spine_meta | 192 | 7 | TEST-INFRA |
@@ -73,16 +59,24 @@ _Regenerate with `python3 scripts/dichotomy_audit.py --write`. A module is FULL
 | inference_trace | 164 | 6 | RUST |
 | ms_loader | 163 | 16 | RUST |
 | canary_gitea_pipeline | 159 | 2 | TEST-INFRA |
+| dc_loader | 158 | 13 | unclassified |
 | mse_loader | 152 | 12 | unclassified |
+| bb_loader | 151 | 14 | unclassified |
 | forge | 146 | 11 | GO |
 | rune_graphic | 145 | 5 | TUI-LAST |
 | spine_loader | 144 | 13 | RUST |
+| ac_loader | 140 | 11 | unclassified |
 | project_registry | 139 | 11 | RUST |
 | canary_rune_entry | 137 | 3 | TEST-INFRA |
+| do_loader | 137 | 8 | unclassified |
+| ps_loader | 136 | 13 | unclassified |
 | verification_layer | 136 | 11 | RUST |
 | tui_backlog | 125 | 5 | TUI-LAST |
 | state_checkpoint | 123 | 8 | unclassified |
+| cp_loader | 122 | 9 | unclassified |
 | drift_resume | 118 | 3 | RUST |
+| tl_loader | 118 | 9 | unclassified |
+| bh_loader | 117 | 9 | unclassified |
 | canary_conversation_state | 115 | 2 | TEST-INFRA |
 | contracts_loader | 112 | 8 | unclassified |
 | canary_query_memory | 111 | 2 | TEST-INFRA |
@@ -100,9 +94,9 @@ _Regenerate with `python3 scripts/dichotomy_audit.py --write`. A module is FULL
 | broker_loader | 86 | 5 | unclassified |
 | canary_schema_migration | 86 | 1 | TEST-INFRA |
 | ss_loader | 86 | 7 | unclassified |
+| to_loader | 85 | 4 | unclassified |
 | mems_loader | 83 | 5 | RUST |
 | cm_loader | 82 | 4 | unclassified |
-| cp_loader | 77 | 4 | unclassified |
 | dl_loader | 77 | 5 | unclassified |
 | lc_loader | 75 | 5 | unclassified |
 | cd_loader | 72 | 4 | unclassified |
@@ -155,7 +149,7 @@ _Regenerate with `python3 scripts/dichotomy_audit.py --write`. A module is FULL
 | growth_delta | 449 | 5/20 | _ctx_budget, _store_db, _llm, _parse_json, _load_deltas, _save_deltas |
 | provider_budget | 357 | 5/20 | get_tracker, track_usage, check_provider, get_best_provider, failover, get_status |
 | safety_override | 574 | 7/22 | _keyword_stopwords, _cc_probe, cc_loader_pin, self_test, _py_matches, escalation_rate |
-| spine | 965 | 20/35 | _py_sorted, _module_imports, _py_declared_surfaces, _py_project_meta_surfaces, _gather_meta_inputs, _rust_core |
+| spine | 975 | 20/35 | _py_sorted, _module_imports, _py_declared_surfaces, _py_project_meta_surfaces, _gather_meta_inputs, _rust_core |
 | tool_executor | 1836 | 32/47 | _execute_tool_now, serialize_prepared_tool, deserialize_prepared_tool, confirm_write, _resolve_execution_cwd, _run_command |
 | cron_engine | 440 | 13/27 | get_cron_engine, run_task, start_engine, stop_engine, to_dict, fire_on_task_complete |
 | night_mode | 421 | 5/19 | _ctx_budget, get_night_mode, start_night_mode, stop_night_mode, get_status, get_report |
@@ -166,6 +160,7 @@ _Regenerate with `python3 scripts/dichotomy_audit.py --write`. A module is FULL
 | identity_anchors | 542 | 5/16 | show_anchors, show_rules, show_pronouns, _announcement_patterns, _pronoun_patterns, _hedge_patterns |
 | parallel_processor | 194 | 3/14 | _shutdown_processor, get_processor, run_parallel_with_args, get_status, test_parallel, run_parallel_with_args |
 | structural_model | 388 | 9/20 | workspace_record, workspace_advance, _init_project, save, reload, add |
+| token_optimizer | 514 | 16/27 | get_optimizer, _tc_core, _to_core, _py_estimate_tokens, _py_compress_context, to_dict |
 | capability_broker | 798 | 10/20 | _native_tool_schemas, load_registry, _cards_are_stale, load_cards, generate_card, build_cards |
 | capability_fit | 414 | 3/13 | record_semantic_fit, _provides_contracts, _consumers, discover_candidates, _cs_core, decide |
 | claim_audit | 484 | 4/14 | _get_warm_neuron_store, reframe, _graph_has, _warm_has, verify, journal |
@@ -176,7 +171,9 @@ _Regenerate with `python3 scripts/dichotomy_audit.py --write`. A module is FULL
 | drift_compression | 307 | 3/12 | _py_get_compression_ratio, _py_compress_text, _py_compress_entry, _py_should_warn, _py_warn_before_compress, get_compression_ratio |
 | drift_ledger | 471 | 7/16 | ledger_path, read_block, load_ledger, save_ledger, surface_alert, snapshot |
 | scaffold | 868 | 22/31 | init, set_prd, add_maintenance_item, bind_repo, observe_repo, repo_state_from_git |
+| behavior | 537 | 6/14 | _bh_core, connect, _correction_matches_python, _habit_matches_python, _reinforce_rust, _reinforce_python |
 | builder_bridge | 326 | 3/11 | _construction_store, _active_construction_lines, _default_selector, _cached_memory, _get_selector, _retrieve_memory |
+| construction_pattern | 547 | 7/14 | _core, _need_dict, _surface_text, _path_matches, _entity_matches, _obligation_when_hit |
 | contracts | 743 | 4/11 | _ct_core, _resolve_and_project, _run_contract_python, capture_signatures, save_signatures, load_signatures |
 | graph_memory | 474 | 2/9 | now_iso, connect, resolve_entity, add_edge, query, stats |
 | intent_continuity | 168 | 2/9 | to_dict, from_dict, _load, save, create_detour, resolve_detour |
@@ -198,8 +195,11 @@ _Regenerate with `python3 scripts/dichotomy_audit.py --write`. A module is FULL
 | tiered_executor | 291 | 3/8 | _ctx_budget, test_executor, to_dict, to_dict, _execute_model |
 | adaptive_router | 625 | 15/19 | _floor_applies_from_flags, _floor_reason, _record_transition_with, _detect_external_refs |
 | construction_commands | 268 | 1/5 | _refuse, _validate, _accept, dispatch |
+| drift_cache | 357 | 12/16 | _dc_core, _py_compute_priority, _py_qualifies_for_cache, main |
+| drift_observability | 399 | 5/9 | _do_core, main, _load_alerts, _save_alerts |
 | layout_contract | 336 | 2/6 | define_region, bind_component, amend_layout_contract, record_layout_observation |
 | model_connect | 471 | 7/11 | _ctx_budget, _now, _get_cloud_models, handle_hook |
+| prompt_sync | 487 | 11/15 | _now, _rust_available, get_store, main |
 | salience_gate | 276 | 1/5 | _sg_core, test_salience, handle_hook, to_dict |
 | self_improve | 246 | 1/5 | _now, suggest_improvements, handle_hook, main |
 | stage_guards | 303 | 7/11 | _py_liveness_canary, _py_reconcile, to_dict, chk |
@@ -227,13 +227,23 @@ _Regenerate with `python3 scripts/dichotomy_audit.py --write`. A module is FULL
 
 | Module | Lines | Fns | Side |
 |---|---|---|---|
+| abstraction | 656 | 17/17 | GO |
 | system_registry | 603 | 14/16 | RUST |
 | construction_intent | 580 | 13/16 | RUST |
+| capability_ledger | 571 | 18/19 | RUST |
 | event_wait | 518 | 14/14 | GO |
 | tool_protocol | 461 | 22/27 | GO |
 | auto_tasks | 460 | 14/14 | GO |
+| programme_continuation | 458 | 11/12 | RUST |
 | goal | 453 | 20/20 | GO |
 | release_notes | 446 | 16/16 | GO |
+| model_routing | 432 | 13/13 | GO |
+| build_basis | 375 | 11/13 | RUST |
+| attention_capsule | 365 | 6/7 | RUST |
+| message_queue | 316 | 13/13 | GO |
+| toolchain | 295 | 5/6 | RUST |
+| auto_context | 273 | 8/8 | GO |
+| evidence_probe | 213 | 9/9 | GO |
 | recall_uplift | 141 | 1/1 | RUST |
 | rune_spinner | 130 | 4/5 | RUST |
 | wait_guard | 116 | 1/1 | RUST |

scripts/gen_evidence_probe_truth.py

@@ -0,0 +1,599 @@
+#!/usr/bin/env python3
+"""gen_evidence_probe_truth.py — evidence_probe FULL-port truth battery
+(dichotomy wave #2479).
+
+Freezes the LIVE src/rune/evidence_probe.py behavior (validate_probe and
+count_probes served by the Rust evidence_provenance core through ep_loader)
+for every public function and branch:
+
+  - build_probe_prompt: baseline / unicode / path-like question / empty args
+  - validate_probe: the 5-key schema gate — missing fields, non-string and
+    blank fields, status rendering (repr, ints, floats, bools, None, list,
+    dict incl. nested key order), GROUNDED source-citation boundary (7/8
+    chars, missing/non-string source), extra-field sorting (meta fields
+    allowed), non-object artifacts, combined-error ordering
+  - count_probes: absent dir/file, empty, blank-only, Rust `lines()`
+    semantics (trailing \\r, no trailing newline, trailing newline,
+    whitespace-only lines)
+  - check_budget: 0/1 ok, 2/3 ProbeBudgetExceeded
+  - record_probe: fresh dir creation, probe.jsonl append, unicode
+    ensure_ascii bytes, returned path
+  - run_probe: injected runner — success (grounded / still_unknown), model
+    fallback chain (explicit / MIMO_MODEL env / unset -> UNKNOWN /
+    set-but-empty -> ""), budget exhausted BEFORE the runner fires, silent
+    runner -> declared-artifact failure, gate-failure joins, and the exact
+    prompt + absolute artifact path the runner receives
+  - ProbeResult.to_dict: field order + ensure_ascii
+  - _default_runner: stubbed MIMO_BIN subprocess — argv parity, stdout log,
+    exit-code error, MIMO_AGENT default/override, require_model
+    resolution + ProviderUnresolved (set/unset/blank)
+
+Clock frozen at FROZEN_TS (probe_id + decided_at deterministic). The only
+excluded branch is the invalid-JSON artifact crash (Python json.JSONDecodeError
+rendering is interpreter-specific, same policy as the auto_context batch).
+
+Usage: python3 scripts/gen_evidence_probe_truth.py   (from repo root)
+"""
+
+import json
+import os
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO / "src"))
+
+from rune import evidence_probe as ep  # noqa: E402
+import rune.provider_context as provider_context  # noqa: E402
+
+FROZEN_TS = 1791753600
+ARTDIR = "@ARTDIR@"
+STUB_NAME = "ep_stub.sh"
+
+_real_gmtime = time.gmtime
+
+GROUNDED = {
+    "question": "Does the API support streaming?",
+    "source": "https://platformdirs.readthedocs.io/changelog.html",
+    "answer": "Current version guarantees the assumed behaviour.",
+    "implication": "Local wiring is sufficient; no version guard needed.",
+    "status": "GROUNDED",
+}
+
+
+def norm(s, base):
+    return s.replace(base, ARTDIR) if isinstance(s, str) else s
+
+
+def write_files(cdir, files):
+    os.makedirs(cdir, exist_ok=True)
+    for rel, content in (files or {}).items():
+        p = os.path.join(cdir, rel)
+        os.makedirs(os.path.dirname(p), exist_ok=True)
+        with open(p, "w") as f:
+            f.write(content)
+
+
+def exc_row(fn, base, types=None):
+    """Call fn(), freeze either {error: None, value: ...} or the exception."""
+    try:
+        return {"error": None, "value": fn()}
+    except Exception as e:  # noqa: BLE001 — the frozen point IS the exception
+        t = type(e).__name__
+        if types and t not in types:
+            raise
+        return {"error": {"type": t, "msg": norm(str(e), base)}}
+
+
+def main():
+    base = tempfile.mkdtemp(prefix="ep_truth_")
+    rows = []
+    real_time = time.time
+    ep.time.time = lambda: FROZEN_TS
+    ep.time.gmtime = lambda *a: _real_gmtime(FROZEN_TS)
+    try:
+        _build_rows(base, rows)
+    finally:
+        ep.time.time = real_time
+        ep.time.gmtime = _real_gmtime
+
+def _build_rows(base, rows):
+    # ── build_probe_prompt ────────────────────────────────────────────────
+    prompt_cases = [
+        ("prompt_baseline", "Does the API support streaming?", "/tmp/probe.json"),
+        ("prompt_unicode", "Wird die Größenbeschränkung überschritten?", "/tmp/probe.json"),
+        (
+            "prompt_path_like_question",
+            "Is the file at /tmp/probe.json overwritten per probe?",
+            "/tmp/artifacts/evidence_probe.json",
+        ),
+        ("prompt_empty_question", "", ""),
+        ("prompt_multiline_question", "line one\nline two\nline three", "/tmp/x.json"),
+    ]
+    for name, q, ap in prompt_cases:
+        rows.append(
+            {
+                "name": name,
+                "op": "prompt",
+                "setup": {"question": q, "artifact_path": ap},
+                "want": ep.build_probe_prompt(q, ap),
+            }
+        )
+
+    # ── validate_probe ────────────────────────────────────────────────────
+    def val(name, artifact):
+        rows.append(
+            {
+                "name": name,
+                "op": "validate",
+                "setup": {"artifact": artifact},
+                "want": ep.validate_probe(artifact),
+            }
+        )
+
+    val("valid_grounded", dict(GROUNDED))
+    val(
+        "valid_grounded_with_meta",
+        dict(GROUNDED, probe_id="probe-1", decided_at="2026-10-11T00:00:00Z", model="m"),
+    )
+    val(
+        "valid_still_unknown_empty_source",
+        {
+            "question": "probe",
+            "source": "",
+            "answer": "Could not find authoritative evidence.",
+            "implication": "Uncertainty remains.",
+            "status": "STILL_UNKNOWN",
+        },
+    )
+    val(
+        "valid_contradicted",
+        dict(GROUNDED, status="CONTRADICTED", source="task assumption"),
+    )
+    for field in ("question", "source", "answer", "implication", "status"):
+        art = dict(GROUNDED)
+        del art[field]
+        val(f"missing_{field}", art)
+    val("empty_artifact", {})
+    val("nonstring_question", dict(GROUNDED, question=7))
+    val("nonstring_answer", dict(GROUNDED, answer=["a"]))
+    val("nonstring_implication", dict(GROUNDED, implication={"k": 1}))
+    val("nonstring_source", dict(GROUNDED, source=3.5))
+    val("blank_question", dict(GROUNDED, question=""))
+    val("whitespace_question", dict(GROUNDED, question="   \t\n"))
+    val("blank_answer", dict(GROUNDED, answer=""))
+    val("blank_implication", dict(GROUNDED, implication=""))
+    val("blank_source_allowed", dict(GROUNDED, source=""))
+    val("status_empty_string", dict(GROUNDED, status=""))
+    val("status_lowercase", dict(GROUNDED, status="grounded"))
+    val("status_int", dict(GROUNDED, status=1))
+    val("status_float", dict(GROUNDED, status=1.5))
+    val("status_float_integral", dict(GROUNDED, status=2.0))
+    val("status_float_small", dict(GROUNDED, status=0.1))
+    val("status_big_float", dict(GROUNDED, status=1e30))
+    val("status_huge_int", dict(GROUNDED, status=9007199254740993))
+    val("status_true", dict(GROUNDED, status=True))
+    val("status_false", dict(GROUNDED, status=False))
+    val("status_null", dict(GROUNDED, status=None))
+    val("status_list", dict(GROUNDED, status=[1, "x"]))
+    val("status_obj_order", dict(GROUNDED, status={"b": 1, "a": {"z": 2, "y": 3}}))
+    val("status_repr_quote", dict(GROUNDED, status="it's"))
+    val("status_repr_newline", dict(GROUNDED, status="a\nb"))
+    val("status_repr_unicode", dict(GROUNDED, status="größe"))
+    val("grounded_short_source_7", dict(GROUNDED, source="1234567"))
+    val("grounded_source_8_ok", dict(GROUNDED, source="12345678"))
+    val("grounded_missing_source", {k: v for k, v in GROUNDED.items() if k != "source"})
+    val("grounded_nonstring_source", dict(GROUNDED, source=42))
+    val("extra_single", dict(GROUNDED, extra=1))
+    val("extra_sorted", dict(GROUNDED, z_extra=1, a_extra=2, m_extra=3))
+    val("extra_meta_allowed", dict(GROUNDED, model="m", probe_id="p", decided_at="d"))
+    val("extra_unicode_key", dict(GROUNDED, **{"ключ": 1}))
+    val("nonobject_list", ["a", 1])
+    val("nonobject_string", "nope")
+    val("nonobject_int", 3)
+    val("nonobject_bool", True)
+    val("nonobject_null", None)
+    val(
+        "combined_ordering",
+        {"status": 5, "extra_b": 1, "extra_a": 2, "answer": ""},
+    )
+
+    # ── count_probes ──────────────────────────────────────────────────────
+    count_cases = [
+        ("count_absent_dir", None),
+        ("count_absent_file", {}),
+        ("count_empty_file", {"probe.jsonl": ""}),
+        ("count_blank_only", {"probe.jsonl": "\n\n  \n"}),
+        ("count_three_lines", {"probe.jsonl": "a\nb\nc\n"}),
+        ("count_no_trailing_newline", {"probe.jsonl": "a\nb"}),
+        ("count_interleaved_blank", {"probe.jsonl": "a\n\nb\n\n\nc"}),
+        ("count_cr_lines", {"probe.jsonl": "a\r\nb\r\n"}),
+        ("count_whitespace_lines", {"probe.jsonl": "  \n x\n\t\n"}),
+        ("count_cr_only_line", {"probe.jsonl": "\r\r\nx"}),
+        ("count_single_line", {"probe.jsonl": "one"}),
+    ]
+    for name, files in count_cases:
+        cdir = os.path.join(base, name)
+        if files is not None:
+            write_files(cdir, files)
+        rows.append(
+            {
+                "name": name,
+                "op": "count",
+                "setup": {"dir": norm(cdir, base), "files": files or {}},
+                "want": ep.count_probes(cdir),
+            }
+        )
+
+    # ── check_budget ──────────────────────────────────────────────────────
+    for n, name in ((0, "budget_zero"), (1, "budget_one"), (2, "budget_two"), (3, "budget_three")):
+        cdir = os.path.join(base, name)
+        write_files(cdir, {"probe.jsonl": "x\n" * n})
+        out = exc_row(lambda d=cdir: ep.check_budget(d), base, types={"ProbeBudgetExceeded"})
+        rows.append(
+            {
+                "name": name,
+                "op": "budget",
+                "setup": {"dir": norm(cdir, base), "files": {"probe.jsonl": "x\n" * n}},
+                "want": out,
+            }
+        )
+
+    # ── record_probe ──────────────────────────────────────────────────────
+    def rec(name, result, files=None):
+        cdir = os.path.join(base, name)
+        if files:
+            write_files(cdir, files)
+        path = ep.record_probe(ep.ProbeResult(**result), cdir)
+        with open(os.path.join(cdir, "probe.json")) as f:
+            pj = f.read()
+        with open(os.path.join(cdir, "probe.jsonl")) as f:
+            pjl = f.read()
+        rows.append(
+            {
+                "name": name,
+                "op": "record",
+                "setup": {"result": result, "files": files or {}},
+                "want": {
+                    "path": norm(path, base),
+                    "probe_json": pj,
+                    "probe_jsonl": pjl,
+                },
+            }
+        )
+
+    rec(
+        "record_fresh_dir_created",
+        {
+            "question": "q",
+            "source": "https://example.com/a",
+            "answer": "a",
+            "implication": "i",
+            "status": "GROUNDED",
+            "probe_id": "probe-1791753600",
+            "decided_at": "2026-10-12T04:00:00Z",
+            "model": "UNKNOWN",
+        },
+    )
+    rec(
+        "record_appends_existing",
+        {
+            "question": "q2",
+            "source": "",
+            "answer": "a2",
+            "implication": "i2",
+            "status": "STILL_UNKNOWN",
+            "probe_id": "probe-1791753600",
+            "decided_at": "2026-10-12T04:00:00Z",
+            "model": "m",
+        },
+        files={"probe.jsonl": '{"question": "old"}\n'},
+    )
+    rec(
+        "record_unicode_ensure_ascii",
+        {
+            "question": "Größe überschritten — warum?",
+            "source": "https://example.com/ä",
+            "answer": "ja ✓",
+            "implication": "🚀",
+            "status": "GROUNDED",
+            "probe_id": "probe-1791753600",
+            "decided_at": "2026-10-12T04:00:00Z",
+            "model": "model-ü",
+        },
+    )
+
+    # ── run_probe ─────────────────────────────────────────────────────────
+    def run(name, *, question, runner_spec, model="UNKNOWN", env=None, files=None):
+        cdir = os.path.join(base, name)
+        write_files(cdir, files or {})
+        old_env = {k: os.environ.get(k) for k in ("MIMO_MODEL",)}
+        try:
+            for k, v in (env or {}).items():
+                if v is None:
+                    os.environ.pop(k, None)
+                else:
+                    os.environ[k] = v
+            captures = []
+
+            def runner(prompt, artifact_path):
+                captures.append((prompt, artifact_path))
+                kind = runner_spec["type"]
+                if kind == "write":
+                    with open(artifact_path, "w") as f:
+                        json.dump(runner_spec["artifact"], f)
+                elif kind == "silent":
+                    pass
+                else:
+                    raise AssertionError(f"unknown runner spec {kind}")
+
+            try:
+                result = ep.run_probe(question, cdir, runner=runner, model=model)
+                out = {
+                    "error": None,
+                    "result": result.to_dict(),
+                }
+                with open(os.path.join(cdir, "probe.json")) as f:
+                    out["probe_json"] = f.read()
+                with open(os.path.join(cdir, "probe.jsonl")) as f:
+                    out["probe_jsonl"] = f.read()
+            except Exception as e:  # noqa: BLE001
+                out = {"error": {"type": type(e).__name__, "msg": norm(str(e), base)}}
+            out["runner_called"] = bool(captures)
+            if captures:
+                out["prompt"] = norm(captures[0][0], base)
+                out["artifact_path"] = norm(captures[0][1], base)
+            rows.append(
+                {
+                    "name": name,
+                    "op": "run",
+                    "setup": {
+                        "question": question,
+                        "model": model,
+                        "env": env or {},
+                        "files": files or {},
+                        "runner": runner_spec,
+                    },
+                    "want": out,
+                }
+            )
+        finally:
+            for k, v in old_env.items():
+                if v is None:
+                    os.environ.pop(k, None)
+                else:
+                    os.environ[k] = v
+
+    STILL_UNKNOWN = {
+        "question": "probe",
+        "source": "",
+        "answer": "Could not find authoritative evidence.",
+        "implication": "Uncertainty remains.",
+        "status": "STILL_UNKNOWN",
+    }
+    run(
+        "run_success_grounded",
+        question="Does the API support streaming?",
+        runner_spec={"type": "write", "artifact": dict(GROUNDED)},
+        model="m-1",
+    )
+    run(
+        "run_success_still_unknown",
+        question="probe",
+        runner_spec={"type": "write", "artifact": dict(STILL_UNKNOWN)},
+        model="m-1",
+    )
+    run(
+        "run_model_from_env",
+        question="probe",
+        runner_spec={"type": "write", "artifact": dict(GROUNDED)},
+        model="",
+        env={"MIMO_MODEL": "env-model"},
+    )
+    run(
+        "run_model_unset_unknown",
+        question="probe",
+        runner_spec={"type": "write", "artifact": dict(GROUNDED)},
+        model="",
+        env={"MIMO_MODEL": None},
+    )
+    run(
+        "run_model_env_blank_is_empty_string",
+        question="probe",
+        runner_spec={"type": "write", "artifact": dict(GROUNDED)},
+        model="",
+        env={"MIMO_MODEL": ""},
+    )
+    run(
+        "run_budget_exhausted_before_runner",
+        question="one too many",
+        runner_spec={"type": "silent"},
+        model="m",
+        files={"probe.jsonl": "a\nb\n"},
+    )
+    run(
+        "run_budget_one_left",
+        question="still allowed",
+        runner_spec={"type": "write", "artifact": dict(GROUNDED)},
+        model="m",
+        files={"probe.jsonl": "a\n"},
+    )
+    run(
+        "run_silent_runner_missing_artifact",
+        question="q",
+        runner_spec={"type": "silent"},
+        model="m",
+    )
+    run(
+        "run_gate_failure_joined",
+        question="q",
+        runner_spec={
+            "type": "write",
+            "artifact": {"question": "q", "source": "s", "answer": "a", "implication": "i", "status": "MAYBE"},
+        },
+        model="m",
+    )
+    run(
+        "run_gate_failure_non_dict",
+        question="q",
+        runner_spec={"type": "write", "artifact": [1, 2]},
+        model="m",
+    )
+    run(
+        "run_gate_failure_status_non_string",
+        question="q",
+        runner_spec={
+            "type": "write",
+            "artifact": {"question": "q", "source": "s", "answer": "a", "implication": "i", "status": 9},
+        },
+        model="m",
+    )
+
+    # ── ProbeResult.to_dict ───────────────────────────────────────────────
+    def todict(name, fields):
+        d = ep.ProbeResult(**fields).to_dict()
+        rows.append(
+            {
+                "name": name,
+                "op": "to_dict",
+                "setup": {"fields": fields},
+                "want": json.dumps(d, ensure_ascii=True),
+            }
+        )
+
+    todict(
+        "to_dict_defaults",
+        {
+            "question": "q",
+            "source": "s",
+            "answer": "a",
+            "implication": "i",
+            "status": "GROUNDED",
+        },
+    )
+    todict(
+        "to_dict_full",
+        {
+            "question": "q2",
+            "source": "s2",
+            "answer": "a2",
+            "implication": "i2",
+            "status": "STILL_UNKNOWN",
+            "probe_id": "probe-1791753600",
+            "decided_at": "2026-10-12T04:00:00Z",
+            "model": "m",
+        },
+    )
+    todict(
+        "to_dict_unicode",
+        {
+            "question": "Größe",
+            "source": "s — dash",
+            "answer": "✓",
+            "implication": "🚀",
+            "status": "CONTRADICTED",
+        },
+    )
+
+    # ── _default_runner (stubbed subprocess) ──────────────────────────────
+    stub_path = os.path.join(base, STUB_NAME)
+    with open(stub_path, "w") as f:
+        f.write(
+            "#!/bin/sh\n"
+            'for a in "$@"; do printf \'%s\\n\' "$a" >> "$EP_CAPTURE"; done\n'
+            'printf \'stub-out\\n\'\n'
+            'exit "$EP_EXIT"\n'
+        )
+    os.chmod(stub_path, 0o755)
+
+    real_require_model = provider_context.require_model
+
+    def defrun(name, *, exit_code, agent_env, model_mode):
+        cdir = os.path.join(base, name)
+        os.makedirs(cdir, exist_ok=True)
+        capture = os.path.join(cdir, "argv_capture.txt")
+        artifact_path = os.path.join(cdir, "evidence_probe.json")
+        old_env = {k: os.environ.get(k) for k in ("MIMO_BIN", "MIMO_AGENT", "MIMO_MODEL", "EP_CAPTURE", "EP_EXIT")}
+        try:
+            if os.path.exists(capture):
+                os.unlink(capture)
+            os.environ["MIMO_BIN"] = stub_path
+            os.environ["EP_CAPTURE"] = capture
+            os.environ["EP_EXIT"] = str(exit_code)
+            if agent_env is None:
+                os.environ.pop("MIMO_AGENT", None)
+            else:
+                os.environ["MIMO_AGENT"] = agent_env
+            if model_mode == "unresolved":
+                os.environ.pop("MIMO_MODEL", None)
+                provider_context.require_model = real_require_model
+            elif model_mode == "blank":
+                os.environ["MIMO_MODEL"] = "   "
+                provider_context.require_model = real_require_model
+            else:
+                provider_context.require_model = (
+                    lambda env_var="MIMO_MODEL": "test-model-v9"
+                )
+            try:
+                ep._default_runner("PROMPT-TEXT", artifact_path)
+                err = None
+            except Exception as e:  # noqa: BLE001
+                err = {"type": type(e).__name__, "msg": norm(str(e), base)}
+            argv = None
+            if os.path.exists(capture):
+                with open(capture) as f:
+                    argv = [norm(line, base) for line in f.read().splitlines()]
+            log_path = os.path.join(cdir, "evidence_probe.stdout.log")
+            log = None
+            if os.path.exists(log_path):
+                with open(log_path) as f:
+                    log = f.read()
+            rows.append(
+                {
+                    "name": name,
+                    "op": "default_runner",
+                    "setup": {
+                        "exit_code": exit_code,
+                        "agent_env": agent_env,
+                        "model_mode": model_mode,
+                    },
+                    "want": {"error": err, "argv": argv, "stdout_log": log},
+                }
+            )
+        finally:
+            provider_context.require_model = real_require_model
+            for k, v in old_env.items():
+                if v is None:
+                    os.environ.pop(k, None)
+                else:
+                    os.environ[k] = v
+
+    defrun("defrun_exit0_default_agent", exit_code=0, agent_env=None, model_mode="patched")
+    defrun("defrun_exit3", exit_code=3, agent_env=None, model_mode="patched")
+    defrun("defrun_custom_agent", exit_code=0, agent_env="custom-agent", model_mode="patched")
+    defrun("defrun_model_unresolved", exit_code=0, agent_env=None, model_mode="unresolved")
+    defrun("defrun_model_blank_env", exit_code=0, agent_env=None, model_mode="blank")
+
+    out = {
+        "oracle": "src/rune/evidence_probe.py (live; validate_probe/count_probes via Rust evidence_provenance core)",
+        "generated": "2026-10-12",
+        "frozen_ts": FROZEN_TS,
+        "frozen_decided_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", _real_gmtime(FROZEN_TS)),
+        "rows": rows,
+    }
+    target = REPO / "src/go/evidence-probe" / "truth.json"
+    target.write_text(json.dumps(out, indent=1, ensure_ascii=False) + "\n")
+    print(f"wrote {target} ({len(rows)} rows)")
+
+    # determinism self-check: probe_id/decided_at must be frozen
+    assert all(
+        r.get("want", {}).get("result", {}).get("probe_id") in (None, f"probe-{FROZEN_TS}")
+        for r in rows
+        if r["op"] == "run"
+    ), "clock freeze failed"
+
+
+if __name__ == "__main__":
+    main()

src/go/evidence-probe/main.go

@@ -1,110 +1,517 @@
-// Package main — evidence-probe: the GO port of
-// src/rune/evidence_probe.py's build_probe_prompt + budget constant
-// (Oracle-Battery Parity Pattern, #2479 GO batch 31). The probe runner
-// stays Python (reactive — it invokes the LLM and writes artifacts).
+// Package main — evidence_probe: the FULL GO port of
+// src/rune/evidence_probe.py (dichotomy wave, issue #2479 tranche).
 //
-// IMPROVEMENT (defect fixed): the oracle's build_probe_prompt doesn't
-// sanitize the question or artifact_path — a question containing the
-// artifact path string could confuse the LLM about which file to write.
-// The Go version embeds them but marks them with explicit delimiters.
+// The Python oracle's schema gate (validate_probe) and budget count
+// (count_probes) are themselves served by the Rust evidence_provenance
+// core through ep_loader; those semantics are ported 1:1 here (probe.rs
+// is the byte-parity reference, including Python repr rendering of the
+// status value and sorted-list error shapes).
+//
+// Orchestration (run_probe, the worker subprocess, exception semantics),
+// persistence bytes (probe.json indent=2 / probe.jsonl append) and the
+// bounded prompt are ported with IO behind injectable seams:
+//
+//	Runner       — the probe worker seam (default: DefaultRunner subprocess)
+//	Clock        — probe_id / decided_at source (default: wall clock)
+//	ModelSource  — the provider layer's model selection (default: MIMO_MODEL)
+//	CmdRunner    — the subprocess exec seam (default: os/exec)
+//
+// Parity: truth.json (frozen vs the LIVE oracle) replays verbatim through
+// replay.go; `go run . replay truth.json` and the battery in truth_test.go
+// gate every public function and branch.
 package main
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"os"
+	"os/exec"
+	"path/filepath"
+	"sort"
 	"strings"
+	"time"
 )
 
+// ── Constants (oracle surface) ──────────────────────────────────────────────
+
+var ProbeStatuses = []string{"GROUNDED", "STILL_UNKNOWN", "CONTRADICTED"}
+var ProbeFields = []string{"question", "source", "answer", "implication", "status"}
+
+// MetaFields — to_dict's bookkeeping keys; excluded from the extra-field gate.
+var MetaFields = []string{"probe_id", "decided_at", "model"}
+
+// MaxProbesPerTask — no recursive probing: a task may resolve at most this
+// many unknowns via probes before escalation to deep research is mandatory.
 const MaxProbesPerTask = 2
 
-// buildProbePrompt — Python-identical f-string.
-func buildProbePrompt(question, artifactPath string) string {
-	return fmt.Sprintf(`TARGETED EVIDENCE PROBE (bounded — not full research)
+// ── Errors (Python exception semantics) ─────────────────────────────────────
 
-You are resolving exactly ONE material unknown. Do NOT research the wider
-task, do NOT design anything, do NOT modify code.
+// ValueError mirrors the oracle's ValueError raises.
+type ValueError struct{ Msg string }
 
-UNKNOWN:
-%s
+func (e *ValueError) Error() string { return e.Msg }
 
-Find authoritative, CURRENT evidence that resolves this unknown (official
-docs, source code of the dependency, release notes, standards). One or two
-focused lookups are enough — this is a probe, not a survey.
+// ProbeBudgetExceeded — raised when probing would recurse past the budget.
+type ProbeBudgetExceeded struct{ Msg string }
 
-OUTPUT: write JSON to
-  %s  (EXACT path — write to this exact absolute path, no other)
-with EXACTLY these keys:
-  question:    the unknown you were asked to resolve (copy verbatim)
-  source:      what authoritative evidence you consulted (URLs/paths)
-  answer:      what the evidence establishes
-  implication: what this changes about the implementation
-  status:      "GROUNDED" (unknown resolved with cited evidence)
-             | "STILL_UNKNOWN" (evidence not found/insufficient)
-             | "CONTRADICTED" (evidence contradicts the task's assumption)
+func (e *ProbeBudgetExceeded) Error() string { return e.Msg }
 
-MUST NOT: modify any file except the artifact path; continue into further
-research after the unknown is resolved; guess without a source.`, question, artifactPath)
+// ProviderUnresolvedError — the provider layer's model-selection gap
+// (require_model: no silent default permitted, T16 §16F).
+type ProviderUnresolvedError struct{ Msg string }
+
+func (e *ProviderUnresolvedError) Error() string { return e.Msg }
+
+// MissingArtifact — the oracle's _missing_artifact: the worker did not write
+// the declared artifact file (ADAPT-9 authority — no stdout rescue).
+func MissingArtifact() error {
+	return &ValueError{
+		Msg: "probe failed: worker did not write the declared artifact file " +
+			"(ADAPT-9 authority — no stdout rescue for probes)",
+	}
 }
 
-func main() {
-	if len(os.Args) < 2 || os.Args[1] != "replay" {
-		fmt.Fprintln(os.Stderr, "usage: evidence-probe replay <truth.json>")
-		os.Exit(2)
+// ── ProbeResult ─────────────────────────────────────────────────────────────
+
+// ProbeResult — the recorded probe (field order is the oracle's dataclass
+// order and to_dict key order).
+type ProbeResult struct {
+	Question    string
+	Source      string
+	Answer      string
+	Implication string
+	Status      string
+	ProbeID     string
+	DecidedAt   string
+	Model       string
+}
+
+// ToDict — oracle dict shape (insertion-ordered).
+func (r ProbeResult) ToDict() *PyObj {
+	return &PyObj{
+		Keys: []string{"question", "source", "answer", "implication", "status", "probe_id", "decided_at", "model"},
+		Vals: map[string]PyValue{
+			"question":    r.Question,
+			"source":      r.Source,
+			"answer":      r.Answer,
+			"implication": r.Implication,
+			"status":      r.Status,
+			"probe_id":    r.ProbeID,
+			"decided_at":  r.DecidedAt,
+			"model":       r.Model,
+		},
 	}
-	raw, err := os.ReadFile(os.Args[2])
-	if err != nil {
-		fmt.Fprintln(os.Stderr, err)
-		os.Exit(1)
+}
+
+// ── build_probe_prompt ──────────────────────────────────────────────────────
+
+// BuildProbePrompt — bounded prompt: research THE UNKNOWN, write the 5-key
+// JSON file. Byte-identical to the oracle f-string.
+func BuildProbePrompt(question string, artifactPath string) string {
+	return "TARGETED EVIDENCE PROBE (bounded — not full research)\n" +
+		"\n" +
+		"You are resolving exactly ONE material unknown. Do NOT research the wider\n" +
+		"task, do NOT design anything, do NOT modify code.\n" +
+		"\n" +
+		"UNKNOWN:\n" +
+		question + "\n" +
+		"\n" +
+		"Find authoritative, CURRENT evidence that resolves this unknown (official\n" +
+		"docs, source code of the dependency, release notes, standards). One or two\n" +
+		"focused lookups are enough — this is a probe, not a survey.\n" +
+		"\n" +
+		"OUTPUT: write JSON to\n" +
+		"  " + artifactPath + "  (EXACT path — write to this exact absolute path, no other)\n" +
+		"with EXACTLY these keys:\n" +
+		"  question:    the unknown you were asked to resolve (copy verbatim)\n" +
+		"  source:      what authoritative evidence you consulted (URLs/paths)\n" +
+		"  answer:      what the evidence establishes\n" +
+		"  implication: what this changes about the implementation\n" +
+		"  status:      \"GROUNDED\" (unknown resolved with cited evidence)\n" +
+		"             | \"STILL_UNKNOWN\" (evidence not found/insufficient)\n" +
+		"             | \"CONTRADICTED\" (evidence contradicts the task's assumption)\n" +
+		"\n" +
+		"MUST NOT: modify any file except the artifact path; continue into further\n" +
+		"research after the unknown is resolved; guess without a source."
+}
+
+// ── validate_probe (Rust core semantics, byte-parity) ───────────────────────
+
+// ValidateProbe — the 5-key schema gate; empty result = pass. Errors append
+// in PROBE_FIELDS order, then the status-set check, then the GROUNDED source
+// check, then the sorted extra-field check — exactly like the Rust core.
+func ValidateProbe(artifact PyValue) []string {
+	obj, ok := artifact.(*PyObj)
+	if !ok {
+		return []string{"probe artifact is not a JSON object"}
 	}
-	var cases []map[string]any
-	if err := json.Unmarshal(raw, &cases); err != nil {
-		fmt.Fprintln(os.Stderr, err)
-		os.Exit(1)
+
+	var errs []string
+	for _, f := range ProbeFields {
+		v, present := obj.Get(f)
+		switch {
+		case !present:
+			errs = append(errs, fmt.Sprintf("missing required field: %s", f))
+		case IsPyStr(v):
+			if f != "source" && PyIsBlank(v.(string)) {
+				errs = append(errs, fmt.Sprintf("field %s must be a non-empty string", f))
+			}
+		default:
+			errs = append(errs, fmt.Sprintf("field %s must be a string", f))
+		}
 	}
-	fails := 0
-	for i, c := range cases {
-		op, _ := c["op"].(string)
-		switch op {
-		case "prompt":
-			q, _ := c["question"].(string)
-			ap, _ := c["artifact_path"].(string)
-			want, _ := c["want"].(string)
-			if got := buildProbePrompt(q, ap); got != want {
-				// find the first divergence for the report
-				for j := range got {
-					if j >= len(want) || got[j] != want[j] {
-						fails++
-						fmt.Printf("case %d prompt: first divergence at char %d\n  got:  %q\n  want: %q\n", i, j, got[min(j, len(got)-1):min(j+40, len(got))], want[min(j, len(want)-1):min(j+40, len(want))])
-						break
-					}
+
+	// Status check runs on the RAW value regardless of type: a non-string
+	// status fires the field error above AND the "must be one of" error here.
+	if statusVal, present := obj.Get("status"); present {
+		statusOK := false
+		if s, isStr := statusVal.(string); isStr {
+			for _, ok := range ProbeStatuses {
+				if s == ok {
+					statusOK = true
+					break
 				}
 			}
-		case "max_probes":
-			want := int(c["want"].(float64))
-			if MaxProbesPerTask != want {
-				fails++
-				fmt.Printf("case %d max_probes: got %d want %d\n", i, MaxProbesPerTask, want)
+		}
+		if !statusOK {
+			sorted := append([]string(nil), ProbeStatuses...)
+			sort.Strings(sorted)
+			var rendered string
+			switch v := statusVal.(type) {
+			case string:
+				rendered = PyRepr(v)
+			case json.Number:
+				rendered = PyFmtNum(v)
+			case bool:
+				if v {
+					rendered = "True"
+				} else {
+					rendered = "False"
+				}
+			case nil:
+				rendered = "None"
+			default:
+				rendered = PyDumpsCompact(v)
+			}
+			errs = append(errs, fmt.Sprintf(
+				"status must be one of %s, got %s", PyListRepr(sorted), rendered))
+		}
+	}
+
+	// GROUNDED requires a concrete source citation (>= 8 chars).
+	if s, present := obj.Get("status"); present {
+		if str, isStr := s.(string); isStr && str == "GROUNDED" {
+			sourceLen := 0
+			if sv, present := obj.Get("source"); present {
+				if ss, isStr := sv.(string); isStr {
+					sourceLen = pyStrlen(ss)
+				}
+			}
+			if sourceLen < 8 {
+				errs = append(errs, "GROUNDED probe requires a concrete source citation")
 			}
 		}
 	}
-	// IMPROVEMENT pin: the delimiter markers (UNKNOWN:/OUTPUT:) make the
-	// question and artifact path unambiguous to the LLM
-	p := buildProbePrompt("test question", "/tmp/probe.json")
-	if !strings.Contains(p, "test question") || !strings.Contains(p, "/tmp/probe.json") {
-		fails++
-		fmt.Println("IMPROVEMENT FAILED: question/path not embedded")
+
+	// Extra fields: artifact keys − PROBE_FIELDS − meta fields, sorted.
+	var extra []string
+	inList := func(list []string, k string) bool {
+		for _, x := range list {
+			if x == k {
+				return true
+			}
+		}
+		return false
 	}
-	if fails > 0 {
-		fmt.Fprintf(os.Stderr, "PARITY FAILED: %d divergence(s)\n", fails)
-		os.Exit(1)
+	for _, k := range obj.Keys {
+		if !inList(ProbeFields, k) && !inList(MetaFields, k) {
+			extra = append(extra, k)
+		}
 	}
-	fmt.Printf("PARITY OK: %d cases vs the Python oracle\n", len(cases))
+	sort.Strings(extra)
+	if len(extra) > 0 {
+		errs = append(errs, fmt.Sprintf("unexpected extra fields: %s", PyListRepr(extra)))
+	}
+
+	return errs
 }
 
-func min(a, b int) int {
-	if a < b {
-		return a
+// IsPyStr reports whether the value is a JSON string.
+func IsPyStr(v PyValue) bool {
+	_, ok := v.(string)
+	return ok
+}
+
+// ── count_probes / check_budget ─────────────────────────────────────────────
+
+// CountProbes — non-empty-line count of <dir>/probe.jsonl (0 if absent).
+// Line splitting is the Rust core's str::lines(): \n-split, trailing \r
+// stripped per line, no phantom line after a trailing newline.
+func CountProbes(dir string) int {
+	raw, err := os.ReadFile(filepath.Join(dir, "probe.jsonl"))
+	if err != nil {
+		return 0
 	}
-	return b
+	content := string(raw)
+	if content == "" {
+		return 0
+	}
+	lines := strings.Split(content, "\n")
+	if len(lines) > 0 && lines[len(lines)-1] == "" {
+		lines = lines[:len(lines)-1]
+	}
+	n := 0
+	for _, line := range lines {
+		line = strings.TrimSuffix(line, "\r")
+		if strings.TrimSpace(line) != "" {
+			n++
+		}
+	}
+	return n
+}
+
+// CheckBudget — ProbeBudgetExceeded once the no-recursion budget is spent.
+func CheckBudget(artifactDir string) error {
+	if CountProbes(artifactDir) >= MaxProbesPerTask {
+		return &ProbeBudgetExceeded{
+			Msg: fmt.Sprintf("probe budget exhausted (%d); unresolved unknowns must escalate to deep research", MaxProbesPerTask),
+		}
+	}
+	return nil
+}
+
+// ── record_probe ────────────────────────────────────────────────────────────
+
+// RecordProbe — persist the probe: probe.json (indent=2 artifact) and
+// probe.jsonl (append-only compact line). Returns the artifact path.
+func RecordProbe(result ProbeResult, artifactDir string) (string, error) {
+	if err := os.MkdirAll(artifactDir, 0o777); err != nil {
+		return "", err
+	}
+	doc := PyDumpsIndent2(result.ToDict())
+	if err := os.WriteFile(filepath.Join(artifactDir, "probe.json"), []byte(doc), 0o666); err != nil {
+		return "", err
+	}
+	line := PyDumpsCompact(result.ToDict()) + "\n"
+	f, err := os.OpenFile(filepath.Join(artifactDir, "probe.jsonl"), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o666)
+	if err != nil {
+		return "", err
+	}
+	if _, err := f.WriteString(line); err != nil {
+		f.Close()
+		return "", err
+	}
+	if err := f.Close(); err != nil {
+		return "", err
+	}
+	return artifactDir + "/probe.json", nil
+}
+
+// ── run_probe ───────────────────────────────────────────────────────────────
+
+// Runner — the probe worker seam: receives the bounded prompt and the exact
+// absolute artifact path it must write (ADAPT-9: the file is the authority).
+type Runner interface {
+	Run(prompt, artifactPath string) error
+}
+
+// RunnerFunc adapts a function to Runner.
+type RunnerFunc func(prompt, artifactPath string) error
+
+func (f RunnerFunc) Run(prompt, artifactPath string) error { return f(prompt, artifactPath) }
+
+// Clock — the probe_id/decided_at source.
+type Clock interface{ Now() time.Time }
+
+// ClockFunc adapts a function to Clock.
+type ClockFunc func() time.Time
+
+func (f ClockFunc) Now() time.Time { return f() }
+
+type wallClock struct{}
+
+func (wallClock) Now() time.Time { return time.Now() }
+
+// ClockNow — the active clock (overridable in tests/battery).
+var ClockNow Clock = wallClock{}
+
+// RunProbe — one bounded probe. runner == nil selects the default worker
+// channel (DefaultRunner). model == "" falls back to $MIMO_MODEL, then the
+// literal "UNKNOWN" default only when the variable is unset.
+func RunProbe(question string, artifactDir string, runner Runner, model string) (ProbeResult, error) {
+	if err := CheckBudget(artifactDir); err != nil {
+		return ProbeResult{}, err
+	}
+	artifactPath, err := filepath.Abs(filepath.Join(artifactDir, "evidence_probe.json"))
+	if err != nil {
+		return ProbeResult{}, err
+	}
+
+	if runner == nil {
+		runner = RunnerFunc(DefaultRunner)
+	}
+	if err := runner.Run(BuildProbePrompt(question, artifactPath), artifactPath); err != nil {
+		return ProbeResult{}, err
+	}
+
+	if _, err := os.Stat(artifactPath); err != nil {
+		return ProbeResult{}, MissingArtifact()
+	}
+
+	raw, err := os.ReadFile(artifactPath)
+	if err != nil {
+		return ProbeResult{}, err
+	}
+	artifact, err := PyParse(raw)
+	if err != nil {
+		return ProbeResult{}, err
+	}
+
+	if errs := ValidateProbe(artifact); len(errs) > 0 {
+		return ProbeResult{}, &ValueError{Msg: "probe gate failed: " + strings.Join(errs, "; ")}
+	}
+
+	obj := artifact.(*PyObj)
+	getStr := func(key string) string {
+		if v, ok := obj.Get(key); ok {
+			if s, isStr := v.(string); isStr {
+				return s
+			}
+		}
+		return ""
+	}
+
+	resolvedModel := model
+	if resolvedModel == "" {
+		if env, ok := os.LookupEnv("MIMO_MODEL"); ok {
+			resolvedModel = env
+		} else {
+			resolvedModel = "UNKNOWN"
+		}
+	}
+
+	now := ClockNow.Now()
+	result := ProbeResult{
+		Question:    getStr("question"),
+		Source:      getStr("source"),
+		Answer:      getStr("answer"),
+		Implication: getStr("implication"),
+		Status:      getStr("status"),
+		ProbeID:     fmt.Sprintf("probe-%d", now.Unix()),
+		DecidedAt:   now.UTC().Format("2006-01-02T15:04:05Z"),
+		Model:       resolvedModel,
+	}
+	if _, err := RecordProbe(result, artifactDir); err != nil {
+		return ProbeResult{}, err
+	}
+	return result, nil
+}
+
+// ── _default_runner (the worker subprocess channel) ─────────────────────────
+
+// ModelSource — the provider layer owns model selection (require_model);
+// a missing value is a configuration gap, never papered over.
+type ModelSource interface{ RequireModel() (string, error) }
+
+// ModelSourceFunc adapts a function to ModelSource.
+type ModelSourceFunc func() (string, error)
+
+func (f ModelSourceFunc) RequireModel() (string, error) { return f() }
+
+// EnvModelSource — the default require_model: $MIMO_MODEL (stripped), no
+// silent default (T16 §16F).
+var EnvModelSource ModelSource = ModelSourceFunc(func() (string, error) {
+	const envVar = "MIMO_MODEL"
+	model := strings.TrimSpace(os.Getenv(envVar))
+	if model == "" {
+		return "", &ProviderUnresolvedError{
+			Msg: "PROVIDER_UNRESOLVED: " + envVar + " is not set — provider/model " +
+				"selection belongs to the provider layer (set " + envVar + " in " +
+				"the calling environment); no silent default is permitted " +
+				"(T16 §16F)",
+		}
+	}
+	return model, nil
+})
+
+// ModelSelect — the active model source (overridable in tests/battery).
+var ModelSelect = EnvModelSource
+
+// CmdRunner — the subprocess exec seam.
+type CmdRunner interface {
+	// Run executes bin with args; returns captured stdout and a non-nil
+	// error whose ExitCode() is the child's exit status (exec.ExitError).
+	Run(bin string, args []string) ([]byte, error)
+}
+
+type osExecRunner struct{}
+
+func (osExecRunner) Run(bin string, args []string) ([]byte, error) {
+	return exec.Command(bin, args...).Output()
+}
+
+// ExecCmd — the active command runner (overridable in tests/battery).
+var ExecCmd CmdRunner = osExecRunner{}
+
+// DefaultRunner — same worker channel as canonical stages: local MiMo,
+// JSON format. stdout is observability only; the artifact file is the
+// authority.
+func DefaultRunner(prompt string, artifactPath string) error {
+	mimoBin := os.Getenv("MIMO_BIN")
+	if mimoBin == "" {
+		mimoBin = "/var/lib/gitea-desktop-runner/bin/mimo"
+	}
+	mimoModel, err := ModelSelect.RequireModel()
+	if err != nil {
+		return err
+	}
+	mimoAgent := os.Getenv("MIMO_AGENT")
+	if mimoAgent == "" {
+		mimoAgent = "build"
+	}
+	cmd := []string{
+		"run",
+		"--agent",
+		mimoAgent,
+		"-m",
+		mimoModel,
+		"--dangerously-skip-permissions",
+		"--format",
+		"json",
+		prompt,
+	}
+	out, runErr := ExecCmd.Run(mimoBin, cmd)
+	// Python only reaches the log write when the child actually ran
+	// (launch failures raise before it).
+	var exitErr *exec.ExitError
+	launched := runErr == nil || errors.As(runErr, &exitErr)
+	if launched {
+		logPath := filepath.Dir(artifactPath) + "/evidence_probe.stdout.log"
+		if writeErr := os.WriteFile(logPath, out, 0o666); writeErr != nil && runErr == nil {
+			return writeErr
+		}
+	}
+	if runErr != nil {
+		if launched {
+			return &ValueError{Msg: fmt.Sprintf("probe worker exited %d", exitErr.ExitCode())}
+		}
+		return runErr
+	}
+	return nil
+}
+
+func main() {
+	if len(os.Args) < 3 || os.Args[1] != "replay" {
+		fmt.Fprintln(os.Stderr, "usage: evidence-probe replay <truth.json>")
+		os.Exit(2)
+	}
+	if fails := RunTruth(os.Args[2]); len(fails) > 0 {
+		for _, f := range fails {
+			fmt.Println("DIVERGED:", f)
+		}
+		fmt.Fprintf(os.Stderr, "PARITY FAILED: %d divergence(s)\n", len(fails))
+		os.Exit(1)
+	}
+	fmt.Println("PARITY OK: frozen oracle battery replayed verbatim")
 }

src/go/evidence-probe/pyjson.go

@@ -0,0 +1,301 @@
+// pyjson.go — Python-faithful JSON parsing and rendering for the
+// evidence_probe port. The oracle round-trips artifacts through
+// json.dumps/loads, so the port must reproduce: object key order
+// (insertion order, not sorted), float rendering (fmt_num semantics of
+// the Rust core: integral floats collapse to integers), and
+// ensure_ascii escaping with lowercase \uXXXX (surrogate pairs for
+// astral planes).
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"math"
+	"strconv"
+	"strings"
+	"unicode"
+	"unicode/utf8"
+)
+
+// PyValue — a JSON value with order-preserving objects.
+// nil | bool | json.Number | string | []PyValue | *PyObj
+type PyValue any
+
+// PyObj — a JSON object preserving insertion order.
+type PyObj struct {
+	Keys []string
+	Vals map[string]PyValue
+}
+
+func (o *PyObj) Get(key string) (PyValue, bool) {
+	if o == nil {
+		return nil, false
+	}
+	v, ok := o.Vals[key]
+	return v, ok
+}
+
+// PyParse — strict single-value JSON parse (trailing garbage rejected,
+// like Python json.loads / serde_json).
+func PyParse(raw []byte) (PyValue, error) {
+	dec := json.NewDecoder(bytes.NewReader(raw))
+	dec.UseNumber()
+	v, err := decodeValue(dec)
+	if err != nil {
+		return nil, err
+	}
+	if _, err := dec.Token(); err != io.EOF {
+		return nil, fmt.Errorf("trailing data after JSON value")
+	}
+	return v, nil
+}
+
+func decodeValue(dec *json.Decoder) (PyValue, error) {
+	tok, err := dec.Token()
+	if err != nil {
+		return nil, err
+	}
+	return decodeFromToken(dec, tok)
+}
+
+func decodeFromToken(dec *json.Decoder, tok json.Token) (PyValue, error) {
+	switch t := tok.(type) {
+	case json.Delim:
+		switch t {
+		case '{':
+			obj := &PyObj{Vals: map[string]PyValue{}}
+			for dec.More() {
+				keyTok, err := dec.Token()
+				if err != nil {
+					return nil, err
+				}
+				key, ok := keyTok.(string)
+				if !ok {
+					return nil, fmt.Errorf("object key is not a string")
+				}
+				val, err := decodeValue(dec)
+				if err != nil {
+					return nil, err
+				}
+				if _, dup := obj.Vals[key]; !dup {
+					obj.Keys = append(obj.Keys, key)
+				}
+				obj.Vals[key] = val
+			}
+			if _, err := dec.Token(); err != nil { // closing '}'
+				return nil, err
+			}
+			return obj, nil
+		case '[':
+			arr := []PyValue{}
+			for dec.More() {
+				val, err := decodeValue(dec)
+				if err != nil {
+					return nil, err
+				}
+				arr = append(arr, val)
+			}
+			if _, err := dec.Token(); err != nil { // closing ']'
+				return nil, err
+			}
+			return arr, nil
+		default:
+			return nil, fmt.Errorf("unexpected delim %q", t)
+		}
+	case json.Number:
+		return t, nil
+	case string:
+		return t, nil
+	case bool:
+		return t, nil
+	case nil:
+		return nil, nil
+	default:
+		return nil, fmt.Errorf("unexpected token %v", tok)
+	}
+}
+
+// PyDumpsCompact — Python json.dumps(value): separators ", " / ": ",
+// ensure_ascii=True, insertion-order objects.
+func PyDumpsCompact(v PyValue) string {
+	var b strings.Builder
+	writePy(&b, v, "", false)
+	return b.String()
+}
+
+// PyDumpsIndent2 — Python json.dump(value, indent=2): separators ',' / ': '
+// with 2-space indentation, ensure_ascii=True.
+func PyDumpsIndent2(v PyValue) string {
+	var b strings.Builder
+	writePy(&b, v, "", true)
+	return b.String()
+}
+
+func writePy(b *strings.Builder, v PyValue, indent string, indented bool) {
+	switch t := v.(type) {
+	case nil:
+		b.WriteString("null")
+	case bool:
+		if t {
+			b.WriteString("true")
+		} else {
+			b.WriteString("false")
+		}
+	case json.Number:
+		b.WriteString(PyFmtNum(t))
+	case string:
+		writePyString(b, t)
+	case []PyValue:
+		if len(t) == 0 {
+			b.WriteString("[]")
+			return
+		}
+		b.WriteByte('[')
+		inner := indent
+		if indented {
+			inner = indent + "  "
+		}
+		for i, item := range t {
+			if indented {
+				if i > 0 {
+					b.WriteString(",")
+				}
+				b.WriteString("\n")
+				b.WriteString(inner)
+			} else if i > 0 {
+				b.WriteString(", ")
+			}
+			writePy(b, item, inner, indented)
+		}
+		if indented {
+			b.WriteString("\n")
+			b.WriteString(indent)
+		}
+		b.WriteByte(']')
+	case *PyObj:
+		if len(t.Keys) == 0 {
+			b.WriteString("{}")
+			return
+		}
+		b.WriteByte('{')
+		inner := indent
+		if indented {
+			inner = indent + "  "
+		}
+		for i, k := range t.Keys {
+			if indented {
+				if i > 0 {
+					b.WriteString(",")
+				}
+				b.WriteString("\n")
+				b.WriteString(inner)
+			} else if i > 0 {
+				b.WriteString(", ")
+			}
+			writePyString(b, k)
+			b.WriteString(": ")
+			writePy(b, t.Vals[k], inner, indented)
+		}
+		if indented {
+			b.WriteString("\n")
+			b.WriteString(indent)
+		}
+		b.WriteByte('}')
+	default:
+		b.WriteString(fmt.Sprintf("%v", v))
+	}
+}
+
+// PyFmtNum — the Rust core's fmt_num: integral floats render as integers,
+// everything else via shortest fixed-notation round-trip (Rust f64 Display
+// semantics — never scientific notation).
+func PyFmtNum(n json.Number) string {
+	f, err := n.Float64()
+	if err != nil {
+		return n.String()
+	}
+	if !math.IsInf(f, 0) && !math.IsNaN(f) && f == math.Trunc(f) &&
+		math.Abs(f) < 9.007199254740992e15 {
+		return strconv.FormatInt(int64(f), 10)
+	}
+	return strconv.FormatFloat(f, 'f', -1, 64)
+}
+
+// writePyString — Python json.dumps string escaping, ensure_ascii=True.
+func writePyString(b *strings.Builder, s string) {
+	b.WriteByte('"')
+	for _, c := range s {
+		switch c {
+		case '"':
+			b.WriteString("\\\"")
+		case '\\':
+			b.WriteString("\\\\")
+		case '\n':
+			b.WriteString("\\n")
+		case '\r':
+			b.WriteString("\\r")
+		case '\t':
+			b.WriteString("\\t")
+		case '\b':
+			b.WriteString("\\b")
+		case '\f':
+			b.WriteString("\\f")
+		default:
+			if c < 0x20 {
+				fmt.Fprintf(b, "\\u%04x", c)
+			} else if c <= 0x7E {
+				b.WriteRune(c)
+			} else if cp := uint32(c); cp > 0xFFFF {
+				v := cp - 0x10000
+				fmt.Fprintf(b, "\\u%04x\\u%04x", 0xD800+(v>>10), 0xDC00+(v&0x3FF))
+			} else {
+				fmt.Fprintf(b, "\\u%04x", cp)
+			}
+		}
+	}
+	b.WriteByte('"')
+}
+
+// PyRepr — Python repr() of a string: single quotes, backslash/quote
+// escaping, \n \r \t control renders (the shapes a status/extra key takes).
+func PyRepr(s string) string {
+	var b strings.Builder
+	b.WriteByte('\'')
+	for _, c := range s {
+		switch c {
+		case '\\':
+			b.WriteString("\\\\")
+		case '\'':
+			b.WriteString("\\'")
+		case '\n':
+			b.WriteString("\\n")
+		case '\r':
+			b.WriteString("\\r")
+		case '\t':
+			b.WriteString("\\t")
+		default:
+			b.WriteRune(c)
+		}
+	}
+	b.WriteByte('\'')
+	return b.String()
+}
+
+// PyListRepr — Python repr() of a list of strings: ['a', 'b'].
+func PyListRepr(items []string) string {
+	parts := make([]string, len(items))
+	for i, s := range items {
+		parts[i] = PyRepr(s)
+	}
+	return "[" + strings.Join(parts, ", ") + "]"
+}
+
+// PyIsBlank — Python str.strip() emptiness (any non-whitespace = content).
+func PyIsBlank(s string) bool {
+	return strings.TrimFunc(s, unicode.IsSpace) == ""
+}
+
+// pyStrlen — Python len() of a str: Unicode scalar (rune) count.
+func pyStrlen(s string) int { return utf8.RuneCountInString(s) }

src/go/evidence-probe/replay.go

@@ -0,0 +1,478 @@
+// Replay — the frozen-truth parity engine. Reads truth.json (generated by
+// scripts/gen_evidence_probe_truth.py against the live Python oracle with
+// the clock frozen, the schema gate served by the Rust core, the runner and
+// subprocess seams stubbed) and re-executes every row through this Go port.
+// Prompt/error/file bytes compare exactly; @ARTDIR@ normalizes the temp
+// base dir on both sides.
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"reflect"
+	"strings"
+	"time"
+)
+
+// TruthDoc — the frozen truth file structure.
+type TruthDoc struct {
+	Oracle          string     `json:"oracle"`
+	FrozenTS        int64      `json:"frozen_ts"`
+	FrozenDecidedAt string     `json:"frozen_decided_at"`
+	Rows            []TruthRow `json:"rows"`
+}
+
+// TruthRow — one frozen case.
+type TruthRow struct {
+	Name  string          `json:"name"`
+	Op    string          `json:"op"`
+	Setup json.RawMessage `json:"setup"`
+	Want  json.RawMessage `json:"want"`
+}
+
+// RunTruth replays every row in its own temp dir and returns a divergence
+// report (empty on 100% parity).
+func RunTruth(path string) []string {
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		return []string{fmt.Sprintf("read truth: %v", err)}
+	}
+	var doc TruthDoc
+	if err := json.Unmarshal(raw, &doc); err != nil {
+		return []string{fmt.Sprintf("parse truth: %v", err)}
+	}
+	var fails []string
+	for _, row := range doc.Rows {
+		if msg := replayRow(&doc, &row); msg != "" {
+			fails = append(fails, row.Name+": "+msg)
+		}
+	}
+	return fails
+}
+
+func replayRow(doc *TruthDoc, row *TruthRow) string {
+	base, err := os.MkdirTemp("", "ep_truth_")
+	if err != nil {
+		return fmt.Sprintf("temp dir: %v", err)
+	}
+	defer os.RemoveAll(base)
+
+	oldClock := ClockNow
+	oldModel := ModelSelect
+	oldExec := ExecCmd
+	oldEnv := snapshotEnv([]string{"MIMO_BIN", "MIMO_AGENT", "MIMO_MODEL", "EP_CAPTURE", "EP_EXIT"})
+	defer func() {
+		ClockNow, ModelSelect, ExecCmd = oldClock, oldModel, oldExec
+		restoreEnv(oldEnv)
+	}()
+	ClockNow = ClockFunc(func() time.Time { return time.Unix(doc.FrozenTS, 0) })
+
+	var got any
+	switch row.Op {
+	case "prompt":
+		got, err = replayPrompt(row)
+	case "validate":
+		got, err = replayValidate(row)
+	case "count":
+		got, err = replayCount(row, base)
+	case "budget":
+		got, err = replayBudget(row, base)
+	case "record":
+		got, err = replayRecord(row, base)
+	case "run":
+		got, err = replayRun(row, base)
+	case "to_dict":
+		got, err = replayToDict(row)
+	case "default_runner":
+		got, err = replayDefaultRunner(row, base)
+	default:
+		return "unknown op " + row.Op
+	}
+	if err != nil {
+		return "error: " + err.Error()
+	}
+
+	var want any
+	if err := json.Unmarshal(row.Want, &want); err != nil {
+		return "parse want: " + err.Error()
+	}
+	if !reflect.DeepEqual(normalizeJSON(got), normalizeJSON(want)) {
+		return fmt.Sprintf("got %#v\n  want %#v", got, want)
+	}
+	return ""
+}
+
+// normalizeJSON — round-trip through map[string]any/float64 so got and want
+// numbers compare by value, then stringify json.Number.
+func normalizeJSON(v any) any {
+	switch t := v.(type) {
+	case int:
+		return float64(t)
+	case json.Number:
+		f, err := t.Float64()
+		if err != nil {
+			return t.String()
+		}
+		return f
+	case map[string]any:
+		m := map[string]any{}
+		for k, val := range t {
+			m[k] = normalizeJSON(val)
+		}
+		return m
+	case []any:
+		s := make([]any, len(t))
+		for i, val := range t {
+			s[i] = normalizeJSON(val)
+		}
+		return s
+	default:
+		return v
+	}
+}
+
+func denorm(s, base string) string { return strings.ReplaceAll(s, "@ARTDIR@", base) }
+
+// normOut — normalize a Go-side output path/string for comparison with the
+// frozen truth (the generator replaced its temp base with @ARTDIR@).
+func normOut(s, base string) string { return strings.ReplaceAll(s, base, "@ARTDIR@") }
+
+func snapshotEnv(keys []string) map[string]*string {
+	out := map[string]*string{}
+	for _, k := range keys {
+		if v, ok := os.LookupEnv(k); ok {
+			cp := v
+			out[k] = &cp
+		} else {
+			out[k] = nil
+		}
+	}
+	return out
+}
+
+func restoreEnv(snap map[string]*string) {
+	for k, v := range snap {
+		if v == nil {
+			os.Unsetenv(k)
+		} else {
+			os.Setenv(k, *v)
+		}
+	}
+}
+
+// ── ops ─────────────────────────────────────────────────────────────────────
+
+func replayPrompt(row *TruthRow) (any, error) {
+	var setup struct {
+		Question     string `json:"question"`
+		ArtifactPath string `json:"artifact_path"`
+	}
+	if err := json.Unmarshal(row.Setup, &setup); err != nil {
+		return nil, err
+	}
+	return BuildProbePrompt(setup.Question, setup.ArtifactPath), nil
+}
+
+func replayValidate(row *TruthRow) (any, error) {
+	var setup struct {
+		Artifact json.RawMessage `json:"artifact"`
+	}
+	if err := json.Unmarshal(row.Setup, &setup); err != nil {
+		return nil, err
+	}
+	artifact, err := PyParse(setup.Artifact)
+	if err != nil {
+		return nil, fmt.Errorf("parse artifact: %w", err)
+	}
+	errs := ValidateProbe(artifact)
+	out := make([]any, len(errs))
+	for i, e := range errs {
+		out[i] = e
+	}
+	return out, nil
+}
+
+func replayCount(row *TruthRow, base string) (any, error) {
+	var setup struct {
+		Dir   string            `json:"dir"`
+		Files map[string]string `json:"files"`
+	}
+	if err := json.Unmarshal(row.Setup, &setup); err != nil {
+		return nil, err
+	}
+	dir := denorm(setup.Dir, base)
+	if err := writeTree(dir, setup.Files); err != nil {
+		return nil, err
+	}
+	return CountProbes(dir), nil
+}
+
+func replayBudget(row *TruthRow, base string) (any, error) {
+	var setup struct {
+		Dir   string            `json:"dir"`
+		Files map[string]string `json:"files"`
+	}
+	if err := json.Unmarshal(row.Setup, &setup); err != nil {
+		return nil, err
+	}
+	dir := denorm(setup.Dir, base)
+	if err := writeTree(dir, setup.Files); err != nil {
+		return nil, err
+	}
+	err := CheckBudget(dir)
+	if err == nil {
+		// exc_row froze {"error": null, "value": null} on success.
+		return map[string]any{"error": nil, "value": nil}, nil
+	}
+	return map[string]any{"error": errorValue(err)}, nil
+}
+
+// errorValue — the frozen exception shape; typed nil must stay untyped nil
+// so DeepEqual matches the JSON null on the want side.
+func errorValue(err error) any {
+	if err == nil {
+		return nil
+	}
+	msg := err.Error()
+	typ := fmt.Sprintf("%T", err)
+	switch t := err.(type) {
+	case *ValueError:
+		typ, msg = "ValueError", t.Msg
+	case *ProbeBudgetExceeded:
+		typ, msg = "ProbeBudgetExceeded", t.Msg
+	case *ProviderUnresolvedError:
+		typ, msg = "ProviderUnresolved", t.Msg
+	}
+	return map[string]any{"type": typ, "msg": msg}
+}
+
+func replayRecord(row *TruthRow, base string) (any, error) {
+	var setup struct {
+		Result map[string]string `json:"result"`
+		Files  map[string]string `json:"files"`
+	}
+	if err := json.Unmarshal(row.Setup, &setup); err != nil {
+		return nil, err
+	}
+	dir := denorm("@ARTDIR@/"+row.Name, base)
+	if err := writeTree(dir, setup.Files); err != nil {
+		return nil, err
+	}
+	result := ProbeResult{
+		Question:    setup.Result["question"],
+		Source:      setup.Result["source"],
+		Answer:      setup.Result["answer"],
+		Implication: setup.Result["implication"],
+		Status:      setup.Result["status"],
+		ProbeID:     setup.Result["probe_id"],
+		DecidedAt:   setup.Result["decided_at"],
+		Model:       setup.Result["model"],
+	}
+	path, err := RecordProbe(result, dir)
+	if err != nil {
+		return nil, err
+	}
+	pj, err := os.ReadFile(filepath.Join(dir, "probe.json"))
+	if err != nil {
+		return nil, err
+	}
+	pjl, err := os.ReadFile(filepath.Join(dir, "probe.jsonl"))
+	if err != nil {
+		return nil, err
+	}
+	return map[string]any{
+		"path":        normOut(path, base),
+		"probe_json":  string(pj),
+		"probe_jsonl": string(pjl),
+	}, nil
+}
+
+func writeTree(dir string, files map[string]string) error {
+	for rel, content := range files {
+		p := filepath.Join(dir, filepath.FromSlash(rel))
+		if err := os.MkdirAll(filepath.Dir(p), 0o777); err != nil {
+			return err
+		}
+		if err := os.WriteFile(p, []byte(content), 0o666); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func replayRun(row *TruthRow, base string) (any, error) {
+	var setup struct {
+		Question string            `json:"question"`
+		Model    string            `json:"model"`
+		Env      map[string]any    `json:"env"`
+		Files    map[string]string `json:"files"`
+		Runner   struct {
+			Type     string          `json:"type"`
+			Artifact json.RawMessage `json:"artifact"`
+		} `json:"runner"`
+	}
+	if err := json.Unmarshal(row.Setup, &setup); err != nil {
+		return nil, err
+	}
+	for k, v := range setup.Env {
+		if v == nil {
+			os.Unsetenv(k)
+		} else if s, ok := v.(string); ok {
+			os.Setenv(k, s)
+		}
+	}
+	dir := denorm("@ARTDIR@/"+row.Name, base)
+	if err := os.MkdirAll(dir, 0o777); err != nil {
+		return nil, err
+	}
+	if err := writeTree(dir, setup.Files); err != nil {
+		return nil, err
+	}
+
+	captured := struct {
+		prompt, artifactPath string
+		called               bool
+	}{}
+	runner := RunnerFunc(func(prompt, artifactPath string) error {
+		captured.prompt, captured.artifactPath, captured.called = prompt, artifactPath, true
+		switch setup.Runner.Type {
+		case "write":
+			artifact, err := PyParse(setup.Runner.Artifact)
+			if err != nil {
+				return err
+			}
+			return os.WriteFile(artifactPath, []byte(PyDumpsCompact(artifact)), 0o666)
+		case "silent":
+			return nil
+		default:
+			return fmt.Errorf("unknown runner type %q", setup.Runner.Type)
+		}
+	})
+
+	result, err := RunProbe(setup.Question, dir, runner, setup.Model)
+	out := map[string]any{"error": errorValue(err)}
+	if err == nil {
+		var resultJSON any
+		if uerr := json.Unmarshal([]byte(PyDumpsCompact(result.ToDict())), &resultJSON); uerr != nil {
+			return nil, uerr
+		}
+		out["result"] = resultJSON
+		pj, rerr := os.ReadFile(filepath.Join(dir, "probe.json"))
+		if rerr != nil {
+			return nil, rerr
+		}
+		pjl, rerr := os.ReadFile(filepath.Join(dir, "probe.jsonl"))
+		if rerr != nil {
+			return nil, rerr
+		}
+		out["probe_json"] = string(pj)
+		out["probe_jsonl"] = string(pjl)
+	}
+	out["runner_called"] = captured.called
+	if captured.called {
+		out["prompt"] = normOut(captured.prompt, base)
+		out["artifact_path"] = normOut(captured.artifactPath, base)
+	}
+	return out, nil
+}
+
+func replayToDict(row *TruthRow) (any, error) {
+	var setup struct {
+		Fields map[string]string `json:"fields"`
+	}
+	if err := json.Unmarshal(row.Setup, &setup); err != nil {
+		return nil, err
+	}
+	result := ProbeResult{
+		Question:    setup.Fields["question"],
+		Source:      setup.Fields["source"],
+		Answer:      setup.Fields["answer"],
+		Implication: setup.Fields["implication"],
+		Status:      setup.Fields["status"],
+		ProbeID:     setup.Fields["probe_id"],
+		DecidedAt:   setup.Fields["decided_at"],
+		Model:       setup.Fields["model"],
+	}
+	if _, ok := setup.Fields["model"]; !ok {
+		result.Model = "UNKNOWN" // the oracle dataclass default
+	}
+	return PyDumpsCompact(result.ToDict()), nil
+}
+
+func replayDefaultRunner(row *TruthRow, base string) (any, error) {
+	var setup struct {
+		ExitCode  int     `json:"exit_code"`
+		AgentEnv  *string `json:"agent_env"`
+		ModelMode string  `json:"model_mode"`
+	}
+	if err := json.Unmarshal(row.Setup, &setup); err != nil {
+		return nil, err
+	}
+	stub := filepath.Join(base, "ep_stub.sh")
+	script := "#!/bin/sh\n" +
+		"for a in \"$@\"; do printf '%s\\n' \"$a\" >> \"$EP_CAPTURE\"; done\n" +
+		"printf 'stub-out\\n'\n" +
+		"exit \"$EP_EXIT\"\n"
+	if err := os.WriteFile(stub, []byte(script), 0o755); err != nil {
+		return nil, err
+	}
+	dir := denorm("@ARTDIR@/"+row.Name, base)
+	if err := os.MkdirAll(dir, 0o777); err != nil {
+		return nil, err
+	}
+	capture := filepath.Join(dir, "argv_capture.txt")
+	os.Remove(capture)
+	artifactPath := filepath.Join(dir, "evidence_probe.json")
+
+	os.Setenv("MIMO_BIN", stub)
+	os.Setenv("EP_CAPTURE", capture)
+	os.Setenv("EP_EXIT", fmt.Sprintf("%d", setup.ExitCode))
+	if setup.AgentEnv == nil {
+		os.Unsetenv("MIMO_AGENT")
+	} else {
+		os.Setenv("MIMO_AGENT", *setup.AgentEnv)
+	}
+	switch setup.ModelMode {
+	case "unresolved":
+		os.Unsetenv("MIMO_MODEL")
+		ModelSelect = EnvModelSource
+	case "blank":
+		os.Setenv("MIMO_MODEL", "   ")
+		ModelSelect = EnvModelSource
+	default:
+		ModelSelect = ModelSourceFunc(func() (string, error) { return "test-model-v9", nil })
+	}
+
+	err := DefaultRunner("PROMPT-TEXT", artifactPath)
+	out := map[string]any{"error": errorValue(err)}
+	if lines, rerr := readLines(capture); rerr == nil {
+		norm := make([]any, len(lines))
+		for i, l := range lines {
+			norm[i] = normOut(l, base)
+		}
+		out["argv"] = norm
+	} else {
+		out["argv"] = nil
+	}
+	logPath := filepath.Join(dir, "evidence_probe.stdout.log")
+	if log, rerr := os.ReadFile(logPath); rerr == nil {
+		out["stdout_log"] = string(log)
+	} else {
+		out["stdout_log"] = nil
+	}
+	return out, nil
+}
+
+func readLines(path string) ([]string, error) {
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		return nil, err
+	}
+	content := strings.TrimSuffix(string(raw), "\n")
+	if content == "" {
+		return []string{}, nil
+	}
+	return strings.Split(content, "\n"), nil
+}

src/go/evidence-probe/truth_test.go

@@ -0,0 +1,168 @@
+package main
+
+// Battery — the HARD parity gate: the frozen oracle truth (truth.json,
+// generated against the LIVE Python oracle with the clock frozen and the
+// Rust core serving validate_probe/count_probes) replays VERBATIM. Any
+// divergence fails; the truth file rides the repo so the gate holds in CI
+// without /tmp.
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"reflect"
+	"testing"
+)
+
+func loadTruthDoc(t *testing.T, path string) *TruthDoc {
+	t.Helper()
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("truth.json missing (frozen oracle battery): %v", err)
+	}
+	var doc TruthDoc
+	if err := json.Unmarshal(raw, &doc); err != nil {
+		t.Fatalf("parse truth.json: %v", err)
+	}
+	if len(doc.Rows) == 0 {
+		t.Fatal("truth.json contained no rows")
+	}
+	return &doc
+}
+
+func findRow(t *testing.T, doc *TruthDoc, name string) TruthRow {
+	t.Helper()
+	for _, row := range doc.Rows {
+		if row.Name == name {
+			return row
+		}
+	}
+	t.Fatalf("frozen row %q missing from truth.json — battery/truth drift", name)
+	return TruthRow{}
+}
+
+func mustReplayRow(t *testing.T, doc *TruthDoc, name string) {
+	t.Helper()
+	row := findRow(t, doc, name)
+	if msg := replayRow(doc, &row); msg != "" {
+		t.Errorf("DIVERGED %s: %s", name, msg)
+	}
+}
+
+func TestFullTruthReplayVerbatim(t *testing.T) {
+	doc := loadTruthDoc(t, "truth.json")
+	var fails []string
+	for _, row := range doc.Rows {
+		if msg := replayRow(doc, &row); msg != "" {
+			fails = append(fails, row.Name+": "+msg)
+		}
+	}
+	if len(fails) > 0 {
+		for _, f := range fails {
+			t.Errorf("DIVERGED %s", f)
+		}
+		t.Fatalf("frozen-truth parity: %d divergence(s)", len(fails))
+	}
+}
+
+func TestTruthOpCoverage(t *testing.T) {
+	doc := loadTruthDoc(t, "truth.json")
+	// The frozen battery must keep covering every op family: if a
+	// regeneration drops a family, this fails before parity silently rots.
+	want := map[string]int{
+		"prompt": 5, "validate": 49, "count": 11, "budget": 4,
+		"record": 3, "run": 11, "to_dict": 3, "default_runner": 5,
+	}
+	got := map[string]int{}
+	for _, row := range doc.Rows {
+		got[row.Op]++
+	}
+	if fmt.Sprint(got) != fmt.Sprint(want) {
+		t.Fatalf("truth.json op coverage drifted:\n got  %v\n want %v", got, want)
+	}
+}
+
+func TestPromptBaselineByteParity(t *testing.T) {
+	doc := loadTruthDoc(t, "truth.json")
+	row := findRow(t, doc, "prompt_baseline")
+	got, err := replayPrompt(&row)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var want string
+	if err := json.Unmarshal(row.Want, &want); err != nil {
+		t.Fatal(err)
+	}
+	if got != want {
+		t.Fatalf("prompt drifted from frozen oracle:\n got  %q\n want %q", got, want)
+	}
+}
+
+func TestGateErrorOrderIsFieldOrder(t *testing.T) {
+	doc := loadTruthDoc(t, "truth.json")
+	for _, name := range []string{
+		"combined_ordering", "status_obj_order", "grounded_missing_source",
+		"status_repr_unicode", "status_big_float", "status_huge_int",
+		"valid_grounded", "empty_artifact", "extra_sorted", "nonobject_list",
+	} {
+		mustReplayRow(t, doc, name)
+	}
+}
+
+func TestRunProbeExceptionSemantics(t *testing.T) {
+	doc := loadTruthDoc(t, "truth.json")
+	for _, name := range []string{
+		"run_success_grounded", "run_success_still_unknown",
+		"run_model_from_env", "run_model_unset_unknown",
+		"run_model_env_blank_is_empty_string",
+		"run_budget_exhausted_before_runner", "run_budget_one_left",
+		"run_silent_runner_missing_artifact", "run_gate_failure_joined",
+		"run_gate_failure_non_dict", "run_gate_failure_status_non_string",
+	} {
+		mustReplayRow(t, doc, name)
+	}
+}
+
+func TestDefaultRunnerArgvParity(t *testing.T) {
+	doc := loadTruthDoc(t, "truth.json")
+	for _, name := range []string{
+		"defrun_exit0_default_agent", "defrun_exit3", "defrun_custom_agent",
+		"defrun_model_unresolved", "defrun_model_blank_env",
+	} {
+		mustReplayRow(t, doc, name)
+	}
+}
+
+func TestCountAndBudgetSemantics(t *testing.T) {
+	doc := loadTruthDoc(t, "truth.json")
+	for _, name := range []string{
+		"count_absent_dir", "count_cr_lines", "count_no_trailing_newline",
+		"count_interleaved_blank", "budget_zero", "budget_one",
+		"budget_two", "budget_three",
+	} {
+		mustReplayRow(t, doc, name)
+	}
+}
+
+func TestRecordProbeBytesParity(t *testing.T) {
+	doc := loadTruthDoc(t, "truth.json")
+	for _, name := range []string{
+		"record_fresh_dir_created", "record_appends_existing",
+		"record_unicode_ensure_ascii",
+	} {
+		mustReplayRow(t, doc, name)
+	}
+}
+
+func TestToDictFieldOrderParity(t *testing.T) {
+	doc := loadTruthDoc(t, "truth.json")
+	for _, name := range []string{"to_dict_defaults", "to_dict_full", "to_dict_unicode"} {
+		mustReplayRow(t, doc, name)
+	}
+}
+
+func TestNormalizeJSONIntFloatEquivalence(t *testing.T) {
+	if !reflect.DeepEqual(normalizeJSON(2), normalizeJSON(2.0)) {
+		t.Fatal("int/float normalization broke count comparisons")
+	}
+}