Freeze unseen music stress corpus and independently verified gated DTW experiment

tools/speech-vad/README.md

@@ -36,3 +36,36 @@ VAD report. It preserves all five competitors, no-speech status, original source
 intervals and raw ASR logs. `compare.py BASELINE GATED` reports component timing
 and segment counts only; it rejects incomplete or invalid pools and does not
 score retrieval quality. See the protocol document for actual evidence.
+# Separate frozen VAD + DTW comparison
+
+`gated-dtw-policy.json` combines the already frozen Silero source-intersection
+decisions with the separately declared base DTW sentence-point contract.
+Add `--dtw-sentences` to `gated_asr.py` and select a new output, for example
+`.searchi/speech/gated-base-dtw-results-v1.json`. All five source rows remain
+mandatory; explicit no-speech rows use unchanged actual frozen VAD evidence and
+launch no ASR. Neither relevance labels nor the observed malformed piano output
+selects a gate or a window. Original raw Google audio retrieval stays unchanged.
+Original raw recognized segments and ordinary token errors remain recorded.
+Valid DTW points alone determine the separate aligned candidate times; missing
+or invalid points make accepted comparison scoring ineligible.
+
+The combined manifest pins unchanged VAD report/worker, base worker, DTW policy,
+DTW helper and ordinary helper hashes. It preserves the old gated policy in
+`policy`, with separate `alignment_policy` and `combination_policy` provenance.
+Rankers must independently validate both policies and reconstruct every frozen
+VAD window and DTW point before inference/scoring. Existing gated base output
+is not overwritten. A combination plan or structural tests establish no new
+speech recognition or timestamp quality.
+
+Run only after the coordinator freezes the plan and grants the exclusive CPU
+slot. Use `--no-gpu --threads 1 --processors 1 --dtw base --no-flash-attn`; the
+harness adds these explicitly. The twelve existing speech windows remain
+bounded to 30 seconds, 4 MiB raw output and 600 seconds each. Observed gated
+base processing previously took about 65 seconds including VAD cost; DTW cost
+and memory have not been measured. Native DTW workspace defaults to an additional
+128 MiB, which is not a total memory ceiling. Strict parse failures retain bounded
+raw bytes and atomically persist incomplete status, then stop instead of skipping
+a competitor. No full rerun has established this combined policy.
+DTW native logs are retained byte-exact as bounded base64 plus SHA so malformed
+recognized bytes in diagnostic stderr cannot cause lossy decoding. Recognized
+JSON remains strict UTF-8 and cannot be repaired by this log representation.

tools/speech-vad/gated-dtw-policy.json

@@ -0,0 +1,13 @@
+{
+  "schema": 1,
+  "policy": "gated-dtw-v1;frozen-silero-source-intersections;dtw-token-sentences-v1;all-five-sources",
+  "vad_report_sha256": "093be2005b57fe16a8f85161cc9da8966e70e2050cf19fa4122531e304dd70df",
+  "vad_worker_sha256": "28746d770ed470d13475d9474fcd37f28c38c39f27da09219bf9a15eef2adb19",
+  "asr_worker_sha256": "12c681b4bcd9c7679b647b9a041e121a2f9e72cb2a9c6f189f529ec8b79c36f8",
+  "dtw_policy_sha256": "0ea19b1a2ee9da75f0284de33aeca0fcdff0a77d88d232b0b25821a7115b83b5",
+  "dtw_contract_sha256": "61594d0e77c561d5d3b7c9e63178a8123841516b8f8fa781458b21ab717ad47f",
+  "ordinary_contract_sha256": "aac049c5bf3e17b07f4f2f7d92bce18c27aef5a67364181e47815301e75ca520",
+  "sources": ["clip04", "music01", "music02", "sound01", "sound02"],
+  "rules": "Unchanged frozen VAD decisions/windows; explicit no-speech rows launch no ASR; retained original raw audio retrieval is unchanged; no labels select speech intervals or suppress sources",
+  "production_default": false
+}

tools/speech-vad/gated_asr.py

@@ -5,6 +5,8 @@ import json
 import math
 from pathlib import Path
 import tempfile
+import sys
+import base64
 
 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)
@@ -16,6 +18,22 @@ 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"
+HARNESS_SHA_AT_START = speech.sha256(Path(__file__))
+
+
+def dtw_contract():
+    plan_path = Path(__file__).with_name("gated-dtw-policy.json")
+    plan = json.loads(plan_path.read_text(encoding="utf-8"))
+    directory = Path(__file__).parents[1] / "speech-eval"
+    for filename, key in [("dtw-policy.json", "dtw_policy_sha256"),
+                          ("dtw_alignment.py", "dtw_contract_sha256"),
+                          ("token_alignment.py", "ordinary_contract_sha256")]:
+        if speech.sha256(directory / filename) != plan[key]:
+            raise ValueError("Frozen gated DTW contract mismatch")
+    if (plan["vad_report_sha256"] != VAD_SHA or plan["asr_worker_sha256"] != WHISPER_SHA
+            or plan["vad_worker_sha256"] != VAD_WORKER_SHA or plan["sources"] != list(speech.ASSETS)):
+        raise ValueError("Frozen gated DTW provenance mismatch")
+    return plan, speech.sha256(plan_path)
 
 
 def interval_windows(intervals, samples, origin):
@@ -72,7 +90,13 @@ def main():
     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"))
+    parser.add_argument("--dtw-sentences", action="store_true", help="Separate frozen VAD+DTW comparison")
     args = parser.parse_args()
+    if args.dtw_sentences:
+        combined, combined_sha = dtw_contract()
+        sys.path.insert(0, str(Path(__file__).parents[1] / "speech-eval"))
+        from dtw_alignment import aligned_segments, POLICY as DTW_POLICY
+        from token_alignment import aligned_segments as ordinary_segments
     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:
@@ -96,17 +120,23 @@ def main():
               "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__)),
+              "harness_sha256": HARNESS_SHA_AT_START,
               "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": []}
+    if args.dtw_sentences:
+        report.update(alignment_policy=DTW_POLICY,
+            alignment_policy_sha256=combined["dtw_policy_sha256"],
+            alignment_contract_sha256=combined["dtw_contract_sha256"],
+            combination_policy=combined["policy"], combination_policy_sha256=combined_sha,
+            combination_policy_utf8=Path(__file__).with_name("gated-dtw-policy.json").read_bytes().decode("utf-8"))
     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")
+        speech.checkpoint_report(args.output, report)
 
     with tempfile.TemporaryDirectory(prefix="gated-asr-", dir=args.scratch_parent) as directory:
         scratch = Path(directory)
@@ -137,15 +167,44 @@ def main():
                 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"))
+                    "--output-json-full" if args.dtw_sentences else "--output-json",
+                    *(["--dtw", "base", "--no-flash-attn"] if args.dtw_sentences else []),
+                    "--output-file", str(prefix)], scratch, "asr", [(output, speech.MAX_OUTPUT)], 600)
+                try:
+                    raw = speech.parse_asr_output(output, args.scratch_parent)
+                except speech.AsrOutputError as error:
+                    row["windows"].append({**planned, "decoded_frames": frames,
+                        "error_stage": "asr_output_parse", "validation_error": str(error),
+                        "actual_output_failure": error.failure, "segments": []})
+                    report.update(quality_failed=True, comparison_eligible=False,
+                        failure={"asset":asset["id"], "offset_ms":offset, **error.failure})
+                    checkpoint()
+                    raise
                 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")
+                if args.dtw_sentences:
+                    log_bytes = (scratch / "asr.log").read_bytes()
+                    if len(log_bytes) > speech.MAX_OUTPUT:
+                        raise ValueError("ASR log exceeds bound")
+                    window["actual_asr_log_base64"] = base64.b64encode(log_bytes).decode("ascii")
+                    window["actual_asr_log_sha256"] = speech.sha256(scratch / "asr.log")
+                else:
+                    window["actual_asr_log"] = (scratch / "asr.log").read_text(encoding="utf-8")
+                if args.dtw_sentences:
+                    try:
+                        ordinary_segments(raw, offset, length)
+                    except (ValueError, KeyError, TypeError) as error:
+                        window["ordinary_alignment_validation_error"] = str(error)
+                    try:
+                        window["aligned_segments"] = aligned_segments(raw, offset, length)
+                    except (ValueError, KeyError, TypeError) as error:
+                        window["alignment_validation_error"] = str(error)
                 row["windows"].append(window)
                 if "validation_error" in window:
                     report["validation_failures"].append({"asset": asset["id"], "offset_ms": offset, "error": window["validation_error"]})
+                if "alignment_validation_error" in window:
+                    report["validation_failures"].append({"asset": asset["id"], "offset_ms": offset, "error": window["alignment_validation_error"]})
                 checkpoint()
             if source.stat().st_size != asset["bytes"] or speech.sha256(source) != asset["sha256"]:
                 raise ValueError("Source changed during recognition")
@@ -161,6 +220,8 @@ def main():
             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
+    if args.dtw_sentences and dtw_contract()[1] != combined_sha:
+        raise ValueError("Gated DTW contracts changed during recognition")
     report["quality_failed"] = bool(report["validation_failures"])
     report["comparison_eligible"] = not report["quality_failed"]
     checkpoint()

tools/speech-vad/test_gated_asr.py

@@ -3,11 +3,20 @@ import copy
 import json
 from pathlib import Path
 import unittest
+from unittest.mock import patch
 
 import gated_asr as gated
 
 
 class GatedPlanning(unittest.TestCase):
+    def test_frozen_dtw_combination_preserves_sources_and_component_hashes(self):
+        plan, digest = gated.dtw_contract()
+        self.assertEqual(plan["sources"], list(gated.speech.ASSETS))
+        self.assertEqual(plan["vad_report_sha256"], gated.VAD_SHA)
+        self.assertEqual(len(digest), 64)
+        with patch.object(gated.speech, "sha256", return_value="changed"):
+            with self.assertRaises(ValueError):
+                gated.dtw_contract()
     def test_absolute_offset_and_contiguous_split(self):
         interval = gated.vad.source_intersection([10, 6110], 1000000, 5125)
         windows = gated.interval_windows([interval], 1000000, 5125)