fix(uvc): preserve control event fairness

Cargo.lock

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

client/Cargo.toml

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

common/Cargo.toml

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

docs/architecture-and-media-readiness.md

@@ -314,7 +314,19 @@ into the Lesavka tree, installs it with the server, requires `libgpiod`, and
 runs it with `/usr/bin/python` before capture discovery. This removes the
 unversioned virtualenv from the physical eye-feed power path.
 
-## 7. The Current Downstream Video Failure
+### The `0.27.10` UVC Event-Fairness Fix
+
+The first post-repair synthetic upstream probe reached Tethys but produced no
+decodable MJPEG timeline. Tethys recorded repeated `EOI before SOF` packets and
+two five-second UVC probe-control timeouts. The helper drained completed V4L2
+output buffers without a per-turn bound before servicing control events, so a
+continuously consuming host could starve probe/commit traffic. Release 0.27.10
+bounds dequeue work, reports fairness yields, and makes the native receiver
+probe terminate with evidence even when malformed packets produce no decoded
+timestamps. This is a hardware-testable correction; marked Tethys capture is
+still required before declaring the tear/smear class fixed.
+
+## 7. The Resolved Downstream Video Failure
 
 The current blank downstream feeds fail before transport or decoding.
 
@@ -369,6 +381,10 @@ successfully imported the installed `libgpiod` binding. This proves the common
 downstream failure was capture-card power automation, not two simultaneous
 card failures, H.264 transport, or client decoding. Release 0.27.9 makes the
 working system binding and a versioned helper the installed service contract.
+After installing 0.27.9, GPIO27 stayed asserted, both GC311 devices enumerated,
+and both downstream RPCs decoded 1280x720 H.264. The right eye carried changing
+desktop video; the left eye carried the capture card's explicit `No Signal`
+frame, isolating that remaining input condition to its HDMI source/cable.
 
 ## 8. Recovery And Deployment Sequence
 
@@ -386,7 +402,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.9`, the pushed release revision,
+7. Confirm Theia reports server version `0.27.10`, 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.
@@ -404,7 +420,7 @@ hardware contract is repeatable. The remaining work falls into five groups.
 
 ### A. Install And Version Parity
 
-- Push and deploy `0.27.9` through the client/server install scripts.
+- Push and deploy `0.27.10` 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.
 
@@ -470,6 +486,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.9` client/server pair, the
+Until that sequence passes on the installed `0.27.10` 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/install/server.sh

@@ -418,6 +418,7 @@ LESAVKA_UVC_WIDTH=$(uvc_env_value LESAVKA_UVC_WIDTH 1280)
 LESAVKA_UVC_HEIGHT=$(uvc_env_value LESAVKA_UVC_HEIGHT 720)
 LESAVKA_UVC_CODEC=${INSTALL_UVC_CODEC}
 LESAVKA_UVC_BUFFER_COUNT=$(uvc_env_value LESAVKA_UVC_BUFFER_COUNT 4)
+LESAVKA_UVC_PUMP_DEQUEUE_BUDGET=$(uvc_env_value LESAVKA_UVC_PUMP_DEQUEUE_BUDGET 4)
 LESAVKA_UVC_BLOCKING=$(uvc_env_value LESAVKA_UVC_BLOCKING 1)
 LESAVKA_UVC_CONTROL_READ_ONLY=$(uvc_env_value LESAVKA_UVC_CONTROL_READ_ONLY 0)
 LESAVKA_UVC_QUEUE_PACING=$(uvc_env_value LESAVKA_UVC_QUEUE_PACING 0)

scripts/manual/run_synthetic_rct_uvc_probe.py

@@ -1672,8 +1672,8 @@ def run_capture(args: argparse.Namespace) -> int:
     command, capture_width, capture_height, device = ffmpeg_cmd(args, width, height)
     artifact_dir = pathlib.Path(args.artifact_dir) if args.artifact_dir else pathlib.Path("/tmp") / f"lesavka-synthetic-rct-capture-{timestamp()}"
     artifact_dir.mkdir(parents=True, exist_ok=True)
-    if args.deep_capture and args.source != "uvc":
+    if args.deep_capture and args.source != "device":
-        raise SystemExit("--deep-capture requires --source uvc")
+        raise SystemExit("--deep-capture requires --source device")
     if args.deep_capture:
         write_deep_capture_snapshot(artifact_dir, "before", device)
     dynamic_debug_enabled = (
@@ -1688,6 +1688,7 @@ def run_capture(args: argparse.Namespace) -> int:
     analysis_elapsed = 0.0
     raw_capture_bytes = 0
     ffmpeg_rc: int | None = None
+    capture_timed_out = False
     frame_index = 0
     suspicious_count = 0
     visual_suspicious_count = 0
@@ -1807,11 +1808,26 @@ def run_capture(args: argparse.Namespace) -> int:
                     capture_command[-1:-1] = ["-t", str(args.duration)]
                 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_mjpeg if native_mjpeg is not None else raw_path
+            print(f"capturing RCT frames before analysis: {capture_target}", file=sys.stderr)
             capture_started = time.monotonic()
-            proc = subprocess.run(capture_command, stdout=subprocess.DEVNULL, stderr=err, check=False)
+            try:
+                proc = subprocess.run(
+                    capture_command,
+                    stdout=subprocess.DEVNULL,
+                    stderr=err,
+                    check=False,
+                    timeout=max(5.0, args.duration + 10.0),
+                )
+                ffmpeg_rc = proc.returncode
+            except subprocess.TimeoutExpired:
+                capture_timed_out = True
+                ffmpeg_rc = 124
+                err.write(
+                    f"capture exceeded wall timeout after {args.duration + 10.0:.1f}s\n".encode()
+                )
+                err.flush()
             capture_elapsed = time.monotonic() - capture_started
-            ffmpeg_rc = proc.returncode
             if args.deep_capture and native_mjpeg is not None and native_mjpeg.exists():
                 decode_command = [
                     "ffmpeg", "-hide_banner", "-nostdin", "-loglevel", "warning",
@@ -1819,6 +1835,7 @@ def run_capture(args: argparse.Namespace) -> int:
                 ]
                 (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)
+            raw_path.touch(exist_ok=True)
             raw_capture_bytes = raw_path.stat().st_size if raw_path.exists() else 0
             print(
                 f"analyzing captured raw RCT frames bytes={raw_capture_bytes} capture_s={capture_elapsed:.3f}",
@@ -1855,6 +1872,7 @@ def run_capture(args: argparse.Namespace) -> int:
         "duration_observed_s": round(elapsed, 3),
         "analysis_duration_s": round(analysis_elapsed, 3),
         "ffmpeg_rc": ffmpeg_rc,
+        "capture_timed_out": capture_timed_out,
         "raw_capture_bytes": raw_capture_bytes,
         "frames": frame_index,
         "fps_observed": round(frame_index / elapsed, 3),

server/Cargo.toml

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

server/src/bin/lesavka-uvc.real.inc

@@ -53,6 +53,7 @@ const IDLE_MJPEG_FRAME: &[u8] = include_bytes!("lesavka_uvc/idle_1280x720_black.
 // Keep enough output buffers queued that the USB gadget is not starved while a
 // browser is consuming high-speed isochronous MJPEG.
 const DEFAULT_UVC_BUFFER_COUNT: u32 = 4;
+const DEFAULT_UVC_PUMP_DEQUEUE_BUDGET: usize = 4;
 const DEFAULT_UVC_IDLE_PUMP_MS: u64 = 2;
 const DEFAULT_UVC_FRAME_MAX_AGE_MS: u64 = 1_000;
 const DEFAULT_UVC_QUEUE_PACING: bool = false;
@@ -292,6 +293,7 @@ struct UvcVideoStats {
     last_rejected_oversize_cap: usize,
     paced_sleeps: u64,
     paced_sleep_ms: u64,
+    pump_budget_yields: u64,
     last_report: Option<Instant>,
 }
 
@@ -380,7 +382,7 @@ impl UvcVideoStream {
         if !self.streaming {
             return Ok(());
         }
-        loop {
+        for _ in 0..uvc_pump_dequeue_budget() {
             let mut buf = empty_v4l2_buffer();
             let req = ioctl_readwrite::<V4l2Buffer>(b'V', 17);
             let rc = unsafe { libc::ioctl(self.fd, req, &mut buf) };
@@ -399,6 +401,8 @@ impl UvcVideoStream {
             }
             self.queue_buffer(buf.index, buf.flags)?;
         }
+        self.stats.pump_budget_yields += 1;
+        Ok(())
     }
 
     fn set_format(&self, cfg: UvcConfig) -> Result<()> {
@@ -684,7 +688,7 @@ impl UvcVideoStream {
         }
         self.stats.last_report = Some(now);
         eprintln!(
-            "[lesavka-uvc] video stats queued={} reloaded={} stale_replay={} rejected_oversize={} rejected_invalid={} fallback_idle={} held_last_good={} read_errors={} strict_validation_failures={} dqbuf_ioctl_errors={} dqbuf_flag_errors={} qbuf_ioctl_errors={} latest_bytes={} frame_cap={} last_rejected_oversize_bytes={} last_rejected_oversize_cap={} paced_sleeps={} paced_sleep_ms={}",
+            "[lesavka-uvc] video stats queued={} reloaded={} stale_replay={} rejected_oversize={} rejected_invalid={} fallback_idle={} held_last_good={} read_errors={} strict_validation_failures={} dqbuf_ioctl_errors={} dqbuf_flag_errors={} qbuf_ioctl_errors={} latest_bytes={} frame_cap={} last_rejected_oversize_bytes={} last_rejected_oversize_cap={} paced_sleeps={} paced_sleep_ms={} pump_budget_yields={}",
             self.stats.queued,
             self.stats.reloaded,
             self.stats.replayed_stale,
@@ -702,7 +706,8 @@ impl UvcVideoStream {
             self.stats.last_rejected_oversize_bytes,
             self.stats.last_rejected_oversize_cap,
             self.stats.paced_sleeps,
-            self.stats.paced_sleep_ms
+            self.stats.paced_sleep_ms,
+            self.stats.pump_budget_yields
         );
         if let Some(path) = uvc_stats_path() {
             let _ = write_atomic_text(
@@ -778,6 +783,14 @@ fn uvc_buffer_count() -> u32 {
     env_u32("LESAVKA_UVC_BUFFER_COUNT", DEFAULT_UVC_BUFFER_COUNT).clamp(1, 8)
 }
 
+fn uvc_pump_dequeue_budget() -> usize {
+    env_u64(
+        "LESAVKA_UVC_PUMP_DEQUEUE_BUDGET",
+        DEFAULT_UVC_PUMP_DEQUEUE_BUDGET as u64,
+    )
+    .clamp(1, 64) as usize
+}
+
 fn uvc_idle_pump_sleep() -> Duration {
     Duration::from_millis(env_u64(
         "LESAVKA_UVC_IDLE_PUMP_MS",
@@ -903,7 +916,7 @@ fn write_atomic_text(path: &std::path::Path, text: &str) -> Result<()> {
 fn uvc_stats_snapshot_json(stats: &UvcVideoStats, frame_cap: usize) -> String {
     let kernel = uvc_kernel_stats_json();
     format!(
-        "{{\"generated_unix_ms\":{},\"queued\":{},\"reloaded\":{},\"stale_replay\":{},\"rejected_oversize\":{},\"rejected_invalid\":{},\"fallback_idle\":{},\"held_last_good\":{},\"read_errors\":{},\"strict_validation_failures\":{},\"dqbuf_ioctl_errors\":{},\"dqbuf_flag_errors\":{},\"qbuf_ioctl_errors\":{},\"latest_bytes\":{},\"frame_cap\":{},\"last_rejected_oversize_bytes\":{},\"last_rejected_oversize_cap\":{},\"paced_sleeps\":{},\"paced_sleep_ms\":{},\"kernel\":{}}}\n",
+        "{{\"generated_unix_ms\":{},\"queued\":{},\"reloaded\":{},\"stale_replay\":{},\"rejected_oversize\":{},\"rejected_invalid\":{},\"fallback_idle\":{},\"held_last_good\":{},\"read_errors\":{},\"strict_validation_failures\":{},\"dqbuf_ioctl_errors\":{},\"dqbuf_flag_errors\":{},\"qbuf_ioctl_errors\":{},\"latest_bytes\":{},\"frame_cap\":{},\"last_rejected_oversize_bytes\":{},\"last_rejected_oversize_cap\":{},\"paced_sleeps\":{},\"paced_sleep_ms\":{},\"pump_budget_yields\":{},\"kernel\":{}}}\n",
         unix_now_ns() / 1_000_000,
         stats.queued,
         stats.reloaded,
@@ -923,6 +936,7 @@ fn uvc_stats_snapshot_json(stats: &UvcVideoStats, frame_cap: usize) -> String {
         stats.last_rejected_oversize_cap,
         stats.paced_sleeps,
         stats.paced_sleep_ms,
+        stats.pump_budget_yields,
         kernel
     )
 }

tests/contract/scripts/install/server_install_script_contract.rs

@@ -276,6 +276,7 @@ fn server_install_pins_hdmi_camera_and_display_defaults() {
     assert!(SERVER_INSTALL.contains("uvc_env_value LESAVKA_UVC_HEIGHT 720"));
     assert!(SERVER_INSTALL.contains("uvc_env_value LESAVKA_UVC_CONTROL_READ_ONLY 0"));
     assert!(SERVER_INSTALL.contains("uvc_env_value LESAVKA_UVC_BUFFER_COUNT 4"));
+    assert!(SERVER_INSTALL.contains("uvc_env_value LESAVKA_UVC_PUMP_DEQUEUE_BUDGET 4"));
     assert!(SERVER_INSTALL.contains("uvc_env_value LESAVKA_UVC_QUEUE_PACING 0"));
     assert!(SERVER_INSTALL.contains("uvc_env_value LESAVKA_UVC_BULK 1"));
     assert!(SERVER_INSTALL.contains("uvc_env_value LESAVKA_UVC_FRAME_SIZE_GUARD 1"));

tests/contract/server/uvc/server_uvc_binary_contract.rs

@@ -376,6 +376,24 @@ mod uvc_binary {
         assert!(source.contains("streaming_bulk is not 1; using isochronous payload sizing"));
     }
 
+    #[test]
+    #[serial]
+    fn uvc_video_pump_is_bounded_so_control_requests_cannot_starve() {
+        with_var("LESAVKA_UVC_PUMP_DEQUEUE_BUDGET", None::<&str>, || {
+            assert_eq!(uvc_pump_dequeue_budget(), 4);
+        });
+        with_var("LESAVKA_UVC_PUMP_DEQUEUE_BUDGET", Some("1"), || {
+            assert_eq!(uvc_pump_dequeue_budget(), 1);
+        });
+        with_var("LESAVKA_UVC_PUMP_DEQUEUE_BUDGET", Some("999"), || {
+            assert_eq!(uvc_pump_dequeue_budget(), 64);
+        });
+
+        let source = include_str!("../../../../server/src/bin/lesavka-uvc.real.inc");
+        assert!(source.contains("for _ in 0..uvc_pump_dequeue_budget()"));
+        assert!(source.contains("self.stats.pump_budget_yields += 1"));
+    }
+
     #[test]
     #[serial]
     fn uvc_stats_snapshot_can_be_disabled_or_written() {
@@ -397,6 +415,7 @@ mod uvc_binary {
             last_rejected_oversize_cap: 300_000,
             paced_sleeps: 5,
             paced_sleep_ms: 123,
+            pump_budget_yields: 11,
             last_report: None,
         };
         let json = uvc_stats_snapshot_json(&stats, 333_333);
@@ -407,6 +426,7 @@ mod uvc_binary {
         assert!(json.contains("\"last_rejected_oversize_cap\":300000"));
         assert!(json.contains("\"paced_sleeps\":5"));
         assert!(json.contains("\"paced_sleep_ms\":123"));
+        assert!(json.contains("\"pump_budget_yields\":11"));
 
         let kernel_path = tempfile::NamedTempFile::new().expect("kernel stats");
         fs::write(

tests/manual/server/rct/synthetic_rct_uvc_probe_manual_contract.rs

@@ -152,6 +152,36 @@ fn synthetic_probe_keeps_bundled_network_ingress_and_rct_comparison_markers() {
     }
 }
 
+#[test]
+fn synthetic_probe_deep_capture_accepts_its_device_source_mode() {
+    assert!(
+        PROBE_SRC.contains("if args.deep_capture and args.source != \"device\":")
+            && PROBE_SRC.contains("--deep-capture requires --source device"),
+        "deep capture must accept the parser's UVC device source instead of requiring an unavailable source name"
+    );
+    assert!(
+        !PROBE_SRC.contains("--deep-capture requires --source uvc"),
+        "the probe must not demand a source name its argument parser rejects"
+    );
+}
+
+#[test]
+fn synthetic_probe_capture_has_a_wall_clock_timeout() {
+    for expected in [
+        "timeout=max(5.0, args.duration + 10.0)",
+        "except subprocess.TimeoutExpired:",
+        "capture_timed_out = True",
+        "ffmpeg_rc = 124",
+        "raw_path.touch(exist_ok=True)",
+        "\"capture_timed_out\": capture_timed_out",
+    ] {
+        assert!(
+            PROBE_SRC.contains(expected),
+            "receiver capture must terminate and report evidence even when malformed UVC packets produce no decoded timeline: {expected}"
+        );
+    }
+}
+
 #[test]
 fn synthetic_injector_enters_the_public_bundled_media_rpc() {
     for expected in [