probe: pause injector upstream during synthetic rct runs

Cargo.lock

@@ -1652,7 +1652,7 @@ checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2"
 
 [[package]]
 name = "lesavka_client"
-version = "0.22.57"
+version = "0.22.58"
 dependencies = [
  "anyhow",
  "async-stream",
@@ -1686,7 +1686,7 @@ dependencies = [
 
 [[package]]
 name = "lesavka_common"
-version = "0.22.57"
+version = "0.22.58"
 dependencies = [
  "anyhow",
  "base64",
@@ -1698,7 +1698,7 @@ dependencies = [
 
 [[package]]
 name = "lesavka_server"
-version = "0.22.57"
+version = "0.22.58"
 dependencies = [
  "anyhow",
  "base64",

client/Cargo.toml

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

common/Cargo.toml

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

scripts/manual/run_synthetic_rct_uvc_probe.py

@@ -27,6 +27,123 @@ MARKER_COLUMNS = 16
 CADENCE_REASONS = {"frame_repeat", "frame_gap", "frame_backwards"}
 NON_VISUAL_REASONS = CADENCE_REASONS | {"sequence_marker_mismatch"}
 
+REMOTE_MEDIA_CONTROL_PAUSE = r"""
+import base64
+import json
+import pathlib
+import sys
+import time
+
+DEFAULT_MEDIA_CONTROL_PATH = "/tmp/lesavka-media.control"
+
+
+def media_control_with_camera(raw, enabled):
+    tokens = raw.split() if raw else []
+    rendered = []
+    saw_camera = False
+    saw_microphone = False
+    saw_audio = False
+    for token in tokens:
+        key, sep, _value = token.partition("=")
+        if sep and key == "camera":
+            rendered.append(f"camera={1 if enabled else 0}")
+            saw_camera = True
+        else:
+            rendered.append(token)
+            saw_microphone = saw_microphone or (sep and key in {"microphone", "mic"})
+            saw_audio = saw_audio or (sep and key in {"audio", "speaker"})
+    if not saw_camera:
+        rendered.insert(0, f"camera={1 if enabled else 0}")
+    if not saw_microphone:
+        rendered.append("microphone=1")
+    if not saw_audio:
+        rendered.append("audio=1")
+    return " ".join(rendered) + "\n"
+
+
+def discover_media_control_paths():
+    candidates = set()
+    proc = pathlib.Path("/proc")
+    if not proc.exists():
+        return []
+    for entry in proc.iterdir():
+        if not entry.name.isdigit():
+            continue
+        try:
+            environ = (entry / "environ").read_bytes()
+            cmdline = (entry / "cmdline").read_bytes().replace(b"\0", b" ")
+        except (FileNotFoundError, PermissionError, ProcessLookupError, OSError):
+            continue
+        if b"lesavka" not in cmdline and b"LESAVKA_MEDIA_CONTROL=" not in environ:
+            continue
+        for token in environ.split(b"\0"):
+            if token.startswith(b"LESAVKA_MEDIA_CONTROL="):
+                raw_path = token.split(b"=", 1)[1].decode(errors="replace")
+                if raw_path:
+                    candidates.add(pathlib.Path(raw_path))
+    return sorted(
+        candidates,
+        key=lambda path: (
+            not path.exists(),
+            -path.stat().st_mtime if path.exists() else 0,
+            str(path),
+        ),
+    )
+
+
+request = json.loads(sys.argv[1])
+state_path = pathlib.Path(request["state_path"])
+explicit_path = request.get("media_control_path") or ""
+discovered = [] if explicit_path else discover_media_control_paths()
+path = (
+    pathlib.Path(explicit_path)
+    if explicit_path
+    else (discovered[0] if discovered else pathlib.Path(DEFAULT_MEDIA_CONTROL_PATH))
+)
+original = path.read_bytes() if path.exists() else None
+original_text = original.decode(errors="replace") if original is not None else None
+path.write_text(media_control_with_camera(original_text, False))
+state_path.write_text(
+    json.dumps(
+        {
+            "path": str(path),
+            "had_original": original is not None,
+            "original_b64": base64.b64encode(original or b"").decode(),
+        },
+        sort_keys=True,
+    )
+    + "\n"
+)
+time.sleep(0.5)
+print(
+    json.dumps(
+        {
+            "path": str(path),
+            "state_path": str(state_path),
+            "discovered": [str(path) for path in discovered],
+        }
+    )
+)
+"""
+
+REMOTE_MEDIA_CONTROL_RESTORE = r"""
+import base64
+import json
+import pathlib
+import sys
+
+request = json.loads(sys.argv[1])
+state_path = pathlib.Path(request["state_path"])
+state = json.loads(state_path.read_text())
+path = pathlib.Path(state["path"])
+if state.get("had_original"):
+    path.write_bytes(base64.b64decode(state.get("original_b64") or ""))
+else:
+    path.unlink(missing_ok=True)
+state_path.unlink(missing_ok=True)
+print(json.dumps({"path": str(path), "state_path": str(state_path)}))
+"""
+
 
 def parse_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser(
@@ -57,7 +174,7 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument(
         "--pause-local-live-upstream",
         action="store_true",
-        help="temporarily write camera=0 to the local Lesavka media control file so a live client does not preempt the synthetic injector",
+        help="temporarily write camera=0 to the injector host's Lesavka media control file so a live client does not preempt the synthetic injector",
     )
     parser.add_argument(
         "--media-control-path",
@@ -262,6 +379,44 @@ def restore_local_live_upstream(path: pathlib.Path, original: bytes | None) -> N
     print(f"restored local live media control at {path}", file=sys.stderr)
 
 
+def run_remote_python(host: str, script: str, payload: dict[str, Any]) -> dict[str, Any]:
+    output = subprocess.check_output(
+        ["ssh", host, f"python3 - {shlex.quote(json.dumps(payload, sort_keys=True))}"],
+        input=script,
+        text=True,
+    )
+    return json.loads(output.strip().splitlines()[-1])
+
+
+def pause_remote_live_upstream(host: str, args: argparse.Namespace) -> dict[str, Any]:
+    state_path = f"/tmp/lesavka-synthetic-rct-media-control-{os.getpid()}.json"
+    state = run_remote_python(
+        host,
+        REMOTE_MEDIA_CONTROL_PAUSE,
+        {
+            "media_control_path": args.media_control_path,
+            "state_path": state_path,
+        },
+    )
+    print(
+        f"paused injector-host live camera upstream on {host} via {state['path']}",
+        file=sys.stderr,
+    )
+    return state
+
+
+def restore_remote_live_upstream(host: str, state: dict[str, Any]) -> None:
+    restored = run_remote_python(
+        host,
+        REMOTE_MEDIA_CONTROL_RESTORE,
+        {"state_path": state["state_path"]},
+    )
+    print(
+        f"restored injector-host live media control on {host} at {restored['path']}",
+        file=sys.stderr,
+    )
+
+
 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(
@@ -442,9 +597,14 @@ def run_remote_orchestrated(args: argparse.Namespace) -> int:
     capture: subprocess.Popen[Any] | None = None
     diagnosis: list[str] = []
     paused_control: tuple[pathlib.Path, bytes | None] | None = None
+    paused_remote_control: tuple[str, dict[str, Any]] | None = None
     try:
         if args.pause_local_live_upstream:
-            paused_control = pause_local_live_upstream(args)
+            if args.local_inject:
+                paused_control = pause_local_live_upstream(args)
+            else:
+                remote_state = pause_remote_live_upstream(args.inject_host, args)
+                paused_remote_control = (args.inject_host, remote_state)
         if args.capture_before_inject:
             capture = start_capture()
             time.sleep(1.0)
@@ -464,6 +624,8 @@ def run_remote_orchestrated(args: argparse.Namespace) -> int:
                 capture = start_capture()
                 capture_rc, inject_rc = wait_capture_or_inject_exit(capture, inject)
     finally:
+        if paused_remote_control is not None:
+            restore_remote_live_upstream(*paused_remote_control)
         if paused_control is not None:
             restore_local_live_upstream(*paused_control)
     local_capture = artifact_dir / "capture"
@@ -494,7 +656,15 @@ def run_remote_orchestrated(args: argparse.Namespace) -> int:
                 )
             frames = int(capture_data.get("frames") or 0)
             reason_counts = capture_data.get("reason_counts") or {}
+            visual_reasons = capture_data.get("visual_reason_counts") or {}
+            visual_frames = int(capture_data.get("visual_suspicious_frames") or 0)
+            suspicious_frames = int(capture_data.get("suspicious_frames") or 0)
             repeats = int(reason_counts.get("frame_repeat") or 0)
+            cadence_only = suspicious_frames > 0 and visual_frames == 0 and not visual_reasons
+            if cadence_only:
+                diagnosis.append(
+                    "RCT capture had cadence-only repeat/gap events; no visual tear/mixed-frame corruption was detected in aligned synthetic frames"
+                )
             if frames > 0 and repeats >= max(3, int(frames * 0.9)):
                 diagnosis.append(
                     "RCT capture repeated nearly every decoded synthetic marker; the received UVC stream was stale/frozen instead of advancing"

server/Cargo.toml

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

tests/manual/server/rct/synthetic_rct_uvc_probe_manual_contract.rs

@@ -97,10 +97,17 @@ fn synthetic_probe_keeps_bundled_network_ingress_and_rct_comparison_markers() {
         "synthetic uplink exited before capture warmup completed",
         "paused local live camera upstream",
         "restored local live media control",
+        "paused injector-host live camera upstream",
+        "restored injector-host live media control",
         "media_control_with_camera",
         "discover_media_control_paths",
         "resolve_media_control_path",
+        "pause_remote_live_upstream",
+        "restore_remote_live_upstream",
+        "run_remote_python",
         "discovered live Lesavka media control path",
+        "cadence-only repeat/gap",
+        "no visual tear/mixed-frame corruption was detected",
         "max_lower_mae",
         "ffmpeg",
         "v4l2",