test(uvc): make synthetic receiver proof blocking

Cargo.lock

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

client/Cargo.toml

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

common/Cargo.toml

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

docs/architecture-and-media-readiness.md

@@ -344,6 +344,12 @@ 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.
 
+Release 0.27.14 makes the marked synthetic receiver probe a blocking gate
+instead of a diagnostic that could pass after decoding one frame. It now
+requires at least 90% duration coverage, at least 99% marker coverage, zero
+marked cadence or visual corruption, a bounded successful deep decode, intact
+native MJPEG, and clean mode-matched server-boundary evidence.
+
 ## 7. The Resolved Downstream Video Failure
 
 The current blank downstream feeds fail before transport or decoding.
@@ -420,7 +426,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.13`, the pushed release revision,
+7. Confirm Theia reports server version `0.27.14`, 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.
@@ -438,7 +444,7 @@ hardware contract is repeatable. The remaining work falls into five groups.
 
 ### A. Install And Version Parity
 
-- Push and deploy `0.27.13` through the client/server install scripts.
+- Push and deploy `0.27.14` 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.
 
@@ -504,6 +510,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.13` client/server pair, the
+Until that sequence passes on the installed `0.27.14` 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_synthetic_rct_uvc_probe.py

@@ -709,6 +709,37 @@ def summarize_server_uvc_audit(
     return summary
 
 
+def synthetic_capture_acceptance(
+    *,
+    frames: int,
+    expected_frames: int,
+    decoded_frames: int,
+    suspicious_frames: int,
+    ffmpeg_rc: int | None,
+    capture_timed_out: bool,
+    stream_analyze: bool,
+    decode_rc: int | None,
+    decode_timed_out: bool,
+) -> tuple[str, list[str]]:
+    reasons: list[str] = []
+    if capture_timed_out:
+        reasons.append("capture_timeout")
+    capture_rc_ok = ffmpeg_rc == 0 or (stream_analyze and ffmpeg_rc in {-15, -9, 137, 143})
+    if not capture_rc_ok:
+        reasons.append("capture_process_failed")
+    if decode_timed_out:
+        reasons.append("decode_timeout")
+    if decode_rc not in {None, 0}:
+        reasons.append("decode_process_failed")
+    if frames < max(1, int(expected_frames * 0.9)):
+        reasons.append("duration_coverage_below_90pct")
+    if frames <= 0 or decoded_frames < frames * 0.99:
+        reasons.append("marker_coverage_below_99pct")
+    if suspicious_frames:
+        reasons.append("marked_frame_corruption_or_cadence_failure")
+    return ("pass" if not reasons else "fail"), reasons
+
+
 def run_remote_orchestrated(args: argparse.Namespace) -> int:
     if (not args.inject_host and not args.local_inject) or not args.rct_host:
         raise SystemExit(
@@ -937,22 +968,37 @@ def run_remote_orchestrated(args: argparse.Namespace) -> int:
     local_capture = artifact_dir / "capture"
     local_inject = artifact_dir / "inject"
     local_server_audit = artifact_dir / "server-uvc-audit"
+    integrity_report: dict[str, Any] | None = None
+    integrity_rc: int | None = None
     if capture is not None:
         subprocess.run(["scp", "-r", f"{args.rct_host}:{remote_rct_dir}", str(local_capture)], check=False)
         if args.deep_capture:
             integrity_script = pathlib.Path(__file__).with_name("analyze_marked_capture.py")
             native_capture = local_capture / "capture.mjpg"
             if integrity_script.exists() and native_capture.exists():
-                subprocess.run(
+                try:
-                    [
+                    integrity_process = subprocess.run(
-                        sys.executable,
+                        [
-                        str(integrity_script),
+                            sys.executable,
-                        str(native_capture),
+                            str(integrity_script),
-                        "--output-dir",
+                            str(native_capture),
-                        str(local_capture / "mjpeg-integrity"),
+                            "--output-dir",
-                    ],
+                            str(local_capture / "mjpeg-integrity"),
-                    check=False,
+                        ],
-                )
+                        check=False,
+                        timeout=max(30.0, args.duration + 120.0),
+                    )
+                    integrity_rc = integrity_process.returncode
+                except subprocess.TimeoutExpired:
+                    integrity_rc = 124
+                integrity_report_path = local_capture / "mjpeg-integrity" / "report.json"
+                if integrity_report_path.exists():
+                    try:
+                        integrity_report = json.loads(integrity_report_path.read_text())
+                    except (OSError, json.JSONDecodeError):
+                        integrity_report = {"verdict": "fail", "error": "invalid integrity report"}
+                else:
+                    integrity_report = {"verdict": "fail", "error": "missing integrity report"}
     if args.local_inject:
         if pathlib.Path(remote_inject_dir).exists():
             if local_inject.exists():
@@ -1029,22 +1075,41 @@ def run_remote_orchestrated(args: argparse.Namespace) -> int:
     if server_boundary_summary:
         for item in server_boundary_summary.get("diagnosis") or []:
             diagnosis.append(str(item))
+    acceptance_failures: list[str] = []
+    if capture_rc != 0:
+        acceptance_failures.append("receiver_capture_gate_failed")
+    if inject_rc != 0:
+        acceptance_failures.append("synthetic_injector_failed")
+    if args.deep_capture and (
+        integrity_rc != 0 or not integrity_report or integrity_report.get("verdict") != "pass"
+    ):
+        acceptance_failures.append("native_mjpeg_integrity_failed")
+    if args.server_uvc_audit and (
+        not server_boundary_summary
+        or server_boundary_summary.get("status") != "no_visual_corruption_observed"
+    ):
+        acceptance_failures.append("server_uvc_boundary_gate_failed")
+    verdict = "pass" if not acceptance_failures else "fail"
     summary = {
         "schema": "lesavka.synthetic-rct-probe.orchestrator.v1",
         "mode": args.mode,
         "capture_rc": capture_rc,
         "inject_rc": inject_rc,
         "diagnosis": diagnosis,
+        "verdict": verdict,
+        "acceptance_failures": acceptance_failures,
         "artifact_dir": str(artifact_dir),
         "capture_artifacts": str(local_capture),
         "inject_artifacts": str(local_inject),
         "server_uvc_boundary": server_boundary_summary,
         "server_uvc_audit_artifacts": str(local_server_audit) if copied_server_audit else None,
+        "native_integrity": integrity_report,
+        "native_integrity_rc": integrity_rc,
     }
     (artifact_dir / "run-summary.json").write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n")
     print(json.dumps(summary, indent=2, sort_keys=True))
     print(f"artifact_dir: {artifact_dir}")
-    return 0 if capture_rc == 0 and inject_rc == 0 else 1
+    return 0 if verdict == "pass" else 1
 
 
 def detect_video_device(label: str) -> str:
@@ -1689,6 +1754,9 @@ def run_capture(args: argparse.Namespace) -> int:
     raw_capture_bytes = 0
     ffmpeg_rc: int | None = None
     capture_timed_out = False
+    decode_timed_out = False
+    decode_rc: int | None = None
+    native_capture_bytes = 0
     frame_index = 0
     suspicious_count = 0
     visual_suspicious_count = 0
@@ -1829,12 +1897,25 @@ def run_capture(args: argparse.Namespace) -> int:
                 err.flush()
             capture_elapsed = time.monotonic() - capture_started
             if args.deep_capture and native_mjpeg is not None and native_mjpeg.exists():
+                native_capture_bytes = native_mjpeg.stat().st_size
                 decode_command = [
                     "ffmpeg", "-hide_banner", "-nostdin", "-loglevel", "warning",
                     "-i", str(native_mjpeg), "-an", "-pix_fmt", "gray", "-f", "rawvideo", str(raw_path),
                 ]
                 (artifact_dir / "decode-command.txt").write_text(" ".join(shlex.quote(part) for part in decode_command) + "\n")
-                subprocess.run(decode_command, stdout=subprocess.DEVNULL, stderr=err, check=False)
+                try:
+                    decode_process = subprocess.run(
+                        decode_command,
+                        stdout=subprocess.DEVNULL,
+                        stderr=err,
+                        check=False,
+                        timeout=max(30.0, args.duration + 120.0),
+                    )
+                    decode_rc = decode_process.returncode
+                except subprocess.TimeoutExpired:
+                    decode_timed_out = True
+                    decode_rc = 124
+                    err.write(b"native MJPEG decode exceeded wall timeout\n")
             raw_path.touch(exist_ok=True)
             raw_capture_bytes = raw_path.stat().st_size if raw_path.exists() else 0
             print(
@@ -1859,6 +1940,18 @@ def run_capture(args: argparse.Namespace) -> int:
             if dynamic_debug_enabled:
                 set_uvc_dynamic_debug(artifact_dir / "deep-capture", False)
     elapsed = max(0.001, capture_elapsed)
+    expected_frames = max(1, int(args.duration * fps))
+    verdict, acceptance_failures = synthetic_capture_acceptance(
+        frames=frame_index,
+        expected_frames=expected_frames,
+        decoded_frames=decoded_frames,
+        suspicious_frames=suspicious_count,
+        ffmpeg_rc=ffmpeg_rc,
+        capture_timed_out=capture_timed_out,
+        stream_analyze=args.stream_analyze,
+        decode_rc=decode_rc,
+        decode_timed_out=decode_timed_out,
+    )
     summary = {
         "schema": "lesavka.synthetic-rct-capture.v1",
         "source": args.source,
@@ -1873,8 +1966,13 @@ def run_capture(args: argparse.Namespace) -> int:
         "analysis_duration_s": round(analysis_elapsed, 3),
         "ffmpeg_rc": ffmpeg_rc,
         "capture_timed_out": capture_timed_out,
+        "decode_timed_out": decode_timed_out,
+        "decode_rc": decode_rc,
         "raw_capture_bytes": raw_capture_bytes,
+        "native_capture_bytes": native_capture_bytes,
         "frames": frame_index,
+        "expected_frames": expected_frames,
+        "coverage_ok": frame_index >= int(expected_frames * 0.9),
         "fps_observed": round(frame_index / elapsed, 3),
         "decoded_frames": decoded_frames,
         "decoded_pct": round(decoded_frames / frame_index * 100.0, 3) if frame_index else 0.0,
@@ -1897,12 +1995,14 @@ def run_capture(args: argparse.Namespace) -> int:
         "suspicious_artifacts": suspicious_artifacts,
         "artifact_dir": str(artifact_dir),
         "ffmpeg_stderr": str(stderr_path),
+        "verdict": verdict,
+        "acceptance_failures": acceptance_failures,
     }
     (artifact_dir / "summary.json").write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n")
     (artifact_dir / "summary.txt").write_text(format_summary(summary))
     print(format_summary(summary), end="")
     print(f"artifact_dir: {artifact_dir}")
-    return 0 if frame_index > 0 else 2
+    return 0 if verdict == "pass" else 2
 
 
 def format_summary(summary: dict[str, Any]) -> str:
@@ -1913,6 +2013,8 @@ def format_summary(summary: dict[str, Any]) -> str:
             f"device: {summary['device']}",
             f"mode: {summary['mode']} capture={summary['width']}x{summary['height']}@{summary['fps_requested']}",
             f"frames: {summary['frames']} ({summary['fps_observed']} fps observed)",
+            f"verdict: {summary.get('verdict', 'unknown')}",
+            f"coverage: {summary.get('frames', 0)}/{summary.get('expected_frames', 0)} ok={summary.get('coverage_ok', False)}",
             f"decoded markers: {summary['decoded_frames']} ({summary['decoded_pct']}%)",
             f"suspicious: {summary['suspicious_frames']} ({summary['suspicious_pct']}%)",
             f"visual suspicious: {summary['visual_suspicious_frames']} ({summary['visual_suspicious_pct']}%)",
@@ -1969,6 +2071,32 @@ 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")
+    passing, passing_reasons = synthetic_capture_acceptance(
+        frames=600,
+        expected_frames=600,
+        decoded_frames=600,
+        suspicious_frames=0,
+        ffmpeg_rc=0,
+        capture_timed_out=False,
+        stream_analyze=False,
+        decode_rc=0,
+        decode_timed_out=False,
+    )
+    failing, failing_reasons = synthetic_capture_acceptance(
+        frames=599,
+        expected_frames=600,
+        decoded_frames=599,
+        suspicious_frames=1,
+        ffmpeg_rc=0,
+        capture_timed_out=False,
+        stream_analyze=False,
+        decode_rc=0,
+        decode_timed_out=False,
+    )
+    if passing != "pass" or passing_reasons:
+        raise AssertionError(f"clean full capture did not pass: {passing_reasons}")
+    if failing != "fail" or "marked_frame_corruption_or_cadence_failure" not in failing_reasons:
+        raise AssertionError(f"corrupt capture did not fail: {failing_reasons}")
     print(json.dumps(summary, indent=2, sort_keys=True))
     return 0 if suspicious >= 3 else 1
 

server/Cargo.toml

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

tests/manual/server/rct/synthetic_rct_uvc_probe_manual_contract.rs

@@ -84,6 +84,18 @@ fn synthetic_probe_keeps_bundled_network_ingress_and_rct_comparison_markers() {
         "suspicious_",
         "decoded_pct",
         "capture_mode",
+        "synthetic_capture_acceptance",
+        "duration_coverage_below_90pct",
+        "marker_coverage_below_99pct",
+        "marked_frame_corruption_or_cadence_failure",
+        "decode_timed_out",
+        "decode_rc",
+        "native_integrity",
+        "integrity_rc = 124",
+        "native_mjpeg_integrity_failed",
+        "server_uvc_boundary_gate_failed",
+        "acceptance_failures",
+        "verdict",
         "raw_capture_bytes",
         "analysis_duration_s",
         "visual_suspicious",
@@ -174,6 +186,9 @@ fn synthetic_probe_capture_has_a_wall_clock_timeout() {
         "ffmpeg_rc = 124",
         "raw_path.touch(exist_ok=True)",
         "\"capture_timed_out\": capture_timed_out",
+        "timeout=max(30.0, args.duration + 120.0)",
+        "decode_timed_out = True",
+        "\"decode_timed_out\": decode_timed_out",
     ] {
         assert!(
             PROBE_SRC.contains(expected),
@@ -182,6 +197,23 @@ fn synthetic_probe_capture_has_a_wall_clock_timeout() {
     }
 }
 
+#[test]
+fn synthetic_probe_cannot_pass_from_one_decoded_or_corrupt_frame() {
+    for expected in [
+        "expected_frames = max(1, int(args.duration * fps))",
+        "frames * 0.99",
+        "suspicious_frames",
+        "return 0 if verdict == \"pass\" else 2",
+        "integrity_report.get(\"verdict\") != \"pass\"",
+        "server_boundary_summary.get(\"status\") != \"no_visual_corruption_observed\"",
+    ] {
+        assert!(
+            PROBE_SRC.contains(expected),
+            "synthetic receiver gate must reject incomplete, corrupt, or structurally invalid evidence: {expected}"
+        );
+    }
+}
+
 #[test]
 fn synthetic_injector_enters_the_public_bundled_media_rpc() {
     for expected in [