P28-01-RESTART: 現HEADの性能基準：途中保存・別PID再開の再現

docs/plans/original-completion-2026-10-02/execution/p28-01/restart/probe/check_restart.py

@@ -5,6 +5,7 @@ import copy
 import hashlib
 import importlib.util
 import json
+import math
 from pathlib import Path
 import re
 import sys
@@ -23,6 +24,13 @@ DEPENDENCIES = {
 }
 LOCAL = {BASE + "restart/probe/" + n for n in
          ("restart_pair.py", "check_restart.py", "test_restart.py", "restart_probe.gd")}
+SIDECAR = "p28_restart/evidence/report_full.json"
+TIMINGS = ("duration_us", "runtime_duration_us")
+SAVE_EVENTS = {
+    "SAVE_CHECKPOINT_SUCCEEDED", "SAVE_CHECKPOINT_FAILED", "SAVE_LOADED", "SAVE_LOAD_FAILED",
+    "SAVE_MIGRATION_SUCCEEDED", "SAVE_MIGRATION_FAILED", "SAVE_RECOVERY_LOADED",
+    "SAVE_RECOVERY_CANDIDATE_LOADED", "SAVE_RECOVERY_FAILED",
+}
 PHASES = ["import", "parse-base", "parse-wrapper", "R-writer", "R-reader", "O-writer", "O-reader"]
 CROSS = [
     {"condition": "原本条件とUI操作が一致", "applicable": False, "reference": "collector C1/C2",
@@ -120,6 +128,7 @@ def check_probe(text):
     function = None
     blocks = []
     pauses = 0
+    watchdog_calls = 0
     for raw in text.splitlines():
         line = raw.split("#", 1)[0]
         match = re.match(r"^func (\w+)\(", line)
@@ -137,9 +146,16 @@ def check_probe(text):
         if "set_paused(" in line:
             require(function == "on_tick" and line.strip() == "GameClock.set_paused(true)", "pause outside tick")
             pauses += 1
-        require(not re.search(r"Time\.get_ticks_usec|\b\w+_us\b|Performance\.|get_performance_report", line),
+        if "Time." in line:
+            require(function == "deadline_timer_seconds" and
+                    line.strip() == "return Time.get_ticks_msec() / 1000.0", "timing API outside watchdog")
+            watchdog_calls += 1
+        # Literal names are used only by the finite evidence redaction routine.
+        timing_line = line.strip() == 'for key: String in ["duration_us", "runtime_duration_us"]:'
+        require(not re.search(r"\b\w+_(?:us|ms)\b|Performance\.|get_performance_report", line)
+                or (function == "redact_payload" and timing_line),
                 "performance instrumentation")
-    require(len(blocks) == 1 and pauses == 1, "deterministic/tick functions required")
+    require(len(blocks) == 1 and pauses == 1 and watchdog_calls == 1, "deterministic/tick/watchdog functions required")
     block = "\n".join(blocks[0])
     require(block.count("GameClock.advance_minutes(1)") == 1 and
             "io_ready" in block and "GameClock.paused" in block, "step admission")
@@ -211,8 +227,180 @@ def check_io(row, kind):
         require(type(failure["migration_attempted"]) is bool, "migration observation")
 
 
-def check_report(report):
+def save_event(event):
+    return isinstance(event, dict) and event.get("source_service") == "save_service" and \
+        event.get("event_type") in SAVE_EVENTS
+
+
+def payload_paths(report):
+    """The complete W2 closure; no recursive payload-name exemption."""
+    result = {}
+    for i, row in enumerate(report["io"]):
+        prefix = "/report/io/%d" % i
+        for name in ("signal_payload", "latest"):
+            result[prefix + "/" + name] = row[name]
+        events = [(prefix + "/event", row["event"]), (prefix + "/dropped", row["dropped"])]
+        for side in ("before", "after"):
+            events.extend((prefix + "/" + side + "/recent_events/%d" % j, event)
+                          for j, event in enumerate(row[side]["recent_events"]))
+        for path, event in events:
+            if save_event(event):
+                result[path + "/payload"] = event["payload"]
+    for k, row in enumerate(report["minutes"]):
+        if save_event(row["replacement"]):
+            result["/report/minutes/%d/replacement/payload" % k] = row["replacement"]["payload"]
+    return result
+
+
+def walk(value, path=""):
+    if isinstance(value, dict):
+        for key, child in value.items():
+            pointer = path + "/" + key.replace("~", "~0").replace("/", "~1")
+            yield pointer, key, child
+            yield from walk(child, pointer)
+    elif isinstance(value, list):
+        for i, child in enumerate(value):
+            yield from walk(child, path + "/" + str(i))
+
+
+def redact_report(verbatim):
+    """Also validates the full sidecar report before producing its unique public form."""
+    full = json.loads(json.dumps(verbatim))
+    allowed = payload_paths(full)
+    for path, key, _ in walk(full, "/report"):
+        require(key != "timing_redactions", "reserved marker in verbatim evidence")
+        require(not key.endswith(("_us", "_ms")) or
+                (path.rsplit("/", 1)[0] in allowed and key in TIMINGS), "timing outside W2 closure")
+        require(key not in ("performance", "statistics", "fps", "budget_verdict"),
+                "sidecar performance use")
+    for path, payload in allowed.items():
+        require(isinstance(payload, dict) and all(k in payload for k in TIMINGS), "both timing keys required")
+        payload["timing_redactions"] = [
+            {"original_key": key, "sidecar_ref": path + "/" + key} for key in TIMINGS]
+        for key in TIMINGS:
+            del payload[key]
+    check_no_metrics(full)
+    return full
+
+
+def resolve_pointer(sidecar, pointer):
+    require(pointer.startswith("/report/"), "pointer outside report")
+    value = sidecar
+    for part in pointer[1:].split("/"):
+        part = part.replace("~1", "/").replace("~0", "~")
+        value = value[int(part)] if isinstance(value, list) else value[part]
+    return value
+
+
+def json_sha(value):
+    # JSON numbers are represented uniformly across Godot's JSON float parser
+    # and Python. Fixed decimal form is valid JSON; keys and strings remain verbatim.
+    def encode(v):
+        if isinstance(v, dict):
+            return "{" + ",".join(json.dumps(k, ensure_ascii=False) + ":" + encode(v[k])
+                                  for k in sorted(v)) + "}"
+        if isinstance(v, list):
+            return "[" + ",".join(encode(x) for x in v) + "]"
+        if type(v) in (int, float):
+            require(math.isfinite(v), "nonfinite JSON")
+            return format(v, ".17f")
+        return json.dumps(v, ensure_ascii=False, separators=(",", ":"), allow_nan=False)
+    return hashlib.sha256(encode(value).encode()).hexdigest()
+
+
+def canonical_event(event, eid):
+    return {"event_id": eid, "replaced_for_cross_pid_compare": True} \
+        if event["event_id"] == eid else event
+
+
+def check_sidecar(public, sidecar):
+    require(isinstance(sidecar, dict) and set(sidecar) == {"report", "at_390_event_verbatim"},
+            "sidecar root/missing")
+    check_no_metrics(public)
+    full = sidecar["report"]
+    require(public == redact_report(full), "public redaction differs from W2")
+    restored = copy.deepcopy(public)
+    for path, payload in payload_paths(restored).items():
+        expected = [{"original_key": k, "sidecar_ref": path + "/" + k} for k in TIMINGS]
+        require(payload.pop("timing_redactions") == expected, "redaction marker/pointer mismatch")
+        for marker in expected:
+            payload[marker["original_key"]] = resolve_pointer(sidecar, marker["sidecar_ref"])
+    require(restored == full, "entire restored report differs")
+    events = sidecar["at_390_event_verbatim"]
+    require(isinstance(events, list), "at390 verbatim array")
+    observation = full.get("at_390")
+    summaries = observation["events"] if observation is not None else []
+    require(len(events) == len(summaries), "at390 verbatim count")
+    for event, summary in zip(events, summaries):
+        for path, key, _ in walk(event):
+            require(key != "timing_redactions" and key not in
+                    ("performance", "statistics", "fps", "budget_verdict"), "at390 forbidden field")
+            require(not key.endswith(("_us", "_ms")) or
+                    (save_event(event) and path == "/payload/" + key and key in TIMINGS),
+                    "at390 timing outside SAVE payload")
+        if save_event(event):
+            require(all(k in event["payload"] for k in TIMINGS), "at390 both timing keys")
+        require(set(summary) == {"event_id", "event_type", "absolute_minute",
+                                "raw_event_sha256", "canonical_sha256"}, "at390 summary shape")
+        require(all(event[k] == summary[k] for k in ("event_id", "event_type", "absolute_minute")),
+                "at390 verbatim order/metadata")
+        require(json_sha(event) == summary["raw_event_sha256"] and
+                json_sha(canonical_event(event, full["event_id"])) == summary["canonical_sha256"],
+                "at390 raw/canonical SHA")
+    return full
+
+
+def check_deadline(value, complete=True):
+    require(set(value) == {"limit_seconds_from_engine_start", "engine_elapsed_at_begin_seconds",
+                           "timer_seconds"}, "deadline shape")
+    elapsed, timer = value["engine_elapsed_at_begin_seconds"], value["timer_seconds"]
+    require(type(elapsed) in (int, float) and type(timer) in (int, float) and
+            math.isfinite(elapsed) and math.isfinite(timer) and elapsed >= 0 and
+            value["limit_seconds_from_engine_start"] == 1740 and timer == 1740 - elapsed,
+            "engine elapsed deadline formula")
+    require(not complete or timer > 0, "expired deadline")
+
+
+def overload_case(report):
+    backlog = report["paused_idle"]["before"]["backlog"]
+    overloads = report["overloads"]
+    require(isinstance(overloads, list) and type(backlog) is int and backlog >= 0, "overload facts")
+    expected = {"exercised": bool(overloads and backlog > 0),
+                "overloads": len(overloads), "backlog_at_save": backlog}
+    require(report["overload_case"] == expected and
+            type(report["overload_case"]["exercised"]) is bool and
+            set(report["overload_case"]) == set(expected), "overload disclosure")
+    for _, key, value in walk(report):
+        if key in ("nonzero_backlog_preserved", "overload_persistence",
+                   "across_restart_backlog_persistence_claimed"):
+            require(value is False, "unsupported overload/persistence claim")
+    return expected
+
+
+def coverage(reports):
+    detail = {r["role"]: overload_case(r) for r in reports if r["role"].endswith("writer")}
+    return {"overload_nonzero_backlog":
+            "exercised_recorded" if detail and all(x["exercised"] for x in detail.values()) else "not_exercised",
+            "per_role": detail, "across_restart_backlog_persistence_claimed": False}
+
+
+def check_no_arrival_diagnostic(report):
+    require(report["role"] == "O-writer" and report["boundary"] is None and
+            report["boundary_reason"] == report["stop_reason"] == "no_arrival_within_cap",
+            "no-arrival reason/null")
+    require(report["ui_ticks"] and
+            report["ui_ticks"][-1]["clock"] - report["start_minute"] == 120 and
+            all(t["guest_auto_count"] == 0 and t["guest_auto_ids"] == [] for t in report["ui_ticks"]),
+            "no-arrival cap/facts")
+    return True
+
+
+def check_report(report, sidecar=None):
     check_no_metrics(report)
+    require(sidecar is not None, "sidecar required")
+    report = check_sidecar(report, sidecar)
+    if report["stop_reason"] == "no_arrival_within_cap":
+        check_no_arrival_diagnostic(report)
     require(report["schema"] == "p28-restart-v1" and report["stop_reason"] == "complete", "incomplete probe")
     require(report["role"] in PHASES[3:] and type(report["pid"]) is int and report["pid"] > 0, "role/PID")
     require(report["main_added"] is True and report["k_target"] == 30 and report["actual_k"] == 30, "main/K")
@@ -220,12 +408,14 @@ def check_report(report):
             "partial digest/path exclusion")
     require(report["transfer_before_guard_writes"] == 0 and report["probe_deadline_seconds"] == 1740,
             "profile/deadline")
+    check_deadline(report["probe_deadline"])
     writer = report["role"].endswith("writer")
     kinds = ["save"] if writer else ["tamper", "load"]
     require(len(report["io"]) == len(kinds), "IO sequence")
     for row, kind in zip(report["io"], kinds):
         check_io(row, kind)
     if writer:
+        overload_case(report)
         exact(report["package_matches_before"])
         require(report["package_matches_before"]["actual_sha256"] == report["states"]["P"]["sha256"],
                 "P not package snapshot")
@@ -269,6 +459,32 @@ def check_report(report):
         require(report["states"]["after_tamper"]["sha256"] == report["io"][0]["comparison"]["actual_sha256"],
                 "tamper state file differs")
     if report["role"].startswith("O-"):
+        observation = report["at_390"]
+        require(isinstance(observation, dict) and observation["clock"] == {"absolute_minute": 390},
+                "at390 missing/clock")
+        require(isinstance(observation["operations"], dict) and
+                type(observation["domain_event_counter"]) is int and
+                type(observation["published_ids_count"]) is int and
+                observation["domain_event_counter"] >= observation["published_ids_count"] >= 0,
+                "at390 report/counter")
+        require(isinstance(observation["processed_arrival_minutes"], list) and
+                observation["processed_arrival_minutes"] == sorted(set(observation["processed_arrival_minutes"])) and
+                all(type(m) is int and m <= 390 for m in observation["processed_arrival_minutes"]),
+                "at390 processed minutes")
+        ids = observation["guest_auto_ids"]
+        require(isinstance(ids, list) and ids == sorted(set(ids)) and
+                type(observation["guest_auto_count"]) is int and
+                len(ids) == observation["guest_auto_count"] and
+                all(x.startswith("guest-auto-") for x in ids), "at390 guests")
+        require(all(type(e["absolute_minute"]) is int and
+                    report["saved_minute"] < e["absolute_minute"] <= 390
+                    for e in observation["events"]), "at390 event window")
+        event_ids = [e["event_id"] for e in observation["events"]]
+        require(len(event_ids) == len(set(event_ids)), "at390 duplicate event")
+        at_rows = [r for r in rows if r["after"] == 390]
+        require(len(at_rows) == 1 and at_rows[0]["operations_observation"] ==
+                {k: v for k, v in observation.items() if k != "events"}, "at390 minute consistency")
+        require(all(e in at_rows[0]["recent_ids"] for e in event_ids), "at390 event window IDs")
         boundary = report["boundary"]
         require(boundary is not None and boundary["guest_auto_count"] > 0, "arrival not processed")
         require(len(boundary["guest_auto_ids"]) == len(set(boundary["guest_auto_ids"])) ==
@@ -292,17 +508,14 @@ def check_report(report):
         require([r["minute"] for r in next_boundary["arrival_reports"]] == minutes,
                 "arrival report duplicates/omissions")
     else:
-        require(report["boundary"] is None and report["next_boundary"] is None, "R Operations boundary")
+        require(report["boundary"] is None and report["next_boundary"] is None and
+                report["at_390"] is None, "R Operations boundary")
     return True
 
 
 def check_no_metrics(value):
     if isinstance(value, dict):
         for key, child in value.items():
-            # Mandatory verbatim domain/IO payloads contain product diagnostics.
-            # They are evidence, never performance measurements or verdicts.
-            if key in ("payload", "signal_payload", "latest"):
-                continue
             require(not key.endswith(("_us", "_ms")) and key not in
                     ("performance", "statistics", "fps", "budget_verdict"), "performance field forbidden")
             check_no_metrics(child)
@@ -311,9 +524,20 @@ def check_no_metrics(value):
             check_no_metrics(child)
 
 
-def check_pair(writer, reader):
+def canonical_observation(observation, eid):
-    check_report(writer)
+    result = copy.deepcopy(observation)
-    check_report(reader)
+    if result is not None:
+        result["events"] = [
+            {"event_id": eid, "replaced_for_cross_pid_compare": True,
+             "canonical_sha256": e["canonical_sha256"]} if e["event_id"] == eid else e
+            for e in result["events"]]
+    return result
+
+
+def check_pair(writer, reader, writer_sidecar=None, reader_sidecar=None):
+    check_report(writer, writer_sidecar)
+    check_report(reader, reader_sidecar)
+    writer, reader = writer_sidecar["report"], reader_sidecar["report"]
     require(writer["role"].replace("writer", "reader") == reader["role"] and writer["pid"] != reader["pid"],
             "pair roles/PID")
     require(writer["event_id"] == reader["event_id"] and writer["saved_minute"] == reader["saved_minute"],
@@ -326,6 +550,8 @@ def check_pair(writer, reader):
                 "cross-PID complete typed state mismatch")
     require(writer["boundary"] == reader["boundary"] and writer["next_boundary"] == reader["next_boundary"],
             "Operations boundary differs")
+    require(canonical_observation(writer["at_390"], writer["event_id"]) ==
+            canonical_observation(reader["at_390"], reader["event_id"]), "at390 pair differs")
 
 
 def bound_raw(folder, row):
@@ -385,10 +611,21 @@ def bound_raw(folder, row):
             close.get("deadline_reached") is False and close.get("clock_process_held_stopped") is True and
             close.get("window_input_held_disabled") is True and close.get("restored_before_quit") is False and
             close.get("shared_bus_deleted") is False and close.get("settings_mutated") is False, "audio close")
-    check_report(report)
+    check_report(report, bound_sidecar(folder.parent, row))
     return report
 
 
+def bound_sidecar(out, row):
+    binding = row["sidecar"]
+    require(set(binding) == {"path", "sha256", "bytes"} and binding["path"] == SIDECAR,
+            "sidecar binding/path")
+    expected = {k: binding[k] for k in ("sha256", "bytes")}
+    path = Path(out) / (row["phase"] + "-states") / "report_full.json"
+    require(pin(path) == expected and pin(Path(row["runtime_profile"]) / SIDECAR) == expected,
+            "sidecar SHA/private profile binding")
+    return json.loads(path.read_text())
+
+
 def check_receipt(receipt, out):
     from restart_pair import COLLECTOR_PINS, check_transfer
     out = Path(out)
@@ -432,12 +669,15 @@ def check_receipt(receipt, out):
             for name, evidence in report["states"].items():
                 require(pin(out / (row["phase"] + "-states") / (name + ".bin")) ==
                         {k: evidence[k] for k in ("sha256", "bytes")}, "state bytes SHA")
+            require(receipt["profiles"][row["runtime_profile"]].get(SIDECAR) ==
+                    {k: row["sidecar"][k] for k in ("sha256", "bytes")}, "sidecar absent from file map")
     require(len(set(pids)) == len(pids), "reused PID")
     require({r["pid"] for r in receipt["cleanup"]} == set(pids) and all(
         r["engine_ended"] is True and r["outer_cleanup_errors"] == [] and
         r["foreign_process_or_group_signals"] == 0 for r in receipt["cleanup"]), "cleanup mismatch")
     for start in (3, 5):
-        check_pair(rows[start]["report"], rows[start + 1]["report"])
+        check_pair(rows[start]["report"], rows[start + 1]["report"],
+                   bound_sidecar(out, rows[start]), bound_sidecar(out, rows[start + 1]))
         transfer = out / (rows[start]["phase"][0] + "-transfer")
         manifest_path = transfer / "manifest.json"
         manifest = json.loads(manifest_path.read_text())
@@ -457,6 +697,8 @@ def check_receipt(receipt, out):
     require(sha(out / "restart_probe.gd") == receipt["worker_before"]["restart_probe.gd"],
             "executed probe differs from submitted source")
     require(receipt["owned_writes"] == [], "probe/driver wrote owned folder")
+    require(receipt["coverage"] == coverage([bound_sidecar(out, r)["report"] for r in rows[3:]]),
+            "overload coverage/persistence claim")
     return True
 
 

docs/plans/original-completion-2026-10-02/execution/p28-01/restart/probe/restart_pair.py

@@ -149,6 +149,7 @@ def bind_report(out, label, row):
     reports = [s.partition("=")[2] for s in text.splitlines() if s.startswith("P28_RESTART_REPORT=")]
     check.require(len(reports) == 1, "one raw report required")
     row["report"] = json.loads(reports[0])
+    copy_states(out, row)
     check.bound_raw(out / label, row)
     return row["report"]
 
@@ -157,6 +158,10 @@ def copy_states(out, row):
     folder = out / (row["phase"] + "-states")
     folder.mkdir()
     profile = Path(row["runtime_profile"])
+    source = profile / check.SIDECAR
+    row["sidecar"] = {"path": check.SIDECAR, **check.pin(source)}
+    shutil.copyfile(source, folder / "report_full.json")
+    check.require(check.pin(source) == check.pin(folder / "report_full.json"), "sidecar copy differs")
     for name, expected in row["report"]["states"].items():
         source = profile / expected["path"]
         check.require(source.resolve().is_relative_to(profile.resolve()) and
@@ -267,14 +272,14 @@ def main():
                 row.update(runtime_profile=str(runtime), runtime_name=runtime_name,
                            guard=observation, guard_sha256=check.sha(guard_path))
                 report = bind_report(out, label, row)
-                copy_states(out, row)
                 collector.write(out / (label + "-report.json"), report)
                 if label.endswith("writer"):
                     create_transfer(runtime / "p28_restart/good", transfer, collector.write)
                 else:
                     check.require(report["transfer_manifest_sha256"] == check.sha(transfer / "manifest.json"),
                                   "reader transfer manifest binding")
-                    check.check_pair(rows[-2]["report"], report)
+                    check.check_pair(rows[-2]["report"], report, check.bound_sidecar(out, rows[-2]),
+                                     check.bound_sidecar(out, row))
             else:
                 check.bound_raw(out / label, row)
             check.require(row["source_preserved"] and row["player_preserved"] and
@@ -297,6 +302,13 @@ def main():
                    "profiles": {str(p): collector.profile_files(p) for p in profiles}, "owned_writes": [],
                    "cross_cutting": check.CROSS, "original_credit": 0,
                    "windows_verified": False, "operations_200_guests_verified": False}
+        try:
+            receipt["coverage"] = check.coverage([
+                check.bound_sidecar(out, row)["report"] for row in rows if "sidecar" in row])
+        except (ValueError, KeyError, TypeError, OSError) as caught:
+            receipt["coverage"] = {"overload_nonzero_backlog": "not_exercised",
+                                   "error": repr(caught)}
+            receipt.update(passed=False, error=receipt["error"] or "coverage: " + repr(caught))
         if receipt["passed"]:
             try:
                 check.check_receipt(receipt, out)

docs/plans/original-completion-2026-10-02/execution/p28-01/restart/probe/restart_probe.gd

@@ -16,6 +16,65 @@ var E := ""
 var package: Dictionary = {}
 var failure_event: Dictionary = {}
 var boundary: Variant = null
+var at_390_event_verbatim: Array = []
+var at_390_pending: Array = []
+var redaction_ok := true
+const SAVE_EVENTS := ["SAVE_CHECKPOINT_SUCCEEDED", "SAVE_CHECKPOINT_FAILED", "SAVE_LOADED",
+	"SAVE_LOAD_FAILED", "SAVE_MIGRATION_SUCCEEDED", "SAVE_MIGRATION_FAILED",
+	"SAVE_RECOVERY_LOADED", "SAVE_RECOVERY_CANDIDATE_LOADED", "SAVE_RECOVERY_FAILED"]
+
+func canonical_json(value: Variant) -> String:
+	if value is Dictionary:
+		var keys: Array = value.keys()
+		keys.sort()
+		var parts: PackedStringArray = []
+		for key: Variant in keys:
+			parts.append(JSON.stringify(str(key)) + ":" + canonical_json(value[key]))
+		return "{" + ",".join(parts) + "}"
+	if value is Array:
+		var parts: PackedStringArray = []
+		for child: Variant in value:
+			parts.append(canonical_json(child))
+		return "[" + ",".join(parts) + "]"
+	if value is int or value is float:
+		return "%.17f" % float(value)
+	return JSON.stringify(value)
+
+func event_summary(event: Dictionary) -> Dictionary:
+	var canonical: Dictionary = {"event_id": E, "replaced_for_cross_pid_compare": true} \
+		if field(event, "event_id") == E else event
+	return {"event_id": field(event, "event_id"), "event_type": field(event, "event_type"),
+		"absolute_minute": field(event, "absolute_minute"),
+		"raw_event_sha256": bytes_hash(canonical_json(event).to_utf8_buffer()),
+		"canonical_sha256": bytes_hash(canonical_json(canonical).to_utf8_buffer())}
+
+func observe_390() -> void:
+	if not role.begins_with("O-") or GameClock.absolute_minute != 390 or report.at_390 != null:
+		return
+	var observed := operations_observation()
+	var state := witness(runtime.create_snapshot())
+	report.at_390 = {"clock": {"absolute_minute": GameClock.absolute_minute},
+		"operations": observed.operations, "processed_arrival_minutes": observed.processed_arrival_minutes,
+		"guest_auto_ids": observed.guest_auto_ids, "guest_auto_count": observed.guest_auto_count,
+		"published_ids_count": state.published_ids.size(), "domain_event_counter": state.counter,
+		"events": []}
+	at_390_pending = state.recent_events.duplicate(true)
+	if report.has("saved_minute"):
+		bind_390_events()
+
+func bind_390_events() -> void:
+	if report.at_390 == null:
+		return
+	at_390_event_verbatim = []
+	report.at_390.events = []
+	for event: Dictionary in at_390_pending:
+		var minute: int = field(event, "absolute_minute")
+		if minute > report.saved_minute and minute <= 390:
+			at_390_event_verbatim.append(event.duplicate(true))
+			report.at_390.events.append(event_summary(event))
+
+func deadline_timer_seconds() -> float:
+	return Time.get_ticks_msec() / 1000.0
 
 func field(d: Variant, name: String) -> Variant:
 	if d is Dictionary:
@@ -117,14 +176,14 @@ func operations_observation() -> Dictionary:
 		"arrival_reports": arrivals}
 
 func on_tick(minute: int) -> void:
+	# During deterministic stepping the observation is made after advance returns.
+	if not io_ready:
+		observe_390()
 	if io_ready or finished or selected_tick:
 		return
 	var observed := operations_observation()
 	if role == "O-writer":
 		report.ui_ticks.append(observed)
-		if minute == 390:
-			report["at_390"] = observed.duplicate(true)
-			report.at_390["domain_events"] = field(runtime_part(runtime.create_snapshot()), "domain_events")
 	var stop: bool = (role == "R-writer" and minute - start_minute >= 10) or \
 		(role == "O-writer" and (observed.guest_auto_count > 0 or minute - start_minute >= 120))
 	if stop:
@@ -147,10 +206,18 @@ func begin() -> void:
 	report = {"schema": "p28-restart-v1", "role": role, "pid": OS.get_process_id(),
 		"main_added": false, "k_target": 30, "actual_k": 0, "io": [], "states": {},
 		"minutes": [], "ui_ticks": [], "overloads": [], "boundary": null, "next_boundary": null,
+		"at_390": null,
 		"comparison_scope": "complete_typed_var_to_bytes", "excluded_paths": [],
 		"transfer_before_guard_writes": 0, "transfer_verified": false,
 		"probe_deadline_seconds": 1740}
-	create_timer(1740.0).timeout.connect(func() -> void: finish("probe_deadline"))
+	var elapsed := deadline_timer_seconds()
+	var interval := 1740.0 - elapsed
+	report["probe_deadline"] = {"limit_seconds_from_engine_start": 1740,
+		"engine_elapsed_at_begin_seconds": elapsed, "timer_seconds": interval}
+	if interval <= 0:
+		finish("probe_deadline")
+		return
+	create_timer(interval).timeout.connect(func() -> void: finish("probe_deadline"))
 	if OS.get_environment("THE_TREE_RESTART_K") != "30" or role not in ["R-writer", "R-reader", "O-writer", "O-reader"]:
 		finish("invalid_environment")
 		return
@@ -217,6 +284,7 @@ func begin() -> void:
 			return
 	report["event_id"] = E
 	report["saved_minute"] = GameClock.absolute_minute
+	bind_390_events()
 	boundary = operations_observation() if role.begins_with("O-") else null
 	report["boundary"] = boundary
 	var idle_before: Dictionary = runtime.create_snapshot()
@@ -245,6 +313,9 @@ func begin() -> void:
 		return
 	report["final_minute"] = GameClock.absolute_minute
 	if role.begins_with("O-"):
+		if report.at_390 == null:
+			finish("at_390_not_reached")
+			return
 		var last := operations_observation()
 		var added: Array = []
 		for minute: Variant in last.processed_arrival_minutes:
@@ -328,6 +399,9 @@ func io_proof(base: Dictionary, after: Dictionary, kind: String, returned: Varia
 
 func writer_save() -> bool:
 	var before: Dictionary = runtime.create_snapshot()
+	var backlog := GameClock.get_backlog_minutes()
+	report["overload_case"] = {"exercised": not report.overloads.is_empty() and backlog > 0,
+		"overloads": report.overloads.size(), "backlog_at_save": backlog}
 	signals = []
 	var returned: bool = main.call("_save_manual_at_path", "user://p28_restart/good", "P28 RESTART")
 	var after: Dictionary = runtime.create_snapshot()
@@ -409,6 +483,7 @@ func deterministic_step() -> bool:
 	var after: int = GameClock.absolute_minute
 	if after != before + 1:
 		return false
+	observe_390()
 	var state: Dictionary = runtime.create_snapshot()
 	var normalized := state.duplicate(true)
 	var recent: Array = field(field(runtime_part(normalized), "domain_events"), "recent_events")
@@ -429,15 +504,86 @@ func deterministic_step() -> bool:
 		"excluded_paths": [], "replacement": replacement, "replacement_count": count,
 		"recent_count": recent.size(), "recent_ids": recent_ids})
 	report.actual_k += after - before
+	if role.begins_with("O-") and after == 390:
+		var observation: Dictionary = report.at_390.duplicate(true)
+		observation.erase("events")
+		report.minutes[-1]["operations_observation"] = observation
 	return GameClock.paused and backlog == GameClock.get_backlog_minutes()
 
+func is_save_event(event: Variant) -> bool:
+	return event is Dictionary and field(event, "source_service") == "save_service" and \
+		field(event, "event_type") in SAVE_EVENTS
+
+func redact_payload(payload: Dictionary, pointer: String) -> void:
+	if payload.has("timing_redactions"):
+		redaction_ok = false
+		return
+	var markers: Array = []
+	for key: String in ["duration_us", "runtime_duration_us"]:
+		if not payload.has(key):
+			redaction_ok = false
+			return
+		markers.append({"original_key": key, "sidecar_ref": pointer + "/" + key})
+		payload.erase(key)
+	payload["timing_redactions"] = markers
+
+func redact_event(event: Variant, pointer: String) -> void:
+	if is_save_event(event):
+		redact_payload(event.payload, pointer + "/payload")
+
+func public_report(full: Dictionary) -> Dictionary:
+	var result: Dictionary = full.duplicate(true)
+	for i in range(result.io.size()):
+		var row: Dictionary = result.io[i]
+		var pointer := "/report/io/%d" % i
+		for name: String in ["signal_payload", "latest"]:
+			if row.has(name):
+				redact_payload(row[name], pointer + "/" + name)
+		for name: String in ["event", "dropped"]:
+			if row.has(name):
+				redact_event(row[name], pointer + "/" + name)
+		for side: String in ["before", "after"]:
+			if row.has(side):
+				for j in range(row[side].recent_events.size()):
+					redact_event(row[side].recent_events[j], pointer + "/" + side + "/recent_events/%d" % j)
+	for k in range(result.minutes.size()):
+		redact_event(result.minutes[k].replacement, "/report/minutes/%d/replacement" % k)
+	return result
+
+func public_has_forbidden(value: Variant) -> bool:
+	if value is Dictionary:
+		for key: Variant in value:
+			if str(key).ends_with("_" + "us") or str(key).ends_with("_" + "ms") or \
+				str(key) in ["performance", "statistics", "fps", "budget_verdict"]:
+				return true
+			if public_has_forbidden(value[key]):
+				return true
+	elif value is Array:
+		for child: Variant in value:
+			if public_has_forbidden(child):
+				return true
+	return false
+
 func finish(reason: String) -> void:
 	if finished:
 		return
 	finished = true
 	report["stop_reason"] = reason
 	report["key_forms"] = key_forms
-	print("P28_RESTART_REPORT=" + JSON.stringify(report))
+	var sidecar_path := "user://p28_restart/evidence/report_full.json"
+	DirAccess.make_dir_recursive_absolute(ProjectSettings.globalize_path(sidecar_path).get_base_dir())
+	var file := FileAccess.open(sidecar_path, FileAccess.WRITE)
+	if file == null:
+		quit(1)
+		return
+	file.store_string(JSON.stringify({"report": report, "at_390_event_verbatim": at_390_event_verbatim}, "", true, true))
+	file.close()
+	# Partial failure evidence stays in the sidecar. Never emit a leaking public payload.
+	var public: Dictionary = public_report(report)
+	if not redaction_ok or public_has_forbidden(public) or public.is_empty():
+		print("P28_RESTART_EVIDENCE_FAILURE=" + reason)
+	else:
+		print("P28_RESTART_REPORT=" + JSON.stringify(public, "", true, true))
 	if is_instance_valid(main) and main.is_inside_tree():
 		main.ui_audio_window_close_drained.connect(emit_audio_close, CONNECT_ONE_SHOT)
 		get_root().close_requested.emit()

docs/plans/original-completion-2026-10-02/execution/p28-01/restart/probe/test_restart.py

@@ -37,7 +37,8 @@ def io(kind, minute=612, count=256):
     payload = {"ok": kind != "tamper", "code": "OK" if kind != "tamper" else "CORRUPT_WORLD",
                "operation": "save" if kind == "save" else "load",
                "save_path": "user://p28_restart/" + ("tampered" if kind == "tamper" else "good"),
-               "save_id": "synthetic", "saved_absolute_minute": minute}
+               "save_id": "synthetic", "saved_absolute_minute": minute,
+               "duration_us": 124, "runtime_duration_us": 57}
     event = {"event_id": check.event_id(count),
              "event_type": {"save": "SAVE_CHECKPOINT_SUCCEEDED", "load": "SAVE_LOADED",
                             "tamper": "SAVE_LOAD_FAILED"}[kind],
@@ -57,20 +58,24 @@ def io(kind, minute=612, count=256):
 
 def report(role="R-writer", pid=100):
     writer, ops = role.endswith("writer"), role.startswith("O-")
-    minute = 391 if ops else 612
+    minute = 361 if ops else 612
     rows = [io("save", minute)] if writer else [io("tamper", 360, 3), io("load", minute)]
     names = ["P", "after_save", "final"] if writer else ["P", "B", "after_tamper", "after_load", "final"]
     event = rows[-1]["event"]
     boundary = {"clock": minute, "guest_auto_ids": ["guest-auto-00000001"], "guest_auto_count": 1,
-                "operations": {"enabled": True}, "processed_arrival_minutes": [390],
+                "operations": {"enabled": True}, "processed_arrival_minutes": [360],
-                "arrival_reports": [{"minute": 390}]} if ops else None
+                "arrival_reports": [{"minute": 360}]} if ops else None
-    next_boundary = {"minute": 420, "processed_once": True, "before": [390], "after": [390, 420],
+    next_boundary = {"minute": 390, "processed_once": True, "before": [360], "after": [360, 390],
-                     "arrival_reports": [{"minute": 390}, {"minute": 420}],
+                     "arrival_reports": [{"minute": 360}, {"minute": 390}],
                      "guest_auto_ids": ["guest-auto-00000001", "guest-auto-00000002"]} if ops else None
     clock = {"minute": minute, "paused": True, "backlog": 121, "accumulator_hex": "00"}
-    return {"schema": "p28-restart-v1", "role": role, "pid": pid, "stop_reason": "complete",
+    result = {"schema": "p28-restart-v1", "role": role, "pid": pid, "stop_reason": "complete",
             "main_added": True, "k_target": 30, "actual_k": 30, "comparison_scope": "complete_typed_var_to_bytes",
             "excluded_paths": [], "transfer_before_guard_writes": 0, "probe_deadline_seconds": 1740,
+            "probe_deadline": {"limit_seconds_from_engine_start": 1740,
+                               "engine_elapsed_at_begin_seconds": 12.25, "timer_seconds": 1727.75},
+            "overloads": [],
+            "overload_case": {"exercised": False, "overloads": 0, "backlog_at_save": 121},
             "io": rows, "package_matches_before": comparison(), "failure_event_absent": True,
             "transfer_verified": True, "event_id": event["event_id"], "saved_minute": minute,
             "start_minute": 360 if ops else 602, "final_minute": minute + 30,
@@ -85,7 +90,37 @@ def report(role="R-writer", pid=100):
                          "replacement_count": 1 if n < 5 else 0, "recent_count": 256,
                          "recent_ids": history(255)["published_ids"] +
                          ([event["event_id"]] if n < 5 else ["domain_event.000000000256"])}
-                        for n in range(30)]}
+                        for n in range(30)], "at_390": None}
+    if ops:
+        result["at_390"] = {"clock": {"absolute_minute": 390}, "operations": {"enabled": True},
+                            "processed_arrival_minutes": [360],
+                            "guest_auto_ids": ["guest-auto-00000001"], "guest_auto_count": 1,
+                            "domain_event_counter": 300, "published_ids_count": 300, "events": []}
+        result["minutes"][28]["operations_observation"] = {
+            k: copy.deepcopy(v) for k, v in result["at_390"].items() if k != "events"}
+    return result
+
+
+def evidence(full, events=None):
+    sidecar = {"report": copy.deepcopy(full), "at_390_event_verbatim": copy.deepcopy(events or [])}
+    return check.redact_report(full), sidecar
+
+
+def validate_report(full):
+    public, sidecar = evidence(full)
+    return check.check_report(public, sidecar)
+
+
+def validate_pair(writer, reader):
+    public_writer, sw = evidence(writer)
+    public_reader, sr = evidence(reader)
+    return check.check_pair(public_writer, public_reader, sw, sr)
+
+
+def summarize(event, eid):
+    return {**{k: event[k] for k in ("event_id", "event_type", "absolute_minute")},
+            "raw_event_sha256": check.json_sha(event),
+            "canonical_sha256": check.json_sha(check.canonical_event(event, eid))}
 
 
 def write(path, value):
@@ -143,8 +178,9 @@ def synthetic_receipt(out):
             observed = report(label, pid)
             if label.endswith("reader"):
                 observed["transfer_manifest_sha256"] = check.sha(out / (label[0] + "-transfer/manifest.json"))
-            row["report"] = observed
+            public, sidecar = evidence(observed)
-            raw = "P28_RESTART_REPORT=" + json.dumps(observed) + "\nP28_RESTART_AUDIO_CLOSE=" + json.dumps(audio(pid)) + "\n"
+            row["report"] = public
+            raw = "P28_RESTART_REPORT=" + json.dumps(public) + "\nP28_RESTART_AUDIO_CLOSE=" + json.dumps(audio(pid)) + "\n"
             profile = str(out / (label + "-private"))
             row.update(runtime_profile=profile, runtime_name=label)
             profiles[profile] = {}
@@ -161,6 +197,12 @@ def synthetic_receipt(out):
             row.update(guard=guard, guard_sha256=check.sha(guard_path))
             states = out / (label + "-states")
             states.mkdir()
+            sidecar_path = Path(profile) / check.SIDECAR
+            sidecar_path.parent.mkdir(parents=True)
+            write(sidecar_path, sidecar)
+            (states / "report_full.json").write_bytes(sidecar_path.read_bytes())
+            row["sidecar"] = {"path": check.SIDECAR, **check.pin(sidecar_path)}
+            profiles[profile][check.SIDECAR] = check.pin(sidecar_path)
             for name in observed["states"]:
                 (states / (name + ".bin")).write_bytes(DATA)
         for name, content in (("stdout.log", raw), ("stderr.log", ""), ("engine-native.log", "native\n")):
@@ -185,6 +227,7 @@ def synthetic_receipt(out):
             "worker_before": worker, "worker_after": worker.copy(), "profiles": profiles,
             "cross_cutting": copy.deepcopy(check.CROSS), "original_credit": 0,
             "windows_verified": False, "operations_200_guests_verified": False,
+            "coverage": check.coverage([report("R-writer"), report("O-writer")]),
             "script_pins": {str(probe): check.sha(probe)}, "owned_writes": [],
             "cleanup": [{"pid": 100 + n, "engine_ended": True, "outer_cleanup_errors": [],
                          "foreign_process_or_group_signals": 0} for n in range(7)]}
@@ -269,11 +312,11 @@ class RestartTests(unittest.TestCase):
             "failed event retained": lambda x: x.update(failure_event_absent=False),
             "backlog consumed": lambda x: x["paused_idle"]["after"].update(backlog=120),
             "reader not paused": lambda x: x["io"][1].update(paused_after=False),
-        }, check.check_report)
+        }, validate_report)
 
     def test_rs3_actual_k_full_typed_cross_pid(self):
         writer, reader = report(), report("R-reader", 101)
-        check.check_pair(writer, reader)
+        validate_pair(writer, reader)
         self.rejects(reader, {
             "K call count": lambda x: x["minutes"][0].update(after=x["saved_minute"]),
             "different K": lambda x: x.update(actual_k=29),
@@ -285,11 +328,11 @@ class RestartTests(unittest.TestCase):
             "after window": lambda x: x["minutes"][6].update(replacement=x["io"][-1]["event"], replacement_count=1),
             "excluded path": lambda x: x["minutes"][0].update(excluded_paths=["clock"]),
             "same PID": lambda x: x.update(pid=100),
-        }, lambda x: check.check_pair(writer, x))
+        }, lambda x: validate_pair(writer, x))
 
     def test_rs4_rs5_arrival_boundary(self):
         writer, reader = report("O-writer"), report("O-reader", 101)
-        check.check_pair(writer, reader)
+        validate_pair(writer, reader)
         self.rejects(writer, {
             "boundary only": lambda x: x["boundary"].update(guest_auto_count=0),
             "null arrival": lambda x: x.update(boundary=None),
@@ -299,7 +342,7 @@ class RestartTests(unittest.TestCase):
             "next missing": lambda x: x["next_boundary"].update(processed_once=False),
             "next duplicate": lambda x: x["next_boundary"]["after"].append(420),
             "cap exceeded": lambda x: x.update(start_minute=200),
-        }, check.check_report)
+        }, validate_report)
 
     def test_d1_import_closure(self):
         check.check_dependencies()
@@ -479,6 +522,267 @@ class RestartTests(unittest.TestCase):
             _, errors = outer.finish_captured(outer.OWNED)
             self.assertEqual(errors, [])
 
+    def test_v1_at390_required_and_pair_consistency(self):
+        writer, reader = report("O-writer"), report("O-reader", 101)
+        validate_pair(writer, reader)
+        self.rejects(writer, {
+            "absent": lambda x: x.pop("at_390"),
+            "null": lambda x: x.update(at_390=None),
+            "not reached": lambda x: x.update(stop_reason="at_390_not_reached"),
+            "clock": lambda x: x["at_390"]["clock"].update(absolute_minute=391),
+            "counter type": lambda x: x["at_390"].update(domain_event_counter="300"),
+            "row differs": lambda x: x["minutes"][28]["operations_observation"].update(guest_auto_count=2),
+        }, validate_report)
+        for mutate in (lambda x: x.pop("at_390"), lambda x: x.update(at_390=None)):
+            w, r = copy.deepcopy(writer), copy.deepcopy(reader)
+            mutate(w)
+            mutate(r)
+            with self.assertRaises((ValueError, KeyError)):
+                validate_pair(w, r)
+        different = copy.deepcopy(reader)
+        different["at_390"]["operations"]["population"] = 3
+        different["minutes"][28]["operations_observation"]["operations"]["population"] = 3
+        validate_report(different)
+        with self.assertRaises(ValueError):
+            validate_pair(writer, different)
+        with self.assertRaises((ValueError, KeyError)):
+            validate_pair(writer, {**reader, "at_390": None})
+
+    def test_v2_all_seven_families_restore_verbatim(self):
+        full = report()
+        # T4 and T6 are conditional: an earlier SAVE event is the dropped head.
+        old = io("save", minute=600, count=0)["event"]
+        row = full["io"][0]
+        row["before"]["recent_events"][0] = old
+        row["dropped"] = copy.deepcopy(old)
+        row["after"] = check.transform(row["before"], row["event"])
+        public, sidecar = evidence(full)
+        check.check_report(public, sidecar)
+        paths = check.payload_paths(public)
+        for path in ("/report/io/0/signal_payload", "/report/io/0/latest",
+                     "/report/io/0/event/payload", "/report/io/0/before/recent_events/0/payload",
+                     "/report/io/0/after/recent_events/255/payload", "/report/io/0/dropped/payload",
+                     "/report/minutes/0/replacement/payload"):
+            self.assertEqual(paths[path]["timing_redactions"], [
+                {"original_key": key, "sidecar_ref": path + "/" + key} for key in check.TIMINGS])
+            self.assertNotIn("duration_us", paths[path])
+        self.assertEqual(public["io"][0]["after"]["recent_events"][0],
+                         full["io"][0]["after"]["recent_events"][0])
+        self.assertEqual(check.check_sidecar(public, sidecar), full)
+        for kind in ("save", "load", "tamper"):
+            full_row = io(kind)
+            check.check_io(full_row, kind)
+        self.rejects(sidecar, {
+            "root extra": lambda x: x.update(extra={}),
+            "performance use": lambda x: x["report"].update(statistics={"average": 1}),
+            "outside closure": lambda x: x["report"].update(elapsed_ms=1),
+            "nested timing": lambda x: x["report"]["io"][0]["event"]["payload"].update(nested={"latency_us": 1}),
+            "missing timing": lambda x: x["report"]["io"][0]["latest"].pop("duration_us"),
+            "tamper value": lambda x: x["report"]["io"][0]["latest"].update(code="CHANGED"),
+            "nonSAVE": lambda x: x["report"]["io"][0]["before"]["recent_events"][0].update(event_type="OLD"),
+            "wrong service": lambda x: x["report"]["io"][0]["before"]["recent_events"][0].update(source_service="other"),
+        }, lambda x: check.check_sidecar(public, x))
+        self.rejects(public, {
+            "leaked key": lambda x: x["io"][0]["signal_payload"].update(duration_us=124),
+            "hidden payload leak": lambda x: x.update(payload={"hidden_ms": 1}),
+            "extra marker": lambda x: x["io"][0]["latest"]["timing_redactions"].append(
+                {"original_key": "other", "sidecar_ref": "/report/other"}),
+            "missing marker": lambda x: x["io"][0]["latest"].pop("timing_redactions"),
+            "one marker": lambda x: x["io"][0]["latest"]["timing_redactions"].pop(),
+            "wrong pointer": lambda x: x["io"][0]["latest"]["timing_redactions"][0].update(
+                sidecar_ref="/report/io/0/signal_payload/duration_us"),
+            "unresolved pointer": lambda x: x["io"][0]["latest"]["timing_redactions"][0].update(
+                sidecar_ref="/report/io/999/latest/duration_us"),
+            "extra marker field": lambda x: x["io"][0]["latest"]["timing_redactions"][0].update(extra=1),
+            "nonSAVE marker": lambda x: x["io"][0]["after"]["recent_events"][0]["payload"].update(
+                timing_redactions=copy.deepcopy(x["io"][0]["latest"]["timing_redactions"])),
+            "restore differs": lambda x: x.update(pid=999),
+        }, lambda x: check.check_sidecar(x, sidecar))
+        with self.assertRaises(ValueError):
+            check.check_report(public)
+        # The raw equality check must see differences hidden by public redaction.
+        tampered = copy.deepcopy(sidecar)
+        tampered["report"]["io"][0]["latest"]["duration_us"] += 1
+        self.assertEqual(check.redact_report(tampered["report"]), public)
+        with self.assertRaises(ValueError):
+            check.check_report(public, tampered)
+
+    def test_v2_at390_full_array_binding(self):
+        full = report("O-writer")
+        events = [
+            {"event_id": "domain_event.%012d" % n, "event_type": "ARRIVAL",
+             "absolute_minute": 380 + i, "source_service": "operations", "payload": {"count": i + 1}}
+            for i, n in enumerate((250, 251))]
+        full["at_390"]["events"] = [summarize(e, full["event_id"]) for e in events]
+        public, sidecar = evidence(full, events)
+        check.check_report(public, sidecar)
+        self.rejects(sidecar, {
+            "missing array": lambda x: x.pop("at_390_event_verbatim"),
+            "missing entry": lambda x: x["at_390_event_verbatim"].pop(),
+            "extra entry": lambda x: x["at_390_event_verbatim"].append(copy.deepcopy(events[0])),
+            "reordered": lambda x: x["at_390_event_verbatim"].reverse(),
+            "raw differs": lambda x: x["at_390_event_verbatim"][0]["payload"].update(count=7),
+            "nonSAVE timing": lambda x: x["at_390_event_verbatim"][0]["payload"].update(duration_us=1),
+        }, lambda x: check.check_sidecar(public, x))
+        for field in ("raw_event_sha256", "canonical_sha256"):
+            changed = copy.deepcopy(full)
+            changed["at_390"]["events"][0][field] = "f" * 64
+            pub, sc = evidence(changed, events)
+            with self.assertRaises(ValueError):
+                check.check_sidecar(pub, sc)
+        changed = copy.deepcopy(full)
+        changed["at_390"]["events"][0]["canonical_sha256"] = check.json_sha(
+            {"event_id": events[0]["event_id"], "replaced_for_cross_pid_compare": True})
+        pub, sc = evidence(changed, events)
+        with self.assertRaises(ValueError):
+            check.check_sidecar(pub, sc)
+        # D5 operates on an entire E element, including type and timing differences.
+        # This exercises the binding independently of the strict > saved_minute filter.
+        canonical_observations = []
+        raw_shas = []
+        for kind in ("save", "load"):
+            raw_event = io(kind, 361)["event"]
+            raw_event["payload"]["duration_us"] += 1 if kind == "load" else 0
+            value = copy.deepcopy(full)
+            value["at_390"]["events"] = [summarize(raw_event, value["event_id"])]
+            pub, sc = evidence(value, [raw_event])
+            check.check_sidecar(pub, sc)
+            canonical_observations.append(check.canonical_observation(value["at_390"], value["event_id"]))
+            raw_shas.append(value["at_390"]["events"][0]["raw_event_sha256"])
+            with self.assertRaises(ValueError):
+                check.check_report(pub, sc)  # E at saved_minute is outside this observation's window.
+        self.assertNotEqual(raw_shas[0], raw_shas[1])
+        self.assertEqual(*canonical_observations)
+        # All other event fields and canonical hashes remain significant cross PID.
+        reader = report("O-reader", 101)
+        reader["at_390"]["events"] = copy.deepcopy(full["at_390"]["events"])
+        rp, rs = evidence(reader, events)
+        check.check_pair(public, rp, sidecar, rs)
+        changed_events = copy.deepcopy(events)
+        changed_events[0]["payload"]["count"] = 9
+        reader["at_390"]["events"] = [summarize(e, reader["event_id"]) for e in changed_events]
+        rp, rs = evidence(reader, changed_events)
+        check.check_report(rp, rs)
+        with self.assertRaises(ValueError):
+            check.check_pair(public, rp, sidecar, rs)
+
+    def test_v2_receipt_sidecar_binding_rejections(self):
+        with tempfile.TemporaryDirectory(dir=os.environ.get("BUILD_OUTPUT_DIR") or None) as temporary:
+            out = Path(temporary)
+            receipt = synthetic_receipt(out)
+            check.check_receipt(receipt, out)
+            self.rejects(receipt, {
+                "unbound": lambda x: x["phases"][3].pop("sidecar"),
+                "wrong SHA": lambda x: x["phases"][3]["sidecar"].update(sha256="f" * 64),
+                "missing map": lambda x: x["profiles"][x["phases"][3]["runtime_profile"]].pop(check.SIDECAR),
+                "wrong path": lambda x: x["phases"][3]["sidecar"].update(path="other.json"),
+                "coverage omitted": lambda x: x.pop("coverage"),
+                "false coverage": lambda x: x["coverage"].update(overload_nonzero_backlog="exercised_recorded"),
+                "persistence claim": lambda x: x["coverage"].update(across_restart_backlog_persistence_claimed=True),
+            }, lambda x: check.check_receipt(x, out))
+            path = out / "R-writer-states/report_full.json"
+            saved = path.read_bytes()
+            path.write_bytes(b"{}")
+            with self.assertRaises(ValueError):
+                check.check_receipt(receipt, out)
+            path.unlink()
+            with self.assertRaises(ValueError):
+                check.check_receipt(receipt, out)
+            path.write_bytes(saved)
+            private = Path(receipt["phases"][3]["runtime_profile"]) / check.SIDECAR
+            private.write_bytes(b"{}")
+            with self.assertRaises(ValueError):
+                check.check_receipt(receipt, out)
+
+    def test_v3_overload_is_observation_not_persistence(self):
+        for count, backlog in ((0, 0), (0, 121), (1, 0), (1, 121)):
+            full = report()
+            full["overloads"] = [{"minute": 612, "backlog": backlog}] * count
+            full["paused_idle"]["before"]["backlog"] = backlog
+            full["paused_idle"]["after"]["backlog"] = backlog
+            full["overload_case"] = {"exercised": bool(count and backlog > 0),
+                                     "overloads": count, "backlog_at_save": backlog}
+            validate_report(full)
+            observed = check.coverage([full])
+            self.assertEqual(observed["overload_nonzero_backlog"],
+                             "exercised_recorded" if count and backlog else "not_exercised")
+            self.assertFalse(observed["across_restart_backlog_persistence_claimed"])
+            self.rejects(full, {
+                "missing disclosure": lambda x: x.pop("overload_case"),
+                "wrong exercised": lambda x: x["overload_case"].update(exercised=not bool(count and backlog)),
+                "wrong count": lambda x: x["overload_case"].update(overloads=count + 1),
+                "wrong backlog": lambda x: x["overload_case"].update(backlog_at_save=backlog + 1),
+                "consumption": lambda x: x["paused_idle"]["after"].update(backlog=backlog + 1),
+                "persistence claim": lambda x: x.update(across_restart_backlog_persistence_claimed=True),
+                "nonzero claim": lambda x: x.update(nonzero_backlog_preserved=True),
+            }, validate_report)
+        writer, reader = report(), report("R-reader", 101)
+        reader["paused_idle"]["before"]["backlog"] = 0
+        reader["paused_idle"]["after"]["backlog"] = 0
+        for row in reader["minutes"]:
+            row.update(backlog_before=0, backlog_after=0)
+        validate_pair(writer, reader)  # No across-restart accumulator persistence requirement.
+
+    def test_v4_no_arrival_is_diagnostic_failure(self):
+        full = report("O-writer")
+        full.update(boundary=None, boundary_reason="no_arrival_within_cap",
+                    stop_reason="no_arrival_within_cap", next_boundary=None, io=[], minutes=[], actual_k=0)
+        full["ui_ticks"] = [{"clock": n, "guest_auto_ids": [], "guest_auto_count": 0}
+                            for n in range(361, 481)]
+        self.assertTrue(check.check_no_arrival_diagnostic(full))
+        with self.assertRaisesRegex(ValueError, "incomplete probe"):
+            validate_report(full)
+        self.rejects(full, {
+            "missing reason": lambda x: x.pop("boundary_reason"),
+            "wrong reason": lambda x: x.update(boundary_reason="complete"),
+            "not null": lambda x: x.update(boundary={}),
+            "short cap": lambda x: x["ui_ticks"].pop(),
+        }, check.check_no_arrival_diagnostic)
+        with tempfile.TemporaryDirectory(dir=os.environ.get("BUILD_OUTPUT_DIR") or None) as temporary:
+            out = Path(temporary)
+            receipt = synthetic_receipt(out)
+            row = receipt["phases"][5]
+            full["pid"] = row["report"]["pid"]
+            public, sidecar = evidence(full)
+            row["report"] = public
+            text = "P28_RESTART_REPORT=" + json.dumps(public) + "\nP28_RESTART_AUDIO_CLOSE=" + json.dumps(audio(full["pid"])) + "\n"
+            (out / "O-writer/stdout.log").write_text(text)
+            row["raw"]["stdout.log"] = check.pin(out / "O-writer/stdout.log")
+            row["process"]["stdout_sha256"] = row["raw"]["stdout.log"]["sha256"]
+            (out / "O-writer/process.json").write_text(json.dumps(row["process"]))
+            row["owned_files"]["process.json"] = check.pin(out / "O-writer/process.json")
+            for path in (out / "O-writer-states/report_full.json",
+                         Path(row["runtime_profile"]) / check.SIDECAR):
+                path.write_text(json.dumps(sidecar))
+            binding = check.pin(out / "O-writer-states/report_full.json")
+            row["sidecar"] = {"path": check.SIDECAR, **binding}
+            receipt["profiles"][row["runtime_profile"]][check.SIDECAR] = binding
+            with self.assertRaisesRegex(ValueError, "incomplete probe"):
+                check.check_receipt(receipt, out)
+
+    def test_v5_engine_elapsed_deadline(self):
+        for elapsed in (0, .25, 37, 1739.5):
+            full = report()
+            full["probe_deadline"].update(engine_elapsed_at_begin_seconds=elapsed, timer_seconds=1740-elapsed)
+            validate_report(full)
+        value = report()["probe_deadline"]
+        self.rejects(value, {
+            "old relative timer": lambda x: x.update(timer_seconds=1740),
+            "wrong limit": lambda x: x.update(limit_seconds_from_engine_start=1800),
+            "negative elapsed": lambda x: x.update(engine_elapsed_at_begin_seconds=-1, timer_seconds=1741),
+            "missing elapsed": lambda x: x.pop("engine_elapsed_at_begin_seconds"),
+            "expired": lambda x: x.update(engine_elapsed_at_begin_seconds=1740, timer_seconds=0),
+            "nonfinite": lambda x: x.update(engine_elapsed_at_begin_seconds=float("inf")),
+        }, check.check_deadline)
+        check.check_deadline({**value, "engine_elapsed_at_begin_seconds": 1750, "timer_seconds": -10}, complete=False)
+        text = (check.HERE / "restart_probe.gd").read_text()
+        for altered in (text + "\nfunc forbidden():\n\tTime.get_ticks_msec()\n",
+                        text.replace("deadline_timer_seconds() ->", "other_watchdog() ->"),
+                        text.replace("GameClock.advance_minutes(1)", "Time.get_ticks_msec()"),
+                        text + "\nfunc forbidden():\n\tTime.get_unix_time_from_system()\n"):
+            with self.assertRaises(ValueError):
+                check.check_probe(altered)
+
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)