src/turn_detection/runtime/replay.py

"""Continuous checkpoint replay for product-level turn metrics."""

from __future__ import annotations

import hashlib
import json
import math
from collections.abc import Iterable
from dataclasses import asdict, dataclass
from pathlib import Path
from statistics import median
from typing import Any

from .controller import ControllerConfig, TurnController
from .types import Prediction, TurnDecision

_EVIDENCE_NOTES = {
    "synthetic_integration": (
        "Synthetic controller integration fixture only: probabilities and targets are "
        "hand-authored and latency values are synthetic. This is not model, conversation, "
        "latency, or product-quality evidence."
    ),
    "unqualified_sequence": (
        "User-supplied sequence replay; source provenance and annotation quality are not "
        "verified by this tool."
    ),
}


@dataclass(frozen=True, slots=True)
class PauseCheckpoint:
    turn_id: str
    timestamp_ms: float
    silence_ms: float
    endpoint_probability: float
    target_endpoint: bool
    inference_ms: float = 0.0


@dataclass(frozen=True, slots=True)
class ReplayRecord:
    checkpoint: PauseCheckpoint
    decision: TurnDecision


def replay_checkpoints(
    checkpoints: Iterable[PauseCheckpoint],
    config: ControllerConfig | None = None,
) -> list[ReplayRecord]:
    """Replay ordered VAD checkpoints, resetting state between user turns."""

    controller = TurnController(config)
    records: list[ReplayRecord] = []
    current_turn: str | None = None
    last_timestamp = -math.inf
    for checkpoint in checkpoints:
        if checkpoint.turn_id != current_turn:
            controller.reset()
            current_turn = checkpoint.turn_id
            last_timestamp = -math.inf
        if checkpoint.timestamp_ms < last_timestamp:
            raise ValueError(f"checkpoints are not ordered within turn {checkpoint.turn_id!r}")
        last_timestamp = checkpoint.timestamp_ms
        prediction = Prediction(
            endpoint_probability=checkpoint.endpoint_probability,
            inference_ms=checkpoint.inference_ms,
            model_name="replay",
        )
        decision = controller.evaluate_pause(
            prediction,
            silence_ms=checkpoint.silence_ms,
            timestamp_ms=checkpoint.timestamp_ms,
        )
        records.append(ReplayRecord(checkpoint=checkpoint, decision=decision))
    return records


def _percentile(values: list[float], quantile: float) -> float | None:
    if not values:
        return None
    ordered = sorted(values)
    location = (len(ordered) - 1) * quantile
    lower = math.floor(location)
    upper = math.ceil(location)
    if lower == upper:
        return ordered[lower]
    weight = location - lower
    return ordered[lower] * (1.0 - weight) + ordered[upper] * weight


def _file_evidence(path: Path) -> dict[str, int | str]:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(block)
    return {"bytes": path.stat().st_size, "sha256": digest.hexdigest()}


def summarize_replay(records: Iterable[ReplayRecord]) -> dict[str, float | int | None]:
    """Compute checkpoint and per-turn interruption/latency metrics."""

    materialized = list(records)
    tp = fp = tn = fn = 0
    turns: dict[str, list[ReplayRecord]] = {}
    for record in materialized:
        predicted = record.decision.emit_response
        actual = record.checkpoint.target_endpoint
        tp += int(predicted and actual)
        fp += int(predicted and not actual)
        tn += int(not predicted and not actual)
        fn += int(not predicted and actual)
        turns.setdefault(record.checkpoint.turn_id, []).append(record)

    interrupted_turns = 0
    detected_turns = 0
    delays: list[float] = []
    for turn_records in turns.values():
        if any(
            item.decision.emit_response and not item.checkpoint.target_endpoint
            for item in turn_records
        ):
            interrupted_turns += 1
        true_endpoint_times = [
            item.checkpoint.timestamp_ms for item in turn_records if item.checkpoint.target_endpoint
        ]
        if not true_endpoint_times:
            continue
        true_time = min(true_endpoint_times)
        end_times = [
            item.checkpoint.timestamp_ms
            for item in turn_records
            if item.decision.emit_response and item.checkpoint.timestamp_ms >= true_time
        ]
        if end_times:
            detected_turns += 1
            delays.append(min(end_times) - true_time)

    false_interruption_rate = fp / (fp + tn) if fp + tn else None
    missed_end_rate = fn / (fn + tp) if fn + tp else None
    turn_count = len(turns)
    endpoint_turn_count = sum(
        1 for values in turns.values() if any(item.checkpoint.target_endpoint for item in values)
    )
    emissions_by_turn = [
        sum(item.decision.emit_response for item in values) for values in turns.values()
    ]
    return {
        "checkpoints": len(materialized),
        "turns": turn_count,
        "true_positives": tp,
        "false_positives": fp,
        "true_negatives": tn,
        "false_negatives": fn,
        "false_interruption_rate": false_interruption_rate,
        "missed_end_rate": missed_end_rate,
        "turns_with_premature_interruption": interrupted_turns,
        "response_emissions": sum(emissions_by_turn),
        "duplicate_response_emissions": sum(max(0, count - 1) for count in emissions_by_turn),
        "premature_interruption_probability_per_turn": (
            interrupted_turns / turn_count if turn_count else None
        ),
        "endpoint_turn_detection_rate": (
            detected_turns / endpoint_turn_count if endpoint_turn_count else None
        ),
        "endpoint_delay_p50_ms": median(delays) if delays else None,
        "endpoint_delay_p90_ms": _percentile(delays, 0.90),
        "endpoint_delay_p95_ms": _percentile(delays, 0.95),
    }


def replay_jsonl(
    source: str | Path,
    destination: str | Path,
    config: ControllerConfig | None = None,
    *,
    evidence_scope: str = "unqualified_sequence",
) -> dict[str, Any]:
    if evidence_scope not in _EVIDENCE_NOTES:
        raise ValueError(f"unsupported replay evidence scope: {evidence_scope!r}")
    source_path = Path(source)
    checkpoints = []
    with source_path.open(encoding="utf-8") as handle:
        for line_number, line in enumerate(handle, 1):
            if not line.strip():
                continue
            try:
                checkpoints.append(PauseCheckpoint(**json.loads(line)))
            except (TypeError, ValueError, json.JSONDecodeError) as exc:
                raise ValueError(f"invalid checkpoint JSON on line {line_number}") from exc
    resolved_config = config or ControllerConfig()
    records = replay_checkpoints(checkpoints, resolved_config)
    output = Path(destination)
    output.parent.mkdir(parents=True, exist_ok=True)
    with output.open("w", encoding="utf-8") as handle:
        for record in records:
            handle.write(
                json.dumps(
                    {
                        "checkpoint": asdict(record.checkpoint),
                        "decision": asdict(record.decision),
                    },
                    default=lambda value: value.value,
                    sort_keys=True,
                )
                + "\n"
            )
    summary = summarize_replay(records)
    summary["format_version"] = 1
    summary["controller_config"] = asdict(resolved_config)
    summary["evidence_scope"] = evidence_scope
    summary["evidence_note"] = _EVIDENCE_NOTES[evidence_scope]
    summary["input"] = _file_evidence(source_path)
    summary["decisions"] = _file_evidence(output)
    output.with_suffix(".summary.json").write_text(
        json.dumps(summary, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )
    return summary