File size: 4,743 Bytes
35d483e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | 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()
|