Validate complete speech decode windows and retain rejected ASR output

tools/speech-eval/README.md

@@ -26,7 +26,11 @@ does not implement overlap/dedup or production budgets. Segment JSON offsets are
 already milliseconds and get the window offset added exactly once.
 
 Each completed window writes an incomplete checkpoint; the final report marks
-complete. Errors retain bounded owned logs outside temporary scratch, which cleans
+complete evaluation, with `quality_failed` and explicit `validation_failures` for
+invalid ASR timestamps. Bounded raw ASR JSON remains in each checkpoint before
+validity is decided; invalid times are never clamped into evidence, and later
+windows continue. The final report records the executed harness SHA. Process
+errors retain bounded owned logs and exit/time diagnostics outside temporary scratch, which cleans
 decoded WAV/JSON on exit. Cancellation terminates only owned subprocess trees.
 Limits are supervisory checks, not OS resource sandbox guarantees. Recognized text
 is private local diagnostic data; never commit raw logs/transcripts by default.

tools/speech-eval/run.py

@@ -97,6 +97,36 @@ def segments(document, offset_ms, duration_ms):
     return parsed
 
 
+def recognized_window(raw, offset, length, decode_ms, asr_ms):
+    report = {"offset_ms": offset, "duration_ms": length, "decode_ms": decode_ms,
+              "asr_process_ms": asr_ms, "actual_whisper_json": raw}
+    try:
+        report["segments"] = segments(raw, offset, length)
+    except (ValueError, KeyError, TypeError) as error:
+        report["segments"] = []
+        report["validation_error"] = str(error)
+    return report
+
+
+def decode_command(ffmpeg, source, offset, length, wav):
+    return [str(ffmpeg.resolve()), "-nostdin", "-hide_banner", "-v", "error",
+            "-protocol_whitelist", "file,pipe", "-threads", "1", "-ss", str(offset / 1000), "-t", str(length / 1000),
+            "-i", str(source.resolve()), "-map", "0:a:0", "-vn", "-ac", "1", "-ar", "16000",
+            "-af", f"aresample=16000,atrim=end_sample={length * 16},asetpts=PTS-STARTPTS",
+            "-c:a", "pcm_s16le", "-threads", "1", "-y", str(wav)]
+
+
+def decoded_frames(wav, length):
+    with wave.open(str(wav), "rb") as audio:
+        if audio.getframerate() != 16000 or audio.getnchannels() != 1 or audio.getsampwidth() != 2:
+            raise ValueError("Invalid decoded PCM")
+        frames = audio.getnframes()
+    # Container duration rounding allows at most one millisecond's shortfall.
+    if not max(1, length * 16 - 16) <= frames <= length * 16:
+        raise ValueError(f"Decoded sample coverage mismatch: expected {length * 16}, got {frames}")
+    return frames
+
+
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--whisper", type=Path, required=True)
@@ -145,20 +175,16 @@ def main():
                 prefix = scratch / "recognized"
                 output = prefix.with_suffix(".json")
                 output.unlink(missing_ok=True)
-                decode_ms = run_owned([str(args.ffmpeg.resolve()), "-nostdin", "-hide_banner", "-v", "error",
-                    "-protocol_whitelist", "file,pipe", "-threads", "1", "-ss", str(offset / 1000), "-t", str(length / 1000),
-                    "-i", str(source.resolve()), "-map", "0:a:0", "-vn", "-ac", "1", "-ar", "16000",
-                    "-af", f"atrim=end_sample={length * 16}", "-c:a", "pcm_s16le", "-threads", "1", "-y", str(wav)],
+                decode_ms = run_owned(decode_command(args.ffmpeg, source, offset, length, wav),
                     scratch, "decode", [(wav, 970000)], 60)
-                with wave.open(str(wav), "rb") as audio:
-                    if audio.getframerate() != 16000 or audio.getnchannels() != 1 or audio.getsampwidth() != 2:
-                        raise ValueError("Invalid decoded PCM")
+                frames = decoded_frames(wav, length)
                 asr_ms = 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, MAX_OUTPUT)], 600)
                 raw = json.loads(output.read_text(encoding="utf-8"))
-                result["windows"].append({"offset_ms": offset, "duration_ms": length, "decode_ms": decode_ms,
-                    "asr_process_ms": asr_ms, "segments": segments(raw, offset, length), "actual_whisper_json": raw})
+                # Preserve genuine invalid output, never clamp it into valid evidence.
+                result["windows"].append(recognized_window(raw, offset, length, decode_ms, asr_ms))
+                result["windows"][-1]["decoded_frames"] = frames
                 args.output.parent.mkdir(parents=True, exist_ok=True)
                 checkpoint = {"complete": False, "fixture_sha256": sha256(args.fixture),
                               "model_sha256": digest, "worker_sha256": args.whisper_sha256,
@@ -167,12 +193,16 @@ def main():
             if source.stat().st_size != asset["bytes"] or sha256(source) != asset["sha256"]:
                 raise ValueError("Source changed during recognition")
             rows.append(result)
-    report = {"version": 1, "complete": True, "scope": "Actual ASR diagnostic, not Google retrieval or production integration",
+    failures = [{"asset": row["asset"], "offset_ms": window["offset_ms"], "error": window["validation_error"]}
+                for row in rows for window in row["windows"] if "validation_error" in window]
+    report = {"version": 1, "complete": True, "quality_failed": bool(failures), "validation_failures": failures,
+              "scope": "Actual ASR diagnostic, not Google retrieval or production integration",
               "fixture_sha256": sha256(args.fixture), "upstream_commit": UPSTREAM,
               "worker_sha256": args.whisper_sha256, "model_sha256": digest, "candidate": args.candidate,
               "ffmpeg_sha256": sha256(args.ffmpeg), "ffprobe_sha256": sha256(args.ffprobe),
               "backend": "forced CPU", "threads": 1,
-              "policy": "nonoverlapping30s;independent-process-per-window;auto-language;segment-times",
+              "harness_sha256": sha256(Path(__file__)),
+              "policy": "aresample16000-before-atrim-v2;samplecoverage-1ms;nonoverlapping30s;independent-process-per-window;auto-language;segment-times",
               "rows": rows}
     args.output.parent.mkdir(parents=True, exist_ok=True)
     args.output.write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")

tools/speech-eval/test_contract.py

@@ -4,11 +4,53 @@ from pathlib import Path
 import unittest
 import tempfile
 import sys
-from run import segments, run_owned
+import os
+import wave
+from run import segments, run_owned, recognized_window, decode_command, decoded_frames
 from score import score
 
 
 class ContractTests(unittest.TestCase):
+    @unittest.skipUnless(os.environ.get("SEARCHI_TEST_FFMPEG"), "Requires verified CPU FFmpeg")
+    def test_actual_ffmpeg_full_window_and_partial_tail_sample_coverage(self):
+        ffmpeg = Path(os.environ["SEARCHI_TEST_FFMPEG"])
+        with tempfile.TemporaryDirectory() as parent:
+            scratch = Path(parent)
+            for rate in [44100, 48000]:
+                with self.subTest(rate=rate):
+                    source = scratch / f"source-{rate}.wav"
+                    with wave.open(str(source), "wb") as audio:
+                        audio.setnchannels(1)
+                        audio.setsampwidth(2)
+                        audio.setframerate(rate)
+                        audio.writeframes(b"\x00\x00" * (rate * 31))
+                    for offset, length, expected in [(0, 30000, 480000), (30000, 1000, 16000)]:
+                        output = scratch / "decoded.wav"
+                        run_owned(decode_command(ffmpeg, source, offset, length, output), scratch,
+                                  "samplecoverage", [(output, 970000)], 30)
+                        self.assertEqual(decoded_frames(output, length), expected)
+
+    def test_threefold_truncation_rejected(self):
+        with tempfile.TemporaryDirectory() as parent:
+            path = Path(parent) / "truncated.wav"
+            with wave.open(str(path), "wb") as audio:
+                audio.setnchannels(1)
+                audio.setsampwidth(2)
+                audio.setframerate(16000)
+                audio.writeframes(b"\x00\x00" * 160000)
+            with self.assertRaises(ValueError):
+                decoded_frames(path, 30000)
+
+    def test_invalid_actual_json_is_retained_and_next_window_parses(self):
+        invalid = {"transcription": [{"offsets": {"from": 0, "to": 30010}, "text": "synthetic test"}]}
+        valid = {"transcription": [{"offsets": {"from": 0, "to": 10}, "text": "synthetic test"}]}
+        rows = [recognized_window(raw, offset, 30000, 1, 2) for raw, offset in [(invalid, 0), (valid, 30000)]]
+        self.assertEqual(rows[0]["actual_whisper_json"], invalid)
+        self.assertEqual(rows[0]["segments"], [])
+        self.assertIn("validation_error", rows[0])
+        self.assertNotIn("validation_error", rows[1])
+        self.assertEqual(rows[1]["segments"][0]["start_ms"], 30000)
+
     def test_timeout_and_output_overflow_terminate_owned_child(self):
         for name, code, error, timeout in [
             ("timeout", "import time; time.sleep(10)", TimeoutError, 0.1),