fix(release): fail a demo take whose capture stalls to a dead grey screen

This commit is contained in:
Hermes committed 2026-11-04 12:44:44 +00:00
1 parent 664a909957
commit 9704e95100
1 file changed
+71
+71
View File
@@ -105,6 +105,70 @@ def sha256(path: Path) -> str:
return digest.hexdigest()
def measure_dead_capture(path: Path, fps: int = 4) -> dict:
"""Measure flat, unreadable stretches in the delivered demo.
A headless compositor that stalls on a contended host emits uniform grey frames: the screen, the
caption bar, and every control disappear for as long as the stall lasts. Sampling a shrunken grey
raster is enough to tell a dead stretch (near-zero variance) from a terminal card, which is dark
but full of text (high variance). A take with a multi-second dead stretch is not reviewable.
"""
width, height = 72, 128
frame_bytes = width * height
raw = subprocess.run(
["ffmpeg", "-v", "error", "-i", str(path), "-vf", f"fps={fps},scale={width}:{height}",
"-pix_fmt", "gray", "-f", "rawvideo", "pipe:1"],
check=True, stdout=subprocess.PIPE).stdout
frames = len(raw) // frame_bytes
runs: list[tuple[int, int]] = []
current: list[int] | None = None
for index in range(frames):
sample = raw[index * frame_bytes:(index + 1) * frame_bytes]
mean = sum(sample) / frame_bytes
variance = sum((value - mean) ** 2 for value in sample) / frame_bytes
flat = variance ** 0.5 < 8 and 90 < mean < 170
if flat:
if current is None:
current = [index, index]
else:
current[1] = index
elif current is not None:
runs.append((current[0], current[1]))
current = None
if current is not None:
runs.append((current[0], current[1]))
seconds = 1 / fps
worst = max(((end - start + 1) * seconds for start, end in runs), default=0.0)
total = sum((end - start + 1) * seconds for start, end in runs)
duration = frames * seconds
return {
"sample_fps": fps,
"duration_s": round(duration, 2),
"runs": [{"start_s": round(start * seconds, 2), "length_s": round((end - start + 1) * seconds, 2)}
for start, end in runs],
"worst_run_s": round(worst, 2),
"total_s": round(total, 2),
}
def assert_readable_demo(measurement: dict) -> None:
"""Refuse a take whose capture went dead: the recording is the release requirement, not a garnish.
A single sampled frame of flat grey at a scene change is structural in this pipeline and every
accepted take has a dozen of them. A stall that spans several samples is not: it is a dead screen
the reviewer is supposed to be watching, so it fails the release.
"""
stalled = [run["length_s"] for run in measurement["runs"] if run["length_s"] > 0.5]
stalled_total = sum(stalled)
budget = max(1.0, 0.005 * measurement["duration_s"])
if stalled or stalled_total > budget:
raise SystemExit(
"Release demo capture went dead for "
f"{measurement['worst_run_s']}s in one stretch ({stalled_total}s stalled of "
f"{measurement['duration_s']}s): re-record on a quiet host instead of shipping a grey screen"
)
def assert_port_free(host: str, port: int) -> None:
"""Fail closed when something is already listening where the acceptance server must bind."""
with socket.socket() as sock:
@@ -277,6 +341,11 @@ def main() -> int:
video_streams = [stream for stream in probe.get("streams", []) if stream.get("codec_type") == "video"]
if len(video_streams) != 1 or video_streams[0].get("codec_name") != "h264" or video_streams[0].get("pix_fmt") != "yuv420p" or video_streams[0].get("width") != 720 or video_streams[0].get("height") != 1280:
raise SystemExit("Release demo media probe failed")
# A stalled compositor paints uniform grey over the whole screen, caption bar included, so the
# take reads as a dead channel for as long as the stall lasts. Measure it; do not eyeball it.
dead_capture = measure_dead_capture(demo)
assert_readable_demo(dead_capture)
print(json.dumps({"dead_capture": dead_capture}), flush=True)
contact_sheet = release_dir / f"timmy-talking-turd-{version}-demo-contact-sheet.jpg"
# Span the whole recording: a fixed 1/2 fps sheet only covered the first 24s of a longer demo.
demo_duration = float(run(["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "csv=p=0", str(demo)], tree, capture=True))
@@ -412,7 +481,9 @@ def main() -> int:
"feature_demo_full_decode": "passed",
"feature_demo_media_probe": "passed",
"release_delta_visible_on_camera": "passed",
"feature_demo_no_dead_capture": "passed",
},
"demo_dead_capture": dead_capture,
}
manifest_path = release_dir / "manifest.json"
manifest_path.write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8")