From ed9faecb20af38001d8c17e17334ba1212481069 Mon Sep 17 00:00:00 2001 From: Brad Stein Date: Thu, 13 Aug 2026 04:42:09 -0300 Subject: [PATCH] test(uvc): make synthetic receiver proof blocking --- Cargo.lock | 6 +- client/Cargo.toml | 2 +- common/Cargo.toml | 2 +- docs/architecture-and-media-readiness.md | 12 +- scripts/manual/run_synthetic_rct_uvc_probe.py | 154 ++++++++++++++++-- server/Cargo.toml | 2 +- ...synthetic_rct_uvc_probe_manual_contract.rs | 32 ++++ 7 files changed, 188 insertions(+), 22 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index fe7cbb6..a5375fc 100644 --- a/Cargo.lock +++ b/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", diff --git a/client/Cargo.toml b/client/Cargo.toml index c0e548c..a097314 100644 --- a/client/Cargo.toml +++ b/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] diff --git a/common/Cargo.toml b/common/Cargo.toml index 33854df..8f0f74a 100644 --- a/common/Cargo.toml +++ b/common/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "lesavka_common" -version = "0.27.13" +version = "0.27.14" edition = "2024" build = "build.rs" diff --git a/docs/architecture-and-media-readiness.md b/docs/architecture-and-media-readiness.md index 28078e3..e1cd330 100644 --- a/docs/architecture-and-media-readiness.md +++ b/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. diff --git a/scripts/manual/run_synthetic_rct_uvc_probe.py b/scripts/manual/run_synthetic_rct_uvc_probe.py index 7221f12..ba6f1bb 100755 --- a/scripts/manual/run_synthetic_rct_uvc_probe.py +++ b/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( - [ - sys.executable, - str(integrity_script), - str(native_capture), - "--output-dir", - str(local_capture / "mjpeg-integrity"), - ], - check=False, - ) + try: + integrity_process = subprocess.run( + [ + sys.executable, + str(integrity_script), + str(native_capture), + "--output-dir", + str(local_capture / "mjpeg-integrity"), + ], + 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 diff --git a/server/Cargo.toml b/server/Cargo.toml index 2de02f5..5188bec 100644 --- a/server/Cargo.toml +++ b/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 diff --git a/tests/manual/server/rct/synthetic_rct_uvc_probe_manual_contract.rs b/tests/manual/server/rct/synthetic_rct_uvc_probe_manual_contract.rs index 671a01d..0904582 100644 --- a/tests/manual/server/rct/synthetic_rct_uvc_probe_manual_contract.rs +++ b/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 [