fix: tighten output freshness clock sampling

AGENTS.md

@@ -109,6 +109,11 @@ path.
       confirmation probe that must pass sync before the run can be trusted.
 - [x] Refuse freshness passes when Theia/Tethys clock alignment is too uncertain
       or would imply impossible negative media age.
+- [x] Use persistent SSH midpoint clock sampling for freshness checks so SSH
+      startup latency does not masquerade as seconds of timing uncertainty.
+- [x] Measure freshness clock alignment from the server host to Tethys directly
+      instead of routing both clock samples through the client laptop.
+- [x] Include clock uncertainty as a margin in freshness pass/fail decisions.
 - [ ] Keep UI/profile controls authoritative for UVC output profiles beyond
       `640x480@20`; validate `1280x720@30` and `1920x1080@20/30` after sync is
       locked.

Cargo.lock

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

client/Cargo.toml

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

common/Cargo.toml

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

docs/operational-env.md

@@ -56,7 +56,7 @@ from `LESAVKA_CLIENT_PKI_SSH_SOURCE` over SSH. Runtime clients require the insta
 | `LESAVKA_CAPTURE_POWER_GRACE_SECS` | runtime/install/session override |
 | `LESAVKA_CAPTURE_POWER_UNIT` | runtime/install/session override |
 | `LESAVKA_CAPTURE_REMOTE` | runtime/install/session override |
-| `LESAVKA_CLOCK_ALIGNMENT_SAMPLES` | manual direct UVC/UAC probe freshness trust gate; number of SSH midpoint clock samples per host, using the lowest-uncertainty sample, defaults to `5` |
+| `LESAVKA_CLOCK_ALIGNMENT_SAMPLES` | manual direct UVC/UAC probe freshness trust gate; number of server-to-capture persistent SSH midpoint clock samples, using the lowest-uncertainty sample, defaults to `5` |
 | `LESAVKA_CLIENT_APP_SRC` | test/build contract variable; not runtime operator config |
 | `LESAVKA_CLIENT_BUNDLE` | server installer output path for the generated client TLS enrollment bundle |
 | `LESAVKA_CLIENT_CAPTURE_DIR` | client installer capture folder override; defaults to `~/Pictures/lesavka` |
@@ -189,7 +189,7 @@ from `LESAVKA_CLIENT_PKI_SSH_SOURCE` over SSH. Runtime clients require the insta
 | `LESAVKA_OUTPUT_DELAY_SAVE` | manual direct UVC/UAC probe override; after applying a ready measured correction, persist it as the server default calibration |
 | `LESAVKA_OUTPUT_DELAY_TARGET` | manual direct UVC/UAC probe override; choose whether measured skew is corrected by shifting `video` or `audio`, defaults to `video` |
 | `LESAVKA_OUTPUT_FRESHNESS_MAX_AGE_MS` | manual direct UVC/UAC probe freshness gate; maximum clock-corrected server-feed-to-Tethys-observed p95 age, defaults to `1000` |
-| `LESAVKA_OUTPUT_FRESHNESS_MAX_CLOCK_UNCERTAINTY_MS` | manual direct UVC/UAC probe freshness trust gate; do not pass freshness when host clock alignment uncertainty exceeds this, defaults to `100` |
+| `LESAVKA_OUTPUT_FRESHNESS_MAX_CLOCK_UNCERTAINTY_MS` | manual direct UVC/UAC probe freshness trust gate; do not pass freshness when host clock alignment uncertainty exceeds this, defaults to `250` |
 | `LESAVKA_OUTPUT_FRESHNESS_MAX_DRIFT_MS` | manual direct UVC/UAC probe freshness gate; maximum allowed freshness drift across paired probe events, defaults to `100` |
 | `LESAVKA_PASTE_DELAY_MS` | input routing/clipboard override |
 | `LESAVKA_PASTE_KEY` | input routing/clipboard override |

scripts/manual/run_upstream_av_sync.sh

@@ -67,7 +67,7 @@ LESAVKA_OUTPUT_DELAY_GAIN=${LESAVKA_OUTPUT_DELAY_GAIN:-1.0}
 LESAVKA_OUTPUT_DELAY_MAX_STEP_US=${LESAVKA_OUTPUT_DELAY_MAX_STEP_US:-1500000}
 LESAVKA_OUTPUT_FRESHNESS_MAX_AGE_MS=${LESAVKA_OUTPUT_FRESHNESS_MAX_AGE_MS:-1000}
 LESAVKA_OUTPUT_FRESHNESS_MAX_DRIFT_MS=${LESAVKA_OUTPUT_FRESHNESS_MAX_DRIFT_MS:-100}
-LESAVKA_OUTPUT_FRESHNESS_MAX_CLOCK_UNCERTAINTY_MS=${LESAVKA_OUTPUT_FRESHNESS_MAX_CLOCK_UNCERTAINTY_MS:-100}
+LESAVKA_OUTPUT_FRESHNESS_MAX_CLOCK_UNCERTAINTY_MS=${LESAVKA_OUTPUT_FRESHNESS_MAX_CLOCK_UNCERTAINTY_MS:-250}
 LESAVKA_CLOCK_ALIGNMENT_SAMPLES=${LESAVKA_CLOCK_ALIGNMENT_SAMPLES:-5}
 CAPTURE_READY_MARKER="__LESAVKA_CAPTURE_READY__"
 
@@ -133,6 +133,85 @@ PY
 sample_best_host_clock_offset_ns() {
   local host=$1
   local samples=$2
+  if python3 - <<'PY' "${host}" "${samples}" "${SSH_OPTS}"; then
+import shlex
+import subprocess
+import sys
+import time
+
+host = sys.argv[1]
+try:
+    samples = max(1, int(sys.argv[2]))
+except Exception:
+    samples = 5
+ssh_opts = shlex.split(sys.argv[3])
+remote_code = (
+    "import sys,time\n"
+    "for _line in sys.stdin:\n"
+    "    print(time.time_ns(), flush=True)\n"
+)
+cmd = [
+    "ssh",
+    *ssh_opts,
+    host,
+    "python3 -u -c " + shlex.quote(remote_code),
+]
+try:
+    proc = subprocess.Popen(
+        cmd,
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.DEVNULL,
+        text=True,
+        bufsize=1,
+    )
+except Exception:
+    raise SystemExit(1)
+
+rows = []
+try:
+    for _ in range(samples):
+        if proc.stdin is None or proc.stdout is None:
+            break
+        before_ns = time.time_ns()
+        try:
+            proc.stdin.write("x\n")
+            proc.stdin.flush()
+            line = proc.stdout.readline()
+        except (BrokenPipeError, OSError):
+            break
+        after_ns = time.time_ns()
+        if not line:
+            break
+        try:
+            remote_ns = int(line.strip())
+        except Exception:
+            continue
+        local_mid_ns = (before_ns + after_ns) // 2
+        rtt_ns = after_ns - before_ns
+        rows.append((rtt_ns // 2, remote_ns - local_mid_ns, rtt_ns))
+finally:
+    try:
+        if proc.stdin is not None:
+            proc.stdin.close()
+    except Exception:
+        pass
+    try:
+        proc.wait(timeout=2)
+    except Exception:
+        try:
+            proc.kill()
+        except Exception:
+            pass
+
+if not rows:
+    raise SystemExit(1)
+uncertainty_ns, offset_ns, rtt_ns = min(rows)
+print(f"{offset_ns} {uncertainty_ns} {rtt_ns} {len(rows)} persistent-ssh-python")
+PY
+    return 0
+  fi
+
   local tmp
   tmp="$(mktemp)"
   local i=0
@@ -154,15 +233,173 @@ for line in pathlib.Path(sys.argv[1]).read_text().splitlines():
 if not rows:
     raise SystemExit(1)
 uncertainty_ns, offset_ns, rtt_ns = min(rows)
-print(f"{offset_ns} {uncertainty_ns} {rtt_ns} {len(rows)}")
+print(f"{offset_ns} {uncertainty_ns} {rtt_ns} {len(rows)} ssh-date")
 PY
   local rc=$?
   rm -f "${tmp}"
   return "${rc}"
 }
 
+sample_server_to_capture_clock_offset_ns() {
+  local server_host=$1
+  local capture_host=$2
+  local samples=$3
+  python3 - <<'PY' "${server_host}" "${capture_host}" "${samples}" "${SSH_OPTS}"
+import shlex
+import subprocess
+import sys
+
+server_host = sys.argv[1]
+capture_host = sys.argv[2]
+try:
+    samples = max(1, int(sys.argv[3]))
+except Exception:
+    samples = 5
+ssh_opts_text = sys.argv[4]
+ssh_opts = shlex.split(ssh_opts_text)
+remote_code = r'''
+import shlex
+import subprocess
+import sys
+import time
+
+capture_host = sys.argv[1]
+samples = max(1, int(sys.argv[2]))
+ssh_opts = shlex.split(sys.argv[3])
+capture_code = (
+    "import sys,time\n"
+    "for _line in sys.stdin:\n"
+    "    print(time.time_ns(), flush=True)\n"
+)
+cmd = [
+    "ssh",
+    *ssh_opts,
+    capture_host,
+    "python3 -u -c " + shlex.quote(capture_code),
+]
+try:
+    proc = subprocess.Popen(
+        cmd,
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.DEVNULL,
+        text=True,
+        bufsize=1,
+    )
+except Exception:
+    raise SystemExit(1)
+
+rows = []
+try:
+    for _ in range(samples):
+        if proc.stdin is None or proc.stdout is None:
+            break
+        before_ns = time.time_ns()
+        try:
+            proc.stdin.write("x\n")
+            proc.stdin.flush()
+            line = proc.stdout.readline()
+        except (BrokenPipeError, OSError):
+            break
+        after_ns = time.time_ns()
+        if not line:
+            break
+        try:
+            remote_ns = int(line.strip())
+        except Exception:
+            continue
+        local_mid_ns = (before_ns + after_ns) // 2
+        rtt_ns = after_ns - before_ns
+        rows.append((rtt_ns // 2, remote_ns - local_mid_ns, rtt_ns))
+finally:
+    try:
+        if proc.stdin is not None:
+            proc.stdin.close()
+    except Exception:
+        pass
+    try:
+        proc.wait(timeout=2)
+    except Exception:
+        try:
+            proc.kill()
+        except Exception:
+            pass
+
+if not rows:
+    raise SystemExit(1)
+uncertainty_ns, offset_ns, rtt_ns = min(rows)
+print(f"{offset_ns} {uncertainty_ns} {rtt_ns} {len(rows)} server-to-capture-persistent-ssh-python")
+'''
+remote_cmd = (
+    "python3 -u -c "
+    + shlex.quote(remote_code)
+    + " "
+    + shlex.quote(capture_host)
+    + " "
+    + shlex.quote(str(samples))
+    + " "
+    + shlex.quote(ssh_opts_text)
+)
+result = subprocess.run(
+    ["ssh", *ssh_opts, server_host, remote_cmd],
+    check=False,
+    stdout=subprocess.PIPE,
+    stderr=subprocess.DEVNULL,
+    text=True,
+)
+if result.returncode != 0:
+    raise SystemExit(result.returncode)
+for line in result.stdout.splitlines():
+    if line.strip():
+        print(line.strip())
+        raise SystemExit(0)
+raise SystemExit(1)
+PY
+}
+
 write_clock_alignment() {
   echo "==> sampling Theia/Tethys clock alignment for freshness"
+  local direct_sample
+  if direct_sample="$(sample_server_to_capture_clock_offset_ns "${LESAVKA_SERVER_HOST}" "${TETHYS_HOST}" "${LESAVKA_CLOCK_ALIGNMENT_SAMPLES}")"; then
+    python3 - <<'PY' \
+      "${LESAVKA_SERVER_HOST}" \
+      "${TETHYS_HOST}" \
+      "${direct_sample}" \
+      "${LOCAL_CLOCK_ALIGNMENT_JSON}"
+import json
+import pathlib
+import sys
+
+server_host, capture_host, sample, output_path = sys.argv[1:]
+parts = sample.split()
+offset_ns, uncertainty_ns, rtt_ns, sample_count = (int(value) for value in parts[:4])
+method = parts[4] if len(parts) > 4 else "server-to-capture-ssh"
+artifact = {
+    "schema": "lesavka.clock-alignment.v1",
+    "available": True,
+    "method": "server host to capture host persistent ssh midpoint",
+    "server_host": server_host,
+    "capture_host": capture_host,
+    "server_clock_offset_from_local_ns": None,
+    "capture_clock_offset_from_local_ns": None,
+    "theia_to_tethys_offset_ns": offset_ns,
+    "uncertainty_ns": uncertainty_ns,
+    "uncertainty_ms": uncertainty_ns / 1_000_000.0,
+    "server_sample_rtt_ns": 0,
+    "capture_sample_rtt_ns": rtt_ns,
+    "server_samples": sample_count,
+    "capture_samples": sample_count,
+    "server_sample_method": "server-host-local-clock",
+    "capture_sample_method": method,
+}
+pathlib.Path(output_path).write_text(json.dumps(artifact, indent=2, sort_keys=True) + "\n")
+print(f"   ↪ theia_to_tethys_offset_ms={offset_ns / 1_000_000.0:+.3f}")
+print(f"   ↪ clock_alignment_uncertainty_ms={uncertainty_ns / 1_000_000.0:.3f}")
+PY
+    return 0
+  fi
+  echo "   ↪ server-to-capture clock alignment unavailable; falling back to client-mediated SSH samples"
+
   local theia_sample tethys_sample
   if ! theia_sample="$(sample_best_host_clock_offset_ns "${LESAVKA_SERVER_HOST}" "${LESAVKA_CLOCK_ALIGNMENT_SAMPLES}")"; then
     echo "   ↪ clock alignment unavailable: failed to sample ${LESAVKA_SERVER_HOST}"
@@ -186,14 +423,18 @@ import pathlib
 import sys
 
 server_host, capture_host, server_sample, capture_sample, output_path = sys.argv[1:]
-server_offset_ns, server_uncertainty_ns, server_rtt_ns, server_samples = (int(value) for value in server_sample.split())
+server_parts = server_sample.split()
-capture_offset_ns, capture_uncertainty_ns, capture_rtt_ns, capture_samples = (int(value) for value in capture_sample.split())
+capture_parts = capture_sample.split()
+server_offset_ns, server_uncertainty_ns, server_rtt_ns, server_samples = (int(value) for value in server_parts[:4])
+capture_offset_ns, capture_uncertainty_ns, capture_rtt_ns, capture_samples = (int(value) for value in capture_parts[:4])
+server_method = server_parts[4] if len(server_parts) > 4 else "ssh-date"
+capture_method = capture_parts[4] if len(capture_parts) > 4 else "ssh-date"
 theia_to_tethys_offset_ns = capture_offset_ns - server_offset_ns
 uncertainty_ns = server_uncertainty_ns + capture_uncertainty_ns
 artifact = {
     "schema": "lesavka.clock-alignment.v1",
     "available": True,
-    "method": "ssh remote date midpoint",
+    "method": "persistent ssh midpoint" if server_method == capture_method == "persistent-ssh-python" else "ssh midpoint",
     "server_host": server_host,
     "capture_host": capture_host,
     "server_clock_offset_from_local_ns": server_offset_ns,
@@ -205,6 +446,8 @@ artifact = {
     "capture_sample_rtt_ns": capture_rtt_ns,
     "server_samples": server_samples,
     "capture_samples": capture_samples,
+    "server_sample_method": server_method,
+    "capture_sample_method": capture_method,
 }
 pathlib.Path(output_path).write_text(json.dumps(artifact, indent=2, sort_keys=True) + "\n")
 print(
@@ -895,13 +1138,13 @@ elif freshness_worst_p95_ms < -clock_uncertainty_ms:
         f"worst p95 {freshness_worst_p95_ms:.1f} ms, uncertainty "
         f"{clock_uncertainty_ms:.1f} ms"
     )
-elif freshness_worst_p95_ms <= max_freshness_age_ms and (
+elif (freshness_worst_p95_ms + clock_uncertainty_ms) <= max_freshness_age_ms and (
     freshness_worst_drift_ms is None or freshness_worst_drift_ms <= max_freshness_drift_ms
 ):
     freshness_status = "pass"
     freshness_reason = (
-        f"worst p95 freshness {freshness_worst_p95_ms:.1f} ms <= "
+        f"worst p95 freshness {freshness_worst_p95_ms:.1f} ms + clock uncertainty "
-        f"{max_freshness_age_ms:.1f} ms and worst freshness drift "
+        f"{clock_uncertainty_ms:.1f} ms <= {max_freshness_age_ms:.1f} ms and worst freshness drift "
         f"{(freshness_worst_drift_ms or 0.0):.1f} ms <= {max_freshness_drift_ms:.1f} ms"
     )
 else:
@@ -909,6 +1152,7 @@ else:
     freshness_reason = (
         f"worst p95 freshness "
         f"{freshness_worst_p95_ms if freshness_worst_p95_ms is not None else 0.0:.1f} ms "
+        f"+ clock uncertainty {clock_uncertainty_ms:.1f} ms "
         f"(limit {max_freshness_age_ms:.1f} ms), worst freshness drift "
         f"{freshness_worst_drift_ms if freshness_worst_drift_ms is not None else 0.0:.1f} ms "
         f"(limit {max_freshness_drift_ms:.1f} ms)"

server/Cargo.toml

@@ -10,7 +10,7 @@ bench                   = false
 
 [package]
 name                    = "lesavka_server"
-version                 = "0.19.7"
+version                 = "0.19.8"
 edition                 = "2024"
 autobins                = false
 

testing/tests/client_manual_sync_script_contract.rs

@@ -53,9 +53,13 @@ fn upstream_sync_script_tunnels_auto_server_addr_through_ssh() {
         "LESAVKA_OUTPUT_DELAY_MAX_STEP_US=${LESAVKA_OUTPUT_DELAY_MAX_STEP_US:-1500000}",
         "LESAVKA_OUTPUT_FRESHNESS_MAX_AGE_MS=${LESAVKA_OUTPUT_FRESHNESS_MAX_AGE_MS:-1000}",
         "LESAVKA_OUTPUT_FRESHNESS_MAX_DRIFT_MS=${LESAVKA_OUTPUT_FRESHNESS_MAX_DRIFT_MS:-100}",
-        "LESAVKA_OUTPUT_FRESHNESS_MAX_CLOCK_UNCERTAINTY_MS=${LESAVKA_OUTPUT_FRESHNESS_MAX_CLOCK_UNCERTAINTY_MS:-100}",
+        "LESAVKA_OUTPUT_FRESHNESS_MAX_CLOCK_UNCERTAINTY_MS=${LESAVKA_OUTPUT_FRESHNESS_MAX_CLOCK_UNCERTAINTY_MS:-250}",
+        "server-to-capture clock alignment unavailable; falling back to client-mediated SSH samples",
         "LESAVKA_CLOCK_ALIGNMENT_SAMPLES=${LESAVKA_CLOCK_ALIGNMENT_SAMPLES:-5}",
         "sample_best_host_clock_offset_ns",
+        "sample_server_to_capture_clock_offset_ns",
+        "persistent-ssh-python",
+        "server-to-capture-persistent-ssh-python",
         "LESAVKA_OUTPUT_DELAY_PROBE_AUDIO_DELAY_US=${LESAVKA_OUTPUT_DELAY_PROBE_AUDIO_DELAY_US:-0}",
         "LESAVKA_OUTPUT_DELAY_PROBE_VIDEO_DELAY_US=${LESAVKA_OUTPUT_DELAY_PROBE_VIDEO_DELAY_US:-0}",
         "write_clock_alignment",