"""Frozen VAD-gated CPU ASR diagnostic; never changes product search.""" import argparse import importlib.util import json import math from pathlib import Path import tempfile spec = importlib.util.spec_from_file_location("gated_speech_contract", Path(__file__).parents[1] / "speech-eval/run.py") speech = importlib.util.module_from_spec(spec) spec.loader.exec_module(speech) vad_spec = importlib.util.spec_from_file_location("gated_vad_contract", Path(__file__).with_name("run.py")) vad = importlib.util.module_from_spec(vad_spec) vad_spec.loader.exec_module(vad) VAD_SHA = "093be2005b57fe16a8f85161cc9da8966e70e2050cf19fa4122531e304dd70df" VAD_WORKER_SHA = "28746d770ed470d13475d9474fcd37f28c38c39f27da09219bf9a15eef2adb19" WHISPER_SHA = "12c681b4bcd9c7679b647b9a041e121a2f9e72cb2a9c6f189f529ec8b79c36f8" POLICY = "gated-vad-v1;source-intersections;integer-ms-tail-remainder-recorded;contiguous30s;independent-process;auto-language;segment-times" def interval_windows(intervals, samples, origin): """Never concatenate intervals or reapply padding; retain sub-ms tail loss.""" result = [] previous = 0 for index, interval in enumerate(intervals): expected = vad.source_intersection(interval["raw_cs"], samples, origin) if interval != expected: raise ValueError("VAD source intersection differs from raw evidence") start, end = interval["source_samples"] if start < previous: raise ValueError("Overlapping or unordered source intersections") previous = end # Raw starts are integer centiseconds; only the final bounded tail can # have a sample fraction that whisper's integer-ms serializer cannot hold. first, last = origin + start // 16, origin + end // 16 for offset in range(first, last, 30000): result.append({"offset_ms": offset, "duration_ms": min(30000, last - offset), "vad_interval_index": index, "vad_source_intersection": interval, "unrepresented_tail_samples": end % 16}) return result def validated_rows(document, fixture): if (document.get("schema") != 1 or document.get("complete") is not True or document.get("fixture_sha256") != speech.FIXTURE_SHA or document.get("model_sha256") != vad.MODEL_SHA or document.get("worker_sha256") != VAD_WORKER_SHA): raise ValueError("Unrecognized frozen VAD provenance") rows = document.get("rows", []) if sorted(row["asset"] for row in rows) != sorted(speech.ASSETS): raise ValueError("Incomplete or duplicate VAD competitor pool") assets = {asset["id"]: asset for asset in fixture["assets"]} for row in rows: if row["source_sha256"] != assets[row["asset"]]["sha256"]: raise ValueError("VAD source hash mismatch") native = row["native"] start, end, _ = speech.primary_audio_range(row["probe"]) samples = native["samples"] probabilities = native["probabilities"] if (type(samples) is not int or samples <= 0 or abs(samples - (end-start)*16) > 16 or native["sample_rate"] != 16000 or native["threads"] != 1 or native["gpu"] is not False or len(probabilities) != (samples + 511) // 512 or any(type(p) not in (int, float) or not math.isfinite(p) or not 0 <= p <= 1 for p in probabilities) or [interval["raw_cs"] for interval in row["intervals"]] != native["intervals_cs"]): raise ValueError("Invalid VAD native coverage or CPU contract") interval_windows(row["intervals"], samples, start) return rows def main(): parser = argparse.ArgumentParser(description=__doc__) for option in ("whisper", "model", "ffmpeg", "ffprobe", "vad-report", "scratch-parent", "output"): parser.add_argument("--" + option, type=Path, required=True) parser.add_argument("--fixture", type=Path, default=Path("fixtures/media/fixture.json")) args = parser.parse_args() if args.output.exists(): raise ValueError("Never overwrite experiment evidence") if speech.sha256(args.vad_report) != VAD_SHA or speech.sha256(args.fixture) != speech.FIXTURE_SHA: raise ValueError("Frozen VAD/fixture SHA mismatch") size, model_sha = speech.MODELS["base"] if args.model.stat().st_size != size or speech.sha256(args.model) != model_sha or speech.sha256(args.whisper) != WHISPER_SHA: raise ValueError("Frozen ASR worker/model mismatch") if args.vad_report.stat().st_size > 16 * 1024 * 1024: raise ValueError("Oversized VAD evidence") vad_bytes = args.vad_report.read_bytes() document = json.loads(vad_bytes) for name in ("ffmpeg", "ffprobe"): if speech.sha256(getattr(args, name)) != document[name + "_sha256"]: raise ValueError("Decoder provenance differs from frozen VAD") fixture = json.loads(args.fixture.read_text(encoding="utf-8")) vad_rows = validated_rows(document, fixture) assets = {asset["id"]: asset for asset in fixture["assets"]} report = {"version": 1, "complete": False, "quality_failed": False, "validation_failures": [], "comparison_eligible": False, "fixture_sha256": speech.FIXTURE_SHA, "upstream_commit": speech.UPSTREAM, "candidate": "base", "model_sha256": model_sha, "worker_sha256": WHISPER_SHA, "backend": "forced CPU", "threads": 1, "policy": POLICY, "vad_report_sha256": VAD_SHA, "actual_vad_report": document, "vad_report_utf8": vad_bytes.decode("utf-8"), "harness_sha256": speech.sha256(Path(__file__)), "speech_contract_sha256": speech.sha256(Path(spec.origin)), "ffmpeg_sha256": document["ffmpeg_sha256"], "ffprobe_sha256": document["ffprobe_sha256"], "scope": "Gated ASR diagnostic; no product or word-alignment claim", "rows": []} args.output.parent.mkdir(parents=True, exist_ok=True) args.scratch_parent.mkdir(parents=True, exist_ok=True) with args.output.open("x", encoding="utf-8") as stream: json.dump(report, stream) def checkpoint(): args.output.write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") with tempfile.TemporaryDirectory(prefix="gated-asr-", dir=args.scratch_parent) as directory: scratch = Path(directory) for detected in vad_rows: asset = assets[detected["asset"]] source = args.fixture.parent / asset["path"] if source.stat().st_size != asset["bytes"] or speech.sha256(source) != asset["sha256"]: raise ValueError("Frozen source mismatch") speech.run_owned([str(args.ffprobe.resolve()), "-v", "error", "-protocol_whitelist", "file,pipe", "-show_entries", "stream=index,codec_type,duration,start_time", "-of", "json", str(source.resolve())], scratch, "probe", [], 30) probe = json.loads((scratch / "probe.log").read_text(encoding="utf-8")) if probe != detected["probe"]: raise ValueError("Audio probe changed after VAD") start, end, index = speech.primary_audio_range(probe) windows = interval_windows(detected["intervals"], detected["native"]["samples"], start) row = {"asset": asset["id"], "source_sha256": asset["sha256"], "audio_start_ms": start, "audio_end_ms": end, "duration_ms": end-start, "audio_stream_index": index, "actual_probe": probe, "status": "speech" if windows else "no_speech", "asr_not_run": not windows, "windows": []} report["rows"].append(row) checkpoint() for planned in windows: wav, prefix = scratch / "window.wav", scratch / "recognized" output = prefix.with_suffix(".json") output.unlink(missing_ok=True) offset, length = planned["offset_ms"], planned["duration_ms"] decode_ms = speech.run_owned(speech.decode_command(args.ffmpeg, source, offset, length, wav, index), scratch, "decode", [(wav, 970000)], 60) frames = speech.decoded_frames(wav, length) asr_ms = speech.run_owned([str(args.whisper.resolve()), "--no-gpu", "--threads", "1", "--processors", "1", "--language", "auto", "--model", str(args.model.resolve()), "--file", str(wav), "--output-json", "--output-file", str(prefix)], scratch, "asr", [(output, speech.MAX_OUTPUT)], 600) raw = json.loads(output.read_text(encoding="utf-8")) window = speech.recognized_window(raw, offset, length, decode_ms, asr_ms) window.update(planned) window["decoded_frames"] = frames window["actual_asr_log"] = (scratch / "asr.log").read_text(encoding="utf-8") row["windows"].append(window) if "validation_error" in window: report["validation_failures"].append({"asset": asset["id"], "offset_ms": offset, "error": window["validation_error"]}) checkpoint() if source.stat().st_size != asset["bytes"] or speech.sha256(source) != asset["sha256"]: raise ValueError("Source changed during recognition") # Reject changed inputs/tools across the entire experiment, including earlier # no-speech rows whose data was verified before later recognition work. for detected in vad_rows: asset = assets[detected["asset"]] source = args.fixture.parent / asset["path"] if source.stat().st_size != asset["bytes"] or speech.sha256(source) != asset["sha256"]: raise ValueError("Source changed before comparison completion") if (speech.sha256(args.vad_report) != VAD_SHA or speech.sha256(args.fixture) != speech.FIXTURE_SHA or speech.sha256(args.whisper) != WHISPER_SHA or speech.sha256(args.model) != model_sha or any(speech.sha256(getattr(args, name)) != document[name + "_sha256"] for name in ("ffmpeg", "ffprobe"))): raise ValueError("Experiment tools/provenance changed during recognition") report["complete"] = True report["quality_failed"] = bool(report["validation_failures"]) report["comparison_eligible"] = not report["quality_failed"] checkpoint() if __name__ == "__main__": main()