| |
| """Replay model probabilities through the production turn controller.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| from pathlib import Path |
|
|
| from turn_detection.runtime.controller import ControllerConfig |
| from turn_detection.runtime.predictor import ModelMetadata |
| from turn_detection.runtime.replay import replay_jsonl |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| DEFAULT_METADATA_CANDIDATES = ( |
| ROOT / "artifacts" / "model_metadata.json", |
| ROOT / "model_metadata.json", |
| ) |
|
|
|
|
| def parse_args() -> argparse.Namespace: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--input", required=True, help="JSONL of PauseCheckpoint fields") |
| parser.add_argument("--output", required=True, help="Destination decision JSONL") |
| parser.add_argument( |
| "--metadata", |
| help="Exported model_metadata.json; packaged layouts are auto-detected", |
| ) |
| parser.add_argument("--threshold", type=float) |
| parser.add_argument("--long-pause-threshold", type=float) |
| parser.add_argument("--min-silence-ms", type=float) |
| parser.add_argument("--relax-after-ms", type=float) |
| parser.add_argument("--max-silence-ms", type=float) |
| parser.add_argument("--confirmations", type=int) |
| parser.add_argument( |
| "--evidence-scope", |
| choices=("synthetic_integration", "unqualified_sequence"), |
| default="unqualified_sequence", |
| help="Label replay provenance; this does not promote a replay to test evidence", |
| ) |
| return parser.parse_args() |
|
|
|
|
| def _metadata_path(value: str | None) -> Path | None: |
| if value: |
| path = Path(value) |
| if not path.is_absolute(): |
| path = ROOT / path |
| if not path.is_file(): |
| raise SystemExit(f"model metadata does not exist: {path}") |
| return path |
| return next((path for path in DEFAULT_METADATA_CANDIDATES if path.is_file()), None) |
|
|
|
|
| def _controller_config(args: argparse.Namespace) -> ControllerConfig: |
| metadata_path = _metadata_path(args.metadata) |
| if metadata_path is not None: |
| base = ModelMetadata.from_path(metadata_path).controller |
| elif args.threshold is not None: |
| base = ControllerConfig( |
| endpoint_threshold=args.threshold, |
| long_pause_threshold=max(0.0, args.threshold - 0.18), |
| ) |
| else: |
| raise SystemExit("provide --metadata or --threshold; refusing an unbound replay policy") |
| endpoint_threshold = base.endpoint_threshold if args.threshold is None else args.threshold |
| relaxation_delta = base.endpoint_threshold - base.long_pause_threshold |
| long_pause_threshold = ( |
| max(0.0, endpoint_threshold - relaxation_delta) |
| if args.long_pause_threshold is None |
| else args.long_pause_threshold |
| ) |
| return ControllerConfig( |
| endpoint_threshold=endpoint_threshold, |
| long_pause_threshold=long_pause_threshold, |
| min_silence_ms=( |
| base.min_silence_ms if args.min_silence_ms is None else args.min_silence_ms |
| ), |
| relax_after_ms=( |
| base.relax_after_ms if args.relax_after_ms is None else args.relax_after_ms |
| ), |
| max_silence_ms=( |
| base.max_silence_ms if args.max_silence_ms is None else args.max_silence_ms |
| ), |
| required_confirmations=( |
| base.required_confirmations if args.confirmations is None else args.confirmations |
| ), |
| ) |
|
|
|
|
| def main() -> None: |
| args = parse_args() |
| config = _controller_config(args) |
| summary = replay_jsonl( |
| args.input, |
| args.output, |
| config, |
| evidence_scope=args.evidence_scope, |
| ) |
| print(json.dumps(summary, indent=2, sort_keys=True)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
