test(uvc): catch flat-frame receiver freezes

Cargo.lock

@@ -1658,7 +1658,7 @@ checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2"
 
 [[package]]
 name = "lesavka_client"
-version = "0.27.12"
+version = "0.27.13"
 dependencies = [
  "anyhow",
  "async-stream",
@@ -1692,7 +1692,7 @@ dependencies = [
 
 [[package]]
 name = "lesavka_common"
-version = "0.27.12"
+version = "0.27.13"
 dependencies = [
  "anyhow",
  "base64",
@@ -1704,7 +1704,7 @@ dependencies = [
 
 [[package]]
 name = "lesavka_server"
-version = "0.27.12"
+version = "0.27.13"
 dependencies = [
  "anyhow",
  "base64",

client/Cargo.toml

@@ -4,7 +4,7 @@ path                    = "src/main.rs"
 
 [package]
 name                    = "lesavka_client"
-version                 = "0.27.12"
+version                 = "0.27.13"
 edition                 = "2024"
 
 [dependencies]

common/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name                    = "lesavka_common"
-version                 = "0.27.12"
+version                 = "0.27.13"
 edition                 = "2024"
 build                   = "build.rs"
 

docs/architecture-and-media-readiness.md

@@ -338,6 +338,12 @@ slabs and lower-frame tears fail independently from motion/blur flags that need
 human review; incomplete duration, decoder failure, timeouts, and structural
 corruption also fail instead of producing an easy success.
 
+Release 0.27.13 closes the remaining real-video detector blind spot exposed by
+the preserved 310-second receiver failure. A sudden whole-frame black/grey
+collapse and an exact-frame replay lasting two seconds are now hard failures.
+The historical Tethys recording fails at its known `103.467s` collapse while
+the simultaneous clean Theia spool control reports zero hard corruption.
+
 ## 7. The Resolved Downstream Video Failure
 
 The current blank downstream feeds fail before transport or decoding.
@@ -414,7 +420,7 @@ The safe completion sequence for this incident is:
 6. Run `scripts/install/server.sh` as the trusted deployment path. Preserve the
    already-attached USB gadget unless a controlled rebuild is explicitly
    required.
-7. Confirm Theia reports server version `0.27.12`, the pushed release revision,
+7. Confirm Theia reports server version `0.27.13`, the pushed release revision,
    direct MJPEG normalizer timeout `0`, and a coherent UVC contract.
 8. Open both downstream RPCs and prove that each emits changing, decodable H.264
    frames.
@@ -432,7 +438,7 @@ hardware contract is repeatable. The remaining work falls into five groups.
 
 ### A. Install And Version Parity
 
-- Push and deploy `0.27.12` through the client/server install scripts.
+- Push and deploy `0.27.13` through the client/server install scripts.
 - Confirm client/server version and revision in every hardware probe artifact.
 - Eliminate the current state where a fixed client talks to an unfixed server.
 
@@ -498,6 +504,6 @@ host repair:
 9. disconnect/reconnect and device changes recover without stale media; and
 10. diagnostics identify the failed physical stage when any item breaks.
 
-Until that sequence passes on the installed `0.27.12` client/server pair, the
+Until that sequence passes on the installed `0.27.13` client/server pair, the
 current release should be described as a validated code correction awaiting
 hardware deployment and end-to-end acceptance, not as a completed product fix.

scripts/manual/run_rct_uvc_artifact_probe.py

@@ -16,7 +16,9 @@ import time
 from typing import Any
 
 DEFAULT_DEVICE_LABEL = "Lesavka Composite"
-HARD_CORRUPTION_REASONS = frozenset({"lower_boundary_jump", "lower_flat_flash", "lower_slab"})
+HARD_CORRUPTION_REASONS = frozenset(
+    {"lower_boundary_jump", "lower_flat_flash", "lower_slab", "global_flat_flash", "exact_freeze"}
+)
 
 
 def parse_args() -> argparse.Namespace:
@@ -51,6 +53,12 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--blur-delta-threshold", type=float, default=6.0)
     parser.add_argument("--blur-var-ratio", type=float, default=0.45)
     parser.add_argument("--change-threshold", type=float, default=1.0)
+    parser.add_argument(
+        "--max-exact-repeat-seconds",
+        type=float,
+        default=2.0,
+        help="fail when identical decoded frames persist this long after capture starts",
+    )
     parser.add_argument("--max-suspicious-artifacts", type=int, default=40)
     parser.add_argument("--max-reference-artifacts", type=int, default=12)
     parser.add_argument("--reference-every", type=int, default=900)
@@ -150,6 +158,8 @@ def run_remote(args: argparse.Namespace) -> int:
         str(args.blur_var_ratio),
         "--change-threshold",
         str(args.change_threshold),
+        "--max-exact-repeat-seconds",
+        str(args.max_exact_repeat_seconds),
         "--max-suspicious-artifacts",
         str(args.max_suspicious_artifacts),
         "--max-reference-artifacts",
@@ -343,6 +353,19 @@ def analyze_frame(frame: bytes, previous: bytes | None, args: argparse.Namespace
         means.append(mean)
         variances.append(variance)
 
+    global_mean = sum(means) / max(1, len(means))
+    global_variance = max(
+        0.0,
+        sum(variance + mean * mean for mean, variance in zip(means, variances, strict=True))
+        / max(1, len(means))
+        - global_mean * global_mean,
+    )
+    previous_global_variance = 0.0
+    if previous is not None:
+        _, previous_global_variance = band_stats(
+            previous, width, 0, height, args.x_step, args.y_step
+        )
+
     half = band_count // 2
     lower_flags = [var < args.flat_var for var in variances[half:]]
     lower_flat_pct = sum(lower_flags) / max(1, len(lower_flags))
@@ -373,6 +396,11 @@ def analyze_frame(frame: bytes, previous: bytes | None, args: argparse.Namespace
     )
     lower_flat_flash = lower_flat_pct >= 0.25 and temporal_lower_jump
     lower_slab = lower_flat_run_pct >= 0.33 and max_lower_jump > args.jump_threshold and temporal_lower_jump
+    global_flat_flash = (
+        previous is not None
+        and global_variance < args.flat_var
+        and previous_global_variance > max(args.flat_var * 4.0, 72.0)
+    )
     temporal_delta_spike = (
         previous is not None
         and max_temporal_delta > args.tear_threshold
@@ -398,6 +426,8 @@ def analyze_frame(frame: bytes, previous: bytes | None, args: argparse.Namespace
         reasons.append("lower_flat_flash")
     if lower_slab:
         reasons.append("lower_slab")
+    if global_flat_flash:
+        reasons.append("global_flat_flash")
     if temporal_delta_spike:
         reasons.append("temporal_delta_spike")
     if horizontal_shift:
@@ -410,6 +440,7 @@ def analyze_frame(frame: bytes, previous: bytes | None, args: argparse.Namespace
         or lower_boundary_jump
         or lower_flat_flash
         or lower_slab
+        or global_flat_flash
         or temporal_delta_spike
         or horizontal_shift
         or lower_blur_flash
@@ -432,9 +463,17 @@ def analyze_frame(frame: bytes, previous: bytes | None, args: argparse.Namespace
         "shift_improvement": round(shift_improvement, 3),
         "lower_variance_min": round(min(variances[half:] or [0.0]), 3),
         "lower_variance_mean": round(sum(variances[half:]) / max(1, len(variances[half:])), 3),
+        "global_mean": round(global_mean, 3),
+        "global_variance": round(global_variance, 3),
+        "previous_global_variance": round(previous_global_variance, 3),
     }
 
 
+def exact_repeat_is_hard(run_frames: int, fps: int, max_seconds: float) -> bool:
+    threshold = max(1, round(max(0.0, max_seconds) * max(1, fps)))
+    return run_frames >= threshold
+
+
 def write_pgm(path: pathlib.Path, frame: bytes, width: int, height: int) -> None:
     path.write_bytes(f"P5\n{width} {height}\n255\n".encode() + frame)
 
@@ -537,11 +576,20 @@ def run_capture(args: argparse.Namespace) -> int:
     max_upper_delta = 0.0
     max_lower_delta = 0.0
     max_lower_jump_seen = 0.0
+    exact_repeat_run = 0
+    max_exact_repeat_run = 0
     def analyze_captured_frame(frame: bytes, elapsed_s: float, metrics: Any) -> None:
         nonlocal previous, frame_index, suspicious_count, hard_corruption_count, artifacts_written, reference_artifacts_written
         nonlocal changed_frames, static_frames, max_upper_delta, max_lower_delta, max_lower_jump_seen, worst
+        nonlocal exact_repeat_run, max_exact_repeat_run
         frame_index += 1
+        exact_repeat_run = exact_repeat_run + 1 if previous is not None and frame == previous else 0
+        max_exact_repeat_run = max(max_exact_repeat_run, exact_repeat_run)
         result = analyze_frame(frame, previous, args)
+        result["exact_repeat_run"] = exact_repeat_run
+        if exact_repeat_is_hard(exact_repeat_run, args.fps, args.max_exact_repeat_seconds):
+            result["suspicious"] = True
+            result["reasons"].append("exact_freeze")
         previous = frame
         result.update({"frame": frame_index, "elapsed_s": round(elapsed_s, 3)})
         max_upper_delta = max(max_upper_delta, float(result["upper_delta"]))
@@ -778,6 +826,8 @@ def run_capture(args: argparse.Namespace) -> int:
         "changed_frames": changed_frames,
         "static_frames": static_frames,
         "static_pct": round((static_frames / max(1, frame_index - 1) * 100.0) if frame_index > 1 else 0.0, 3),
+        "max_exact_repeat_frames": max_exact_repeat_run,
+        "max_exact_repeat_seconds": round(max_exact_repeat_run / max(1, args.fps), 3),
         "max_upper_delta": round(max_upper_delta, 3),
         "max_lower_delta": round(max_lower_delta, 3),
         "max_lower_jump_seen": round(max_lower_jump_seen, 3),
@@ -811,6 +861,7 @@ def format_summary(summary: dict[str, Any]) -> str:
         f"hard corruption: {summary.get('hard_corruption_frames', 0)}",
         f"review only: {summary.get('review_only_frames', 0)}",
         f"static: {summary.get('static_frames', 0)} ({summary.get('static_pct', 0.0)}%)",
+        f"max exact repeat: {summary.get('max_exact_repeat_frames', 0)} frames ({summary.get('max_exact_repeat_seconds', 0.0)} s)",
         f"max deltas: upper={summary.get('max_upper_delta', 0.0)} lower={summary.get('max_lower_delta', 0.0)}",
         f"reasons: {summary['reason_counts']}",
         f"reference artifacts: {summary.get('reference_artifacts', 0)}",
@@ -865,6 +916,7 @@ def run_self_test(args: argparse.Namespace) -> int:
     rich = synthetic_rich_frame(args.width, args.height)
     frames.append(rich)
     frames.append(synthetic_shift_frame(rich, args.width, args.height, 24))
+    frames.append(bytes([96]) * (args.width * args.height))
     previous = None
     suspicious = 0
     records = []
@@ -889,6 +941,10 @@ def run_self_test(args: argparse.Namespace) -> int:
         "artifact_dir": str(artifact_dir),
     }
     (artifact_dir / "summary.json").write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n")
+    if "global_flat_flash" not in records[-1]["reasons"]:
+        raise AssertionError("full-frame flat collapse was not classified as hard corruption")
+    if not exact_repeat_is_hard(40, 20, 2.0) or exact_repeat_is_hard(39, 20, 2.0):
+        raise AssertionError("exact-repeat freeze threshold is not duration bounded")
     print(json.dumps(summary, indent=2, sort_keys=True))
     return 0 if suspicious >= 1 else 1
 

server/Cargo.toml

@@ -16,7 +16,7 @@ bench                   = false
 
 [package]
 name                    = "lesavka_server"
-version                 = "0.27.12"
+version                 = "0.27.13"
 edition                 = "2024"
 autobins                = false
 

tests/manual/server/rct/rct_uvc_artifact_probe_manual_contract.rs

@@ -27,6 +27,8 @@ fn rct_uvc_artifact_probe_documents_late_path_lower_half_detection() {
         "lower_boundary_jump",
         "lower_flat_flash",
         "lower_slab",
+        "global_flat_flash",
+        "exact_freeze",
         "temporal_delta_spike",
         "horizontal_shift",
         "lower_blur_flash",
@@ -37,6 +39,7 @@ fn rct_uvc_artifact_probe_documents_late_path_lower_half_detection() {
         "--shift-threshold",
         "--blur-delta-threshold",
         "--change-threshold",
+        "--max-exact-repeat-seconds",
         "--reference-every",
         "static_pct",
         "reference_",
@@ -102,15 +105,15 @@ fn rct_uvc_artifact_probe_self_test_flags_synthetic_lower_half_slab() {
         summary["schema"],
         "lesavka.rct-uvc-artifact-probe.self-test.v1"
     );
-    assert_eq!(summary["frames"], 7);
+    assert_eq!(summary["frames"], 8);
     assert!(
         summary["suspicious_frames"].as_u64().unwrap_or_default() >= 1,
         "self-test should detect the synthetic lower-half slab: {summary}"
     );
     assert_eq!(
         summary["hard_corruption_frames"].as_u64(),
-        Some(1),
+        Some(2),
-        "only the synthetic slab should be a hard corruption failure: {summary}"
+        "the synthetic slab and full-frame flat collapse should be hard failures: {summary}"
     );
     assert!(
         dir.path().join("reference_000001.pgm").exists(),
@@ -145,6 +148,14 @@ fn rct_uvc_artifact_probe_self_test_flags_synthetic_lower_half_slab() {
         }),
         "self-test should catch transient tear spikes: {summary}"
     );
+    assert!(
+        records.iter().any(|record| {
+            record["reasons"]
+                .as_array()
+                .is_some_and(|reasons| reasons.iter().any(|reason| reason == "global_flat_flash"))
+        }),
+        "self-test should catch a full-frame grey/black collapse: {summary}"
+    );
     assert!(
         dir.path().join("selftest_suspicious_000005.pgm").exists(),
         "probe should save visual evidence for suspicious frames"