"""Homepage evidence from approved aggregate snapshots; never private run artifacts."""
from __future__ import annotations
from html import escape
import json
import math
from collections.abc import Callable
SFT = "docs/use-cases/german-reasoning-results.json"
AGENT = "docs/use-cases/german-agent-rl-results.json"
GYMS = "docs/agent-gyms-validation.json"
OPENENV = "docs/openenv-validation.json"
def _load(snapshots: dict[str, bytes], path: str) -> dict | None:
def unique(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError("duplicate aggregate key")
value[key] = item
return value
try:
raw = snapshots[path]
if not isinstance(raw, bytes):
return None
value = json.loads(raw.decode("utf-8"), object_pairs_hook=unique,
parse_constant=lambda item: (_ for _ in ()).throw(ValueError(item)))
return value if isinstance(value, dict) else None
except (KeyError, ValueError, UnicodeError, TypeError):
return None
def _count(value, *, positive=False) -> int:
if type(value) is not int or not (int(positive) <= value <= 1_000_000):
raise ValueError("invalid aggregate count")
return value
def _number(value, low=-1.0, high=1.0) -> float:
if type(value) not in (int, float) or not math.isfinite(value) or not low <= value <= high:
raise ValueError("invalid aggregate number")
return float(value)
def _comparison(before, after, paired, wins_key, losses_key, interval):
total = _count(before["total"], positive=True)
baseline, candidate = _count(before["correct"]), _count(after["correct"])
wins, losses = _count(paired[wins_key]), _count(paired[losses_key])
low, high = (_number(value) for value in interval)
if (_count(after["total"], positive=True) != total or max(baseline, candidate, wins, losses) > total
or _count(paired["items"], positive=True) != total
or _count(paired["trials"], positive=True) != total
or candidate - baseline != wins - losses
or baseline < losses or total - baseline < wins or low > high
or not math.isclose(_number(paired["accuracy_delta"]), (candidate - baseline) / total)):
raise ValueError("inconsistent paired aggregate")
for score, correct in ((before, baseline), (after, candidate)):
if not math.isclose(_number(score["strict_final_answer_protocol_accuracy"], 0, 1), correct / total):
raise ValueError("aggregate accuracy disagrees with counts")
if "both_correct" in paired or "both_incorrect" in paired:
both = _count(paired["both_correct"])
neither = _count(paired["both_incorrect"])
if both + losses != baseline or both + wins != candidate or both + neither + wins + losses != total:
raise ValueError("paired coverage disagrees with counts")
return {"before": baseline, "after": candidate, "total": total,
"wins": wins, "losses": losses, "low": low, "high": high}
def _sft(data):
if data["status"] != "completed_research_pilot" or data["release_eligible"] is not False:
raise ValueError("not a completed research pilot")
result = _comparison(data["baseline"], data["candidate"], data["paired"],
"candidate_wins", "candidate_losses", data["paired"]["bootstrap"]["ci_95"])
result["malformed_before"] = _count(data["baseline"]["malformed_outputs"])
result["malformed_after"] = _count(data["candidate"]["malformed_outputs"])
if max(result["malformed_before"], result["malformed_after"]) > result["total"]:
raise ValueError("malformed coverage exceeds pilot size")
return result
def _agent(data):
if (data["status"] != "completed_research_pilot" or data["release_eligible"] is not False
or data["agent_eval"]["source_kind"] != "captured"):
raise ValueError("not a captured completed pilot")
agent, retention = data["agent_eval"], data["MGSM_retention"]
scores = []
for score in (agent["baseline"], agent["candidate"]):
if (_count(score["infrastructure_errors"]) != 0
or _count(score["scored"], positive=True) != _count(score["episodes"], positive=True)):
raise ValueError("incomplete agent coverage")
scores.append({"correct": score["successful"], "total": score["episodes"],
"strict_final_answer_protocol_accuracy": score["success_rate"]})
success = _comparison(*scores, agent["paired"], "wins", "losses",
agent["paired"]["descriptive_bootstrap_95"])
math_result = _comparison(retention["baseline"], retention["candidate"], retention["paired"],
"candidate_wins", "candidate_losses", retention["paired"]["bootstrap"]["ci_95"])
passed = retention["gate"]["passed"]
if type(passed) is not bool or retention["gate"]["release_eligible"] is not False:
raise ValueError("missing retention gate")
return success, math_result, passed
def _anchor(link: Callable[[str], str | None], path: str, label: str) -> str:
target = link(path)
return (f'{escape(label)}'
if isinstance(target, str) else escape(label))
def _bars(key: str, title: str, result: dict, labels: tuple[str, str]) -> str:
rows = []
for series, label in zip(("before", "after"), labels, strict=True):
value, total = result[series], result["total"]
identifier = f"{key}-{series}"
rows.append(
f'
See recorded evidence: '
+ _anchor(link, f"docs/use-cases/{stem}-results.md", "Read the pilot results")
+ '. Scores are shown only when complete aggregate counts are available.
')
def _sft_panel(data, link):
heading = '
German reasoning SFT
'
try:
result = _sft(data)
badge = ('Inconclusive'
if result["low"] <= 0 <= result["high"] else 'Research pilot')
content = (heading + badge
+ f'
A fixed {result["total"]}-item German MGSM pilot, one stochastic trial per item.
'
try:
success, retention, passed = _agent(data)
badge = ('Retention gate failed' if not passed
else 'Observed retention gate passed')
inconclusive = success["low"] <= 0 <= success["high"]
gain = ('Inconclusive agent gain' if inconclusive
else 'Small research pilot')
content = (heading + badge + gain
+ f'
{success["total"]} synthetic lookup/calculator episodes and the fixed '
+ f'{retention["total"]}-item German MGSM retention check; one stochastic trial per item.
'
+ ('Agent gain remains inconclusive. ' if inconclusive else 'A bounded pilot comparison. ')
+ escape(_interval(success)) + ' Strict success includes final-answer format; it does not demonstrate improved arithmetic or tool workflow.
'
+ '
'
+ ('The candidate failed the observed no-drop MGSM gate. ' if not passed else '')
+ escape(_interval(retention)) + ' This candidate is not qualified for release.
' + _anchor(link, target, "Read qualification scope") + '')
def _qualifications(gyms, openenv, link):
cards = []
try:
nemo = gyms["NeMo_Gym"]["online_training"]
if (nemo["status"] != "completed" or nemo["validation_only"] is not True
or nemo["release_eligible"] is not False
or _count(nemo["gpu"]["devices"], positive=True) != 1
or _count(nemo["update_evidence"]["changed_trainable_tensors"], positive=True) < 1):
raise ValueError("unqualified NeMo execution")
steps = _count(nemo["parameters"]["max_steps"], positive=True)
rollouts = _count(nemo["counts"]["rollouts"], positive=True)
cards.append(_qualification(link, "NeMo Gym", "GPU integration", f"{steps} steps · {rollouts} rollouts",
"Verified adapter update. No held-out quality or MGSM retention claim.", "docs/agent-gyms.md#nemo-gym-with-online-trl"))
except (KeyError, ValueError, TypeError):
cards.append(_qualification(link, "NeMo Gym", "Recorded execution", "See recorded evidence",
"Read the qualification before drawing a quality conclusion.", "docs/agent-gyms.md"))
try:
if (openenv["status"] != "cpu-qualified-gpu-pending" or openenv["release_eligible"] is not False
or openenv["model_inference"] is not False or openenv["gpu_training"] is not False):
raise ValueError("not CPU-only qualification")
checks = openenv["native_service"]["checks"]
client_checks = openenv["native_client"]["checks"]
if any(not isinstance(group, dict) or not group or any(value is not True for value in group.values())
for group in (checks, client_checks)):
raise ValueError("incomplete native service/client checks")
sessions = _count(openenv["native_service"]["simultaneous_sessions"]["simultaneously_open_before_actions"], positive=True)
cards.append(_qualification(link, "OpenEnv", "Native CPU qualification", f"{len(checks)} checks · {sessions} simultaneous sessions",
"Native SDK and client exercised. No model inference or GPU training.", "docs/use-cases/german-openenv-comparison.md"))
except (KeyError, ValueError, TypeError):
cards.append(_qualification(link, "OpenEnv", "Recorded execution", "See recorded evidence",
"Read the native CPU and pending GPU qualification scope.", "docs/use-cases/german-openenv-comparison.md"))
try:
verifiers = gyms["Verifiers"]
if (verifiers["policy_kind"] != "scripted_fixture" or verifiers["model_inference"] is not False
or verifiers["training"] is not False or not isinstance(verifiers["cases"], list)
or not verifiers["cases"]):
raise ValueError("not scripted CPU qualification")
tests = _count(gyms["software_validation"]["optional_Verifiers_tests_passed"], positive=True)
cards.append(_qualification(link, "Prime Verifiers", "CPU fixtures", f'{len(verifiers["cases"])} scenarios · {tests} optional tests',
"Native MCP tools exercised. Model inference and Prime-RL training remain unqualified.", "docs/agent-gyms.md#prime-intellect-verifiers"))
except (KeyError, ValueError, TypeError):
cards.append(_qualification(link, "Prime Verifiers", "Recorded execution", "See recorded evidence",
"Read the scripted fixture scope and remaining model/training work.", "docs/agent-gyms.md#prime-intellect-verifiers"))
return '
' + ''.join(cards) + '
'
def render_evidence(snapshots: dict[str, bytes], link: Callable[[str], str | None]) -> str:
"""Render only four approved aggregate snapshots captured by the hub builder.
Missing, malformed or inconsistent aggregates show method links, never invented
scores. Both panels are visible in the HTML; progressive JS can select one.
No file, run-directory, model-output, network or live-status access occurs.
"""
sft, agent, gyms, openenv = (_load(snapshots, path) for path in (SFT, AGENT, GYMS, OPENENV))
return (''
'
03 / MEASURED OUTCOMES
'
'
Recorded experimental results
'
'
German Qwen pilot results, with sample sizes and limitations. '
'Integration checks are reported separately from model-quality assessments.
These small, fixed pilots do not establish broad German reasoning ability. '
'Descriptive paired intervals exclude training-seed and generation-seed variance. '
'No recorded candidate here is qualified for release. '
+ _anchor(link, "src/hessian_recipes/evaluation.py", "Inspect the comparison checks") + ' · '
+ _anchor(link, "src/hessian_recipes/agent_rl_report.py", "Inspect agent evidence verification") + '
'
'
Integration qualification
' + _qualifications(gyms, openenv, link)
+ '
Longer-run quality results require their own recorded assessment. '
'This is saved evidence, not a live job or queue monitor.