diff --git a/.github/workflows/skill-eval-pr.yml b/.github/workflows/skill-eval-pr.yml
index eefd7c3d..85141f06 100644
--- a/.github/workflows/skill-eval-pr.yml
+++ b/.github/workflows/skill-eval-pr.yml
@@ -140,6 +140,8 @@ jobs:
- uses: actions/checkout@v4
with:
ref: ${{ github.workflow_sha }}
+ # Report-only retries authenticate the original verifier as a main ancestor.
+ fetch-depth: 0
persist-credentials: false
- uses: actions/setup-python@v5
with:
@@ -160,6 +162,7 @@ jobs:
name: skill-eval-report
path: |
prepared/pr-plan.json
+ prepared/report-renderer.json
prepared/*/report.json
prepared/*/report.md
retention-days: 30
diff --git a/docs/skill-eval-pr.md b/docs/skill-eval-pr.md
index 43829b5d..147ac8cf 100644
--- a/docs/skill-eval-pr.md
+++ b/docs/skill-eval-pr.md
@@ -383,3 +383,32 @@ that a small case covers every mechanism in those files. Unmapped generated
projections, navigation documents and tests remain visible coverage gaps.
Source-only review dispositions must be explicit and must not be represented as
model-evaluated passes. The maintained mapping is not expanded automatically.
+
+## Retained-result explanations
+
+The automated comment leads with bounded baseline/candidate components, submitted
+finite diagnostic fields, and fixed failure reasons derived after trusted regrading.
+A component improvement does not change the frozen aggregate: a strict citation
+failure remains `FAIL` even when diagnosis and next-check components improve.
+Missing, invalid or unverified components are unavailable, not failures. No raw
+model explanation, hidden oracle answer or arbitrary answer field is published.
+Synthetic diagnostic coverage is not an engine benchmark or general efficacy claim.
+
+An explicit `publish` retry with the original `source_run` can refresh presentation
+without a new paid evaluation. It authenticates the original workflow, complete
+plan digest and verified ledger record, current PR base/head, retention and frozen
+inputs. Historical controller compatibility is established against the original
+trusted-main ancestor, not by relaxing hashes: an isolated report-only subprocess
+runs that Git source with no provider/GitHub environment and never executes
+artifact Python. The comment distinguishes the original execution implementation
+from the current presentation implementation. Current explanatory projection also
+requires the identical builtin grader and schema; otherwise components are unavailable.
+Failures stop the retry rather than replacing available evidence with a fabricated
+new result. New-run authorization/accounting and retained-result validity are separate.
+
+Use the existing workflow dispatch with `action=publish`, the current PR number
+and original `source_run`; do not request `execute` or provide a new approval.
+This bridge supports completed first-attempt automatic (`pull_request_target`)
+source runs with a verified ledger binding. Missing Git ancestry, mismatched live
+revisions, expired artifacts or unauthenticated provenance stop publication.
+The uploaded report contains the original plan and separate renderer metadata.
diff --git a/tests/test_skill_eval_feedback.py b/tests/test_skill_eval_feedback.py
new file mode 100644
index 00000000..88823514
--- /dev/null
+++ b/tests/test_skill_eval_feedback.py
@@ -0,0 +1,250 @@
+"""Offline public projection checks: no model/provider calls or grader changes."""
+import copy
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+from inferrsi.harness import skill_eval_diagnostic as diagnostic
+from inferrsi.infra.artifacts import sha256, write_json
+from tests.fixtures.skill_eval_compare_support import FixedOutputClient
+from tests.fixtures.skill_eval_pr_support import approval, setup
+from tools import skill_eval_feedback as feedback
+from tools import skill_eval_pr as integration
+
+ROOT = Path(__file__).resolve().parents[1]
+TASK_PATH = ROOT / 'docs/examples/skill-comparison/pr133-diagnostic-case05.json'
+TASK = json.loads(TASK_PATH.read_text())[0]
+OLD = dict(diagnosis='capacity', evidence=['E4'], next_check='inspect_allocator_accounting',
+ reuse_authorized=False)
+NEW = dict(diagnosis='readiness', evidence=['E2', 'E4'], next_check='check_receive_completion',
+ reuse_authorized=False)
+
+
+def trial(answer):
+ checks = diagnostic.verify(TASK, answer)
+ return dict(verdict='PASS' if checks['passed'] else 'FAIL', grader_trust='reverified',
+ feedback=feedback.projection(diagnostic.ADAPTER, TASK, answer, checks))
+
+
+def fixture():
+ exam = dict(cases='f' * 64, schema='a' * 64, grader='9' * 64)
+ plan = dict(repository='example/fixture', pr=7, base='b' * 40, head='c' * 40,
+ version='d' * 40, model=dict(name='offline', effort='test'),
+ limits=dict(repeats=1, max_task_calls=2, timeout_seconds=60,
+ campaign_seconds=600, token_stop_new_trials=1000),
+ estimated_calls=2, coverage=[], comparisons=[dict(
+ skill='selected', mode='update', calls=2, plan_sha256='e' * 64,
+ cases=[dict(id='case-05', split='holdout')], exam=exam)])
+ result = dict(skill='selected', plan_sha256='e' * 64, exam=exam,
+ arms={r: dict(git_revision=plan[k]) for r, k in
+ [('baseline', 'base'), ('candidate', 'head')]},
+ status=dict(execution_state='completed'), discovery=[],
+ pairs=[dict(case_id='case-05', repeat=1, outcome='unchanged_incorrect',
+ baseline=trial(OLD), candidate=trial(NEW))],
+ counts={k: int(k == 'unchanged_incorrect') for k in
+ ('fix', 'regression', 'unchanged_correct', 'unchanged_incorrect',
+ 'inconclusive', 'missing_pair')})
+ return plan, [result]
+
+
+def render(plan, results, **kwargs):
+ return integration.summary(plan, results, 'https://example.invalid/evidence', 100, 1, **kwargs)
+
+
+class FeedbackTests(unittest.TestCase):
+ def test_retained_case_changes_components_not_total(self):
+ plan, results = fixture()
+ text = render(plan, results)
+ self.assertIn('unchanged_incorrect', text)
+ self.assertEqual(text.count('diagnosis=FAIL'), 1)
+ self.assertEqual(text.count('diagnosis=PASS'), 1)
+ self.assertEqual(text.count('citation=FAIL'), 2)
+ self.assertEqual(text.count('reuse_authorized=false'), 2)
+ self.assertIn('Extra/nondecisive citation', text)
+ self.assertIn('capacity', text)
+ self.assertIn('readiness', text)
+ self.assertNotIn('expected', text.lower())
+ self.assertLess(text.index('Retained baseline'), text.index('### Advisory'))
+
+ def test_component_regression_with_unchanged_strict_failure(self):
+ plan, results = fixture()
+ pair = results[0]['pairs'][0]
+ pair['baseline'], pair['candidate'] = pair['candidate'], pair['baseline']
+ text = render(plan, results)
+ self.assertLess(text.index('diagnosis=PASS'), text.index('diagnosis=FAIL'))
+ self.assertIn('unchanged_incorrect', text)
+
+ def test_execution_block_does_not_obscure_evidence(self):
+ plan, results = fixture()
+ plan['execution_blocked'] = 'accounting unavailable'
+ text = render(plan, results)
+ self.assertLess(text.index('readiness'), text.index('Paid execution blocked'))
+ self.assertIn('does not invalidate available retained results', text)
+
+ def test_bounded_block_reason_keeps_explicit_no_fallback_safety_clause(self):
+ plan, results = fixture()
+ plan["execution_blocked"] = "@someone `unsafe` " * 1000
+ text = render(plan, results)
+ self.assertIn("no alternate provider/model is selected", text)
+ self.assertIn("does not invalidate available retained results", text)
+ self.assertEqual(text.count(""), 2)
+ self.assertNotIn("@someone", text)
+ self.assertNotIn("`unsafe`", text)
+ self.assertLess(len(text), 10000)
+
+ def test_unavailable_and_schema_failure_are_not_component_failures(self):
+ for t in ({}, dict(verdict='INCONCLUSIVE', grader_trust='reverified'),
+ dict(verdict='FAIL', grader_trust='unverified')):
+ self.assertIn('unavailable', feedback.trial_text(t))
+ self.assertNotIn('diagnosis=FAIL', feedback.trial_text(t))
+ text = feedback.trial_text(trial({'reason': 'secret-canary'}))
+ self.assertIn('schema=FAIL', text)
+ self.assertIn('diagnosis=unavailable', text)
+ self.assertNotIn('secret-canary', text)
+
+ def test_citation_reason_classification(self):
+ for ids, present, absent in [(['E2', 'E4'], 'extra_citation', 'missing_citation'),
+ ([], 'missing_citation', 'extra_citation'),
+ (['E2', 'E2'], 'duplicate_citation', 'extra_citation')]:
+ data = trial({**NEW, 'evidence': ids})['feedback']
+ self.assertIn(present, data['reasons'])
+ self.assertNotIn(absent, data['reasons'])
+
+ def test_oversized_and_arbitrary_fields_are_not_published(self):
+ for ids in [['E' + '1' * 10000], ['E2'] * 10000, ['secret-canary']]:
+ text = feedback.trial_text(trial({**NEW, 'evidence': ids}))
+ self.assertNotIn('secret-canary', text)
+ self.assertLess(len(text), 1500)
+ for field, value in [('diagnosis', '@secret-canary'),
+ ('reuse_authorized', 1), ('next_check', {'secret-canary': 1})]:
+ text = feedback.trial_text(trial({**NEW, field: value}))
+ self.assertNotIn('secret-canary', text)
+ t = trial(NEW)
+ t['feedback']['components']['citation'] = 1
+ self.assertIn('invalid projection', feedback.trial_text(t))
+
+ def test_mismatched_and_unavailable_evidence_suppresses_details_and_counts(self):
+ for mutation in ('head', 'base', 'exam', 'plan_sha256', 'duplicate'):
+ plan, results = fixture()
+ if mutation in ('head', 'base'):
+ plan[mutation] = 'f' * 40
+ elif mutation == 'duplicate':
+ results.append(copy.deepcopy(results[0]))
+ else:
+ results[0][mutation] = 'wrong'
+ text = render(plan, results)
+ self.assertNotIn('readiness', text)
+ self.assertNotIn('"unchanged_incorrect":1', text)
+ for state in ('expired', 'missing', 'invalid'):
+ plan, results = fixture()
+ text = render(plan, results, evidence_state=state)
+ self.assertNotIn('readiness', text)
+ self.assertNotIn('"unchanged_incorrect":1', text)
+
+ def test_projection_is_bounded_and_never_echoes_freeform_or_oracle(self):
+ t = trial(NEW)
+ t['feedback']['answer']['reason'] = 'secret-canary'
+ t['feedback']['oracle'] = 'hidden-secret-canary'
+ self.assertNotIn('secret-canary', feedback.trial_text(t))
+ plan, results = fixture()
+ results[0]['pairs'][0]['case_id'] = '@secret-canary'
+ self.assertNotIn('secret-canary', '\n'.join(feedback.comparison_lines(
+ plan, results, 'available')))
+ self.assertEqual(feedback.code('@secret-canary'), 'unavailable')
+
+ def test_actual_reports_regrade_and_replace_untrusted_projection(self):
+ with tempfile.TemporaryDirectory() as temporary:
+ root = Path(temporary)
+ repo, mapping, identity = setup(root, mode="addition", ids=("one",))
+ output = root / "evaluation"
+ plan = integration.prepare(repo, mapping, identity, output)
+ integration.execute(output, approval(plan), client=FixedOutputClient(root))
+ result = integration.reports(output)[0]
+ self.assertEqual(result["pairs"][0]["candidate"]["feedback"]["adapter"], "audit")
+ text = render(plan, [result])
+ self.assertIn("report_verdict=PASS", text)
+ path = output / "selected/results/one-candidate-1/result.json"
+ record = json.loads(path.read_text())
+ record["feedback"] = {"answer": "secret-canary", "components": {"injected": True}}
+ write_json(path, record)
+ write_json(path.with_name("receipt.json"), {"result_sha256": sha256(path)})
+ text = render(plan, integration.reports(output))
+ self.assertNotIn("secret-canary", text)
+ self.assertNotIn("injected", text)
+ record["checks"]["assertions"]["report_verdict"] = False
+ write_json(path, record)
+ write_json(path.with_name("receipt.json"), {"result_sha256": sha256(path)})
+ with self.assertRaisesRegex(ValueError, "Untrusted grading"):
+ integration.reports(output)
+
+ def test_missing_pair_and_consistently_hostile_metadata(self):
+ plan, results = fixture()
+ results[0]["pairs"][0]["candidate"] = {"verdict": "INCONCLUSIVE", "missing": True}
+ text = render(plan, results)
+ self.assertIn("Candidate: unavailable", text)
+ self.assertNotIn("Candidate: FAIL", text)
+ bad = " \n@someone `evil` | # forged"
+ plan["comparisons"][0]["skill"] = results[0]["skill"] = bad
+ plan["comparisons"][0]["cases"][0]["id"] = bad
+ results[0]["pairs"][0]["case_id"] = bad
+ text = render(plan, results)
+ self.assertEqual(text.count(" "), 2)
+ self.assertNotIn("@someone", text)
+ self.assertNotIn("`evil`", text)
+ self.assertNotIn("\n# forged", text)
+
+ def test_changed_grader_cannot_generate_current_explanations(self):
+ result = {"pairs": [{"baseline": trial(OLD), "candidate": trial(NEW)}],
+ "discovery": []}
+ context = dict(adapter=diagnostic.ADAPTER, schema=diagnostic.SCHEMA,
+ exam={"grader": "f" * 64}, tasks={TASK["id"]: TASK})
+ integration.attach_feedback(result, context)
+ self.assertNotIn("feedback", result["pairs"][0]["baseline"])
+
+ def test_historical_publish_preserves_original_plan_and_new_writer(self):
+ plan, results = fixture()
+ context = dict(adapter=diagnostic.ADAPTER, schema=diagnostic.SCHEMA,
+ exam={"grader": "f" * 64}, tasks={TASK["id"]: TASK})
+ retained = dict(plan=plan, reports=results, contexts={"selected": context},
+ expires_at=100, evidence_url="https://example.invalid/evidence")
+ api = Mock()
+ api.publish.return_value = {"html_url": "https://example.invalid/comment"}
+ with tempfile.TemporaryDirectory() as temporary, \
+ patch("tools.skill_eval_retained.retained_reports", return_value=retained), \
+ patch.object(integration, "git", return_value=b"a" * 40):
+ output = Path(temporary)
+ integration.publish_historical(api, 7, 10, 20, now=1, output=output)
+ self.assertEqual(json.loads((output / "pr-plan.json").read_text()), plan)
+ self.assertEqual(json.loads((output / "report-renderer.json").read_text())[
+ "original_plan_sha256"], integration.fingerprint(plan))
+ self.assertTrue((output / "selected/report.json").is_file())
+ args = api.publish.call_args.args
+ self.assertIs(args[0], plan)
+ self.assertEqual(args[2], 20)
+ self.assertIn("Presentation implementation: `" + "a" * 40, args[1])
+ self.assertIn("starts no model execution or budget reservation", args[1])
+
+ def test_copyable_json_and_metadata_escape_are_not_double_escaped(self):
+ value = [{"outcome": "unchanged_incorrect", "case": "[one]```\n@someone"}]
+ text = feedback.json_for_code(value)
+ self.assertEqual(json.loads(text), value)
+ self.assertIn("unchanged_incorrect", text)
+ self.assertNotIn("```", text)
+ self.assertNotIn("", text)
+ self.assertNotIn("@someone", text)
+ self.assertEqual(feedback.safe_text("@someone_[x]"),
+ "@someone_[x]")
+
+ def test_historical_validation_failure_never_overwrites_comment(self):
+ api = Mock()
+ with patch("tools.skill_eval_retained.retained_reports", side_effect=ValueError("stale")), \
+ patch.object(integration, "git", return_value=b"a" * 40):
+ with self.assertRaisesRegex(ValueError, "stale"):
+ integration.publish_historical(api, 7, 10, 20, now=1)
+ api.publish.assert_not_called()
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/tests/test_skill_eval_pr.py b/tests/test_skill_eval_pr.py
index f96bb335..71b46a4a 100644
--- a/tests/test_skill_eval_pr.py
+++ b/tests/test_skill_eval_pr.py
@@ -633,9 +633,10 @@ def test_publish_dispatch_rejects_planning_source_before_artifact_access(
monkeypatch.setattr(integration, "GitHub", lambda repository, token: api)
dispatch_event(monkeypatch, tmp_path, output,
{"action": "publish", "pr": "7", "source_run": "10"}, "publish")
- with pytest.raises(ValueError, match="Untrusted or mismatched execution workflow"):
+ with pytest.raises(ValueError, match="Untrusted retained source workflow"):
integration.main()
- request.assert_called_once_with("GET", "/actions/runs/10")
+ assert request.call_args_list[0].args == ("GET", "/actions/runs/10")
+ assert not any("/artifacts" in call.args[1] for call in request.call_args_list)
assert api.writes == [] and api.comments == []
diff --git a/tests/test_skill_eval_pr_summary.py b/tests/test_skill_eval_pr_summary.py
index a9cf07c7..ff70ba75 100644
--- a/tests/test_skill_eval_pr_summary.py
+++ b/tests/test_skill_eval_pr_summary.py
@@ -74,7 +74,15 @@ def test_planned_summary_keeps_visible_status_and_folds_full_approval_fields(pla
def test_mixed_results_keep_all_aggregate_outcome_counts_visible(plan):
- text = render(plan, [outcome(fix=1, inconclusive=2), outcome(regression=3, missing_pair=4)])
+ first = {**outcome(fix=1, inconclusive=2), "skill": "selected",
+ "plan_sha256": plan["comparisons"][0]["plan_sha256"],
+ "exam": plan["comparisons"][0]["exam"],
+ "arms": {"baseline": {"git_revision": plan["base"]},
+ "candidate": {"git_revision": plan["head"]}}}
+ second = {**copy.deepcopy(first), **outcome(regression=3, missing_pair=4),
+ "skill": "another"}
+ plan["comparisons"].append({**copy.deepcopy(plan["comparisons"][0]), "skill": "another"})
+ text = render(plan, [first, second])
assert_boundaries(text)
main = visible(text)
assert "**RETAINED RESULTS**" in main and "**PLANNED:" not in main
diff --git a/tests/test_skill_eval_retained.py b/tests/test_skill_eval_retained.py
new file mode 100644
index 00000000..375993cc
--- /dev/null
+++ b/tests/test_skill_eval_retained.py
@@ -0,0 +1,286 @@
+"""Offline security tests for the historical report-only bridge."""
+import copy
+import io
+import json
+import os
+import stat
+import subprocess
+import tempfile
+import unittest
+import zipfile
+from pathlib import Path
+from unittest.mock import patch
+
+from tools import skill_eval_retained as bridge
+
+
+def zipped(files):
+ stream = io.BytesIO()
+ with zipfile.ZipFile(stream, "w") as archive:
+ for name, value in files.items():
+ archive.writestr(name, value)
+ return stream.getvalue()
+
+
+class RetainedArchiveTests(unittest.TestCase):
+ def test_archive_reads_only_data(self):
+ raw = b"raise RuntimeError('never execute artifact code')"
+ self.assertEqual(bridge._archive(zipped({"validator/verify.py": raw})),
+ {"validator/verify.py": raw})
+
+ def test_unsafe_names(self):
+ for name in ("../escape", "/absolute", "a\\b", "a//b", "a/./b", "a/../b"):
+ with self.subTest(name=name), self.assertRaises(ValueError):
+ bridge._archive(zipped({name: b"x"}))
+
+ def test_symlink_and_special(self):
+ for mode in (stat.S_IFLNK, stat.S_IFIFO, stat.S_IFSOCK):
+ stream = io.BytesIO()
+ with zipfile.ZipFile(stream, "w") as archive:
+ item = zipfile.ZipInfo("entry")
+ item.external_attr = (mode | 0o600) << 16
+ archive.writestr(item, "target")
+ with self.assertRaises(ValueError):
+ bridge._archive(stream.getvalue())
+
+ def test_duplicate_and_oversize(self):
+ stream = io.BytesIO()
+ with zipfile.ZipFile(stream, "w") as archive:
+ archive.writestr("one", "first")
+ archive.writestr("one", "second")
+ with self.assertRaises(ValueError):
+ bridge._archive(stream.getvalue())
+ with patch.object(bridge, "MAX_BYTES", 1), self.assertRaises(ValueError):
+ bridge._archive(zipped({"one": b"xx"}))
+
+
+class HistoricalGitTests(unittest.TestCase):
+ def setUp(self):
+ self.temporary = tempfile.TemporaryDirectory()
+ self.addCleanup(self.temporary.cleanup)
+ self.root = Path(self.temporary.name)
+ self.repo = self.root / "repo"
+ self.repo.mkdir()
+ self.git("init", "--quiet")
+ (self.repo / "tools").mkdir()
+ (self.repo / "src").mkdir()
+ (self.repo / "src/empty.py").write_text("")
+ self.adapter = b"# trusted historical adapter\n"
+ (self.repo / "tools/skill_eval_pr.py").write_bytes(self.adapter)
+ (self.repo / "tools/skill_eval_pr_cases.json").write_text("{}")
+ self.git("add", ".")
+ self.git("commit", "--quiet", "-m", "historical main")
+ self.old = self.git("rev-parse", "HEAD")
+ (self.repo / "src/empty.py").write_text("# new controller bytes\n")
+ self.git("add", ".")
+ self.git("commit", "--quiet", "-m", "current main")
+ self.current = self.git("rev-parse", "HEAD")
+ self.plan = {"adapter_sha256": bridge._hash(self.adapter),
+ "mapping_sha256": bridge._hash(b"{}")}
+
+ def git(self, *args):
+ return subprocess.check_output(["git", "-c", "user.name=Offline Fixture", "-c",
+ "user.email=fixture@example.invalid", "-c", "commit.gpgsign=false", *args],
+ cwd=self.repo, text=True).strip()
+
+ def test_historical_bytes_not_current_controller(self):
+ dest = self.root / "code"
+ dest.mkdir()
+ bridge._trusted_code(self.repo, self.old, self.current, self.plan, dest)
+ self.assertEqual((dest / "src/empty.py").read_bytes(), b"")
+ self.assertEqual((dest / "tools/skill_eval_pr.py").read_bytes(), self.adapter)
+
+ def test_invalid_revision_and_blob_bindings(self):
+ for field in ("adapter_sha256", "mapping_sha256"):
+ changed = {**self.plan, field: "0" * 64}
+ with self.subTest(field=field), self.assertRaises(ValueError):
+ bridge._trusted_code(self.repo, self.old, self.current, changed, self.root / field)
+ with self.assertRaises(ValueError):
+ bridge._trusted_code(self.repo, "--help", self.current, self.plan, self.root / "bad")
+ with self.assertRaises(ValueError):
+ bridge._trusted_code(self.repo, self.old, self.old, self.plan, self.root / "bad")
+
+ def test_nonancestor_rejected(self):
+ self.git("checkout", "--quiet", "--orphan", "untrusted")
+ self.git("commit", "--quiet", "-m", "unrelated")
+ unrelated = self.git("rev-parse", "HEAD")
+ self.git("checkout", "--quiet", self.current)
+ with self.assertRaises(ValueError):
+ bridge._trusted_code(self.repo, unrelated, self.current, self.plan, self.root / "bad")
+
+
+class ReportOnlyTests(unittest.TestCase):
+ def test_clean_environment_no_provider_execute_network_or_artifact_import(self):
+ with tempfile.TemporaryDirectory() as temporary:
+ root = Path(temporary)
+ code, data, home = (root / n for n in ("code", "data", "home"))
+ for directory in (code / "tools", code / "src", data, home):
+ directory.mkdir(parents=True)
+ (code / "tools/__init__.py").write_text("")
+ (code / "tools/skill_eval_compare.py").write_text("")
+ (data / "results.zip").write_bytes(zipped({}))
+ (data / "sitecustomize.py").write_text("raise AssertionError('artifact import')")
+ (code / "tools/skill_eval_pr.py").write_text('''
+import os, socket, subprocess, sys
+from tools import skill_eval_compare as compare
+def import_results(raw, output):
+ assert set(os.environ) <= {"HOME", "LANG", "LC_ALL", "LC_CTYPE"}
+ assert not any(str(output) == p for p in sys.path)
+def reports(output):
+ for call in (lambda: execute(None), lambda: compare.provider_client(None),
+ lambda: subprocess.run([sys.executable, "-c", "pass"]),
+ lambda: socket.socket()):
+ try:
+ call()
+ except RuntimeError:
+ continue
+ raise AssertionError("report-only restriction bypassed")
+ return [{"verified": True}]
+''')
+ with patch.dict(os.environ, {"GITHUB_TOKEN": "fake-secret",
+ "ANTHROPIC_API_KEY": "fake-key",
+ "PYTHONPATH": str(data), "HTTPS_PROXY": "fake-proxy"}):
+ self.assertEqual(bridge._regrade(code, data, home), [{"verified": True}])
+
+
+class FrozenInputTests(unittest.TestCase):
+ def setUp(self):
+ self.temporary = tempfile.TemporaryDirectory()
+ self.addCleanup(self.temporary.cleanup)
+ self.root = Path(self.temporary.name)
+ self.code, self.data = self.root / "code", self.root / "data"
+ (self.code / "tools").mkdir(parents=True)
+ self.data.mkdir()
+ (self.code / "tools/controller.py").write_bytes(b"trusted")
+ self.frozen = {"skill": "selected", "adapter": "audit", "exam": {"grader": "a" * 64},
+ "arms": {"baseline": {"git_revision": "b" * 40},
+ "candidate": {"git_revision": "c" * 40}},
+ "controllers": {"tools/controller.py": bridge._hash(b"trusted")},
+ "files": {"plan/tasks.json": bridge._hash(b"[]"),
+ "plan/schema.json": bridge._hash(b"{}"),
+ "validator/verify.py": bridge._hash(b"raise RuntimeError('do not execute')")}}
+ self.files = {"selected/plan/tasks.json": b"[]", "selected/plan/schema.json": b"{}",
+ "selected/validator/verify.py": b"raise RuntimeError('do not execute')"}
+ self.plan = {"head": "c" * 40, "base": "b" * 40, "comparisons": [{"skill": "selected",
+ "exam": self.frozen["exam"]}]}
+ self.freeze()
+
+ def freeze(self):
+ raw = json.dumps(self.frozen).encode()
+ self.files["selected/plan/freeze.json"] = raw
+ self.plan["comparisons"][0]["plan_sha256"] = bridge._hash(raw)
+ self.files["pr-plan.json"] = json.dumps(self.plan).encode()
+
+ def materialize(self):
+ return bridge._materialize(self.files, self.plan, self.data, self.code)
+
+ def test_hash_verified_python_remains_data(self):
+ context = self.materialize()
+ self.assertEqual(context["selected"]["tasks"], [])
+ self.assertEqual((self.data / "selected/validator/verify.py").read_bytes(),
+ self.files["selected/validator/verify.py"])
+
+ def test_mismatched_freeze_input_controller_and_arm(self):
+ for kind in ("freeze", "input", "controller", "arm", "extra"):
+ original_files, original_frozen = copy.deepcopy(self.files), copy.deepcopy(self.frozen)
+ with self.subTest(kind=kind):
+ if kind == "freeze":
+ self.files["selected/plan/freeze.json"] += b" "
+ elif kind == "input":
+ self.files["selected/plan/tasks.json"] = b"[1]"
+ elif kind == "extra":
+ self.files["selected/plan/extra.py"] = b"malicious"
+ elif kind == "controller":
+ self.frozen["controllers"]["tools/controller.py"] = "0" * 64
+ self.freeze()
+ else:
+ self.frozen["arms"]["baseline"]["git_revision"] = "d" * 40
+ self.freeze()
+ with self.assertRaises(ValueError):
+ self.materialize()
+ self.files, self.frozen = original_files, original_frozen
+ self.freeze()
+
+ def test_frozen_path_traversal(self):
+ for field, name in (("files", "../../escape"), ("controllers", "../escape")):
+ with self.subTest(field=field):
+ self.frozen[field][name] = "0" * 64
+ self.freeze()
+ with self.assertRaises(ValueError):
+ self.materialize()
+ del self.frozen[field][name]
+
+
+class SourceAuthenticationTests(unittest.TestCase):
+ def setUp(self):
+ from tests.fixtures.skill_eval_pr_support import SimulatedGitHub
+ self.plan = {"repository": "example/fixture", "pr": 7, "head": "b" * 40,
+ "base": "c" * 40, "version": "d" * 40, "comparisons": []}
+ self.api = SimulatedGitHub(self.plan, zipped({}))
+ self.api.run.update(event="pull_request_target", run_attempt=1,
+ display_title="skill-eval plan or publish PR 7", head_sha=self.plan["head"])
+ self.patches = [patch("tools.skill_eval_auto.retained_source", return_value=10),
+ patch.object(bridge, "_trusted_code"),
+ patch.object(bridge, "_materialize", return_value={}),
+ patch.object(bridge, "_regrade", return_value=[])]
+ self.mocks = [p.start() for p in self.patches]
+ for p in self.patches:
+ self.addCleanup(p.stop)
+
+ def call(self):
+ return bridge.retained_reports(self.api, 7, 10, trusted_repo=Path("."),
+ renderer_revision="e" * 40, now=100)
+
+ def test_authenticated_source_read_only_and_separate_renderer(self):
+ result = self.call()
+ self.assertEqual(result["plan"], self.plan)
+ self.assertEqual(result["renderer_revision"], "e" * 40)
+ self.assertEqual(self.api.writes, [])
+ self.mocks[0].assert_called_once_with(self.api, self.plan)
+ self.mocks[1].assert_called_once()
+
+ def test_bad_source_identity_fails_before_verifier(self):
+ fields = {"id": 11, "run_attempt": 2, "path": "untrusted.yml",
+ "event": "pull_request", "status": "in_progress", "display_title": "other",
+ "head_sha": "f" * 40,
+ "repository": {"id": 999, "full_name": "example/untrusted"}}
+ for name, value in fields.items():
+ old = self.api.run[name]
+ self.api.run[name] = value
+ with self.subTest(name=name), self.assertRaises(ValueError):
+ self.call()
+ self.api.run[name] = old
+ self.mocks[1].assert_not_called()
+ self.assertEqual(self.api.writes, [])
+
+ def test_stale_and_ledger_mismatch(self):
+ self.api.current["head"] = "f" * 40
+ with self.assertRaises(ValueError):
+ self.call()
+ self.api.current["head"] = self.plan["head"]
+ for source in (None, 11):
+ self.mocks[0].return_value = source
+ with self.subTest(source=source), self.assertRaises(ValueError):
+ self.call()
+ self.mocks[0].side_effect = ValueError("ambiguous ledger")
+ with self.assertRaises(ValueError):
+ self.call()
+ self.mocks[1].assert_not_called()
+
+ def test_expired_plan_and_results(self):
+ for artifact in (self.api.plan_artifact, self.api.artifact):
+ artifact["expired"] = True
+ with self.assertRaises(ValueError):
+ self.call()
+ artifact["expired"] = False
+ self.mocks[1].assert_not_called()
+
+ def test_missing_source_plan(self):
+ self.api.plan_artifact = None
+ with self.assertRaises(ValueError):
+ self.call()
+ self.mocks[1].assert_not_called()
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tools/skill_eval_feedback.py b/tools/skill_eval_feedback.py
new file mode 100644
index 00000000..412409e7
--- /dev/null
+++ b/tools/skill_eval_feedback.py
@@ -0,0 +1,216 @@
+"""Bounded public projections, never raw answers or hidden expected answers.
+
+Projection is called only after trusted in-tree regrading. Rendering still
+validates the projection: stored report dictionaries are not publication authority.
+"""
+import html
+import json
+import re
+
+from inferrsi.harness import skill_eval_diagnostic as diagnostic
+
+COMPONENTS = {
+ diagnostic.ADAPTER: ("schema", "diagnosis", "citation", "safe_next_check",
+ "no_invariant_violation"),
+ "audit": ("schema", "report_verdict", "claim_scope", "findings", "reduction",
+ "evidence", "explanation"),
+}
+REASONS = {
+ "schema": "Answer does not satisfy the frozen schema.",
+ "diagnosis": "Diagnosis does not match the recorded observations.",
+ "citation": "Citations do not satisfy the exact decisive-evidence requirement.",
+ "extra_citation": "Extra/nondecisive citation present; strict citation check fails.",
+ "missing_citation": "Required decisive citation missing.",
+ "duplicate_citation": "Duplicate citation present.",
+ "safe_next_check": "Next read-only check does not match the recorded observations.",
+ "no_invariant_violation": "Answer incorrectly authorizes reuse.",
+ "report_verdict": "Report verdict fails the frozen check.",
+ "claim_scope": "Claim scope fails the frozen check.",
+ "findings": "Findings fail the frozen check.",
+ "reduction": "Reduction fails the frozen numeric check.",
+ "evidence": "Evidence references fail the frozen check.",
+ "explanation": "Required explanation is absent or invalid.",
+}
+
+
+def projection(adapter, task, answer, checks):
+ """Select only finite fields from a trusted grade; no oracle answer is emitted."""
+ adapter = diagnostic.ADAPTER if adapter == diagnostic.ADAPTER else "audit"
+ assertions = checks["assertions"]
+ components = {key: assertions[key] for key in COMPONENTS[adapter] if key in assertions}
+ reasons = [key for key, value in components.items() if value is False]
+ selected = {}
+ if adapter == diagnostic.ADAPTER and assertions.get("schema") is True:
+ selected = {key: answer[key] for key in ("diagnosis", "next_check", "reuse_authorized")}
+ citations = answer["evidence"]
+ # Never include arbitrary strings or identifiers that could carry secrets.
+ if (len(citations) <= 4 and all(isinstance(v, str) and
+ re.fullmatch(r"E[0-9]{1,4}", v) for v in citations)):
+ selected["evidence"] = citations
+ if assertions.get("citation") is False:
+ _, required, _ = diagnostic.expected(task)
+ cited = set(citations)
+ reasons += (["extra_citation"] if cited - required else [])
+ reasons += (["missing_citation"] if required - cited else [])
+ reasons += (["duplicate_citation"] if len(cited) != len(citations) else [])
+ return {"adapter": adapter, "components": components, "answer": selected,
+ "reasons": reasons}
+
+
+def code(value):
+ """Render a bounded safe identifier, without Markdown or mentions."""
+ if not isinstance(value, str) or not re.fullmatch(r"[A-Za-z0-9_.:/-]{1,128}", value):
+ return "unavailable"
+ return "" + html.escape(value) + ""
+
+
+def safe_text(value, limit=512):
+ """Bound presentation metadata and neutralize Markdown/HTML/mention syntax."""
+ if not isinstance(value, str):
+ return "unavailable"
+ value = value[:limit].replace("\n", " ").replace("\r", " ")
+ return "".join("&#" + str(ord(char)) + ";" if char in "`@*_[\\]#|"
+ else html.escape(char) for char in value)
+
+
+def json_for_code(value):
+ """Keep valid copyable JSON while neutralizing fence/HTML/mention characters."""
+ text = json.dumps(value, sort_keys=True, separators=(",", ":"))
+ if len(text) > 12000:
+ return "[]"
+ for char in "`<>&@":
+ text = text.replace(char, "\\u" + format(ord(char), "04x"))
+ return text
+
+
+def trial_text(trial):
+ if not isinstance(trial, dict) or trial.get("grader_trust") != "reverified":
+ return "unavailable (no trusted component evidence)"
+ verdict = trial.get("verdict")
+ if verdict not in ("PASS", "FAIL"):
+ return "unavailable (incomplete or invalid retained evidence)"
+ data = trial.get("feedback")
+ if (not isinstance(data, dict) or not isinstance(data.get("adapter"), str)
+ or data["adapter"] not in COMPONENTS):
+ return verdict + "; components unavailable"
+ adapter = data["adapter"]
+ checks = data.get("components")
+ reasons = data.get("reasons")
+ if (not isinstance(checks, dict) or not checks or
+ any(k not in COMPONENTS[adapter] or type(v) is not bool for k, v in checks.items())
+ or not isinstance(reasons, list) or len(reasons) > 12
+ or any(not isinstance(r, str) or r not in REASONS for r in reasons)):
+ return verdict + "; components unavailable (invalid projection)"
+ parts = [verdict]
+ answer = data.get("answer")
+ if adapter == diagnostic.ADAPTER and isinstance(answer, dict):
+ for key, allowed in (("diagnosis", diagnostic.DIAGNOSES),
+ ("next_check", diagnostic.CHECKS)):
+ value = answer.get(key)
+ if type(value) is str and value in allowed:
+ parts.append(key + "=" + code(value))
+ citations = answer.get("evidence")
+ if (isinstance(citations, list) and len(citations) <= 4 and
+ all(isinstance(v, str) and re.fullmatch(r"E[0-9]{1,4}", v) for v in citations)):
+ parts.append("citations=" + (", ".join(code(v) for v in citations) or "none"))
+ if type(answer.get("reuse_authorized")) is bool:
+ parts.append("reuse_authorized=" + str(answer["reuse_authorized"]).lower())
+ parts.append("components: " + ", ".join(
+ key + "=" + ("PASS" if checks[key] else "FAIL") if key in checks else key + "=unavailable"
+ for key in COMPONENTS[adapter]))
+ if reasons:
+ specific = any(r in reasons for r in ("extra_citation", "missing_citation",
+ "duplicate_citation"))
+ parts.append("Reasons: " + " ".join(REASONS[r] for r in dict.fromkeys(reasons)
+ if not (r == "citation" and specific)))
+ return "; ".join(parts)
+
+
+def component_changes(pair):
+ before, after = (pair.get(role, {}) for role in ("baseline", "candidate"))
+ if any(not isinstance(t, dict) or t.get("grader_trust") != "reverified" or
+ t.get("verdict") not in ("PASS", "FAIL") for t in (before, after)):
+ return "Component changes unavailable."
+ old, new = (t.get("feedback", {}) for t in (before, after))
+ if not isinstance(old, dict) or not isinstance(new, dict):
+ return "Component changes unavailable."
+ adapter = old.get("adapter")
+ if not isinstance(adapter, str) or adapter not in COMPONENTS or new.get("adapter") != adapter:
+ return "Component changes unavailable."
+ checks = [v.get("components") for v in (old, new)]
+ if any(not isinstance(c, dict) for c in checks):
+ return "Component changes unavailable."
+ changes = []
+ for key in COMPONENTS[adapter]:
+ a, b = (c.get(key) for c in checks)
+ if type(a) is bool and type(b) is bool and a != b:
+ changes.append(key + ": " + ("PASS" if a else "FAIL") + " → "
+ + ("PASS" if b else "FAIL"))
+ return "Component changes: " + ("; ".join(changes) or "none observed") + "."
+
+
+def comparison_lines(plan, results, evidence_state):
+ """Only exact, unique bindings may expose component explanations."""
+ if evidence_state != "available":
+ return []
+ lines, count = [], 0
+ for item, result in bound_results(plan, results):
+ allowed = {(case["id"], repeat) for case in item["cases"] if case["split"] != "discovery"
+ for repeat in range(1, plan["limits"]["repeats"] + 1)}
+ pairs = result.get("pairs", [])
+ seen = [(p.get("case_id"), p.get("repeat")) for p in pairs]
+ if len(set(seen)) != len(seen):
+ continue
+ for pair in pairs:
+ if (pair.get("case_id"), pair.get("repeat")) not in allowed:
+ continue
+ count += 1
+ if count > 8:
+ continue
+ lines += ["", "**" + code(item["skill"]) + " / " + code(pair["case_id"])
+ + " · repeat " + str(pair["repeat"]) + "**",
+ component_changes(pair),
+ "- Baseline: " + trial_text(pair.get("baseline")),
+ "- Candidate: " + trial_text(pair.get("candidate"))]
+ if any(isinstance(pair.get(role, {}).get("feedback"), dict) and
+ pair[role]["feedback"].get("adapter") == diagnostic.ADAPTER
+ for role in ("baseline", "candidate")):
+ lines += ["Coverage: one synthetic diagnostic case per pair; not a production "
+ "trace, engine benchmark or general efficacy result."]
+ if lines:
+ lines = ["### Retained baseline → candidate", "Strict aggregate outcomes are unchanged "
+ "by component explanations; unavailable is not failure."] + lines
+ gaps = sum(item["coverage"] != "covered" for item in plan["coverage"])
+ lines += ["", f"Showing {min(count, 8)} of {count} bound pairs; discovery excluded. "
+ f"Uncovered/unsupported changes: **{gaps}**; gaps never count as passes."]
+ return lines
+
+
+def evidence_binding_lines(plan, results, evidence_state):
+ if evidence_state != "available":
+ return []
+ lines = []
+ for item, result in bound_results(plan, results):
+ lines += ["Frozen comparison " + code(item["skill"]) + ": " + code(item["plan_sha256"])]
+ for pair in result.get("pairs", [])[:8]:
+ lines += ["Case " + code(pair["case_id"]) + " answer SHA-256 (baseline / candidate): "
+ + " / ".join(code(pair.get(role, {}).get("output_sha256"))
+ for role in ("baseline", "candidate"))]
+ return lines[:40]
+
+
+def bound_results(plan, results):
+ """No aggregate or detail can outlive its unique revision/plan/exam binding."""
+ for item in plan["comparisons"]:
+ if "plan_sha256" not in item:
+ continue
+ matches = [r for r in results if r.get("skill") == item["skill"]]
+ if len(matches) != 1:
+ continue
+ result = matches[0]
+ if (result.get("plan_sha256") != item["plan_sha256"] or
+ result.get("exam") != item["exam"] or
+ any(result.get("arms", {}).get(role, {}).get("git_revision") != plan[key]
+ for role, key in (("baseline", "base"), ("candidate", "head")))):
+ continue
+ yield item, result
diff --git a/tools/skill_eval_pr.py b/tools/skill_eval_pr.py
index 1df0452b..107473d0 100644
--- a/tools/skill_eval_pr.py
+++ b/tools/skill_eval_pr.py
@@ -18,6 +18,15 @@
from inferrsi.model.anthropic_agent import money, reservation
from tools import skill_eval_compare as compare
from tools.check_knowledge_model_policy import GitHubHTTPError
+from tools.skill_eval_feedback import (
+ bound_results,
+ code,
+ comparison_lines,
+ evidence_binding_lines,
+ json_for_code,
+ projection,
+ safe_text,
+)
ROOT = Path(__file__).resolve().parents[1]
MAPPING = ROOT / "tools/skill_eval_pr_cases.json"
@@ -332,10 +341,75 @@ def reports(output):
compare.load_plan(directory),
[p[r] for p in result["pairs"] for r in compare.ROLES] + result["discovery"],
tasks)
+ frozen = compare.load_plan(directory)
+ attach_feedback(result, {"adapter": frozen.get("adapter", "audit"),
+ "tasks": tasks, "exam": frozen["exam"],
+ "schema": compare.read_json(directory / "plan/schema.json")})
results.append(result)
return results
+def attach_feedback(result, context):
+ """Project only an exact current builtin exam, including historical reports.
+
+ A historical report's trusted result remains intact when its grader has
+ changed; explanations are unavailable rather than inferred under new rules.
+ """
+ diagnostic = compare.diagnostic
+ adapter = context["adapter"]
+ if adapter == diagnostic.ADAPTER:
+ verifier, schema, path = diagnostic.verify, diagnostic.SCHEMA, Path(diagnostic.__file__)
+ elif adapter == "audit":
+ verifier, schema = compare.verify, compare.SCHEMA
+ path = ROOT / "tools/skill_eval_verifier.py"
+ else:
+ return
+ trials = [p[r] for p in result["pairs"] for r in compare.ROLES] + result["discovery"]
+ for trial in trials:
+ trial.pop("feedback", None)
+ if context["exam"]["grader"] != sha256(path) or context["schema"] != schema:
+ return
+ tasks = context["tasks"]
+ if isinstance(tasks, list):
+ tasks = {task["id"]: task for task in tasks}
+ for trial in trials:
+ if trial.get("grader_trust") != "reverified" or trial["verdict"] not in ("PASS", "FAIL"):
+ continue
+ answer = json.loads(trial["output"])
+ checks = verifier(tasks[trial["task_id"]], answer)
+ if checks != trial["checks"] or checks["passed"] != (trial["verdict"] == "PASS"):
+ raise ValueError("Historical grading disagrees with the current identical grader")
+ trial["feedback"] = projection(adapter, tasks[trial["task_id"]], answer, checks)
+
+
+def publish_historical(api, pr, source_run, writer_run, *, now, output=None):
+ """Explicit no-execution retry; retain the original plan's complete identity."""
+ from tools.skill_eval_retained import retained_reports
+
+ revision = git(ROOT, "rev-parse", "HEAD").decode().strip()
+ retained = retained_reports(api, pr, source_run, trusted_repo=ROOT,
+ renderer_revision=revision, now=now)
+ plan, results = retained["plan"], retained["reports"]
+ for result in results:
+ attach_feedback(result, retained["contexts"][result["skill"]])
+ from tools.skill_eval_auto import report_control
+
+ body = summary(plan, results, retained["evidence_url"], retained["expires_at"], now,
+ evidence_state="available", execution_control=report_control(api, plan))
+ body += ("\n\nPresentation implementation: `" + revision + "`. "
+ "Original execution plan and frozen verifier bindings are retained; "
+ "this report-only refresh starts no model execution or budget reservation.")
+ if output is not None:
+ write_json(output / "pr-plan.json", plan)
+ write_json(output / "report-renderer.json", {"renderer_revision": revision,
+ "source_run": source_run, "original_plan_sha256": fingerprint(plan)})
+ for result in results:
+ directory = output / result["skill"]
+ write_json(directory / "report.json", result)
+ (directory / "report.md").write_text(body, encoding="utf-8")
+ return api.publish(plan, body, writer_run)
+
+
def review_guidance(plan, results, *, evidence_state):
"""Advisory only, over trusted regenerated reports for this exact plan.
@@ -414,27 +488,29 @@ def summary(plan, results, evidence_url, expires_at, now, *, evidence_state=None
("available" if results else "planned / unexecuted"))
guidance = review_guidance(plan, results, evidence_state=evidence_state)
lines = [MARKER, "## Skill evaluation"]
+ lines += comparison_lines(plan, results, evidence_state)
if plan.get("execution_blocked"):
- lines += ["", "Paid execution blocked: " + plan["execution_blocked"] + ". "
- "This is free planning only; no alternate provider/model is selected."]
+ lines += ["", "Paid execution blocked: " + safe_text(plan["execution_blocked"]) + ". "
+ "New execution only: this does not invalidate available retained results; "
+ "no alternate provider/model is selected."]
if execution_control is not None:
lines += ["", "Automatic execution / cumulative PR budget: `"
+ canonical(execution_control) + "`", ""]
lines += ["### Advisory: " + guidance["disposition"],
"Retained evidence availability: **" + evidence_state + "**. "
"No approval or merge authorization is issued."]
- lines.extend("- " + reason for reason in guidance["reasons"][:6])
+ lines.extend("- " + safe_text(reason) for reason in guidance["reasons"][:6])
if len(guidance["reasons"]) > 6:
lines += [f"- {len(guidance['reasons']) - 6} additional reasons in details below."]
lines += [""]
counts = {k: 0 for k in ("fix", "regression", "unchanged_correct", "unchanged_incorrect",
"inconclusive", "missing_pair")}
- if expired:
+ if expired or evidence_state == "expired":
lines += ["**EXPIRED: retained evidence is no longer verifiable.**"]
elif evidence_state in ("missing", "invalid"):
lines += ["**INSUFFICIENT EVIDENCE: retained artifacts are " + evidence_state + ".**"]
elif results:
- for result in results:
+ for _, result in bound_results(plan, results):
for key in counts:
counts[key] += result["counts"][key]
lines += ["**RETAINED RESULTS**", "", "Outcomes: `" + canonical(counts) + "`",
@@ -464,14 +540,15 @@ def summary(plan, results, evidence_url, expires_at, now, *, evidence_state=None
"Content/source review and authorship admission remain separate.",
"Trusted regrading checks retained answers, not model/hardware provenance or "
"a hostile executor; this is not an efficacy certification.",
- *["- " + reason for reason in guidance["reasons"][6:]],
+ *["- " + safe_text(reason) for reason in guidance["reasons"][6:26]],
"", "Per-case outcomes (discovery excluded; regressions and gaps first):",
- "```json", canonical(sorted(guidance["observations"], key=lambda item:
+ "```json", json_for_code(sorted(guidance["observations"], key=lambda item:
(item["outcome"] not in ("regression", "inconclusive", "missing_pair",
"unchanged_incorrect"), item["skill"]))[:20]), "```",
f"Showing at most 20 of {len(guidance['observations'])} pairs; "
"the evidence link contains all outcomes, including omitted regressions/gaps.",
"",
+ *[line for line in evidence_binding_lines(plan, results, evidence_state)],
f"Plan SHA-256: `{fingerprint(plan)}`",
f"Base: `{plan['base']}` · Head: `{plan['head']}`",
f"Trusted implementation: `{plan['version']}`", "",
@@ -479,21 +556,25 @@ def summary(plan, results, evidence_url, expires_at, now, *, evidence_state=None
"Budget: `" + canonical(budget(plan)) + "`", "",
"| Skill | Comparison | Cases | Estimated calls |",
"| --- | --- | --- | --- |"]
- for item in plan["comparisons"]:
+ for item in plan["comparisons"][:50]:
if "plan_sha256" in item:
- cases = ", ".join(c["id"] + " (" + c["split"] + ")" for c in item["cases"])
- lines.append(f"| {item['skill']} | {item['mode']} | {cases} | {item['calls']} |")
+ cases = ", ".join(safe_text(c["id"], 128) + " (" + safe_text(c["split"], 24) + ")"
+ for c in item["cases"][:20])
+ lines.append(f"| {code(item['skill'])} | {code(item['mode'])} | {cases} | "
+ f"{item['calls']} |")
lines.append("\nFrozen exam hashes: `" + canonical(item["exam"]) + "`\n")
else:
- lines.append(f"| {item['skill']} | unsupported snapshot | — | 0 |")
+ lines.append(f"| {code(item['skill'])} | unsupported snapshot | — | 0 |")
if "provider_config" in plan:
lines += ["", "Provider/settings: `" + canonical(plan["provider_config"]) + "`"]
lines += ["", "", "", "",
f"Uncovered / unsupported paths ({len(gaps)})
", ""]
# A collapsed HTML container still requires data escaping for untrusted Git paths.
- for gap in gaps:
- lines.append("- " + gap["coverage"] + ": "
- + html.escape(canonical(gap["path"])).replace("@", "@") + "")
+ for gap in gaps[:100]:
+ lines.append("- " + safe_text(gap["coverage"], 24) + ": "
+ + html.escape(canonical(gap["path"][:1024])).replace("@", "@") + "")
+ if len(gaps) > 100:
+ lines.append("Additional paths omitted; see retained evidence.")
if not gaps:
lines.append("(none)")
lines += ["", " "]
@@ -890,6 +971,13 @@ def main():
handle.write("client_kind=" + (plan["provider_config"]["client"]["kind"]
if "provider_config" in plan else "codex") + "\n")
return
+ if args.mode == "publish" and inputs.get("action") == "publish":
+ if not inputs.get("source_run"):
+ raise ValueError("Publish-only retry requires the original execution run ID")
+ receipt = publish_historical(api, pr, int(inputs["source_run"]), run_id,
+ now=time.time(), output=output)
+ print(receipt["html_url"])
+ return
plan = compare.read_json(output / "pr-plan.json")
if args.mode == "authorize":
actor = os.environ["GITHUB_ACTOR"]
diff --git a/tools/skill_eval_retained.py b/tools/skill_eval_retained.py
new file mode 100644
index 00000000..9c346250
--- /dev/null
+++ b/tools/skill_eval_retained.py
@@ -0,0 +1,258 @@
+"""Read-only bridge to an authenticated historical main-ancestor verifier.
+
+This is deliberately not an execution/reservation entry point. The caller must run
+from a trusted main checkout and publish through GitHub.publish's live-PR guards.
+Artifact Python is hash-checked data, never an import or executable.
+"""
+
+import hashlib
+import io
+import json
+import re
+import stat
+import subprocess
+import sys
+import tarfile
+import tempfile
+import zipfile
+from pathlib import Path, PurePosixPath
+
+MAX_BYTES = 100 * 1024 * 1024
+MAX_FILES = 10000
+SHA = r"[0-9a-f]{40}"
+DIGEST = r"[0-9a-f]{64}"
+WORKFLOW = ".github/workflows/skill-eval-pr.yml"
+GIT = ("git", "--no-replace-objects", "--no-lazy-fetch")
+
+
+def _hash(raw):
+ return hashlib.sha256(raw).hexdigest()
+
+
+def _path(name):
+ if (not isinstance(name, str) or not name or "\\" in name or "\x00" in name
+ or str(PurePosixPath(name)) != name or name.startswith("/")
+ or any(p in {"", ".", ".."} for p in name.split("/"))):
+ raise ValueError("Unsafe retained path")
+ return name
+
+
+def _archive(raw):
+ if not isinstance(raw, bytes) or len(raw) > MAX_BYTES:
+ raise ValueError("Missing or oversized retained archive")
+ with zipfile.ZipFile(io.BytesIO(raw)) as zipped:
+ entries = zipped.infolist()
+ names = [m.orig_filename for m in entries]
+ if (len(entries) > MAX_FILES or len(set(names)) != len(names)
+ or sum(m.file_size for m in entries) > MAX_BYTES):
+ raise ValueError("Duplicate or oversized retained archive")
+ files = {}
+ for member in entries:
+ name = member.orig_filename
+ mode = stat.S_IFMT(member.external_attr >> 16)
+ if member.is_dir():
+ _path(name[:-1])
+ if mode not in (0, stat.S_IFDIR):
+ raise ValueError("Unsafe retained archive directory")
+ continue
+ _path(name)
+ if mode not in (0, stat.S_IFREG):
+ raise ValueError("Unsafe retained archive file")
+ files[name] = zipped.read(member)
+ return files
+
+
+def _git(repo, *args):
+ return subprocess.check_output([*GIT, *args], cwd=repo, stderr=subprocess.DEVNULL)
+
+
+def _trusted_code(repo, revision, renderer, plan, destination):
+ for value in (revision, renderer):
+ if not isinstance(value, str) or re.fullmatch(SHA, value) is None:
+ raise ValueError("Expected immutable trusted revision")
+ if _git(repo, "rev-parse", "HEAD").decode().strip() != renderer:
+ raise ValueError("Renderer is not the trusted checkout")
+ try:
+ _git(repo, "merge-base", "--is-ancestor", revision, renderer)
+ except subprocess.CalledProcessError as error:
+ raise ValueError("Verifier is not a trusted main ancestor") from error
+ for field, name in (("adapter_sha256", "tools/skill_eval_pr.py"),
+ ("mapping_sha256", "tools/skill_eval_pr_cases.json")):
+ if plan.get(field) != _hash(_git(repo, "show", f"{revision}:{name}")):
+ raise ValueError("Historical Git blob binding mismatch")
+ # Git archive is trusted source; avoid checkout hooks, filters and submodules.
+ raw = _git(repo, "archive", "--format=tar", revision, "src", "tools")
+ if len(raw) > MAX_BYTES:
+ raise ValueError("Oversized historical source archive")
+ with tarfile.open(fileobj=io.BytesIO(raw)) as archive:
+ members = archive.getmembers()
+ if len(members) > MAX_FILES or sum(m.size for m in members) > MAX_BYTES:
+ raise ValueError("Oversized historical source tree")
+ for member in members:
+ if not member.name.endswith(".py"):
+ continue
+ _path(member.name)
+ if not member.isfile():
+ raise ValueError("Historical Python must be a regular Git blob")
+ path = destination / member.name
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_bytes(archive.extractfile(member).read())
+
+
+def _materialize(files, plan, destination, code):
+ """Select only frozen inputs, bound to the authenticated original PR plan."""
+ contexts = {}
+ for item in plan["comparisons"]:
+ skill = item["skill"]
+ if not isinstance(skill, str) or re.fullmatch(r"[a-z0-9][a-z0-9._-]*", skill) is None:
+ raise ValueError("Unsafe retained skill")
+ if skill in contexts:
+ raise ValueError("Duplicate retained comparison")
+ if "plan_sha256" not in item:
+ continue
+ name = skill + "/plan/freeze.json"
+ raw = files[name]
+ if _hash(raw) != item["plan_sha256"]:
+ raise ValueError("Retained freeze binding mismatch")
+ frozen = json.loads(raw)
+ if (frozen["skill"] != skill or frozen["exam"] != item["exam"]
+ or any(frozen["arms"][role]["git_revision"] != plan[key]
+ for role, key in (("baseline", "base"), ("candidate", "head")))):
+ raise ValueError("Retained comparison identity mismatch")
+ selected = {name: raw}
+ for path, digest in frozen["files"].items():
+ _path(path)
+ if not path.startswith(("plan/", "validator/")) or path == "plan/freeze.json":
+ raise ValueError("Unsafe frozen input path")
+ payload = files[skill + "/" + path]
+ if (not isinstance(digest, str) or re.fullmatch(DIGEST, digest) is None
+ or _hash(payload) != digest):
+ raise ValueError("Frozen input hash mismatch")
+ selected[skill + "/" + path] = payload
+ for path, digest in frozen["controllers"].items():
+ _path(path)
+ if not path.startswith(("src/", "tools/")) or not path.endswith(".py"):
+ raise ValueError("Unsafe historical controller")
+ if _hash((code / path).read_bytes()) != digest:
+ raise ValueError("Historical controller hash mismatch")
+ # Unexpected plan/validator files cannot hide behind selective extraction.
+ prefixes = (skill + "/plan/", skill + "/validator/")
+ actual = {name for name in files if name.startswith(prefixes)}
+ if actual != set(selected):
+ raise ValueError("Frozen input file set mismatch")
+ for name, payload in selected.items():
+ path = destination / name
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_bytes(payload)
+ contexts[skill] = {"adapter": frozen.get("adapter", "audit"), "exam": frozen["exam"],
+ "tasks": json.loads(selected[skill + "/plan/tasks.json"]),
+ "schema": json.loads(selected[skill + "/plan/schema.json"])}
+ (destination / "pr-plan.json").write_bytes(files["pr-plan.json"])
+ return contexts
+
+
+# Fixed program: no artifact-derived command, imports, Python flags or executable.
+# Ancestor source is trusted, not hostile sandbox input. The audit hook and denied
+# entry points additionally prevent accidental execution/network in report mode.
+REPORT_PROGRAM = r'''
+import json, sys
+from pathlib import Path
+code, data, output = map(Path, sys.argv[1:])
+sys.dont_write_bytecode = True
+sys.path[:0] = [str(code), str(code / "src")]
+def deny(*args, **kwargs):
+ raise RuntimeError("Historical bridge is report-only")
+def audit(event, args):
+ if (event.startswith(("socket.", "subprocess.", "os.exec", "os.spawn"))
+ or event in {"os.system", "os.fork", "os.posix_spawn"}):
+ deny()
+sys.addaudithook(audit)
+from tools import skill_eval_pr as pr
+from tools import skill_eval_compare as compare
+for module, names in ((pr, ("main", "prepare", "execute", "authorize", "claim")),
+ (compare, ("main", "prepare", "execute", "provider_client", "run_trial"))):
+ for name in names:
+ setattr(module, name, deny)
+pr.import_results((data / "results.zip").read_bytes(), data)
+results = pr.reports(data)
+output.write_text(json.dumps(results, allow_nan=False), encoding="utf-8")
+'''
+
+
+def _regrade(code, data, home):
+ result = home / "report.json"
+ # -I ignores PYTHONPATH/user-site; a fresh HOME and explicit allowlist prevent
+ # all GitHub/provider/proxy/cloud credentials from entering the interpreter.
+ environment = {"HOME": str(home), "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8"}
+ subprocess.run([sys.executable, "-I", "-c", REPORT_PROGRAM, str(code), str(data), str(result)],
+ cwd=code, env=environment, check=True, timeout=120,
+ stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+ if result.stat().st_size > MAX_BYTES:
+ raise ValueError("Oversized historical report")
+ reports = json.loads(result.read_text(encoding="utf-8"))
+ if not isinstance(reports, list):
+ raise ValueError("Invalid historical report")
+ return reports
+
+
+def retained_reports(api, pr_number, source_run, *, trusted_repo, renderer_revision, now):
+ """Authenticate and regrade retained data; performs no GitHub writes or spend.
+
+ ``trusted_repo``/``renderer_revision`` MUST be the caller's trusted main
+ checkout, never a PR checkout. Missing Git ancestry fails closed (no fetch).
+ Failures propagate so a valid existing comment is not replaced by PLANNED.
+ """
+ from tools import skill_eval_pr as pr
+ from tools.skill_eval_auto import retained_source
+
+ if (type(source_run) is not int or source_run <= 0
+ or type(pr_number) is not int or pr_number <= 0):
+ raise ValueError("Expected positive retained run and PR IDs")
+ run = api.request("GET", f"/actions/runs/{source_run}")
+ repository = api.request("GET", "")
+ if (run.get("id") != source_run or run.get("run_attempt") != 1
+ or run.get("path") != WORKFLOW or run.get("event") != "pull_request_target"
+ or run.get("status") != "completed"
+ or run.get("display_title") != f"skill-eval plan or publish PR {pr_number}"
+ or type(repository.get("id")) is not int
+ or run.get("repository", {}).get("id") != repository["id"]
+ or run.get("repository", {}).get("full_name") != repository.get("full_name")):
+ raise ValueError("Untrusted retained source workflow")
+ raw, expires = pr.artifact_download(api, source_run, "skill-eval-plan", now)
+ if raw is None or now >= expires:
+ raise ValueError("Expired retained plan")
+ files = _archive(raw)
+ if len(files.get("pr-plan.json", b"")) > 4_000_000:
+ raise ValueError("Oversized retained plan")
+ plan = json.loads(files["pr-plan.json"])
+ if (plan["pr"] != pr_number or plan["repository"] != repository["full_name"]
+ or plan["head"] != run.get("head_sha")
+ or api.pr(pr_number) != {"head": plan["head"], "base": plan["base"]}
+ or retained_source(api, plan) != source_run):
+ raise ValueError("Stale or unauthenticated retained plan")
+ data, result_expiry = pr.artifact_zip(api, source_run, "skill-eval-results", now)
+ expires = min(expires, result_expiry)
+ if data is None or now >= expires:
+ raise ValueError("Expired retained results")
+ _archive(data) # Validate names/modes/size before historical import's whitelist.
+ with tempfile.TemporaryDirectory(prefix="skill-eval-retained-") as temporary:
+ root = Path(temporary)
+ code, inputs, home = (root / name for name in ("code", "inputs", "home"))
+ for path in (code, inputs, home):
+ path.mkdir()
+ _trusted_code(trusted_repo, plan["version"], renderer_revision, plan, code)
+ contexts = _materialize(files, plan, inputs, code)
+ (inputs / "results.zip").write_bytes(data)
+ reports = _regrade(code, inputs, home)
+ expected = {item["skill"]: item for item in plan["comparisons"] if "plan_sha256" in item}
+ if len(reports) != len(expected) or {r["skill"] for r in reports} != set(expected):
+ raise ValueError("Historical report comparison mismatch")
+ for report in reports:
+ item = expected[report["skill"]]
+ if (report["plan_sha256"] != item["plan_sha256"] or report["exam"] != item["exam"]
+ or any(report["arms"][role]["git_revision"] != plan[key]
+ for role, key in (("baseline", "base"), ("candidate", "head")))):
+ raise ValueError("Historical report binding mismatch")
+ return {"plan": plan, "reports": reports, "contexts": contexts, "expires_at": expires,
+ "evidence_url": f"https://github.com/{plan['repository']}/actions/runs/{source_run}",
+ "renderer_revision": renderer_revision}