P28-01-COLLECTOR: 現HEADの性能基準：windowed x1/x3・Operations ON collector

2026-10-20 23:44:54 +09:00
parent 061df6f120
commit 25198e2924
2 changed files with 106 additions and 1 deletions
@@ -321,9 +321,10 @@ def main():
    planned = execution_plan(args.engine, stage, wrapper)
    admission = {"commit": commit, "source_map": before, "source_map_sha256": digest(before),
                 "readonly": pins, "engine_sha256": ENGINE_SHA, "engine_version": VERSION,
-                 "fixture_sha256": FIXTURE_SHA, "planned_argv": planned, "phases": PHASES,
+                 "fixture_sha256": FIXTURE_SHA, "planned_argv": planned, "phases": [list(p) for p in PHASES],
                 "budget_verdict_scope": VERDICT_SCOPE,
                 "reused_r02": reused_observations(args.readonly, pins)}
+    require(admission == json.loads(json.dumps(admission)), "admission JSON representation drift")
    check_admission(admission)
    if args.check:
        print(json.dumps(dict(admission, stages_created=0, engines_started=0), ensure_ascii=False, indent=2))
@@ -170,6 +170,110 @@ def synthetic_admission():


 class ReportTests(unittest.TestCase):
+    def constructed_admission(self, reused=None):
+        """Run actual CLI construction, stopping before stages or engines exist."""
+        class AdmissionReached(Exception):
+            pass
+
+        captured = {}
+        def capture(admission):
+            checker.check_admission(admission)
+            captured.update(admission)
+            raise AdmissionReached()
+
+        with tempfile.TemporaryDirectory() as temp:
+            base = Path(temp)
+            fixture = base / "fixture"
+            fixture.write_bytes(b"synthetic")
+            engine = base / "engine"
+            argv = ["collect", "--readonly", str(base), "--fixture", str(fixture),
+                    "--engine", str(engine), "--output", str(base / "output"), "--run"]
+            with patch.object(sys, "argv", argv), \
+                    patch.object(collector, "load_tool") as load_tool, \
+                    patch.object(collector, "inputs", return_value={}), \
+                    patch.object(collector, "sha", side_effect=lambda p:
+                                 collector.FIXTURE_SHA if p == fixture else collector.ENGINE_SHA), \
+                    patch.object(collector, "source_map", return_value={}), \
+                    patch.object(collector.subprocess, "check_output", side_effect=["synthetic\n", ""]), \
+                    patch.object(collector, "reused_observations",
+                                 return_value=synthetic_admission()["reused_r02"] if reused is None else reused), \
+                    patch.object(collector, "check_admission", side_effect=capture) as admission_gate, \
+                    patch.object(collector, "no_live_engine") as live_gate, \
+                    patch.object(collector, "run_engine") as engine_gate:
+                try:
+                    try:
+                        collector.main()
+                    except AdmissionReached:
+                        pass
+                finally:
+                    live_gate.assert_not_called()
+                    engine_gate.assert_not_called()
+                    self.assertEqual(load_tool.call_args_list, [
+                        unittest.mock.call(collector.EXTERNAL_IMPORTS[0])])
+                    self.assertFalse((base / "output").exists())
+                    if reused is not None:
+                        admission_gate.assert_not_called()
+        self.assertTrue(captured)
+        return captured
+
+    def admission_receipt(self, base, admission, labels):
+        collector.write(base / "admission.json", admission)
+        probe = (HERE / "measure_p28_01.gd").read_bytes()
+        (base / "measure_p28_01.gd").write_bytes(probe)
+        return {"admission": admission,
+                "phases": [{"phase": label, "phase_stop": {},
+                            "startup_project": {"profile": "synthetic-" + label}} for label in labels],
+                "source_before": {}, "source_after": {},
+                "default_player_before": {}, "default_player_after": {},
+                "cleanup_errors": [], "cleanup": [], "error": None, "passed": True,
+                "probe_check": {"sha256": hashlib.sha256(probe).hexdigest(), "passed": True}}
+
+    def test_n1_admission_roundtrip_receipt_binding(self):
+        admission = self.constructed_admission()
+        self.assertEqual(admission["phases"], [list(p) for p in collector.PHASES])
+        self.assertIsInstance(collector.PHASES, tuple)
+        with tempfile.TemporaryDirectory() as temp:
+            base = Path(temp)
+            receipt = self.admission_receipt(
+                base, admission, ["import", *collector.PARSE_PHASES, *checker.PHASES])
+            self.assertEqual(admission, json.loads((base / "admission.json").read_text()))
+            # Actual binding and full phase-order gates pass. Stop at the next
+            # gate using an intentionally incomplete synthetic phase cap.
+            with self.assertRaisesRegex(ValueError, r"^phase cap origin keys$"):
+                checker.check_receipt(receipt, base)
+
+    def test_n1_persisted_admission_tampering(self):
+        admission = self.constructed_admission()
+        for field in ("phases", "commit"):
+            with self.subTest(field=field), tempfile.TemporaryDirectory() as temp:
+                base = Path(temp)
+                receipt = self.admission_receipt(
+                    base, admission, ["import", *collector.PARSE_PHASES, *checker.PHASES])
+                persisted = json.loads((base / "admission.json").read_text())
+                if field == "phases":
+                    persisted[field][0][2] = 3
+                else:
+                    persisted[field] = "tampered"
+                (base / "admission.json").write_text(json.dumps(persisted))
+                with self.assertRaisesRegex(ValueError, r"^admission binding\." + field + "$"):
+                    checker.check_receipt(receipt, base)
+
+    def test_n2_remaining_tuple_rejected_before_startup(self):
+        reused = synthetic_admission()["reused_r02"]
+        reused["remaining_tuple"] = ("must", "reject")
+        with self.assertRaisesRegex(ValueError, r"^admission JSON representation drift$"):
+            self.constructed_admission(reused)
+
+    def test_n3_subset_receipt_rejected_by_phase_order(self):
+        admission = self.constructed_admission()
+        for labels in (["import", "parse-base", "parse-wrapper", "R0"],
+                       ["import", *collector.PARSE_PHASES, *list(checker.PHASES)[:-1]]):
+            with self.subTest(labels=labels), tempfile.TemporaryDirectory() as temp:
+                base = Path(temp)
+                receipt = self.admission_receipt(base, admission, labels)
+                with self.assertRaisesRegex(ValueError, r"^phase order$"):
+                    checker.check_receipt(receipt, base)
+
    def test_k1_audio_close_receipts(self):
        report = {"main_added": True, "audio_owner_present": True}
        close = {"coordinator": True, "passed": True, "deadline_reached": False,