From 25198e2924948e5ac2c0c4774d5adfcb6267833c Mon Sep 17 00:00:00 2001 From: Agent Control Runner Date: Tue, 20 Oct 2026 23:44:54 +0900 Subject: [PATCH] =?UTF-8?q?P28-01-COLLECTOR:=20=E7=8F=BEHEAD=E3=81=AE?= =?UTF-8?q?=E6=80=A7=E8=83=BD=E5=9F=BA=E6=BA=96=EF=BC=9Awindowed=20x1/x3?= =?UTF-8?q?=E3=83=BBOperations=20ON=20collector?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../p28-01/collector/collect_p28_01.py | 3 +- .../p28-01/collector/test_collect_p28_01.py | 104 ++++++++++++++++++ 2 files changed, 106 insertions(+), 1 deletion(-) diff --git a/docs/plans/original-completion-2026-10-02/execution/p28-01/collector/collect_p28_01.py b/docs/plans/original-completion-2026-10-02/execution/p28-01/collector/collect_p28_01.py index 56fa984..707111f 100644 --- a/docs/plans/original-completion-2026-10-02/execution/p28-01/collector/collect_p28_01.py +++ b/docs/plans/original-completion-2026-10-02/execution/p28-01/collector/collect_p28_01.py @@ -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)) diff --git a/docs/plans/original-completion-2026-10-02/execution/p28-01/collector/test_collect_p28_01.py b/docs/plans/original-completion-2026-10-02/execution/p28-01/collector/test_collect_p28_01.py index 690c113..1a3ea0e 100644 --- a/docs/plans/original-completion-2026-10-02/execution/p28-01/collector/test_collect_p28_01.py +++ b/docs/plans/original-completion-2026-10-02/execution/p28-01/collector/test_collect_p28_01.py @@ -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,