test(uvc): gate sustained receiver artifacts

This commit is contained in:
Brad Stein 2026-08-13 04:04:47 -03:00
parent 9b1bf46cb9
commit 99077a4527
9 changed files with 308 additions and 23 deletions

6
Cargo.lock generated
View File

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

View File

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

View File

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

View File

@ -331,6 +331,13 @@ webcam artifact detector. A malformed stream now yields a bounded report with
`capture_timed_out`, FFmpeg diagnostics, and extracted reference/suspicious
frames instead of blocking the soak indefinitely.
Release 0.27.12 turns that long run into a real acceptance gate. Deep receiver
captures retain native MJPEG on Tethys, validate every JPEG with a bounded-memory
streaming analyzer, and then decode it for visual scoring. High-confidence grey
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.
## 7. The Resolved Downstream Video Failure
The current blank downstream feeds fail before transport or decoding.
@ -407,7 +414,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.11`, the pushed release revision,
7. Confirm Theia reports server version `0.27.12`, 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.
@ -425,7 +432,7 @@ hardware contract is repeatable. The remaining work falls into five groups.
### A. Install And Version Parity
- Push and deploy `0.27.11` through the client/server install scripts.
- Push and deploy `0.27.12` 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.
@ -491,6 +498,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.11` client/server pair, the
Until that sequence passes on the installed `0.27.12` 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.

View File

@ -13,12 +13,15 @@ import time
import warnings
import zlib
from collections import Counter
from typing import Any
from collections.abc import Iterator
from typing import Any, BinaryIO
APP4_MAGIC = b"LSVK"
BAND_ROWS = 64
BAND_BITS = 32
BAND_CELL = 4
STREAM_READ_BYTES = 1024 * 1024
MAX_UNFRAMED_BYTES = 8 * 1024 * 1024
def parse_args() -> argparse.Namespace:
@ -27,6 +30,11 @@ def parse_args() -> argparse.Namespace:
parser.add_argument("--output-dir", default="", help="write report.json, report.txt, and frames.jsonl")
parser.add_argument("--idle-mean-max", type=float, default=8.0)
parser.add_argument("--idle-stddev-max", type=float, default=4.0)
parser.add_argument(
"--structural-only",
action="store_true",
help="validate JPEG/APP4 structure without decoding pixels; suitable for multi-gigabyte real-camera soaks",
)
parser.add_argument("--self-test", action="store_true")
return parser.parse_args()
@ -65,6 +73,53 @@ def split_mjpeg(data: bytes) -> list[bytes]:
return frames
def iter_mjpeg_stream(stream: BinaryIO) -> Iterator[bytes]:
"""Yield MJPEG frames while bounding memory for sustained captures."""
buffer = bytearray()
eof = False
while not eof:
chunk = stream.read(STREAM_READ_BYTES)
if chunk:
buffer.extend(chunk)
else:
eof = True
while buffer:
start = buffer.find(b"\xff\xd8")
if start < 0:
if eof or len(buffer) >= MAX_UNFRAMED_BYTES:
yield bytes(buffer)
buffer.clear()
elif len(buffer) > 1:
del buffer[:-1]
break
if start > 0:
yield bytes(buffer[:start])
del buffer[:start]
continue
end = buffer.find(b"\xff\xd9", 2)
next_start = buffer.find(b"\xff\xd8", 2)
if next_start >= 0 and (end < 0 or next_start < end):
yield bytes(buffer[:next_start])
del buffer[:next_start]
continue
if end >= 0:
frame_end = end + 2
yield bytes(buffer[:frame_end])
del buffer[:frame_end]
continue
if eof or len(buffer) >= MAX_UNFRAMED_BYTES:
yield bytes(buffer)
buffer.clear()
break
def iter_mjpeg_file(path: pathlib.Path) -> Iterator[bytes]:
with path.open("rb") as stream:
yield from iter_mjpeg_stream(stream)
def inspect_jpeg(frame: bytes) -> dict[str, Any]:
if len(frame) < 4 or not frame.startswith(b"\xff\xd8"):
return {"valid": False, "reason": "missing_soi", "app4": None}
@ -294,6 +349,39 @@ def classify_frame(frame: bytes, previous_sequence: int | None, args: argparse.N
}
def classify_structural_frame(frame: bytes, previous_sequence: int | None) -> dict[str, Any]:
inspection = inspect_jpeg(frame)
app4 = inspection.get("app4")
sequence = app4.get("sequence") if isinstance(app4, dict) else None
if not inspection["valid"]:
classification = "truncated"
elif isinstance(app4, dict) and not app4.get("crc_ok", False):
classification = "spliced"
elif sequence is not None and previous_sequence is not None and sequence <= previous_sequence:
classification = "stale_repeat"
else:
classification = "intact"
return {
"classification": classification,
"verdict": classification,
"bytes": len(frame),
"sha256": hashlib.sha256(frame).hexdigest(),
"jpeg_valid": inspection["valid"],
"jpeg_reason": inspection["reason"],
"sequence": sequence,
"app4_crc_ok": app4.get("crc_ok") if isinstance(app4, dict) else None,
"width": None,
"height": None,
"band_markers": [],
"band_sequences": [],
"first_bad_row": None,
"seam_row": None,
"decode_warnings": [],
"luma_mean": None,
"luma_stddev": None,
}
def add_app4(frame: bytes, sequence: int) -> bytes:
payload = APP4_MAGIC + bytes([1]) + sequence.to_bytes(8, "big") + (zlib.crc32(frame) & 0xFFFFFFFF).to_bytes(4, "big")
return frame[:2] + b"\xff\xe4" + (len(payload) + 2).to_bytes(2, "big") + payload + frame[2:]
@ -346,6 +434,14 @@ def self_test(args: argparse.Namespace) -> int:
actual = classify_frame(frame, previous, args)["classification"]
if actual != expected:
raise AssertionError(f"expected {expected}, got {actual}")
streamed = list(iter_mjpeg_stream(io.BytesIO(intact + intact[:-9] + visibly_spliced_frame)))
if len(streamed) != 3:
raise AssertionError(f"streaming MJPEG parser expected 3 frames, got {len(streamed)}")
streamed_classes = [
classify_structural_frame(frame, None)["classification"] for frame in streamed
]
if streamed_classes != ["intact", "truncated", "intact"]:
raise AssertionError(f"streaming structural classes were {streamed_classes}")
localized = classify_frame(visibly_spliced_frame, 6, args)
if localized["seam_row"] != 128 or localized["band_sequences"] != [7, 8]:
raise AssertionError(f"splice localization failed: {localized}")
@ -363,8 +459,12 @@ def main() -> int:
records: list[dict[str, Any]] = []
previous_sequence: int | None = None
for path in files:
for index, frame in enumerate(split_mjpeg(path.read_bytes())):
record = classify_frame(frame, previous_sequence, args)
for index, frame in enumerate(iter_mjpeg_file(path)):
record = (
classify_structural_frame(frame, previous_sequence)
if args.structural_only
else classify_frame(frame, previous_sequence, args)
)
record.update({"source": str(path), "source_frame_index": index})
records.append(record)
if record["sequence"] is not None:
@ -373,6 +473,7 @@ def main() -> int:
report = {
"schema": "lesavka.uvc-mjpeg-integrity.v1",
"generated_unix_ms": int(time.time() * 1000),
"analysis_mode": "structural-only" if args.structural_only else "structural-and-visual",
"frames": len(records),
"classification_counts": dict(sorted(counts.items())),
"intact": counts["intact"],

View File

@ -16,6 +16,7 @@ import time
from typing import Any
DEFAULT_DEVICE_LABEL = "Lesavka Composite"
HARD_CORRUPTION_REASONS = frozenset({"lower_boundary_jump", "lower_flat_flash", "lower_slab"})
def parse_args() -> argparse.Namespace:
@ -59,6 +60,17 @@ def parse_args() -> argparse.Namespace:
action="store_true",
help="debug path: analyze ffmpeg stdout directly instead of spooling raw frames first",
)
parser.add_argument(
"--deep-capture",
action="store_true",
help="capture native MJPEG, validate every JPEG structurally, then decode for visual analysis",
)
parser.add_argument(
"--copy-native-capture",
action="store_true",
help="copy the potentially multi-gigabyte native MJPEG back from --host; default leaves it on the receiver",
)
parser.add_argument("--integrity-analyzer", default="", help=argparse.SUPPRESS)
parser.add_argument("--self-test", action="store_true")
return parser.parse_args()
@ -76,12 +88,21 @@ def run_remote(args: argparse.Namespace) -> int:
remote_artifact_dir = args.remote_artifact_dir or f"/tmp/lesavka-rct-uvc-artifact-probe-{timestamp()}"
remote_script = f"/tmp/lesavka-rct-uvc-artifact-probe-{os.getpid()}.py"
script_text = pathlib.Path(__file__).read_text()
analyzer_source = pathlib.Path(__file__).with_name("analyze_uvc_mjpeg_integrity.py")
remote_analyzer = f"/tmp/lesavka-uvc-mjpeg-integrity-{os.getpid()}.py"
subprocess.run(
["ssh", args.host, f"cat > {shlex.quote(remote_script)} && chmod +x {shlex.quote(remote_script)}"],
input=script_text,
text=True,
check=True,
)
if args.deep_capture:
subprocess.run(
["ssh", args.host, f"cat > {shlex.quote(remote_analyzer)} && chmod +x {shlex.quote(remote_analyzer)}"],
input=analyzer_source.read_text(),
text=True,
check=True,
)
remote_cmd = [
"python3",
remote_script,
@ -140,13 +161,36 @@ def run_remote(args: argparse.Namespace) -> int:
]
if args.stream_analyze:
remote_cmd.append("--stream-analyze")
if args.deep_capture:
remote_cmd.extend(["--deep-capture", "--integrity-analyzer", remote_analyzer])
print(f"running remote RCT UVC probe on {args.host}: {remote_artifact_dir}", file=sys.stderr)
rc = subprocess.run(["ssh", args.host, " ".join(shlex.quote(part) for part in remote_cmd)]).returncode
local_artifact_dir.parent.mkdir(parents=True, exist_ok=True)
subprocess.run(
["scp", "-r", f"{args.host}:{remote_artifact_dir}", str(local_artifact_dir)],
check=False,
)
local_artifact_dir.mkdir(parents=True, exist_ok=True)
if args.deep_capture and not args.copy_native_capture:
remote_tar = subprocess.Popen(
[
"ssh",
args.host,
f"tar -C {shlex.quote(remote_artifact_dir)} --exclude=./capture.mjpg -cf - .",
],
stdout=subprocess.PIPE,
)
assert remote_tar.stdout is not None
extract = subprocess.run(["tar", "-C", str(local_artifact_dir), "-xf", "-"], stdin=remote_tar.stdout)
remote_tar.stdout.close()
tar_rc = remote_tar.wait()
if extract.returncode != 0 or tar_rc != 0:
print("failed to copy remote artifact reports without native MJPEG", file=sys.stderr)
rc = rc or extract.returncode or tar_rc
else:
subprocess.run(
["scp", "-r", f"{args.host}:{remote_artifact_dir}/.", str(local_artifact_dir)],
check=False,
)
if args.deep_capture:
(local_artifact_dir / "remote-native-capture.txt").write_text(
f"host={args.host}\npath={remote_artifact_dir}/capture.mjpg\ncopied={str(args.copy_native_capture).lower()}\n"
)
print(f"artifact_dir: {local_artifact_dir}")
return rc
@ -477,9 +521,13 @@ def run_capture(args: argparse.Namespace) -> int:
raw_capture_bytes = 0
ffmpeg_rc: int | None = None
capture_timed_out = False
native_capture_bytes = 0
native_capture_path: str | None = None
integrity_report: dict[str, Any] | None = None
previous: bytes | None = None
frame_index = 0
suspicious_count = 0
hard_corruption_count = 0
artifacts_written = 0
reference_artifacts_written = 0
changed_frames = 0
@ -490,7 +538,7 @@ def run_capture(args: argparse.Namespace) -> int:
max_lower_delta = 0.0
max_lower_jump_seen = 0.0
def analyze_captured_frame(frame: bytes, elapsed_s: float, metrics: Any) -> None:
nonlocal previous, frame_index, suspicious_count, artifacts_written, reference_artifacts_written
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
frame_index += 1
result = analyze_frame(frame, previous, args)
@ -506,6 +554,8 @@ def run_capture(args: argparse.Namespace) -> int:
static_frames += 1
if result["suspicious"]:
suspicious_count += 1
if HARD_CORRUPTION_REASONS.intersection(result["reasons"]):
hard_corruption_count += 1
reason_counts.update(result["reasons"])
worst.append(result)
worst = sorted(
@ -562,13 +612,28 @@ def run_capture(args: argparse.Namespace) -> int:
else:
raw_path = artifact_dir / "capture.raw"
capture_command = command[:]
if "-an" in capture_command:
native_path = artifact_dir / "capture.mjpg" if args.deep_capture and args.source == "device" else None
if native_path is not None:
input_end = capture_command.index("-an")
capture_command = capture_command[:input_end] + [
"-t",
str(args.duration),
"-an",
"-c:v",
"copy",
"-f",
"mjpeg",
str(native_path),
]
elif "-an" in capture_command:
capture_command[capture_command.index("-an") : capture_command.index("-an")] = ["-t", str(args.duration)]
else:
capture_command[-1:-1] = ["-t", str(args.duration)]
capture_command[-1] = str(raw_path)
if native_path is None:
capture_command[-1] = str(raw_path)
(artifact_dir / "command.txt").write_text(" ".join(shlex.quote(part) for part in capture_command) + "\n")
print(f"capturing raw RCT frames before analysis: {raw_path}", file=sys.stderr)
capture_target = native_path if native_path is not None else raw_path
print(f"capturing RCT frames before analysis: {capture_target}", file=sys.stderr)
started = time.monotonic()
try:
proc = subprocess.run(
@ -587,6 +652,76 @@ def run_capture(args: argparse.Namespace) -> int:
)
err.flush()
capture_elapsed = time.monotonic() - started
if native_path is not None:
native_path.touch(exist_ok=True)
native_capture_bytes = native_path.stat().st_size
native_capture_path = str(native_path)
analyzer = pathlib.Path(args.integrity_analyzer) if args.integrity_analyzer else pathlib.Path(__file__).with_name("analyze_uvc_mjpeg_integrity.py")
integrity_dir = artifact_dir / "mjpeg-integrity"
integrity_command = [
sys.executable,
str(analyzer),
str(native_path),
"--structural-only",
"--output-dir",
str(integrity_dir),
]
(artifact_dir / "integrity-command.txt").write_text(
" ".join(shlex.quote(part) for part in integrity_command) + "\n"
)
with (artifact_dir / "integrity.stdout").open("wb") as integrity_stdout, (artifact_dir / "integrity.stderr").open("wb") as integrity_stderr:
try:
integrity_proc = subprocess.run(
integrity_command,
stdout=integrity_stdout,
stderr=integrity_stderr,
check=False,
timeout=max(30.0, args.duration + 120.0),
)
integrity_rc = integrity_proc.returncode
except subprocess.TimeoutExpired:
integrity_rc = 124
integrity_report_path = integrity_dir / "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",
"frames": 0,
"analyzer_rc": integrity_rc,
"error": "invalid analyzer report",
}
else:
integrity_report = {"verdict": "fail", "frames": 0, "analyzer_rc": integrity_rc}
decode_command = [
"ffmpeg",
"-hide_banner",
"-nostdin",
"-loglevel",
"warning",
"-i",
str(native_path),
"-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"
)
try:
subprocess.run(
decode_command,
stdout=subprocess.DEVNULL,
stderr=err,
check=False,
timeout=max(30.0, args.duration + 120.0),
)
except subprocess.TimeoutExpired:
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(
@ -605,11 +740,21 @@ def run_capture(args: argparse.Namespace) -> int:
raw_path.unlink(missing_ok=True)
analysis_elapsed = time.monotonic() - analysis_started
elapsed = max(0.001, capture_elapsed)
expected_frames = max(1, int(args.duration * args.fps))
coverage_ok = frame_index >= int(expected_frames * 0.9)
structural_ok = integrity_report is None or integrity_report.get("verdict") == "pass"
capture_rc_ok = ffmpeg_rc == 0 or (args.stream_analyze and ffmpeg_rc in {-15, -9, 137, 143})
if capture_timed_out or not capture_rc_ok or not coverage_ok or not structural_ok or hard_corruption_count:
verdict = "fail"
elif suspicious_count:
verdict = "review_required"
else:
verdict = "pass"
summary = {
"schema": "lesavka.rct-uvc-artifact-probe.v1",
"source": args.source,
"device": device,
"capture_mode": "stream" if args.stream_analyze else "rawfile",
"capture_mode": "deep-mjpeg" if args.deep_capture and args.source == "device" else ("stream" if args.stream_analyze else "rawfile"),
"width": args.width,
"height": args.height,
"fps_requested": args.fps,
@ -619,9 +764,16 @@ def run_capture(args: argparse.Namespace) -> int:
"ffmpeg_rc": ffmpeg_rc,
"capture_timed_out": capture_timed_out,
"raw_capture_bytes": raw_capture_bytes,
"native_capture_bytes": native_capture_bytes,
"native_capture_path": native_capture_path,
"native_integrity": integrity_report,
"frames": frame_index,
"expected_frames": expected_frames,
"coverage_ok": coverage_ok,
"fps_observed": round(frame_index / elapsed, 3),
"suspicious_frames": suspicious_count,
"hard_corruption_frames": hard_corruption_count,
"review_only_frames": suspicious_count - hard_corruption_count,
"suspicious_pct": round((suspicious_count / frame_index * 100.0) if frame_index else 0.0, 3),
"changed_frames": changed_frames,
"static_frames": static_frames,
@ -635,12 +787,13 @@ def run_capture(args: argparse.Namespace) -> int:
"suspicious_artifacts": artifacts_written,
"artifact_dir": str(artifact_dir),
"ffmpeg_stderr": str(stderr_path),
"verdict": verdict,
}
(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 (3 if verdict == "review_required" else 2)
def format_summary(summary: dict[str, Any]) -> str:
@ -651,7 +804,12 @@ def format_summary(summary: dict[str, Any]) -> str:
f"mode: {summary['width']}x{summary['height']}@{summary['fps_requested']}",
f"frames: {summary['frames']} ({summary['fps_observed']} fps observed)",
f"capture timed out: {summary.get('capture_timed_out', False)}",
f"verdict: {summary.get('verdict', 'unknown')}",
f"coverage: {summary.get('frames', 0)}/{summary.get('expected_frames', 0)} ok={summary.get('coverage_ok', False)}",
f"native integrity: {(summary.get('native_integrity') or {}).get('verdict', 'not-run')}",
f"suspicious: {summary['suspicious_frames']} ({summary['suspicious_pct']}%)",
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 deltas: upper={summary.get('max_upper_delta', 0.0)} lower={summary.get('max_lower_delta', 0.0)}",
f"reasons: {summary['reason_counts']}",
@ -724,6 +882,9 @@ def run_self_test(args: argparse.Namespace) -> int:
"schema": "lesavka.rct-uvc-artifact-probe.self-test.v1",
"frames": len(frames),
"suspicious_frames": suspicious,
"hard_corruption_frames": sum(
1 for record in records if HARD_CORRUPTION_REASONS.intersection(record["reasons"])
),
"records": records,
"artifact_dir": str(artifact_dir),
}

View File

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

View File

@ -44,6 +44,13 @@ fn rct_uvc_artifact_probe_documents_late_path_lower_half_detection() {
"--crop",
"PGM",
"--host",
"--deep-capture",
"--copy-native-capture",
"HARD_CORRUPTION_REASONS",
"hard_corruption_frames",
"review_required",
"native_integrity",
"--structural-only",
] {
assert!(
PROBE_SRC.contains(expected),
@ -100,6 +107,11 @@ fn rct_uvc_artifact_probe_self_test_flags_synthetic_lower_half_slab() {
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),
"only the synthetic slab should be a hard corruption failure: {summary}"
);
assert!(
dir.path().join("reference_000001.pgm").exists(),
"probe should save a reference frame so no-artifact runs prove the crop"

View File

@ -22,6 +22,10 @@ fn uvc_integrity_analyzer_has_machine_readable_failure_classes() {
"crc_ok",
"report.json",
"frames.jsonl",
"iter_mjpeg_stream",
"MAX_UNFRAMED_BYTES",
"--structural-only",
"analysis_mode",
] {
assert!(
ANALYZER.contains(expected),