| from __future__ import annotations |
|
|
| import json |
| import tempfile |
| import unittest |
| from pathlib import Path |
| from types import SimpleNamespace |
|
|
| from scripts.replay_stream import _controller_config |
| from turn_detection.runtime.controller import ControllerConfig |
| from turn_detection.runtime.predictor import ModelMetadata |
| from turn_detection.runtime.replay import ( |
| PauseCheckpoint, |
| replay_checkpoints, |
| replay_jsonl, |
| summarize_replay, |
| ) |
|
|
|
|
| class ReplayTest(unittest.TestCase): |
| def test_replay_policy_loads_exported_model_metadata(self) -> None: |
| metadata = ModelMetadata( |
| model_name="preview", |
| architecture="tiny_tcn", |
| threshold=0.74, |
| controller=ControllerConfig( |
| endpoint_threshold=0.74, |
| long_pause_threshold=0.56, |
| min_silence_ms=250, |
| relax_after_ms=900, |
| max_silence_ms=2_000, |
| required_confirmations=2, |
| ), |
| ) |
| with tempfile.TemporaryDirectory() as directory: |
| path = Path(directory) / "model_metadata.json" |
| path.write_text(json.dumps(metadata.to_dict()), encoding="utf-8") |
| args = SimpleNamespace( |
| metadata=str(path), |
| threshold=None, |
| long_pause_threshold=None, |
| min_silence_ms=None, |
| relax_after_ms=None, |
| max_silence_ms=None, |
| confirmations=None, |
| ) |
| loaded = _controller_config(args) |
| self.assertEqual(loaded, metadata.controller) |
|
|
| args.threshold = 0.80 |
| overridden = _controller_config(args) |
| self.assertAlmostEqual(overridden.endpoint_threshold, 0.80) |
| self.assertAlmostEqual(overridden.long_pause_threshold, 0.62) |
|
|
| def test_product_metrics_count_interruptions_and_delay(self) -> None: |
| checkpoints = [ |
| PauseCheckpoint("a", 300, 300, 0.9, False), |
| PauseCheckpoint("a", 1000, 300, 0.9, True), |
| PauseCheckpoint("b", 200, 200, 0.1, False), |
| PauseCheckpoint("b", 900, 300, 0.9, True), |
| ] |
| records = replay_checkpoints( |
| checkpoints, |
| ControllerConfig(endpoint_threshold=0.6, required_confirmations=1), |
| ) |
| metrics = summarize_replay(records) |
| self.assertEqual(metrics["turns"], 2) |
| self.assertEqual(metrics["turns_with_premature_interruption"], 1) |
| self.assertAlmostEqual(metrics["premature_interruption_probability_per_turn"], 0.5) |
| self.assertEqual(metrics["endpoint_turn_detection_rate"], 0.5) |
| self.assertEqual(metrics["endpoint_delay_p50_ms"], 0.0) |
| self.assertEqual(metrics["response_emissions"], 2) |
| self.assertEqual(metrics["duplicate_response_emissions"], 0) |
|
|
| def test_order_is_validated_per_turn(self) -> None: |
| checkpoints = [ |
| PauseCheckpoint("a", 400, 300, 0.5, False), |
| PauseCheckpoint("a", 300, 400, 0.5, True), |
| ] |
| with self.assertRaises(ValueError): |
| replay_checkpoints(checkpoints) |
|
|
| def test_jsonl_summary_binds_files_and_labels_synthetic_scope(self) -> None: |
| with tempfile.TemporaryDirectory() as directory: |
| source = Path(directory) / "fixture.jsonl" |
| output = Path(directory) / "decisions.jsonl" |
| source.write_text( |
| json.dumps( |
| { |
| "turn_id": "synthetic-a", |
| "timestamp_ms": 300, |
| "silence_ms": 300, |
| "endpoint_probability": 0.9, |
| "target_endpoint": True, |
| } |
| ) |
| + "\n", |
| encoding="utf-8", |
| ) |
| summary = replay_jsonl( |
| source, |
| output, |
| ControllerConfig(endpoint_threshold=0.6), |
| evidence_scope="synthetic_integration", |
| ) |
| self.assertEqual(summary["evidence_scope"], "synthetic_integration") |
| self.assertIn("not model", summary["evidence_note"]) |
| self.assertEqual(summary["input"]["bytes"], source.stat().st_size) |
| self.assertEqual(summary["decisions"]["bytes"], output.stat().st_size) |
| self.assertEqual(len(summary["input"]["sha256"]), 64) |
| persisted = json.loads(output.with_suffix(".summary.json").read_text()) |
| self.assertEqual(persisted, summary) |
|
|
| with self.assertRaisesRegex(ValueError, "unsupported replay evidence scope"): |
| replay_jsonl(source, output, evidence_scope="official_test") |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
