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

docs/plans/original-completion-2026-10-02/execution/p28-01/collector/check_p28_01_report.py

@@ -346,6 +346,7 @@ def check_report(report):
 
 
 def check_receipt(receipt, directory):
+    from collect_p28_01 import OWNED_FILES, projection, extrapolation, guard_gate
     need(receipt["source_before"] == receipt["source_after"], "source map changed")
     need(receipt["default_player_before"] == receipt["default_player_after"], "player changed")
     need(not receipt["cleanup_errors"] and all(r["engine_ended"] is True for r in receipt["cleanup"]), "cleanup")
@@ -358,31 +359,39 @@ def check_receipt(receipt, directory):
     rows = receipt["phases"]
     need([r["phase"] for r in rows] == ["import", *PHASES], "phase order")
     for row in rows:
+        forecast = None
+        if row["phase"] in ("P1", "P2", "P3", "P4"):
+            source = PHASES[row["phase"]][0]
+            stage0 = next(r for r in rows if r["phase"] == source + "0")
+            forecast = projection(stage0)
+            equal(row["projection"], forecast, "phase projection")
+        if row.get("stop_reason") == "projected_over_limit":
+            need(forecast is not None and forecast["stop_reason"] == "projected_over_limit"
+                 and row["engines_started"] == 0 and row["passed"] is False,
+                 "invalid skipped phase")
+            need("report" not in row and "raw" not in row, "skipped phase claims execution")
+            continue
+        need(forecast is None or forecast["stop_reason"] is None, "started over projected limit")
         need(row["timeout_seconds"] == 1800 and row["passed"] is True, "phase timeout/pass")
         need(row["probe_check"] == probe, "phase execution copy check")
+        need(set(row["owned_files"]) == set(OWNED_FILES), "owned evidence missing")
+        for name, pin in row["owned_files"].items():
+            data = (directory / row["phase"] / name).read_bytes()
+            need(hashlib.sha256(data).hexdigest() == pin["sha256"] and len(data) == pin["bytes"],
+                 "owned evidence changed")
         for name, pin in row["raw"].items():
             data = (directory / row["phase"] / name).read_bytes()
             need(hashlib.sha256(data).hexdigest() == pin["sha256"] and len(data) == pin["bytes"], "raw SHA")
         if row["phase"] == "import":
             continue
+        guard_gate(row["guard"], Path(row["runtime_profile"]), row["runtime_name"])
         stdout = (directory / row["phase"] / "stdout.log").read_text()
         printed = [normalize_report(json.loads(x[len(PREFIX):])) for x in stdout.splitlines() if x.startswith(PREFIX)]
         need(printed == [row["report"]], "raw/report mismatch")
         need(row["capture"]["actual_captured_pids"] == [row["report"]["pid"]], "PID mismatch")
         check_report(row["report"])
-        need(row["report"]["stop_reason"] == "complete", "collection includes stopped phase")
-        if row["phase"] in ("P1", "P2", "P3", "P4"):
-            stage0 = next(r for r in rows if r["phase"] == row["report"]["input"] + "0")
-            duration = stage0["extrapolation"]["seconds_per_phase"]
-            startup = stage0["report"]["startup_wall_seconds"]
-            equal(row["projection"], {"seconds_per_phase": duration, "startup_wall_seconds": startup,
-                  "projected_total_seconds": duration + startup, "limit_seconds": 1800,
-                  "stage0_phase": stage0["phase"], "stop_reason": None}, "phase projection")
-            need(duration + startup <= 1800, "phase exceeds projected limit")
         if row["phase"] in ("R0", "O0"):
-            estimate = row["extrapolation"]
+            equal(row["extrapolation"], extrapolation(row["report"]), "extrapolation")
-            equal(estimate["seconds_per_phase"], row["report"]["elapsed_wall_seconds"] * 150 /
-                  row["report"]["actual_minutes"], "extrapolation")
 
 
 def check_probe(text):

docs/plans/original-completion-2026-10-02/execution/p28-01/collector/collect_p28_01.py

@@ -106,12 +106,68 @@ def checked_probe(path):
 
 def projection(stage0):
     evidence = stage0["extrapolation"]
-    seconds = evidence["seconds_per_phase"] + stage0["report"]["startup_wall_seconds"]
+    seconds = (evidence["seconds_per_phase"] + stage0["report"]["startup_wall_seconds"]
+               if evidence["seconds_per_phase"] is not None else None)
     return {"seconds_per_phase": evidence["seconds_per_phase"],
             "startup_wall_seconds": stage0["report"]["startup_wall_seconds"],
             "projected_total_seconds": seconds, "limit_seconds": LIMIT,
             "stage0_phase": stage0["phase"],
-            "stop_reason": "projected_over_limit" if seconds > LIMIT else None}
+            "stop_reason": "projected_over_limit" if seconds is not None and seconds > LIMIT else None}
+
+
+OWNED_FILES = ("process.json", "started.json", "ended.json", "adapter-owned-cleanup.json",
+               "outer-owned-finally.json")
+
+
+def extrapolation(report):
+    minutes = report["actual_minutes"]
+    return {"observed_minutes": minutes, "observed_seconds": report["elapsed_wall_seconds"],
+            "target_minutes": 150,
+            "seconds_per_phase": report["elapsed_wall_seconds"] * 150 / minutes if minutes else None,
+            "reason": None if minutes else "no_minutes_observed",
+            "method": "stage0 wall seconds per game minute; x1/x3 same conservative estimate"}
+
+
+def stage_gate(before, after, importing=False):
+    require(all(after.get(k) == v for k, v in before.items()), "stage files changed")
+    added = set(after) - set(before)
+    require(not added or importing and all(
+        (n.endswith(".gd.uid") and n[:-4] in before) or
+        (n.endswith(".import") and n[:-7] in before) for n in added), "unexpected stage files")
+
+
+def guard_gate(value, runtime, name):
+    expected = {"verified": True, "conditions": [True] * 5, "conditions_checked": 5,
+                "actual_user_data_dir": str(runtime), "expected_user_data_dir": str(runtime),
+                "globalized_probe": str(runtime / "guard-probe"), "custom_user_dir_name": name,
+                "fresh_before_parent_initialize": True, "profile_writes_before_guard": 0,
+                "project_settings_persisted": False}
+    require(all(k in value and type(value[k]) is type(v) and value[k] == v
+                for k, v in expected.items()), "profile guard failure")
+
+
+def bound_report(folder, row):
+    printed = [line.split("=", 1)[1] for line in (folder / "stdout.log").read_text().splitlines()
+               if line.startswith("P28_01_PERF_REPORT=")]
+    require(len(printed) == 1, "exactly one raw report required")
+    report = normalize_report(json.loads(printed[0]))
+    require(row["capture"]["actual_captured_pids"] == [report["pid"]], "PID binding")
+    return report
+
+
+def phase_schedule(rows):
+    """Caller appends each executed result; exceptions abort the consuming loop."""
+    for index, phase in enumerate((("import", "", 0, 0, 0),) + PHASES):
+        label, source, _, _, _ = phase
+        forecast = None
+        if label in ("P1", "P2", "P3", "P4"):
+            stage0 = next(r for r in rows if r["phase"] == source + "0")
+            forecast = projection(stage0)
+            if forecast["stop_reason"]:
+                rows.append({"phase": label, "passed": False, "engines_started": 0,
+                             "stop_reason": "projected_over_limit", "projection": forecast})
+                continue
+        yield index, phase, forecast
 
 
 def run_engine(outer, command, folder, environment):
@@ -140,7 +196,11 @@ def run_engine(outer, command, folder, environment):
         len(capture.get("actual_captured_pids", [])) == 1, all(raw.values()), not bad))
     row = {"passed": clean, "process": result, "capture": capture, "raw": raw,
            "diagnostics": bad, "timeout_seconds": LIMIT, "argv": command}
-    write(folder / "process.json", row)
+    row["owned_files"] = {name: {"sha256": sha(folder / name),
+                                "bytes": (folder / name).stat().st_size}
+                          for name in OWNED_FILES if (folder / name).is_file()}
+    row["passed"] = clean and set(row["owned_files"]) == set(OWNED_FILES)
+    write(folder.parent / (folder.name + "-collector-receipt.json"), row)
     return row
 
 
@@ -252,23 +312,16 @@ def main():
                 "build": {"engine_string": VERSION, "engine_hash": "ed1daf0bf", "engine_sha256": ENGINE_SHA,
                           "project_commit": commit, "source_map_sha256": digest(before)}}
         public = (stage / "project.godot").read_bytes()
-        expected = before
+        expected = standard.inventory(stage, stage=True)
-        for index, phase in enumerate((("import", "", 0, 0, 0),) + PHASES):
+        require(expected == before, "initial whole stage differs")
+        for index, phase, forecast in phase_schedule(rows):
             label, source, speed, warmup, minutes = phase
             require(source_map(standard, ROOT) == before and standard.default_player(ROOT) == player,
                     "source/player changed before phase")
-            require(source_map(standard, stage) == expected, "stage changed before phase")
+            require(standard.inventory(stage, stage=True) == expected, "stage changed before phase")
             require(all(sha(Path(p)) == pin for p, pin in script_pins.items()), "execution scripts changed")
             require(checked_probe(probe) == probe_check, "probe check changed")
             require(sha(args.fixture) == FIXTURE_SHA, "input_shortfall: changed input")
-            forecast = None
-            if label in ("P1", "P2", "P3", "P4"):
-                stage0 = next(r for r in rows if r["phase"] == source + "0")
-                forecast = projection(stage0)
-                if forecast["stop_reason"]:
-                    rows.append({"phase": label, "passed": False, "engines_started": 0,
-                                 "stop_reason": "projected_over_limit", "projection": forecast})
-                    raise ValueError("projected_over_limit: " + label)
             name = "TheTree-PCK-tests-" + uuid.uuid4().hex + "-p28-startup"
             startup = standard.profile_path(name)
             require(not startup.exists(), "startup profile not fresh")
@@ -293,39 +346,28 @@ def main():
             row["probe_check"] = probe_check
             rows.append(row)
             require(row["passed"], "native phase failed; raw retained: " + label)
-            after = source_map(standard, stage)
+            after = standard.inventory(stage, stage=True)
             if label == "import":
-                require(all(after.get(k) == v for k, v in before.items()), "import changed source")
+                stage_gate(expected, after, importing=True)
-                added = set(after) - set(before)
-                require(all((n.endswith(".gd.uid") and n[:-4] in before) or
-                            (n.endswith(".import") and n[:-7] in before) for n in added),
-                        "unexpected import-generated files")
                 expected = after
             else:
                 require(after == expected, "stage changed during measurement")
                 observation = json.loads(guard_path.read_text())
-                require(observation["verified"] is True and observation["conditions"] == [True] * 5
+                guard_gate(observation, runtime, runtime_name)
-                        and observation["actual_user_data_dir"] == str(runtime)
+                row["runtime_profile"] = str(runtime)
-                        and observation["fresh_before_parent_initialize"] is True
+                row["runtime_name"] = runtime_name
-                        and observation["profile_writes_before_guard"] == 0
+                report = bound_report(out / label, row)
-                        and observation["project_settings_persisted"] is False, "profile guard failure")
-                raw = (out / label / "stdout.log").read_text()
-                printed = [line.split("=", 1)[1] for line in raw.splitlines() if line.startswith("P28_01_PERF_REPORT=")]
-                require(len(printed) == 1, "exactly one raw report required")
-                report = normalize_report(json.loads(printed[0]))
-                require(report["pid"] == row["capture"]["actual_captured_pids"][0], "PID binding")
                 row.update(report=report, report_sha256=digest(report), guard=observation,
                            guard_sha256=sha(guard_path))
-                write(out / label / "report.json", report)
+                write(out / (label + "-report.json"), report)
                 check_report(report)
-                require(report["stop_reason"] == "complete", "phase stopped: " + report["stop_reason"])
+                row["stop_reason"] = report["stop_reason"]
                 if label in ("R0", "O0"):
-                    duration = report["elapsed_wall_seconds"]
+                    row["extrapolation"] = extrapolation(report)
-                    row["extrapolation"] = {"observed_minutes": report["actual_minutes"],
-                                            "observed_seconds": duration, "target_minutes": 150,
-                                            "seconds_per_phase": duration * 150 / report["actual_minutes"],
-                                            "method": "stage0 wall seconds per game minute; x1/x3 same conservative estimate"}
             require(standard.default_player(ROOT) == player, "default player changed")
+            require(source_map(standard, ROOT) == before, "source changed after phase")
+            require(sha(args.fixture) == FIXTURE_SHA, "input changed after phase")
+            require(all(sha(Path(p)) == pin for p, pin in script_pins.items()), "scripts changed after phase")
     except BaseException as caught:
         error = repr(caught)
     finally:

docs/plans/original-completion-2026-10-02/execution/p28-01/collector/measure_p28_01.gd

@@ -533,6 +533,9 @@ func summary(segment: String) -> Dictionary:
 
 func finish(reason: String) -> void:
 	ready_to_sample = false
+	# Early input shortfalls also need observed startup time for stage0 evidence.
+	if not report.has("startup_wall_seconds"):
+		report["startup_wall_seconds"] = float(Time.get_ticks_usec()) / 1000000.0
 	report["stop_reason"] = reason
 	report["actual_minutes"] = int(clock.absolute_minute) - start_minute if start_us > 0 else 0
 	report["elapsed_wall_seconds"] = float(Time.get_ticks_usec() - start_us) / 1000000.0 if start_us > 0 else 0.0

docs/plans/original-completion-2026-10-02/execution/p28-01/collector/test_collect_p28_01.py

@@ -6,6 +6,8 @@ import hashlib
 import json
 from pathlib import Path
 import sys
+import tempfile
+import time
 from types import SimpleNamespace
 import unittest
 from unittest.mock import patch
@@ -17,6 +19,14 @@ import collect_p28_01 as collector
 HERE = Path(__file__).resolve().parent
 
 
+def synthetic_guard(runtime=Path("private"), name="runtime"):
+    return {"verified": True, "conditions": [True] * 5, "conditions_checked": 5,
+            "actual_user_data_dir": str(runtime), "expected_user_data_dir": str(runtime),
+            "globalized_probe": str(runtime / "guard-probe"), "custom_user_dir_name": name,
+            "fresh_before_parent_initialize": True, "profile_writes_before_guard": 0,
+            "project_settings_persisted": False}
+
+
 def synthetic_report():
     meta = {"machine": {"hw.model": "synthetic", "cores": 10, "memory_bytes": 24000000000},
             "build": {"engine_string": collector.VERSION, "engine_hash": "ed1daf0bf",
@@ -213,7 +223,8 @@ class ReportTests(unittest.TestCase):
         row = {"phase": "O0", "passed": True, "timeout_seconds": 1800,
                "raw": {"stdout.log": {"sha256": hashlib.sha256(raw).hexdigest(), "bytes": len(raw)}},
                "report": r, "capture": {"actual_captured_pids": [123]},
-               "extrapolation": {"seconds_per_phase": 30}}
+               "extrapolation": collector.extrapolation(r),
+               "guard": synthetic_guard(), "runtime_profile": "private", "runtime_name": "runtime"}
         receipt = {"source_before": {"x": "a"}, "source_after": {"x": "a"},
                    "default_player_before": {}, "default_player_after": {}, "cleanup_errors": [],
                    "cleanup": [{"engine_ended": True}], "error": None, "passed": True,
@@ -222,6 +233,8 @@ class ReportTests(unittest.TestCase):
         receipt["probe_check"] = {"sha256": hashlib.sha256(probe_text.encode()).hexdigest(), "passed": True}
         for phase in receipt["phases"]:
             phase["probe_check"] = receipt["probe_check"].copy()
+            phase["owned_files"] = {name: {"sha256": hashlib.sha256(raw).hexdigest(), "bytes": len(raw)}
+                                    for name in collector.OWNED_FILES}
         with patch.object(checker, "PHASES", {"O0": ("O", 1, 0, 10)}), \
                 patch.object(Path, "read_bytes", return_value=raw), \
                 patch.object(Path, "read_text", autospec=True,
@@ -356,6 +369,166 @@ class ReportTests(unittest.TestCase):
         self.reject(lambda r: r["measurement"].update(render_gpu_ms=r["measurement"]["render_submit_wall_us"]))
         self.reject(lambda r: r["frames"][0].update(render_time_source="render_submit_wall_us"))
 
+    def test_j1_j2_actual_owned_path(self):
+        outer = collector.load_tool(collector.EXTERNAL_IMPORTS[1])
+        # Child self-observation supplies the public probe_fn in this synthetic
+        # test, so no host ps permission is needed. Real Popen/lifecycle remain.
+        child = (
+            "import os,sys,json,time,pathlib\n"
+            "mode=sys.argv[1]\n"
+            "identity={'pid':os.getpid(),'ppid':os.getppid(),'pgid':os.getpgrp(),"
+            "'args':' '.join([sys.executable]+sys.argv)}\n"
+            "identity_path=pathlib.Path(sys.argv[0]).parent/(mode+'-identity')\n"
+            "tmp=identity_path.with_suffix('.tmp')\n"
+            "tmp.write_text(json.dumps(identity)); tmp.replace(identity_path)\n"
+            "pathlib.Path(sys.argv[sys.argv.index('--log-file')+1]).write_text('native raw\\n')\n"
+            "print('Godot Engine v4.7.2.stable.official.ed1daf0bf - https://godotengine.org',flush=True)\n"
+            "time.sleep(0.15)\n"
+            "if mode=='timeout': time.sleep(5)\n"
+            "for i in range(0 if mode=='missing' else 2 if mode=='duplicate' else 1):\n"
+            " print('P28_01_PERF_REPORT='+json.dumps({'pid':os.getpid()+int(mode=='pid')}),flush=True)\n"
+            "sys.exit(3 if mode=='exit' else 0)\n")
+        with tempfile.TemporaryDirectory() as temp:
+            base = Path(temp)
+            script = base / "child"
+            script.write_text(child)
+            def diagnostic(command, folder, timeout, env, **kwargs):
+                def probe(pid):
+                    identity_path = base / (command[2] + "-identity")
+                    deadline = time.monotonic() + 2
+                    while time.monotonic() < deadline:
+                        if identity_path.exists():
+                            identity = json.loads(identity_path.read_text())
+                            if identity["pid"] == pid:
+                                return identity
+                        time.sleep(.005)
+                    raise RuntimeError("child identity unavailable")
+                return outer.native.run_owned_diagnostic(command, folder, timeout, env,
+                                                         probe_fn=probe, **kwargs)
+            for mode in ("ok", "exit", "timeout", "missing", "duplicate", "pid"):
+                folder = base / mode
+                def run_phase(command, output, limit, env, **kwargs):
+                    self.assertEqual(limit, 1800)
+                    return outer.run_phase(command, output, .3 if mode == "timeout" else 5,
+                                           env, diagnostic_fn=diagnostic, **kwargs)
+                # Record every collector write; the helper gets the original
+                # writer through the public dump_fn parameter below.
+                writes = []
+                original_write = collector.write
+                def guarded_write(path, value):
+                    self.assertFalse(Path(path).is_relative_to(folder),
+                                     "collector wrote inside owned folder")
+                    writes.append(Path(path))
+                    original_write(path, value)
+                with self.assertRaises(AssertionError):
+                    guarded_write(folder / "process.json", {})
+                def real_phase(*args, **kwargs):
+                    kwargs["dump_fn"] = original_write
+                    return run_phase(*args, **kwargs)
+                with self.subTest(mode=mode), patch.object(collector, "live_engines", return_value=[]), \
+                        patch.object(collector, "write", side_effect=guarded_write):
+                    row = collector.run_engine(SimpleNamespace(run_phase=real_phase),
+                                               [sys.executable, str(script), mode], folder, {})
+                self.assertEqual(writes, [base / (mode + "-collector-receipt.json")])
+                self.assertEqual(set(row["owned_files"]), set(collector.OWNED_FILES), row)
+                for name, pin in row["owned_files"].items():
+                    self.assertEqual(collector.sha(folder / name), pin["sha256"])
+                self.assertEqual(json.loads((folder / "process.json").read_text()), row["process"])
+                if mode in ("exit", "timeout"):
+                    self.assertFalse(row["passed"])
+                elif mode == "ok":
+                    self.assertTrue(row["passed"], row)
+                    report = collector.bound_report(folder, row)
+                    self.assertEqual(report["pid"], row["process"]["pid"])
+                else:
+                    with self.assertRaises(ValueError):
+                        collector.bound_report(folder, row)
+            # Actual owned receipts are immutable after return. The collector
+            # detects missing files even when the helper reports success.
+            def missing_file(*args, **kwargs):
+                result = outer.run_phase(*args, diagnostic_fn=diagnostic, **kwargs)
+                (args[1] / "ended.json").unlink()
+                return result
+            with patch.object(collector, "live_engines", return_value=[]):
+                row = collector.run_engine(SimpleNamespace(run_phase=missing_file),
+                                           [sys.executable, str(script), "ok"], base / "lost", {})
+            self.assertFalse(row["passed"])
+        cleanup, errors = outer.finish_captured(outer.OWNED)
+        self.assertFalse(errors)
+        self.assertTrue(all(r["engine_ended"] for r in cleanup))
+
+    def test_j3_guard_and_whole_inventory(self):
+        guard = synthetic_guard()
+        collector.guard_gate(guard, Path("private"), "runtime")
+        for key in guard:
+            for missing in (True, False):
+                bad = copy.deepcopy(guard)
+                if missing:
+                    del bad[key]
+                else:
+                    bad[key] = "wrong"
+                with self.subTest(key=key, missing=missing), self.assertRaises(ValueError):
+                    collector.guard_gate(bad, Path("private"), "runtime")
+        standard = collector.load_tool(collector.EXTERNAL_IMPORTS[0])
+        with tempfile.TemporaryDirectory() as temp:
+            stage = Path(temp)
+            (stage / "source.gd").write_text("source")
+            before = standard.inventory(stage, stage=True)
+            collector.stage_gate(before, before)
+            (stage / "source.gd.uid").write_text("uid")
+            imported = standard.inventory(stage, stage=True)
+            collector.stage_gate(before, imported, importing=True)
+            with self.assertRaises(ValueError):
+                collector.stage_gate(before, imported)
+            (stage / "untracked.txt").write_text("unexpected")
+            with self.assertRaises(ValueError):
+                collector.stage_gate(imported, standard.inventory(stage, stage=True))
+            with self.assertRaises(ValueError):
+                collector.stage_gate(before, standard.inventory(stage, stage=True), importing=True)
+            with self.assertRaises(ValueError):
+                collector.stage_gate(imported, {})
+
+    def test_j4_phase_stop_propagation(self):
+        def execute(minutes=1, duration=1, failure=False):
+            rows, launched = [], []
+            try:
+                for _, phase, forecast in collector.phase_schedule(rows):
+                    label = phase[0]
+                    launched.append(label)
+                    if label == "R0" and failure:
+                        collector.require(False, "native phase failed")
+                    report = {"actual_minutes": minutes,
+                              "elapsed_wall_seconds": duration if phase[1] == "R" else 1,
+                              "startup_wall_seconds": 1, "stop_reason": "simulation_overloaded"}
+                    rows.append({"phase": label, "report": report, "projection": forecast,
+                                 "extrapolation": collector.extrapolation(report)})
+            except ValueError:
+                pass
+            return rows, launched
+        rows, launched = execute()
+        self.assertEqual(launched, ["import", "R0", "P1", "P2", "O0", "P3", "P4"])
+        self.assertEqual(rows[1]["report"]["stop_reason"], "simulation_overloaded")
+        self.assertEqual(execute(failure=True)[1], ["import", "R0"])
+        rows, launched = execute(duration=20)
+        self.assertEqual(launched, ["import", "R0", "O0", "P3", "P4"])
+        self.assertEqual([r["phase"] for r in rows if r.get("engines_started") == 0],
+                         ["P1", "P2"])
+        rows, launched = execute(minutes=0)
+        self.assertEqual(len(launched), 7)
+        self.assertIsNone(rows[1]["extrapolation"]["seconds_per_phase"])
+        self.assertEqual(rows[1]["extrapolation"]["reason"], "no_minutes_observed")
+        self.assertIsNone(rows[2]["projection"]["projected_total_seconds"])
+        shortfall = synthetic_report()
+        shortfall.update(stop_reason="input_shortfall", frames=[], actual_minutes=0)
+        checker.check_report(shortfall)
+        self.assertIsNone(collector.extrapolation(shortfall)["seconds_per_phase"])
+        stopped = synthetic_report()
+        stopped["stop_reason"] = "timeout"
+        checker.check_report(stopped)
+        stopped["stop_reason"] = "unrecognized_stop"
+        with self.assertRaises(ValueError):
+            checker.check_report(stopped)
+
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)