Require frozen policy and ready authorized judging handoffs

roadscore/tools/judging_run.py

@@ -18,6 +18,41 @@ def save(path, value):
     temp.replace(path)
 
 
+
+def validate_frozen_configuration(root):
+    """Refuse official work unless a nonempty local freeze verifies completely."""
+    frozen = root / 'generated/judging_configuration.private.json'
+    try:
+        entries = json.loads(frozen.read_text())
+    except (OSError, ValueError) as error:
+        raise SystemExit('Valid frozen judging configuration is required') from error
+    if not isinstance(entries, dict) or not entries:
+        raise SystemExit('Frozen judging configuration must be a nonempty file/hash mapping')
+    for name, digest in entries.items():
+        if (not isinstance(name, str) or not name or Path(name).is_absolute()
+                or '..' in Path(name).parts or not isinstance(digest, str)
+                or len(digest) != 64 or any(c not in '0123456789abcdef' for c in digest)):
+            raise SystemExit('Invalid frozen judging configuration entry')
+        path = root / name
+        if not path.resolve().is_relative_to(root.resolve()):
+            raise SystemExit('Frozen judging configuration path escapes project')
+        try:
+            actual = hashlib.sha256(path.read_bytes()).hexdigest()
+        except OSError as error:
+            raise SystemExit('Frozen judging configuration file unavailable: ' + name) from error
+        if actual != digest:
+            raise SystemExit('Frozen judging configuration changed: ' + name)
+
+
+def validate_handoff_readiness(manifest, row):
+    """Preparation exports are inert until explicitly authorized and complete."""
+    if manifest.get('schema') == 'roadscore-judging-handoff-v1':
+        if manifest.get('generation_authorized') is not True:
+            raise SystemExit('Judging handoff has not been authorized for generation')
+        if row.get('preparation_ready') is not True or row.get('preparation_blockers') != []:
+            raise SystemExit('Judging handoff preparation is incomplete or blocked')
+
+
 def main():
     parser = argparse.ArgumentParser()
     parser.add_argument('manifest', type=Path)
@@ -31,11 +66,8 @@ def main():
         row.update(json.loads(ranges.read_text()).get(args.label, {}))
     if row.get('range_resolution') == 'pending metadata':
         raise SystemExit('Resolve submitted range before launching this entry')
-    frozen = ROOT / 'generated/judging_configuration.private.json'
-    if frozen.exists():
-        for name, digest in json.loads(frozen.read_text()).items():
-            if hashlib.sha256((ROOT/name).read_bytes()).hexdigest() != digest:
-                raise SystemExit('Frozen judging configuration changed: '+name)
+    validate_handoff_readiness(manifest, row)
+    validate_frozen_configuration(ROOT)
     out = ROOT / 'results/community_judging'
     out.mkdir(exist_ok=True, mode=0o700)
     ledger = out / f'official_{args.label}.json'

roadscore/tools/test_judging_run.py

@@ -0,0 +1,89 @@
+import hashlib
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+import judging_run
+
+
+class OfficialAttemptPreflightTests(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temporary.cleanup)
+        self.root = Path(self.temporary.name)
+        self.freeze = self.root / 'generated/judging_configuration.private.json'
+        self.freeze.parent.mkdir()
+        self.policy = self.root / 'policy.json'
+        self.policy.write_text('{"shared_policy": 1}')
+        self.manifest = self.root / 'manifest.json'
+        self.manifest.write_text(json.dumps({'configuration': {}, 'submissions': [
+            {'label': 'A', 'seed': 123, 'route': 'synthetic/route'}]}))
+
+    def write_freeze(self):
+        self.freeze.write_text(json.dumps({
+            'policy.json': hashlib.sha256(self.policy.read_bytes()).hexdigest()}))
+
+    def assert_preflight_rejected_without_side_effects(self):
+        ledger = self.root / 'results/community_judging/official_A.json'
+        previous = ledger.read_bytes() if ledger.exists() else None
+        with patch.object(judging_run, 'ROOT', self.root), patch.object(judging_run.subprocess, 'run') as run:
+            with patch('sys.argv', ['judging_run', str(self.manifest), '--label', 'A']):
+                with self.assertRaises(SystemExit):
+                    judging_run.main()
+            run.assert_not_called()
+        self.assertEqual(ledger.read_bytes() if ledger.exists() else None, previous)
+        if previous is None:
+            self.assertFalse((self.root / 'results').exists())
+
+    def test_missing_empty_or_malformed_freeze_blocks_before_attempt_creation(self):
+        self.assert_preflight_rejected_without_side_effects()
+        for content in ('{}', '[]', 'null', '{broken', '{"policy.json": "wrong"}'):
+            with self.subTest(content=content):
+                self.freeze.write_text(content)
+                self.assert_preflight_rejected_without_side_effects()
+
+    def test_matching_freeze_passes_but_changed_or_missing_policy_blocks(self):
+        self.write_freeze()
+        judging_run.validate_frozen_configuration(self.root)
+        self.policy.write_text('changed')
+        self.assert_preflight_rejected_without_side_effects()
+        self.policy.unlink()
+        self.assert_preflight_rejected_without_side_effects()
+
+    def test_freeze_cannot_reference_files_outside_project(self):
+        for name in ('../policy.json', '/tmp/policy.json'):
+            with self.subTest(name=name):
+                self.freeze.write_text(json.dumps({name: '0' * 64}))
+                self.assert_preflight_rejected_without_side_effects()
+
+    def test_failed_preflight_preserves_existing_official_attempt(self):
+        ledger = self.root / 'results/community_judging/official_A.json'
+        ledger.parent.mkdir(parents=True)
+        ledger.write_text('{"phase":"finished","attempt":1,"seed":123}')
+        self.assert_preflight_rejected_without_side_effects()
+
+    def test_handoff_requires_explicit_authorization_and_complete_preparation(self):
+        self.write_freeze()
+        manifest = json.loads(self.manifest.read_text())
+        manifest['schema'] = 'roadscore-judging-handoff-v1'
+        row = manifest['submissions'][0]
+        row.update(preparation_ready=True, preparation_blockers=[])
+        for authorized in (None, False, 'true', 1):
+            with self.subTest(authorized=authorized):
+                manifest['generation_authorized'] = authorized
+                self.manifest.write_text(json.dumps(manifest))
+                self.assert_preflight_rejected_without_side_effects()
+        manifest['generation_authorized'] = True
+        for ready, blockers in ((False, []), (True, ['missing camera']), (True, None)):
+            with self.subTest(ready=ready, blockers=blockers):
+                row.update(preparation_ready=ready, preparation_blockers=blockers)
+                self.manifest.write_text(json.dumps(manifest))
+                self.assert_preflight_rejected_without_side_effects()
+        row.update(preparation_ready=True, preparation_blockers=[])
+        judging_run.validate_handoff_readiness(manifest, row)
+
+
+if __name__ == '__main__':
+    unittest.main()