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

@@ -6,6 +6,7 @@ import json
 import math
 from pathlib import Path
 import re
+import struct
 
 PREFIX = "P28_01_PERF_REPORT="
 RAW_SOURCE = "clock.performance_budget (product diagnostic, raw, not a verdict)"
@@ -35,6 +36,9 @@ FORBIDDEN = ("_toggle_pause", "clear(", "enable_production_", "set_paused(false)
 # Only declared collector fields are normalized. Product dictionaries (RNG,
 # startup, raw reports) retain their original JSON representation.
 INTEGER_FIELDS = {
+    "engine_frame_pre", "engine_frame_post", "pre_order", "post_order",
+    "minute_at_frame_start", "minute_after", "speed_at_frame_start",
+    "allowed_ticks_observation", "short_of_allowed_frames",
     "cores", "memory_bytes", "pid", "seed", "frame", "game_start", "game_end", "game_minute",
     "minutes_advanced", "start_minute", "end_minute", "opening_minute", "actual_minutes",
     "warmup_target", "measurement_target", "speed", "speed_after", "backlog",
@@ -233,6 +237,9 @@ def interval(report, name):
     need(set(observation) == {"frame_wall_us", "sim_us_per_minute", "stall_over_50000_us"},
          "frame observation must not contain verdict")
     need(value["frames"] == len(rows), "frame count")
+    need(value.get("short_of_allowed_frames") ==
+         sum(f["minutes_advanced"] < f["allowed_ticks_observation"] for f in rows),
+         "short-of-allowed observation count")
     need(value["actual_minutes"] == sum(f["minutes_advanced"] for f in rows), "false interval duration")
     for field, expected in (("start_minute", rows[0]["game_start"] if rows else None),
                             ("end_minute", rows[-1]["game_end"] if rows else None),
@@ -325,7 +332,15 @@ def check_report(report):
     previous_wall = report["start_wall"]
     previous_drawing = None
     previous_counts = None
+    keys(report, ("baseline",))
+    previous_pair = report["baseline"]
+    check_pair(previous_pair)
+    need(previous_pair.get("excluded_from_statistics") is True and
+         previous_pair["minute_after"] == previous and previous_pair.get("wall") == previous_wall,
+         "missing/exposed baseline")
     for index, frame in enumerate(frames):
+        check_pair(frame, previous_pair, speed)
+        previous_pair = frame
         keys(frame, ("frame", "game_start", "game_end", "minutes_advanced", "frame_wall_us", "wall", "wall_start",
                      "speed", "paused", "segment", "registered", "active", "drawn", "seed", "rng_streams",
                      "metadata", "input_sha256", "preparation_sha256", "render_cpu_raw_ms",
@@ -475,18 +490,107 @@ def check_receipt(receipt, directory):
         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")
+        equal(audio_close(stdout, row["report"]), row["audio_close"], "audio close binding")
         need(row["capture"]["actual_captured_pids"] == [row["report"]["pid"]], "PID mismatch")
         check_report(row["report"])
         if row["phase"] in ("R0", "O0"):
             equal(row["extrapolation"], extrapolation(row["report"]), "extrapolation")
 
 
+def decode_f64(value):
+    need(isinstance(value, str) and re.fullmatch("[0-9a-f]{16}", value) is not None,
+         "missing/malformed f64 evidence")
+    result = struct.unpack("<d", bytes.fromhex(value))[0]
+    need(math.isfinite(result), "nonfinite f64 evidence")
+    return result
+
+
+def check_pair(pair, previous=None, speed=None):
+    keys(pair, ("engine_frame_pre", "engine_frame_post", "pre_order", "post_order",
+                "paused_at_frame_start", "speed_at_frame_start", "skip_at_frame_start",
+                "acc_at_frame_start_f64_hex", "minute_at_frame_start", "process_delta_f64_hex",
+                "post_delta_f64_hex", "paused_after", "minute_after", "acc_after_f64_hex"))
+    for key in ("engine_frame_pre", "engine_frame_post", "pre_order", "post_order",
+                "minute_at_frame_start", "minute_after", "speed_at_frame_start"):
+        need(type(pair[key]) is int and pair[key] >= 0, "invalid pair integer")
+    for key in ("paused_at_frame_start", "paused_after", "skip_at_frame_start"):
+        need(type(pair[key]) is bool, "invalid pair boolean")
+    need(pair["engine_frame_pre"] == pair["engine_frame_post"] and
+         pair["post_order"] == pair["pre_order"] + 1, "unpaired frame")
+    acc = decode_f64(pair["acc_at_frame_start_f64_hex"])
+    after = decode_f64(pair["acc_after_f64_hex"])
+    delta = decode_f64(pair["process_delta_f64_hex"])
+    decode_f64(pair["post_delta_f64_hex"])
+    need(pair["process_delta_f64_hex"] == pair["post_delta_f64_hex"], "different process deltas")
+    if previous is None:
+        # First complete pair is an excluded anchor, never an inferred pre-begin state.
+        return
+    need(pair["engine_frame_pre"] == previous["engine_frame_post"] + 1 and
+         pair["pre_order"] == previous["post_order"] + 1, "frame order gap")
+    need(pair["minute_at_frame_start"] == previous["minute_after"] and
+         pair["acc_at_frame_start_f64_hex"] == previous["acc_after_f64_hex"], "pair continuity")
+    need(pair["speed_at_frame_start"] == speed, "cadence speed")
+    advanced = pair["minute_after"] - pair["minute_at_frame_start"]
+    allowed = 0
+    if not pair["skip_at_frame_start"] and not pair["paused_at_frame_start"] and speed > 0 and delta > 0:
+        acc += delta * speed
+        allowed = min(int(acc / (10.0 / 60.0)), 8)
+    need(0 <= advanced <= allowed, "cadence upper bound")
+    need(type(pair.get("allowed_ticks_observation")) is int and
+         pair["allowed_ticks_observation"] == allowed, "allowed tick observation")
+    for _ in range(advanced):
+        acc -= 10.0 / 60.0
+    need(struct.pack("<d", acc).hex() == pair["acc_after_f64_hex"], "accumulator recurrence")
+    need(advanced == pair["minutes_advanced"] and
+         pair["game_start"] == pair["minute_at_frame_start"] and
+         pair["game_end"] == pair["minute_after"] and pair["paused"] is pair["paused_after"],
+         "pair/frame reconciliation")
+
+
+def audio_close(stdout, report):
+    prefix = "P28_01_AUDIO_CLOSE="
+    rows = [json.loads(line[len(prefix):]) for line in stdout.splitlines() if line.startswith(prefix)]
+    need(len(rows) == 1, "exactly one AUDIO_CLOSE required")
+    row = rows[0]
+    need(type(report.get("main_added")) is bool, "missing main-added observation")
+    need(type(report.get("audio_owner_present")) is bool, "missing audio owner observation")
+    if row.get("coordinator") is True:
+        need(report["main_added"] and report["audio_owner_present"] and row.get("passed") is True and
+             row.get("deadline_reached") is False and row.get("all_owned_targets_released") is True,
+             "audio close drain failed")
+        need(type(row.get("waited_actual_us")) in (int, float) and
+             math.isfinite(row["waited_actual_us"]) and row["waited_actual_us"] >= 0,
+             "audio close wait")
+    else:
+        need(not report["main_added"] or not report["audio_owner_present"], "owner drain bypassed")
+        need(row == {"coordinator": False, "reason":
+                     "no_audio_owner" if report["main_added"] else "main_not_added"},
+             "invalid no-owner receipt")
+    return row
+
+
 def check_probe(text):
     # The SceneTree bootstrap name is required; all other initialization is forbidden.
     body = text.replace("func _initialize()", "func BOOTSTRAP()")
     for token in FORBIDDEN:
         need(token not in body, "forbidden probe token: " + token)
     need('main.call("_set_speed", speed)' in body, "missing UI speed path")
+    close_branch = '''\tif not is_instance_valid(main) or not main.is_inside_tree():
+\t\temit_audio_close({"coordinator": false, "reason": "main_not_added"})
+\t\tquit(0)
+\telif not root.has_meta(UIAudioOwnerShutdownScript.OWNER_META):
+\t\temit_audio_close({"coordinator": false, "reason": "no_audio_owner"})
+\t\tquit(0)
+\telse:
+\t\tmain.ui_audio_window_close_drained.connect(emit_audio_close, CONNECT_ONE_SHOT)
+\t\troot.close_requested.emit()'''
+    need(close_branch in body, "missing product close branch")
+    need(re.search(r"\bquit\s*\(", body.replace(close_branch, "")) is None,
+         "quit outside no-main/no-owner branches")
+    need("tests/helpers/ui_audio_fixture_cleanup" not in body, "fixture close forbidden")
+    for token in ("pre_observer.process_priority = -1000000", "observer.process_priority = 100000",
+                  "PackedFloat64Array([value]).to_byte_array().hex_encode()", "anchor_baseline()"):
+        need(token in body, "missing passive pair observation: " + token)
 
 
 def main():

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

@@ -13,7 +13,7 @@ import uuid
 from datetime import datetime, timezone
 
 sys.dont_write_bytecode = True
-from check_p28_01_report import check_report, check_receipt, check_probe, normalize_report, check_admission, R02_FIELDS, VERDICT_SCOPE, CAP_SOURCE
+from check_p28_01_report import check_report, check_receipt, check_probe, normalize_report, check_admission, audio_close, R02_FIELDS, VERDICT_SCOPE, CAP_SOURCE
 HERE = Path(__file__).resolve().parent
 ROOT = HERE.parents[5]
 FIXTURE_SHA = "eb4f770071eb07cd7e136fe54c3f78768cc8de7a4264e62a9af1f06a8f90f9bd"
@@ -152,6 +152,11 @@ def bound_report(folder, row):
     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")
+    try:
+        row["audio_close"] = audio_close((folder / "stdout.log").read_text(), report)
+    except (ValueError, KeyError, TypeError):
+        row.update(passed=False, stop_reason="engine_failure")
+        raise
     return report
 
 

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

@@ -1,14 +1,28 @@
 extends SceneTree
+const UIAudioOwnerShutdownScript = preload("res://scripts/ui/ui_audio_owner_shutdown.gd")
 ## External, passive observer of the default autoload world.
 ## Raw per-frame lines survive an owned timeout before the final report.
 
+class PreObserver:
+	extends Node
+	var owner_probe: Variant
+	func _process(delta: float) -> void:
+		owner_probe.pre_sample(delta)
+
 class Observer:
 	extends Node
 	var owner_probe: Variant
-	func _process(_delta: float) -> void:
+	func _process(delta: float) -> void:
 		var observer_started_us := Time.get_ticks_usec()
-		owner_probe.sample(observer_started_us)
+		owner_probe.sample(observer_started_us, delta)
 
+var pending_pair := {}
+var baseline := {}
+var last_pair := {}
+var finished := false
+var audio_emitted := false
+var observation_order := 0
+var pre_cost_us := 0
 var clock: Variant
 var runtime: Variant
 var main: Variant
@@ -253,7 +267,7 @@ func begin() -> void:
 	speed = int(OS.get_environment("THE_TREE_P28_SPEED"))
 	warmup = int(OS.get_environment("THE_TREE_P28_WARMUP"))
 	target = int(OS.get_environment("THE_TREE_P28_MINUTES"))
-	report = {"schema": "p28-01-v1", "pid": OS.get_process_id(),
+	report = {"schema": "p28-01-v1", "pid": OS.get_process_id(), "main_added": false,
 		"phase": OS.get_environment("THE_TREE_P28_PHASE"), "input": input,
 		"metadata": JSON.parse_string(OS.get_environment("THE_TREE_P28_META")),
 		"input_sha256": null, "input_sha256_reason": "fresh_autoload_initialization",
@@ -301,6 +315,7 @@ func begin() -> void:
 	var opening_minute: int = clock.absolute_minute
 	main = load("res://scenes/main/main.tscn").instantiate()
 	root.add_child(main)
+	report["main_added"] = true
 	report["startup"] = {"pause_fresh_opening": {"return": null, "reason": "void_function",
 		"paused_before": paused_before, "paused_after": clock.paused},
 		"save_policy": {"return": null, "reason": "return_discarded_by_main（main.gd:222）"},
@@ -333,6 +348,10 @@ func begin() -> void:
 	runtime.save_operation_completed.connect(on_save)
 	RenderingServer.frame_pre_draw.connect(on_pre_draw)
 	RenderingServer.frame_post_draw.connect(on_post_draw)
+	var pre_observer := PreObserver.new()
+	pre_observer.owner_probe = self
+	pre_observer.process_priority = -1000000
+	root.add_child(pre_observer)
 	var observer := Observer.new()
 	observer.owner_probe = self
 	observer.process_priority = 100000
@@ -346,6 +365,9 @@ func begin() -> void:
 	if clock.paused or clock.speed != speed:
 		finish("input_shortfall")
 		return
+	ready_to_sample = true
+
+func anchor_baseline() -> void:
 	start_minute = clock.absolute_minute
 	measurement_start = start_minute if warmup == 0 else -1
 	previous_minute = start_minute
@@ -363,7 +385,12 @@ func begin() -> void:
 	report["start_minute"] = start_minute
 	report["start_wall"] = Time.get_unix_time_from_system()
 	previous_wall = report["start_wall"]
-	ready_to_sample = true
+	baseline["wall"] = report.start_wall
+	baseline["excluded_from_statistics"] = true
+	report["baseline"] = baseline
+	cached_registered = registered()
+	cached_active = active_counts()
+	cached_rng = rng_state()
 
 func on_overload(backlog: int) -> void:
 	overloads.append({"backlog": backlog, "game_minute": clock.absolute_minute, "frame": frames.size()})
@@ -373,7 +400,7 @@ func on_checkpoint(_presentation: Dictionary) -> void:
 		"wall": Time.get_unix_time_from_system(), "cause": "production_checkpoint_presentation_requested"})
 
 func on_tick(minute: int) -> void:
-	if not ready_to_sample:
+	if not ready_to_sample or baseline.is_empty():
 		return
 	var tick_us := Time.get_ticks_usec()
 	var row := {"game_minute": minute, "wall": Time.get_unix_time_from_system(),
@@ -383,9 +410,61 @@ func on_tick(minute: int) -> void:
 	previous_tick_us = tick_us
 	print("P28_01_MINUTE=" + JSON.stringify(row))
 
-func sample(observer_started_us: int) -> void:
+func f64(value: float) -> String:
+	return PackedFloat64Array([value]).to_byte_array().hex_encode()
+
+func pre_sample(delta: float) -> void:
 	if not ready_to_sample:
 		return
+	var began := Time.get_ticks_usec()
+	if not pending_pair.is_empty():
+		report["frame_order_evidence"] = pending_pair
+		finish("frame_order_unverified")
+		return
+	observation_order += 1
+	pending_pair = {"engine_frame_pre": Engine.get_process_frames(), "pre_order": observation_order,
+		"paused_at_frame_start": clock.paused, "speed_at_frame_start": clock.speed,
+		"skip_at_frame_start": clock.get("_skip_next_process"),
+		"acc_at_frame_start_f64_hex": f64(clock.get("_accumulator")),
+		"minute_at_frame_start": clock.absolute_minute, "process_delta_f64_hex": f64(delta)}
+	var allowed := 0
+	if not pending_pair.skip_at_frame_start and not clock.paused and clock.speed > 0 and delta > 0:
+		allowed = mini(int((float(clock.get("_accumulator")) + delta * clock.speed) / (10.0 / 60.0)), 8)
+	pending_pair["allowed_ticks_observation"] = allowed
+	pre_cost_us = Time.get_ticks_usec() - began
+
+func sample(observer_started_us: int, delta: float) -> void:
+	if not ready_to_sample:
+		return
+	if pending_pair.is_empty():
+		if baseline.is_empty():
+			report["leading_partial_frame"] = {"engine_frame_post": Engine.get_process_frames()}
+			return
+		report["frame_order_evidence"] = {"engine_frame_post": Engine.get_process_frames()}
+		finish("frame_order_unverified")
+		return
+	var pair := pending_pair
+	pending_pair = {}
+	observation_order += 1
+	pair.merge({"engine_frame_post": Engine.get_process_frames(), "post_order": observation_order,
+		"post_delta_f64_hex": f64(delta), "paused_after": clock.paused,
+		"minute_after": clock.absolute_minute, "acc_after_f64_hex": f64(clock.get("_accumulator"))})
+	if pair.engine_frame_pre != pair.engine_frame_post or pair.process_delta_f64_hex != pair.post_delta_f64_hex:
+		report["frame_order_evidence"] = pair
+		finish("frame_order_unverified")
+		return
+	if baseline.is_empty():
+		baseline = pair
+		last_pair = pair
+		anchor_baseline()
+		baseline["observer_us"] = pre_cost_us + Time.get_ticks_usec() - observer_started_us
+		return
+	if pair.engine_frame_pre != last_pair.engine_frame_post + 1 or pair.pre_order != last_pair.post_order + 1 \
+	or pair.minute_at_frame_start != last_pair.minute_after or pair.acc_at_frame_start_f64_hex != last_pair.acc_after_f64_hex:
+		report["frame_order_evidence"] = {"previous": last_pair, "current": pair}
+		finish("frame_order_unverified")
+		return
+	last_pair = pair
 	var current_us := Time.get_ticks_usec()
 	var minute: int = clock.absolute_minute
 	var elapsed := current_us - previous_us
@@ -426,6 +505,7 @@ func sample(observer_started_us: int) -> void:
 		"render_cpu_raw_ms": RenderingServer.viewport_get_measured_render_time_cpu(root.get_viewport_rid()),
 		"render_gpu_raw_ms": RenderingServer.viewport_get_measured_render_time_gpu(root.get_viewport_rid()),
 		"time_process_seconds": Performance.get_monitor(Performance.TIME_PROCESS)}
+	row.merge(pair)
 	last_render_serial = render_serial
 	previous_us = current_us
 	previous_wall = row.wall
@@ -458,7 +538,7 @@ func sample(observer_started_us: int) -> void:
 	# summary are disclosed separately and are never subtracted from frame wall.
 	var raw_line := JSON.stringify(row)
 	print("P28_01_FRAME=" + raw_line)
-	row["observer_us"] = Time.get_ticks_usec() - observer_started_us
+	row["observer_us"] = pre_cost_us + Time.get_ticks_usec() - observer_started_us
 	frames.append(row)
 	print("P28_01_OBSERVER=" + JSON.stringify({"frame": row.frame, "observer_us": row.observer_us}))
 	if not reason.is_empty():
@@ -486,6 +566,7 @@ func summary(segment: String) -> Dictionary:
 	var observer: Array = []
 	var submit: Array = []
 	var excluded := 0
+	var short_of_allowed := 0
 	var minutes := 0
 	var cpu_missing := false
 	var gpu_missing := false
@@ -500,6 +581,8 @@ func summary(segment: String) -> Dictionary:
 			submit.append(frame.render_submit_wall_us)
 		process.append(frame.time_process_seconds)
 		minutes += frame.minutes_advanced
+		if frame.minutes_advanced < frame.allowed_ticks_observation:
+			short_of_allowed += 1
 		if frame.frame_wall_us > 50000:
 			stalls.append(frame.frame_wall_us)
 		if frame.minutes_advanced > 0:
@@ -520,6 +603,7 @@ func summary(segment: String) -> Dictionary:
 	var unavailable := "zero_from_api_on_gl_compatibility_unsupported_or_unmeasured"
 	var minute_stats := stats(per_minute)
 	return {"frames": selected.size(), "actual_minutes": minutes,
+		"short_of_allowed_frames": short_of_allowed,
 		"start_minute": selected.front().game_start if not selected.is_empty() else null,
 		"end_minute": selected.back().game_end if not selected.is_empty() else null,
 		"start_wall": selected.front().wall_start if not selected.is_empty() else null,
@@ -534,7 +618,13 @@ func summary(segment: String) -> Dictionary:
 		"time_process_seconds": stats(process), "count_ranges": ranges}
 
 func finish(reason: String) -> void:
+	if finished:
+		return
+	finished = true
 	ready_to_sample = false
+	if not pending_pair.is_empty():
+		report["trailing_partial_frame"] = pending_pair
+		pending_pair = {}
 	# 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
@@ -544,9 +634,25 @@ func finish(reason: String) -> void:
 	report["end_wall"] = Time.get_unix_time_from_system()
 	report["kernel_at_phase_end"] = {"source": "clock.performance_budget (product diagnostic, raw, not a verdict)", "data": clock.get_performance_reports()}
 	report["budget_at_phase_end"] = {"source": "clock.performance_budget (product diagnostic, raw, not a verdict)", "data": budget_snapshot()}
-	report["budget_verdict"] = kernel_verdict(report.budget_at_warmup_end.data, report.budget_at_phase_end.data)
+	if reason != "frame_order_unverified":
+		report["budget_verdict"] = kernel_verdict(report.budget_at_warmup_end.data, report.budget_at_phase_end.data)
 	report["operations_at_end"] = runtime.get_operations_report()
 	report["warmup"] = summary("warmup")
 	report["measurement"] = summary("measurement")
+	report["audio_owner_present"] = root.has_meta(UIAudioOwnerShutdownScript.OWNER_META)
 	print("P28_01_PERF_REPORT=" + JSON.stringify(report))
-	quit(0)
+	if not is_instance_valid(main) or not main.is_inside_tree():
+		emit_audio_close({"coordinator": false, "reason": "main_not_added"})
+		quit(0)
+	elif not root.has_meta(UIAudioOwnerShutdownScript.OWNER_META):
+		emit_audio_close({"coordinator": false, "reason": "no_audio_owner"})
+		quit(0)
+	else:
+		main.ui_audio_window_close_drained.connect(emit_audio_close, CONNECT_ONE_SHOT)
+		root.close_requested.emit()
+
+func emit_audio_close(result: Dictionary) -> void:
+	if audio_emitted:
+		return
+	audio_emitted = true
+	print("P28_01_AUDIO_CLOSE=" + JSON.stringify(result))

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

@@ -4,6 +4,7 @@ import ast
 import copy
 import hashlib
 import json
+import struct
 from pathlib import Path
 import sys
 import tempfile
@@ -19,6 +20,30 @@ import collect_p28_01 as collector
 HERE = Path(__file__).resolve().parent
 
 
+def f64(value):
+    return struct.pack("<d", value).hex()
+
+
+def pair(number, minute, acc, delta=.2, ticks=1, speed=1, paused=False, skip=False,
+         paused_after=False):
+    after = acc
+    allowed = 0
+    if not skip and not paused and speed > 0 and delta > 0:
+        after += delta * speed
+        allowed = min(int(after / (10.0 / 60.0)), 8)
+    for _ in range(ticks):
+        after -= 10.0 / 60.0
+    return {"engine_frame_pre": number, "engine_frame_post": number,
+            "pre_order": number * 2 + 1, "post_order": number * 2 + 2,
+            "paused_at_frame_start": paused, "speed_at_frame_start": speed,
+            "skip_at_frame_start": skip, "acc_at_frame_start_f64_hex": f64(acc),
+            "minute_at_frame_start": minute, "process_delta_f64_hex": f64(delta),
+            "post_delta_f64_hex": f64(delta), "paused_after": paused_after,
+            "minute_after": minute + ticks, "acc_after_f64_hex": f64(after),
+            "minutes_advanced": ticks, "game_start": minute, "game_end": minute + ticks,
+            "paused": paused_after, "allowed_ticks_observation": allowed}
+
+
 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),
@@ -121,6 +146,19 @@ def synthetic_report():
     for name in checker.RAW_FIELDS:
         report[name] = {"source": checker.RAW_SOURCE, "data": report[name]}
     report["probe_deadline"] = {"limit_seconds": 1740, "source": checker.DEADLINE_SOURCE}
+    report["main_added"] = True
+    report["audio_owner_present"] = True
+    report["baseline"] = pair(0, 360, 0, delta=0, ticks=0, skip=True)
+    report["baseline"].update(excluded_from_statistics=True, wall=100)
+    acc = 0.0
+    for i, frame in enumerate(frames):
+        evidence = pair(i + 1, frame["game_start"], acc)
+        frame.update(evidence)
+        acc = checker.decode_f64(evidence["acc_after_f64_hex"])
+    for name in ("warmup", "measurement"):
+        report[name]["short_of_allowed_frames"] = sum(
+            f["minutes_advanced"] < f["allowed_ticks_observation"]
+            for f in frames if f["segment"] == name)
     return report
 
 
@@ -131,6 +169,105 @@ def synthetic_admission():
 
 
 class ReportTests(unittest.TestCase):
+    def test_k1_audio_close_receipts(self):
+        report = {"main_added": True, "audio_owner_present": True}
+        close = {"coordinator": True, "passed": True, "deadline_reached": False,
+                 "all_owned_targets_released": True, "waited_actual_us": 15000}
+        def raw(value):
+            return "P28_01_AUDIO_CLOSE=" + json.dumps(value) + "\n"
+        self.assertEqual(checker.audio_close(raw(close), report), close)
+        no_owner = {"coordinator": False, "reason": "no_audio_owner"}
+        self.assertEqual(checker.audio_close(raw(no_owner),
+                         dict(report, audio_owner_present=False)), no_owner)
+        for text in ("", raw(close) * 2, raw(dict(close, passed=False)),
+                     raw(dict(close, deadline_reached=True)),
+                     raw(dict(close, all_owned_targets_released=False)),
+                     raw({"coordinator": False, "reason": "main_not_added"}), raw(no_owner)):
+            with self.subTest(text=text), self.assertRaises(ValueError):
+                checker.audio_close(text, report)
+        # Binding uses the actual raw marker, independently of a phase's claimed pass.
+        with patch.object(Path, "read_text", return_value=checker.PREFIX +
+                          json.dumps({"pid": 123, **report}) + "\n" + raw(close)):
+            row = {"capture": {"actual_captured_pids": [123]}}
+            collector.bound_report(Path("synthetic"), row)
+            self.assertEqual(row["audio_close"], close)
+
+    def test_k1_product_close_static(self):
+        text = (HERE / "measure_p28_01.gd").read_text()
+        checker.check_probe(text)
+        owner = text.split("\telse:\n\t\tmain.ui_audio_window_close_drained.connect", 1)[1]
+        self.assertNotIn("quit(", owner)
+        for broken in (text + "\nfunc bypass():\n\tquit(0)\n",
+                       text.replace("\t\tquit(0)", "\tquit(0)", 1),
+                       text.replace("root.close_requested.emit()", "quit(0)"),
+                       text + "\n# tests/helpers/ui_audio_fixture_cleanup.gd"):
+            with self.assertRaises(ValueError):
+                checker.check_probe(broken)
+
+    def test_k2_valid_cadence_boundaries(self):
+        for options in ({"acc": 2.0, "delta": .016, "ticks": 8},
+                        {"acc": 2.0, "delta": .016, "ticks": 3, "paused_after": True},
+                        {"acc": 2.0, "delta": .016, "ticks": 0, "skip": True},
+                        {"acc": 30.0, "delta": .016, "ticks": 8, "paused_after": True},
+                        {"acc": 30.0, "delta": .016, "ticks": 0, "paused": True,
+                         "paused_after": True}):
+            baseline = pair(0, 360, options["acc"], delta=0, ticks=0)
+            current = pair(1, 360, **options)
+            with self.subTest(options=options):
+                checker.check_pair(current, baseline, 1)
+        # A real skip may follow an excluded baseline; don't guess it was consumed.
+        baseline = pair(0, 360, 0, delta=0, ticks=0)
+        skipped = pair(1, 360, 0, ticks=0, skip=True)
+        checker.check_pair(skipped, baseline, 1)
+        checker.check_pair(pair(2, 360, 0), skipped, 1)
+
+    def test_k2_reject_cadence_and_missing_evidence(self):
+        baseline = pair(0, 360, 0, delta=0, ticks=0)
+        valid = pair(1, 360, 0)
+        mutations = [
+            lambda x: x.update(**pair(1, 360, 0, delta=.016, ticks=1)),
+            lambda x: x.update(**pair(1, 360, 0, delta=.2, ticks=3)),
+            lambda x: x.update(**pair(1, 360, 0, delta=2, ticks=9)),
+            lambda x: x.update(**pair(1, 360, 0, paused=True)),
+            lambda x: x.update(minute_at_frame_start=361),
+            lambda x: x.update(acc_at_frame_start_f64_hex=f64(.01)),
+            lambda x: x.update(engine_frame_post=2),
+            lambda x: x.update(pre_order=4),
+        ]
+        for key in ("process_delta_f64_hex", "post_delta_f64_hex",
+                    "acc_at_frame_start_f64_hex", "acc_after_f64_hex", "skip_at_frame_start"):
+            mutations.append(lambda x, k=key: x.pop(k))
+            mutations.append(lambda x, k=key: x.update({k: "broken"}))
+        mutations.append(lambda x: x.update(process_delta_f64_hex=f64(float("nan"))))
+        for mutate in mutations:
+            current = copy.deepcopy(valid)
+            mutate(current)
+            with self.subTest(current=current), self.assertRaises(ValueError):
+                checker.check_pair(current, baseline, 1)
+        baseline = pair(0, 360, 2.0, delta=0, ticks=0)
+        current = pair(1, 360, 2.0, delta=.016, ticks=8)
+        bulk = f64((2.0 + .016) - (10.0 / 60.0) * 8)
+        self.assertNotEqual(bulk, current["acc_after_f64_hex"])
+        current["acc_after_f64_hex"] = bulk
+        with self.assertRaises(ValueError):
+            checker.check_pair(current, baseline, 1)
+
+    def test_k3_baseline_and_complete_pairs(self):
+        checker.check_report(synthetic_report())
+        for mutate in (
+            lambda r: r.pop("baseline"),
+            lambda r: r["baseline"].update(excluded_from_statistics=False),
+            lambda r: r["baseline"].update(minute_after=359),
+            lambda r: r["baseline"].update(wall=99),
+            lambda r: r["frames"][0].update(engine_frame_pre=0, engine_frame_post=0,
+                                            pre_order=1, post_order=2),
+            lambda r: r["frames"][1].pop("post_order"),
+            lambda r: r["frames"][1].update(acc_at_frame_start_f64_hex=f64(0)),
+            lambda r: r["frames"][1].update(minute_at_frame_start=360),
+        ):
+            with self.subTest(mutate=mutate):
+                self.reject(mutate)
+
     def test_r1_recursive_verdict_scope(self):
         report = synthetic_report()
         receipt = {"admission": synthetic_admission(), "phases": [{"report": report, "passed": True}],
@@ -306,10 +443,12 @@ class ReportTests(unittest.TestCase):
 
     def test_receipt_rejects_source_player_and_raw_pid(self):
         r = synthetic_report()
-        raw = (checker.PREFIX + json.dumps(r) + "\n").encode()
+        close = {"coordinator": True, "passed": True, "deadline_reached": False,
+                 "all_owned_targets_released": True, "waited_actual_us": 15000}
+        raw = (checker.PREFIX + json.dumps(r) + "\nP28_01_AUDIO_CLOSE=" + json.dumps(close) + "\n").encode()
         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]},
+               "report": r, "audio_close": close, "capture": {"actual_captured_pids": [123]},
                "extrapolation": collector.extrapolation(r),
                "guard": synthetic_guard(), "runtime_profile": "private", "runtime_name": "runtime"}
         receipt = {"admission": synthetic_admission(), "source_before": {"x": "a"}, "source_after": {"x": "a"},
@@ -475,7 +614,8 @@ class ReportTests(unittest.TestCase):
             "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"
+            " print('P28_01_PERF_REPORT='+json.dumps({'pid':os.getpid()+int(mode=='pid'),'main_added':False,'audio_owner_present':False}),flush=True)\n"
+            "print('P28_01_AUDIO_CLOSE='+json.dumps({'coordinator':False,'reason':'main_not_added'}),flush=True)\n"
             "sys.exit(3 if mode=='exit' else 0)\n")
         with tempfile.TemporaryDirectory() as temp:
             base = Path(temp)