lesavka/scripts/ci/media_reliability_gate.sh

#!/usr/bin/env bash
# Run deterministic media/device contracts and emit Atlas quality metrics.
set -euo pipefail

ROOT_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)
REPORT_DIR="${ROOT_DIR}/target/media-reliability-gate"
TEST_LOG="${REPORT_DIR}/cargo-test.log"
SUMMARY_JSON="${REPORT_DIR}/summary.json"
SUMMARY_TXT="${REPORT_DIR}/summary.txt"
METRICS_FILE="${REPORT_DIR}/metrics.prom"
STATUS_FILE="${REPORT_DIR}/gate-status.txt"
PUSHGATEWAY_URL=${QUALITY_GATE_PUSHGATEWAY_URL:-}
PUSHGATEWAY_JOB=${LESAVKA_MEDIA_GATE_PUSHGATEWAY_JOB:-lesavka-media-reliability-gate}
SYNC_PROBE_REPORT_JSON=${LESAVKA_SYNC_PROBE_REPORT_JSON:-}
SYNC_PROBE_REPORT_DIR=${LESAVKA_SYNC_PROBE_REPORT_DIR:-}
REQUIRE_SYNC_PROBE=${LESAVKA_REQUIRE_SYNC_PROBE:-0}

mkdir -p "${REPORT_DIR}"
cd "${ROOT_DIR}"

branch=${BRANCH_NAME:-${GIT_BRANCH:-}}
if [[ -z "${branch}" ]]; then
  branch=$(git rev-parse --abbrev-ref HEAD 2>/dev/null || echo unknown)
fi
commit=${GIT_COMMIT:-}
if [[ -z "${commit}" ]]; then
  commit=$(git rev-parse --short HEAD 2>/dev/null || echo unknown)
fi
build_url=${BUILD_URL:-}

MEDIA_TESTS=(
  --test client_camera_include_contract
  --test client_launcher_runtime_contract
  --test client_microphone_include_contract
  --test client_microphone_source_contract
  --test client_uplink_freshness_contract
  --test client_uplink_performance_contract
  --test client_output_video_include_contract
  --test handshake_camera_contract
  --test server_camera_contract
  --test server_camera_runtime_contract
  --test server_upstream_media_contract
  --test server_uvc_runtime_contract
  --test server_video_include_contract
  --test server_video_sink_smoke_contract
  --test server_video_sinks_include_contract
  --test video_downstream_feed_contract
  --test video_support_contract
)

start_seconds=$(date +%s)
status=0
set +e
{
  echo '==> client camera profile/unit guards'
  cargo test -p lesavka_client --color never input::camera::tests -- --nocapture
  camera_status=${PIPESTATUS[0]}
  echo
  echo '==> media reliability contract tests'
  cargo test -p lesavka_testing --color never "${MEDIA_TESTS[@]}"
  contract_status=${PIPESTATUS[0]}
  if [[ "${camera_status}" -ne 0 || "${contract_status}" -ne 0 ]]; then
    exit 1
  fi
} 2>&1 | tee "${TEST_LOG}"
status=${PIPESTATUS[0]}
set -e
end_seconds=$(date +%s)
duration_seconds=$((end_seconds - start_seconds))

python3 - "${SUMMARY_JSON}" "${SUMMARY_TXT}" "${METRICS_FILE}" "${STATUS_FILE}" "${status}" "${duration_seconds}" "${branch}" "${commit}" "${build_url}" "${REPORT_DIR}" "${SYNC_PROBE_REPORT_JSON}" "${SYNC_PROBE_REPORT_DIR}" "${REQUIRE_SYNC_PROBE}" <<'PY'
import json
import pathlib
import sys
from datetime import datetime, timezone

summary_path = pathlib.Path(sys.argv[1])
text_path = pathlib.Path(sys.argv[2])
metrics_path = pathlib.Path(sys.argv[3])
status_path = pathlib.Path(sys.argv[4])
status = int(sys.argv[5])
duration_seconds = int(sys.argv[6])
branch = sys.argv[7]
commit = sys.argv[8]
build_url = sys.argv[9]
report_dir = pathlib.Path(sys.argv[10])
sync_probe_report_json = sys.argv[11]
sync_probe_report_dir = sys.argv[12]
require_sync_probe = sys.argv[13] == '1'
manual_report = report_dir / 'manual-soak.json'

def esc(value: str) -> str:
    return value.replace('\\', r'\\').replace('\n', r'\n').replace('"', r'\"')

def num(report: dict, path: list[str], default: float = 0.0) -> float:
    current = report
    for key in path:
        if not isinstance(current, dict) or key not in current:
            return default
        current = current[key]
    try:
        return float(current)
    except (TypeError, ValueError):
        return default

def discover_sync_probe_report() -> tuple[pathlib.Path | None, dict | None, str]:
    candidates = []
    if sync_probe_report_json:
        candidates.append(pathlib.Path(sync_probe_report_json))
    if sync_probe_report_dir:
        candidates.append(pathlib.Path(sync_probe_report_dir) / 'report.json')
    candidates.append(report_dir / 'sync-probe' / 'report.json')

    for path in candidates:
        if not path.exists():
            continue
        try:
            return path, json.loads(path.read_text(encoding='utf-8')), ''
        except json.JSONDecodeError as exc:
            return path, None, f'invalid JSON: {exc}'
    return None, None, 'no report.json found; set LESAVKA_SYNC_PROBE_REPORT_JSON or LESAVKA_SYNC_PROBE_REPORT_DIR'

sync_probe_path, sync_probe_report, sync_probe_error = discover_sync_probe_report()
sync_probe_verdict = {}
sync_probe_check_status = 'not_applicable'
sync_probe_why = 'requires real Theia -> Lesavka -> Tethys UVC/UAC hardware evidence'
if sync_probe_report is not None:
    sync_probe_verdict = sync_probe_report.get('verdict', {})
    sync_probe_check_status = 'ok' if bool(sync_probe_verdict.get('passed')) else 'failed'
    sync_probe_why = sync_probe_verdict.get('reason') or 'sync probe report was present'
elif require_sync_probe:
    sync_probe_check_status = 'failed'
    sync_probe_why = sync_probe_error

manual_checks = [
    {
        'name': 'zoom_equivalent_webcam_consumer',
        'status': 'not_applicable' if not manual_report.exists() else 'reported',
        'why': 'requires a real UVC/HDMI consumer such as Zoom, Teams, Slack, or capture software',
    },
    {
        'name': 'ten_minute_soak',
        'status': 'not_applicable' if not manual_report.exists() else 'reported',
        'why': 'requires sustained live hardware output and should be attached as manual-soak.json',
    },
    {
        'name': 'hdmi_capture_adapter_path',
        'status': 'not_applicable' if not manual_report.exists() else 'reported',
        'why': 'requires the Theia HDMI -> UGREEN -> Tethys USB path',
    },
    {
        'name': 'direct_upstream_av_sync_probe',
        'status': sync_probe_check_status,
        'why': sync_probe_why,
    },
]

tracked_signals = [
    'frame_count',
    'effective_fps',
    'dropped_frames',
    'queue_depth',
    'max_inter_frame_gap_ms',
    'decode_errors',
    'appsrc_push_failures',
    'artifact_score',
    'latency_estimate_ms',
    'idr_recovery_after_drop',
    'synthetic_moving_pattern_distortion',
]

final_status = status
if require_sync_probe and sync_probe_check_status != 'ok':
    final_status = 1

summary = {
    'suite': 'lesavka',
    'branch': branch,
    'commit': commit,
    'build_url': build_url,
    'generated_at': datetime.now(timezone.utc).isoformat(),
    'status': 'ok' if final_status == 0 else 'failed',
    'deterministic_status': 'ok' if status == 0 else 'failed',
    'duration_seconds': duration_seconds,
    'deterministic_tests': 'cargo test -p lesavka_testing media reliability contract subset',
    'tracked_media_signals': tracked_signals,
    'manual_checks': manual_checks,
    'sync_probe': {
        'required': require_sync_probe,
        'status': sync_probe_check_status,
        'report_json': '' if sync_probe_path is None else str(sync_probe_path),
        'verdict': sync_probe_verdict,
        'paired_event_count': 0 if sync_probe_report is None else sync_probe_report.get('paired_event_count', 0),
        'median_skew_ms': 0.0 if sync_probe_report is None else sync_probe_report.get('median_skew_ms', 0.0),
        'drift_ms': 0.0 if sync_probe_report is None else sync_probe_report.get('drift_ms', 0.0),
    },
}
summary_path.write_text(json.dumps(summary, indent=2, sort_keys=True) + '\n', encoding='utf-8')
status_path.write_text(str(final_status) + '\n', encoding='utf-8')

lines = [
    'media reliability gate report',
    f'status: {summary["status"]}',
    f'deterministic_status: {summary["deterministic_status"]}',
    f'branch: {branch}',
    f'commit: {commit}',
    f'duration_seconds: {duration_seconds}',
    '',
    'deterministic coverage',
    '- bounded appsrc/appsink queue contracts',
    '- stale-frame/drop-over-latency contracts',
    '- A/V uplink freshness budget contracts',
    '- local monotonic timestamp contracts',
    '- IDR/keyframe recovery contracts',
    '- HDMI/UVC sink construction contracts',
    '- preview telemetry and freeze-signal contracts',
    '',
    'manual hardware evidence slots',
]
for check in manual_checks:
    lines.append(f'- {check["name"]}: {check["status"]} ({check["why"]})')
lines.extend([
    '',
    'sync probe evidence',
    f'- required: {require_sync_probe}',
    f'- status: {sync_probe_check_status}',
    f'- report_json: {summary["sync_probe"]["report_json"] or "none"}',
    f'- p95_abs_skew_ms: {num(sync_probe_report or {}, ["verdict", "p95_abs_skew_ms"]):.1f}',
    f'- max_abs_skew_ms: {num(sync_probe_report or {}, ["max_abs_skew_ms"]):.1f}',
    f'- median_skew_ms: {num(sync_probe_report or {}, ["median_skew_ms"]):.1f}',
    f'- drift_ms: {num(sync_probe_report or {}, ["drift_ms"]):+.1f}',
])
text_path.write_text('\n'.join(lines) + '\n', encoding='utf-8')

labels = f'suite="lesavka",branch="{esc(branch)}",commit="{esc(commit)}"'
ok = 1 if final_status == 0 else 0
failed = 0 if final_status == 0 else 1
probe_ok = 1 if sync_probe_check_status == 'ok' else 0
probe_failed = 1 if sync_probe_check_status == 'failed' else 0
probe_not_applicable = 1 if sync_probe_check_status == 'not_applicable' else 0
metrics = [
    '# HELP platform_quality_gate_checks_total Check outcomes from the latest lesavka gate run.',
    '# TYPE platform_quality_gate_checks_total gauge',
    f'platform_quality_gate_checks_total{{{labels},check="media_reliability",status="ok"}} {ok}',
    f'platform_quality_gate_checks_total{{{labels},check="media_reliability",status="failed"}} {failed}',
    f'platform_quality_gate_checks_total{{{labels},check="sync_probe",status="ok"}} {probe_ok}',
    f'platform_quality_gate_checks_total{{{labels},check="sync_probe",status="failed"}} {probe_failed}',
    f'platform_quality_gate_checks_total{{{labels},check="sync_probe",status="not_applicable"}} {probe_not_applicable}',
    '# HELP lesavka_media_reliability_manual_check_info Manual media reliability evidence slots.',
    '# TYPE lesavka_media_reliability_manual_check_info gauge',
]
for check in manual_checks:
    metrics.append(
        f'lesavka_media_reliability_manual_check_info{{{labels},check="{esc(check["name"])}",status="{esc(check["status"])}"}} 1'
    )
metrics.extend([
    '# HELP lesavka_sync_probe_skew_ms Last direct UVC/UAC sync-probe skew values.',
    '# TYPE lesavka_sync_probe_skew_ms gauge',
    f'lesavka_sync_probe_skew_ms{{{labels},stat="p95_abs"}} {num(sync_probe_report or {}, ["verdict", "p95_abs_skew_ms"]):.3f}',
    f'lesavka_sync_probe_skew_ms{{{labels},stat="max_abs"}} {num(sync_probe_report or {}, ["max_abs_skew_ms"]):.3f}',
    f'lesavka_sync_probe_skew_ms{{{labels},stat="median"}} {num(sync_probe_report or {}, ["median_skew_ms"]):.3f}',
    f'lesavka_sync_probe_skew_ms{{{labels},stat="drift"}} {num(sync_probe_report or {}, ["drift_ms"]):.3f}',
    '# HELP lesavka_sync_probe_events_total Last direct UVC/UAC sync-probe event counts.',
    '# TYPE lesavka_sync_probe_events_total gauge',
    f'lesavka_sync_probe_events_total{{{labels},event_type="paired"}} {int(num(sync_probe_report or {}, ["paired_event_count"]))}',
    f'lesavka_sync_probe_events_total{{{labels},event_type="video"}} {int(num(sync_probe_report or {}, ["video_event_count"]))}',
    f'lesavka_sync_probe_events_total{{{labels},event_type="audio"}} {int(num(sync_probe_report or {}, ["audio_event_count"]))}',
    '# HELP lesavka_sync_probe_verdict_info Last direct UVC/UAC sync-probe verdict.',
    '# TYPE lesavka_sync_probe_verdict_info gauge',
    f'lesavka_sync_probe_verdict_info{{{labels},status="{esc(sync_probe_check_status)}",verdict="{esc(str(sync_probe_verdict.get("status", "")))}",reason="{esc(sync_probe_why)}"}} 1',
])
metrics_path.write_text('\n'.join(metrics) + '\n', encoding='utf-8')
print(text_path.read_text(encoding='utf-8'))
PY

status=$(cat "${STATUS_FILE}")

if [[ -n "${PUSHGATEWAY_URL}" ]]; then
  curl --fail --silent --show-error \
    --data-binary @"${METRICS_FILE}" \
    "${PUSHGATEWAY_URL%/}/metrics/job/${PUSHGATEWAY_JOB}/suite/lesavka" || status=$?
fi

exit "${status}"