P04-01: 同時発生の更新規約

2026-10-18 14:36:16 +09:00
parent befa0ef1f6
commit acab529c8d
23 changed files with 2003 additions and 186 deletions
@@ -2,6 +2,7 @@
 """S2b-only native entry. --check-only never loads ownership helpers or starts Godot."""
 from pathlib import Path
 import argparse
+import copy
 import hashlib
 import importlib.util
 import json
@@ -62,19 +63,42 @@ def main():
    parser.add_argument("--cache-run", type=Path, default=CACHE_RUN)
    parser.add_argument("--godot", type=Path, default=ENGINE)
    parser.add_argument("--check-only", action="store_true")
+    parser.add_argument("--main")
+    parser.add_argument("--submission")
+    parser.add_argument("--tested-tree")
+    parser.add_argument("--historical-r17", action="store_true")
    args = parser.parse_args()
+    shared = module("fresh_q_shared", HERE.parent / "s5/run_s5_native.py")
+    mode = shared.select_mode(args)
+    fresh = mode == "fresh-q"
+    if fresh and "--project-root" not in sys.argv:
+        raise ValueError("fresh-q requires explicit --project-root")
+    if fresh and "--cache-run" not in sys.argv:
+        args.cache_run = None
+    engine_descriptor = shared.engine_pin(args.godot) if fresh else None
+    provenance = None
+
+    if args.project_root.is_symlink():
+        raise ValueError("project root symlink")
    root = args.project_root.resolve()
    out = args.output.resolve()
+    if fresh:
+        if not out.is_relative_to(BUILD.resolve()) or not args.plan.resolve().is_relative_to(BUILD.resolve()):
+            raise ValueError("fresh plan/output must stay inside BUILD")
+        if out.exists() or out.is_symlink() or (args.check_only and args.plan.exists()):
+            raise ValueError("fresh output and plan required")
    if not out.is_relative_to(BUILD.resolve()) or not args.plan.resolve().is_relative_to(BUILD.resolve()):
        raise ValueError("plan/output must stay in task build")
    r = module("s2b_standard_runner", root / "tools/run_all_tests.py")
    # Preserve the reviewed source_refs -> exactly one QA override order.
    source_before = r.source_refs(root)
+    if fresh:
+        provenance = shared.fresh_q_guard(root, args.main, args.submission, args.tested_tree)
    refs = dict(source_before)
    refs[TEST] = pin(QA)
-    template_guard(root)
+    if not fresh: template_guard(root)
    expected = {n: dict(bytes=p["bytes"], sha256=p["sha256"]) for n, p in refs.items()}
-    fixed = [pin(p) for p in MATERIALS + [
+    fixed = [pin(p) for p in ([] if fresh else MATERIALS) + [
        QA, Path(__file__), HERE / "S2B_ROOT_RUN.md",
        root / "tests/transport/test_runtime_transport.gd",
        root / "tests/staff/test_rescue_elevator.gd",
@@ -83,7 +107,7 @@ def main():
    selected = dict(root_before=source_before, map={n: p["sha256"] for n, p in refs.items()}, timeout=1200)
    cache_note = None
    cache_receipts = []
-    for leaf in ["runner-ended.json", "runner-input.json"]:
+    for leaf in (["runner-ended.json", "runner-input.json"] if args.cache_run is not None else []):
        path = args.cache_run / leaf
        try:
            cache_receipts.append(pin(path))
@@ -97,6 +121,16 @@ def main():
                cache_receipts=cache_receipts, cache_access=cache_note,
                phases=["import", "parse", "writer", "reader"],
                native_engine_launches=0, whole_goal_accepted=False, unit_accepted=False)
+    if fresh:
+        uid_plan, uid_snapshot, uid_binding = shared.load_uid_helper(root, r, source_before)
+        uid_source = copy.deepcopy(expected)
+        uid_plan_map = uid_plan(refs)
+        plan.update(protocol="p04-s2b-fresh-q-v1", mode=mode, provenance=provenance, engine=engine_descriptor,
+                    engine_sha256=engine_descriptor["sha256"], uid_helper=uid_binding,
+                    uid_plan=uid_plan_map, import_mode="fresh" if args.cache_run is None else "cache",
+                    historical_r17=False, fresh_q_evidence=False)
+        fixed.append(pin(HERE.parent / "s5/run_s5_native.py"))
+    # No historical execution is admitted.
    # Check-only only reads source/materials and writes the plan in task build.
    # It deliberately does not read default-player data, inspect processes or stat Godot.
    if args.check_only:
@@ -109,20 +143,31 @@ def main():
                              QA_overrides=[TEST], fixed_pins=len(fixed), plan=str(args.plan),
                              plan_sha256=r.sha(args.plan), cache_access=cache_note), ensure_ascii=False))
        return 0
-    r.require(root == PRODUCT.resolve() and len(source_before) == 1338, "Root native requires product 1338; worker plan is not native evidence")
+    r.require(fresh, "historical mode is read-only")
    r.require(args.plan.is_file() and r.read(args.plan) == plan, "run exact check-only plan first")
+    reports, phase_call, uid_state = shared.make_uid_policy(r, uid_snapshot, uid_source, uid_plan_map, out)
+    engine_checks = []
+    def fresh_engine_check(label):
+        value = shared.engine_pin(args.godot, engine_descriptor)
+        engine_checks.append(dict(label=label, descriptor=value))
    def fixed_guard():
-        template_guard(root)
+        fresh_engine_check("fixed")
+        r.require(shared.fresh_q_guard(root, args.main, args.submission, args.tested_tree) == provenance,
+                  "fresh Q provenance changed")
+        r.require(pin(root / uid_binding["helper_path"])["sha256"] == uid_binding["helper_sha256"],
+                  "UID helper changed")
+        if not fresh: template_guard(root)
        for value in fixed + cache_receipts:
            r.require(pin(value["path"]) == value, "fixed input changed: " + value["path"])
    fixed_guard()
    # Native admission is Root-only and fails closed on unreadable cache.
    r.require(cache_note is None, "Root must read exact cache receipts: " + str(cache_note))
    old = args.cache_run
-    r.require(r.read(old / "runner-ended.json")["passed"], "ended cache run must have passed")
-    old_map = r.read(old / "runner-input.json")["selected_source_map"]
-    current_map = {n: p["sha256"] for n, p in source_before.items()}
-    r.require(old_map == current_map, "ended cache selected source must equal current product")
+    if old is not None:
+        r.require(r.read(old / "runner-ended.json")["passed"], "ended cache run must have passed")
+        old_map = r.read(old / "runner-input.json")["selected_source_map"]
+        current_map = {n: p["sha256"] for n, p in source_before.items()}
+        r.require(old_map == current_map, "ended cache selected source must equal current product")
    protected = r.default_player(root)
    outer = module("s2b_owned_outer", r.SUPPORT / "owned_outer.py")
    r.require(not outer.native.helper["live_engines"](), "another Godot owns CPU")
@@ -167,20 +212,21 @@ def main():
            data = value["text"].encode()
            r.require(hashlib.sha256(data).hexdigest() == value["sha256"], "bootstrap pin")
            (out / "bootstrap" / name).write_bytes(data)
-        cache = old / "stage/.godot"
-        cache_before = r.inventory(cache)
-        r.require(bool(cache_before), "nonempty ended cache")
-        for name in cache_before:
-            src = cache / name
-            dst = stage / ".godot" / name
-            dst.parent.mkdir(parents=True, exist_ok=True)
-            subprocess.run(["/bin/cp", "-c", str(src), str(dst)], check=True)
-            r.require((src.stat().st_dev, src.stat().st_ino) != (dst.stat().st_dev, dst.stat().st_ino),
-                      "cache clone is a separate inode")
-        r.require(r.inventory(stage / ".godot") == cache_before, "whole cache clone hashes")
-        r.write(out / "cache-reuse.json", dict(ended_run=str(old), receipts=cache_receipts,
-                whole_cache=cache_before, independent_inode=True,
-                ordinary_import_required=True, ordinary_parse_required=True))
+        if old is not None:
+            cache = old / "stage/.godot"
+            cache_before = r.inventory(cache)
+            r.require(bool(cache_before), "nonempty ended cache")
+            for name in cache_before:
+                src = cache / name
+                dst = stage / ".godot" / name
+                dst.parent.mkdir(parents=True, exist_ok=True)
+                subprocess.run(["/bin/cp", "-c", str(src), str(dst)], check=True)
+                r.require((src.stat().st_dev, src.stat().st_ino) != (dst.stat().st_dev, dst.stat().st_ino),
+                          "cache clone is a separate inode")
+            r.require(r.inventory(stage / ".godot") == cache_before, "whole cache clone hashes")
+            r.write(out / "cache-reuse.json", dict(ended_run=str(old), receipts=cache_receipts,
+                    whole_cache=cache_before, independent_inode=True,
+                    ordinary_import_required=True, ordinary_parse_required=True))
        paths = ["res://" + n for n in sorted(refs) if n.endswith(".gd")]
        r.write(out / "parse-paths.json", paths)
        markers = module("s2b_markers", root / "tools/strict_test_markers.py")
@@ -202,7 +248,10 @@ def main():
                writer = r.read(out / "raw/writer.json")
                formal_before = {n: r.inventory(Path(p)) for n, p in writer["slots"].items()}
                r.require(len(formal_before) == 6, "three formal points for each of two fresh branches")
-            row = r.execute_phase(str(engine), stage, out, phase, test, selected, expected,
+            fresh_engine_check("before-" + phase)
+            if phase == "import" and old is None:
+                r.require(not (stage / ".godot").exists(), "fresh import cache absent")
+            row = phase_call(str(engine), stage, out, phase, test, selected, expected,
                                  previous, policy, contracts, paths, public, protected, root,
                                  outer, markers, diagnostics, owners, audio)
            phases.append(row)
@@ -237,6 +286,8 @@ def main():
    # Fixed guard always follows ownership cleanup, even if setup or an engine failed.
    try:
        fixed_guard()
+        r.require(uid_snapshot(stage, uid_source, uid_plan_map) == uid_state["accepted"], "UID seal after cleanup")
+        r.write(out / "fresh-engine-checks.json", engine_checks)
        after = r.source_refs(root)
        r.write(out / "source_after.json", after)
        r.require(after == source_before, "whole product source unchanged")
@@ -252,9 +303,15 @@ def main():
    except BaseException as caught:
        guards.append(type(caught).__name__ + ": " + str(caught))
    pids = [p.pid for p in outer.OWNED]
-    passed = (error is None and not guards and not cleanup_errors and len(phases) == 4
+    passed = (uid_state["accepted"] is not None and error is None and not guards and not cleanup_errors and len(phases) == 4
              and all(p["passed"] for p in phases) and len(pids) == len(set(pids)) == 4
              and all(p["engine_ended"] for p in cleanup))
+    if passed:
+        try:
+            shared.mirror_formal(r, out, formal_before)
+        except BaseException as caught:
+            guards.append("formal mirror: " + str(caught))
+            passed = False
    execution = dict(passed=passed, error=error, guard_errors=guards, phases=phases,
                     actual_captured_PIDs=pids, cleanup=cleanup, cleanup_errors=cleanup_errors,
                     formal_before=formal_before, formal_after=formal_after,