lesavka/scripts/manual/run_upstream_mirrored_av_sync.sh

#!/usr/bin/env bash
# scripts/manual/run_upstream_mirrored_av_sync.sh
# Manual: full mirrored upstream A/V sync probe.
#
# This probe intentionally uses the normal lesavka-client capture path as the
# sender. A local browser stimulus is captured by the real webcam and real mic,
# Lesavka relays those live captures to Theia, and a Tethys browser records the
# Lesavka UVC/UAC devices via getUserMedia before analysis.

set -euo pipefail

SCRIPT_DIR="$(cd -- "$(dirname "${BASH_SOURCE[0]}")" >/dev/null 2>&1 && pwd)"
REPO_ROOT="$(cd -- "${SCRIPT_DIR}/../.." >/dev/null 2>&1 && pwd)"

LESAVKA_SERVER_ADDR=${LESAVKA_SERVER_ADDR:-auto}
LESAVKA_SERVER_HOST=${LESAVKA_SERVER_HOST:-theia}
LESAVKA_SERVER_SCHEME=${LESAVKA_SERVER_SCHEME:-https}
LESAVKA_SERVER_PORT=${LESAVKA_SERVER_PORT:-50051}
LESAVKA_TLS_DOMAIN=${LESAVKA_TLS_DOMAIN:-lesavka-server}
PROBE_DURATION_SECONDS=${PROBE_DURATION_SECONDS:-20}
PROBE_WARMUP_SECONDS=${PROBE_WARMUP_SECONDS:-4}
PROBE_PULSE_PERIOD_MS=${PROBE_PULSE_PERIOD_MS:-1000}
PROBE_PULSE_WIDTH_MS=${PROBE_PULSE_WIDTH_MS:-120}
PROBE_MARKER_TICK_PERIOD=${PROBE_MARKER_TICK_PERIOD:-5}
PROBE_EVENT_WIDTH_CODES=${PROBE_EVENT_WIDTH_CODES:-1,2,1,3,2,4,1,1,3,1,4,2,1,2,3,4,1,3,2,2,4,1,2,4,3,1,1,4,2,3,1,2}
STIMULUS_PORT=${STIMULUS_PORT:-18444}
STIMULUS_SETTLE_SECONDS=${STIMULUS_SETTLE_SECONDS:-10}
LOCAL_OUTPUT_DIR=${LOCAL_OUTPUT_DIR:-"${REPO_ROOT}/tmp"}
SSH_OPTS=${SSH_OPTS:-"-o BatchMode=yes -o ConnectTimeout=5"}
LOCAL_BROWSER=${LOCAL_BROWSER:-firefox}

mkdir -p "${LOCAL_OUTPUT_DIR}"
STAMP="$(date +%Y%m%d-%H%M%S)"
ARTIFACT_DIR="${LOCAL_OUTPUT_DIR%/}/lesavka-mirrored-av-sync-${STAMP}"
mkdir -p "${ARTIFACT_DIR}"
STIMULUS_STATUS="${ARTIFACT_DIR}/stimulus-status.json"
STIMULUS_PROFILE="${ARTIFACT_DIR}/stimulus-firefox-profile"
CLIENT_LOG="${ARTIFACT_DIR}/lesavka-client.log"
MEDIA_CONTROL="${ARTIFACT_DIR}/media.control"
RESOLVED_LESAVKA_SERVER_ADDR=""
SERVER_TUNNEL_PID=""
STIMULUS_PID=""
STIMULUS_BROWSER_PID=""
CLIENT_PID=""

cleanup() {
  set +e
  [[ -n "${CLIENT_PID}" ]] && kill "${CLIENT_PID}" >/dev/null 2>&1
  [[ -n "${STIMULUS_BROWSER_PID}" ]] && kill "${STIMULUS_BROWSER_PID}" >/dev/null 2>&1
  [[ -n "${STIMULUS_PID}" ]] && kill "${STIMULUS_PID}" >/dev/null 2>&1
  [[ -n "${SERVER_TUNNEL_PID}" ]] && kill "${SERVER_TUNNEL_PID}" >/dev/null 2>&1
}
trap cleanup EXIT

pick_local_port() {
  python3 - <<'PY'
import socket
with socket.socket() as s:
    s.bind(('127.0.0.1', 0))
    print(s.getsockname()[1])
PY
}

wait_for_url() {
  local url=$1
  local timeout_seconds=$2
  local deadline=$(( $(date +%s) + timeout_seconds ))
  until curl -fsS "${url}" >/dev/null 2>&1; do
    if (( $(date +%s) >= deadline )); then
      echo "Timed out waiting for ${url}" >&2
      return 1
    fi
    sleep 0.2
  done
}

wait_for_tcp() {
  local host=$1
  local port=$2
  local timeout_seconds=$3
  python3 - "$host" "$port" "$timeout_seconds" <<'PY'
import socket
import sys
import time

host = sys.argv[1]
port = int(sys.argv[2])
deadline = time.monotonic() + float(sys.argv[3])
last_error = None
while time.monotonic() < deadline:
    try:
        with socket.create_connection((host, port), timeout=0.5):
            sys.exit(0)
    except OSError as exc:
        last_error = exc
        time.sleep(0.2)

print(f"Timed out waiting for TCP {host}:{port}: {last_error}", file=sys.stderr)
sys.exit(1)
PY
}

wait_for_stimulus_page_ready() {
  local timeout_seconds=$1
  local deadline=$(( $(date +%s) + timeout_seconds ))
  local status_json=""
  until status_json="$(curl -fsS "http://127.0.0.1:${STIMULUS_PORT}/status" 2>/dev/null)"; do
    if (( $(date +%s) >= deadline )); then
      echo "Timed out waiting for local stimulus status endpoint" >&2
      return 1
    fi
    sleep 0.2
  done
  while true; do
    if STATUS_JSON="${status_json}" python3 - <<'PY'
import json
import os
import sys

status = json.loads(os.environ["STATUS_JSON"])
sys.exit(0 if status.get("ready") and status.get("page_message") != "booting" else 1)
PY
    then
      return 0
    fi
    if (( $(date +%s) >= deadline )); then
      echo "local stimulus page did not become ready before timeout" >&2
      echo "last stimulus status: ${status_json}" >&2
      echo "stimulus server log: ${ARTIFACT_DIR}/stimulus-server.log" >&2
      echo "stimulus browser log: ${ARTIFACT_DIR}/stimulus-browser.log" >&2
      return 1
    fi
    sleep 0.5
    status_json="$(curl -fsS "http://127.0.0.1:${STIMULUS_PORT}/status" 2>/dev/null || true)"
  done
}

start_server_tunnel_if_needed() {
  if [[ "${LESAVKA_SERVER_ADDR}" != "auto" ]]; then
    RESOLVED_LESAVKA_SERVER_ADDR="${LESAVKA_SERVER_ADDR}"
    return
  fi
  local port
  port="$(pick_local_port)"
  echo "==> opening SSH tunnel to ${LESAVKA_SERVER_HOST}:127.0.0.1:${LESAVKA_SERVER_PORT} on localhost:${port}"
  ssh ${SSH_OPTS} -N \
    -o ExitOnForwardFailure=yes \
    -L "127.0.0.1:${port}:127.0.0.1:${LESAVKA_SERVER_PORT}" \
    "${LESAVKA_SERVER_HOST}" >/tmp/lesavka-mirrored-sync-tunnel.log 2>&1 &
  SERVER_TUNNEL_PID=$!
  wait_for_tcp "127.0.0.1" "${port}" 5
  RESOLVED_LESAVKA_SERVER_ADDR="${LESAVKA_SERVER_SCHEME}://127.0.0.1:${port}"
  echo "==> resolved Lesavka server addr: ${RESOLVED_LESAVKA_SERVER_ADDR}"
  echo "   ↪ tunneled to ${LESAVKA_SERVER_HOST}:127.0.0.1:${LESAVKA_SERVER_PORT}"
}

print_lesavka_versions() {
  echo "==> Lesavka versions under test"
  if [[ ! -x "${REPO_ROOT}/target/debug/lesavka-relayctl" ]]; then
    (cd "${REPO_ROOT}" && cargo build -p lesavka_client --bin lesavka-relayctl >/dev/null)
  fi
  local version_output
  if ! version_output="$(
    LESAVKA_TLS_DOMAIN="${LESAVKA_TLS_DOMAIN}" \
      "${REPO_ROOT}/target/debug/lesavka-relayctl" \
      --server "${RESOLVED_LESAVKA_SERVER_ADDR}" \
      version 2>&1
  )"; then
    echo "failed to query Lesavka versions through ${RESOLVED_LESAVKA_SERVER_ADDR}" >&2
    echo "${version_output}" >&2
    return 1
  fi
  if ! grep -q "^client_version=" <<<"${version_output}"; then
    echo "Lesavka version query did not report client_version=; refusing to run an unattributed probe" >&2
    echo "${version_output}" >&2
    return 1
  fi
  if grep -q "^client_full_version=" <<<"${version_output}"; then
    echo "Lesavka version query reported a combined version+revision; refusing ambiguous probe attribution" >&2
    echo "${version_output}" >&2
    return 1
  fi
  if ! grep -q "^client_revision=" <<<"${version_output}"; then
    echo "Lesavka version query did not report client_revision=; refusing to run an unattributed probe" >&2
    echo "${version_output}" >&2
    return 1
  fi
  if ! grep -q "^server_version=" <<<"${version_output}"; then
    echo "Lesavka version query did not report server_version=; refusing to run an unattributed probe" >&2
    echo "${version_output}" >&2
    return 1
  fi
  if ! grep -q "^server_revision=" <<<"${version_output}"; then
    echo "Lesavka version query did not report server_revision=; refusing to run an unattributed probe" >&2
    echo "${version_output}" >&2
    return 1
  fi
  while IFS= read -r line; do
    [[ -n "${line}" ]] && echo "   ↪ ${line}"
  done <<<"${version_output}"
}

print_upstream_sync_state() {
  local label="$1"
  echo "==> upstream sync planner state (${label})"
  if [[ ! -x "${REPO_ROOT}/target/debug/lesavka-relayctl" ]]; then
    (cd "${REPO_ROOT}" && cargo build -p lesavka_client --bin lesavka-relayctl >/dev/null)
  fi
  local sync_output
  if ! sync_output="$(
    LESAVKA_TLS_DOMAIN="${LESAVKA_TLS_DOMAIN}" \
      "${REPO_ROOT}/target/debug/lesavka-relayctl" \
      --server "${RESOLVED_LESAVKA_SERVER_ADDR}" \
      upstream-sync 2>&1
  )"; then
    echo "   ↪ planner query failed: ${sync_output}"
    return 0
  fi
  while IFS= read -r line; do
    [[ -n "${line}" ]] && echo "   ↪ ${line}"
  done <<<"${sync_output}"
}

start_local_stimulus() {
  echo "==> starting local A/V stimulus server"
  python3 "${REPO_ROOT}/scripts/manual/local_av_stimulus.py" \
    --port "${STIMULUS_PORT}" \
    --status "${STIMULUS_STATUS}" \
    --duration-seconds "${PROBE_DURATION_SECONDS}" \
    --warmup-seconds "${PROBE_WARMUP_SECONDS}" \
    --pulse-period-ms "${PROBE_PULSE_PERIOD_MS}" \
    --pulse-width-ms "${PROBE_PULSE_WIDTH_MS}" \
    --marker-tick-period "${PROBE_MARKER_TICK_PERIOD}" \
    --event-width-codes "${PROBE_EVENT_WIDTH_CODES}" \
    >"${ARTIFACT_DIR}/stimulus-server.log" 2>&1 &
  STIMULUS_PID=$!
  wait_for_url "http://127.0.0.1:${STIMULUS_PORT}/status" 10

  mkdir -p "${STIMULUS_PROFILE}"
  cat >"${STIMULUS_PROFILE}/user.js" <<'PREFS'
user_pref("media.autoplay.default", 0);
user_pref("media.autoplay.blocking_policy", 0);
user_pref("toolkit.telemetry.reportingpolicy.firstRun", false);
user_pref("browser.shell.checkDefaultBrowser", false);
user_pref("browser.tabs.warnOnClose", false);
user_pref("browser.startup.page", 1);
user_pref("browser.aboutwelcome.enabled", false);
PREFS
  printf 'user_pref("browser.startup.homepage", "http://127.0.0.1:%s/");\n' "${STIMULUS_PORT}" >>"${STIMULUS_PROFILE}/user.js"
  echo "==> opening local stimulus browser"
  "${LOCAL_BROWSER}" --new-instance --no-remote --profile "${STIMULUS_PROFILE}" \
    "http://127.0.0.1:${STIMULUS_PORT}/" \
    >"${ARTIFACT_DIR}/stimulus-browser.log" 2>&1 &
  STIMULUS_BROWSER_PID=$!
  wait_for_stimulus_page_ready 15

  echo "==> position check"
  echo "    Point the real webcam at the stimulus window and keep the selected microphone hearing the tone."
  echo "    Waiting ${STIMULUS_SETTLE_SECONDS}s before starting the mirrored capture."
  sleep "${STIMULUS_SETTLE_SECONDS}"
}

start_real_lesavka_client() {
  echo "camera=1 microphone=1 audio=0 $(date +%s%N)" >"${MEDIA_CONTROL}"
  echo "==> starting real headless lesavka-client sender"
  (
    cd "${REPO_ROOT}"
    LESAVKA_HEADLESS=1 \
    LESAVKA_SERVER_ADDR="${RESOLVED_LESAVKA_SERVER_ADDR}" \
    LESAVKA_TLS_DOMAIN="${LESAVKA_TLS_DOMAIN}" \
    LESAVKA_MEDIA_CONTROL="${MEDIA_CONTROL}" \
    RUST_LOG="${RUST_LOG:-warn,lesavka_client::app=info,lesavka_client::input::camera=info,lesavka_client::input::microphone=info}" \
    "${REPO_ROOT}/target/debug/lesavka-client" --no-launcher --server "${RESOLVED_LESAVKA_SERVER_ADDR}"
  ) >"${CLIENT_LOG}" 2>&1 &
  CLIENT_PID=$!
  sleep 4
  if ! kill -0 "${CLIENT_PID}" >/dev/null 2>&1; then
    echo "lesavka-client exited before mirrored probe could start; see ${CLIENT_LOG}" >&2
    exit 1
  fi
}

run_browser_capture_with_real_driver() {
  local record_seconds=$((PROBE_DURATION_SECONDS + 3))
  local wait_seconds=$((PROBE_DURATION_SECONDS + 2))
  local driver_command="curl -fsS -X POST http://127.0.0.1:${STIMULUS_PORT}/start >/dev/null; sleep ${wait_seconds}"
  echo "==> starting Tethys browser consumer and mirrored driver"
  BROWSER_RECORD_SECONDS="${record_seconds}" \
  PROBE_DURATION_SECONDS="${PROBE_DURATION_SECONDS}" \
  BROWSER_SYNC_DRIVER_COMMAND="${driver_command}" \
  SYNC_ANALYZE_EVENT_WIDTH_CODES="${PROBE_EVENT_WIDTH_CODES}" \
  LOCAL_OUTPUT_DIR="${ARTIFACT_DIR}" \
  LESAVKA_SERVER_ADDR="${RESOLVED_LESAVKA_SERVER_ADDR}" \
  "${REPO_ROOT}/scripts/manual/run_upstream_browser_av_sync.sh"
}

echo "==> prebuilding real client and analyzer"
(
  cd "${REPO_ROOT}"
  cargo build -p lesavka_client --bin lesavka-client --bin lesavka-sync-analyze --bin lesavka-relayctl >/dev/null
)

start_server_tunnel_if_needed
print_lesavka_versions
print_upstream_sync_state "before mirrored run"
start_local_stimulus
start_real_lesavka_client
run_browser_capture_with_real_driver
print_upstream_sync_state "after mirrored run"

echo "==> mirrored probe complete"
echo "artifact_dir: ${ARTIFACT_DIR}"
echo "client_log: ${CLIENT_LOG}"
echo "stimulus_status: ${STIMULUS_STATUS}"