Retain bounded invalid ASR output and atomic failure evidence

tools/speech-eval/run.py

@@ -31,6 +31,50 @@ def sha256(path):
     return digest.hexdigest()
 
 
+class AsrOutputError(ValueError):
+    """Bounded raw output retained; callers must keep coverage incomplete."""
+    def __init__(self, failure):
+        super().__init__("ASR output failed strict UTF-8/JSON parsing; raw evidence retained")
+        self.failure = failure
+
+
+def checkpoint_report(path, document):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary_path = None
+    try:
+        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=path.parent,
+                                         prefix="speech-checkpoint-", suffix=".tmp", delete=False) as temporary:
+            temporary_path = Path(temporary.name)
+            json.dump(document, temporary, indent=2, ensure_ascii=False)
+            temporary.write("\n")
+            temporary.flush()
+            os.fsync(temporary.fileno())
+        os.replace(temporary_path, path)
+    finally:
+        if temporary_path is not None:
+            temporary_path.unlink(missing_ok=True)
+
+
+def parse_asr_output(path, retained_parent):
+    with path.open("rb") as stream:
+        raw = stream.read(MAX_OUTPUT + 1)
+    if len(raw) > MAX_OUTPUT:
+        raise ValueError("ASR output exceeds parse budget")
+    try:
+        return json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as error:
+        with tempfile.NamedTemporaryFile(prefix="speech-asr-output-failure-", suffix=".bin",
+                                         dir=retained_parent, delete=False) as retained:
+            retained.write(raw)
+        failure = {"error_type": type(error).__name__, "raw_output_path": retained.name,
+                   "raw_output_bytes": len(raw), "raw_output_sha256": hashlib.sha256(raw).hexdigest()}
+        if isinstance(error, UnicodeDecodeError):
+            failure.update(byte_start=error.start, byte_end=error.end, reason=error.reason)
+        else:
+            failure.update(character_position=error.pos, line=error.lineno, column=error.colno)
+        raise AsrOutputError(failure) from error
+
+
 def stop(child):
     if os.name == "nt":
         subprocess.run(["taskkill", "/PID", str(child.pid), "/T", "/F"],
@@ -228,7 +272,25 @@ def main():
                         "--output-json-full" if alignment_mode else "--output-json",
                         *(["--dtw", "base", "--no-flash-attn"] if args.dtw_sentences else []),
                         "--output-file", str(prefix)], scratch, "asr", [(output, MAX_OUTPUT)], 600)
-                    raw = json.loads(output.read_text(encoding="utf-8"))
+                    try:
+                        raw = parse_asr_output(output, args.scratch_parent)
+                    except AsrOutputError as error:
+                        result["windows"].append({"offset_ms": offset, "duration_ms": length,
+                            "decode_ms": decode_ms, "asr_process_ms": asr_ms,
+                            "decoded_frames": frames, "segments": [],
+                            "error_stage": "asr_output_parse", "validation_error": str(error),
+                            "actual_output_failure": error.failure})
+                        failed = {"complete": False, "quality_failed": True,
+                            "comparison_eligible": False, "fixture_sha256": sha256(args.fixture),
+                            "model_sha256": digest, "worker_sha256": args.whisper_sha256,
+                            "harness_sha256": HARNESS_SHA256_AT_START, "rows": [*rows, result],
+                            "failure": {"asset": asset["id"], "offset_ms": offset, **error.failure}}
+                        if alignment_mode:
+                            failed.update(alignment_policy=policy["policy"],
+                                alignment_policy_sha256=alignment_policy_sha,
+                                alignment_contract_sha256=alignment_contract_sha)
+                        checkpoint_report(args.output, failed)
+                        raise
                     result["windows"].append(recognized_window(raw, offset, length, decode_ms, asr_ms))
                     result["windows"][-1]["decoded_frames"] = frames
                     if alignment_mode:
@@ -250,7 +312,7 @@ def main():
                     checkpoint["alignment_policy"] = policy["policy"]
                     checkpoint["alignment_policy_sha256"] = alignment_policy_sha
                     checkpoint["alignment_contract_sha256"] = alignment_contract_sha
-                args.output.write_text(json.dumps(checkpoint, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+                checkpoint_report(args.output, checkpoint)
             if source.stat().st_size != asset["bytes"] or sha256(source) != asset["sha256"]:
                 raise ValueError("Source changed during recognition")
             rows.append(result)
@@ -275,7 +337,7 @@ def main():
         report["alignment_contract_sha256"] = alignment_contract_sha
         report["alignment_policy_sha256"] = alignment_policy_sha
     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")
+    checkpoint_report(args.output, report)
 
 
 if __name__ == "__main__":