diff --git a/src/eva/assistant/pipeline/observers.py b/src/eva/assistant/pipeline/observers.py
index 5eaee483..57afb3a8 100644
--- a/src/eva/assistant/pipeline/observers.py
+++ b/src/eva/assistant/pipeline/observers.py
@@ -296,11 +296,14 @@ async def on_push_frame(self, data: FramePushed):
         if not isinstance(frame, MetricsFrame):
             return
 
-        # Deduplicate based on frame ID
-        frame_id = id(frame)
-        if frame_id in self._frames_seen:
+        # Deduplicate on the frame's own id, which pipecat assigns from a
+        # process-global monotonic counter. The same frame is observed once per
+        # pipeline hop, so dedup is required; id(frame) is not usable as the key
+        # because CPython reuses the memory address of a collected frame, which
+        # silently drops later, unrelated MetricsFrames.
+        if frame.id in self._frames_seen:
             return
-        self._frames_seen.add(frame_id)
+        self._frames_seen.add(frame.id)
 
         timestamp = self.clock.to_wall_time(data.timestamp)
 
diff --git a/src/eva/metrics/experience/turn_taking.py b/src/eva/metrics/experience/turn_taking.py
index 912c9e2c..059e67ce 100644
--- a/src/eva/metrics/experience/turn_taking.py
+++ b/src/eva/metrics/experience/turn_taking.py
@@ -41,6 +41,23 @@
                         computed from audit_log.json directly so it works uniformly across
                         cascade/S2S/audio-LLM; omitted when there are no tool calls or the
                         audit log is unavailable — see ``_compute_pre_tool_speech_groups``)
+  VAD turn diagnostics: vad_stuck_rate (fraction of classified turns where the turn analyzer
+                        never produced a natural or forced completion signal at all —
+                        includes both non-fatal mid-conversation hangs and, when it's also
+                        why the conversation died via inactivity_timeout, a trailing turn
+                        the analyzer never even opened a VAD-start window for),
+                        forced_completion_rate
+                        (Smart Turn's silence fallback firing rate; null for Krisp — the
+                        mechanism doesn't exist), forced_completion_final_turn_{short,
+                        acknowledgement,spelled_entity}_rate (among forced completions, how
+                        often the STT transcript matched to that window has that input shape
+                        — a candidate explanation for why the turn analyzer needed the
+                        stop_secs fallback; omitted when there are no forced completions with
+                        a matched transcript), mean/p50/p90_time_to_complete_ms
+                        (natural completions only).
+                        All null/omitted for runs with no local turn-analyzer VAD
+                        (non-pipecat frameworks, or turn_stop_strategy == "external").
+                        See src/eva/metrics/experience/vad_end_of_turn.py.
 
 All reported sub-metrics are consistent with the main score: ``mean_overlap_score``,
 ``mean_count_score``, and ``mean_yield_score`` aggregate exactly the per-turn scores
@@ -53,6 +70,7 @@
 from typing import Any
 
 from eva.metrics.base import CodeMetric, MetricContext
+from eva.metrics.experience.vad_end_of_turn import compute_vad_turn_sub_metrics
 from eva.metrics.processor import is_agent_timeout_on_user_turn
 from eva.metrics.registry import register_metric
 from eva.metrics.utils import mean_agent_perf_stat
@@ -67,7 +85,7 @@ class TurnTakingMetric(CodeMetric):
     description = "Turn-taking evaluation based on per-turn latency and interruption behavior"
     category = "experience"
     pass_at_k_threshold = 0.8
-    version = "v0.2"
+    version = "v0.3"
 
     # --- Latency curve (piecewise linear). 0 outside [LATENCY_HARD_EARLY_MS, LATENCY_HARD_LATE_MS]. ---
     # Ramp up 0 → 1 from LATENCY_HARD_EARLY_MS to LATENCY_SWEET_SPOT_LOW_MS.
@@ -585,6 +603,12 @@ async def compute(self, context: MetricContext) -> MetricScore:
                 "missed_turn": missed_turn,
             }
 
+            vad_result = compute_vad_turn_sub_metrics(context)
+            vad_sub_metrics: dict[str, MetricScore] = {}
+            if vad_result is not None:
+                vad_sub_metrics, vad_per_turn = vad_result
+                details["vad_turns"] = vad_per_turn
+
             if not per_turn_score:
                 self.logger.info(
                     f"[{context.record_id}] No turns with both user and assistant audio timestamps; "
@@ -595,10 +619,12 @@ async def compute(self, context: MetricContext) -> MetricScore:
                     score=0.0,
                     normalized_score=0.0,
                     details=details,
+                    sub_metrics=vad_sub_metrics,
                 )
 
             score = 0.0 if missed_turn else round(statistics.mean(per_turn_score.values()), 4)
             sub_metrics = self._build_flat_sub_metrics(context, turn_keys, turns_with_tool_calls, per_turn_evidence)
+            sub_metrics.update(vad_sub_metrics)
 
             return MetricScore(
                 name=self.name,
diff --git a/src/eva/metrics/experience/vad_end_of_turn.py b/src/eva/metrics/experience/vad_end_of_turn.py
new file mode 100644
index 00000000..83322371
--- /dev/null
+++ b/src/eva/metrics/experience/vad_end_of_turn.py
@@ -0,0 +1,569 @@
+"""VAD-analyzer turn diagnostics for the turn_taking metric.
+
+Surfaces Krisp's "no silence-duration fallback" hang and Smart Turn's silence-fallback
+(stop_secs) usage directly in turn_taking.sub_metrics, for any pipecat run with local
+turn-analyzer VAD (Krisp or Smart Turn). Null for runs with no local VAD (other
+frameworks, or external turn strategies like self-endpointing STT/FLUX).
+
+No pipecat modifications: all signals come from files eva already writes
+(pipecat_logs.jsonl, pipecat_metrics.jsonl, audit_log.json) or pipecat's own log
+output (logs.log).
+"""
+
+import json
+import re
+import statistics
+from pathlib import Path
+from typing import Any
+
+from eva.metrics.base import MetricContext
+from eva.metrics.processor import is_agent_timeout_on_user_turn
+from eva.models.results import MetricScore
+from eva.utils.conversation_correctly_finished.final_turn import final_turn_input_flags
+from eva.utils.json_utils import load_jsonl
+from eva.utils.log_processing import load_audit_log
+
+_STOP_SECS_RE = re.compile(r"End of Turn complete due to stop_secs\. Silence in ms: ([\d.]+)")
+_FINAL_TURN_FLAG_KEYS = ("short", "acknowledgement", "spelled_entity")
+
+# Widest gap tolerated between a vad_stop and the dispatched user turn it closed. Two orders of
+# magnitude above the observed +3..+8ms dispatch lag, and well under half the closest observed
+# spacing between consecutive vad_stops (2244ms) - see ``_transcript_for_vad_stop``.
+_DISPATCH_VS_VAD_STOP_TOLERANCE_MS = 1000
+
+
+def _find_run_config(output_dir: str) -> dict[str, Any] | None:
+    """Walk up from output_dir to the run root and load config.json, or None if absent/unreadable."""
+    if not output_dir:
+        return None
+    path = Path(output_dir)
+    while True:
+        candidate = path / "config.json"
+        if candidate.exists():
+            try:
+                return json.loads(candidate.read_text())
+            except (OSError, json.JSONDecodeError):
+                return None
+        if path == path.parent:
+            return None
+        path = path.parent
+
+
+def vad_turn_metrics_applicable(output_dir: str) -> bool:
+    """True when this record's run used local turn-analyzer VAD (Krisp or Smart Turn).
+
+    False for non-pipecat frameworks (openai_realtime, elevenlabs, grok_voice,
+    smallest_hydra) and for pipecat runs with turn_stop_strategy == "external"
+    (self-endpointing STT / external turn detection, e.g. Deepgram FLUX). Also False
+    when config.json can't be found at all — a safe default rather than assuming a
+    pipeline shape we can't confirm.
+    """
+    config = _find_run_config(output_dir)
+    if config is None:
+        return False
+    if config.get("framework", "pipecat") != "pipecat":
+        return False
+    model_cfg = config.get("model") or {}
+    return model_cfg.get("turn_stop_strategy", "turn_analyzer") != "external"
+
+
+def _load_turn_metrics_events(output_dir: str) -> list[dict[str, Any]]:
+    """Return TurnMetricsData entries from pipecat_metrics.jsonl, sorted by timestamp.
+
+    Covers both KrispVivaTurn (always is_complete=True when present) and BaseSmartTurn
+    (is_complete True or False, one entry per VAD-stop evaluation) — same shape for both.
+    """
+    entries = load_jsonl(Path(output_dir) / "pipecat_metrics.jsonl")
+    events = [
+        {
+            "timestamp": e["timestamp"],
+            "is_complete": e["value"]["is_complete"],
+            "e2e_processing_time_ms": e["value"].get("e2e_processing_time_ms"),
+        }
+        for e in entries
+        if e.get("type") == "TurnMetricsData"
+    ]
+    return sorted(events, key=lambda e: e["timestamp"])
+
+
+def _load_vad_events(output_dir: str) -> tuple[list[int], list[int]]:
+    """Return (sorted user_started_speaking timestamps, sorted user_stopped_speaking timestamps).
+
+    Both epoch-ms, from the same WallClock instance that produces pipecat_metrics.jsonl's
+    timestamps (confirmed shared at pipecat_server.py:483-488) — directly comparable.
+    """
+    entries = load_jsonl(Path(output_dir) / "pipecat_logs.jsonl")
+    starts = sorted(e["timestamp"] for e in entries if e.get("type") == "user_started_speaking")
+    stops = sorted(e["timestamp"] for e in entries if e.get("type") == "user_stopped_speaking")
+    return starts, stops
+
+
+def _load_dispatched_user_turns(output_dir: str) -> list[tuple[int, str]]:
+    """Return (timestamp, text) for each LLM-dispatched user turn in audit_log.json, sorted.
+
+    Deliberately *not* pipecat_logs.jsonl's "transcript" events. Those are individual STT
+    fragments, stamped when the fragment finalized, and neither property survives contact with
+    real data: a single user turn routinely spans several fragments (confirmed in the fixtures -
+    one turn arrived as "My employee ID is e n" + "p zero six six six six six." + "The last four
+    of my phone are four four two five."), and a fragment can finalize *before* its own turn's
+    user_started_speaking fires, so slicing fragments by VAD-start boundaries attributes them to
+    the previous turn.
+
+    audit_log.json's transcript entries instead carry the aggregated text actually handed to the
+    LLM, one entry per real turn, stamped in the same epoch-ms clock as the VAD events.
+    """
+    audit = load_audit_log(Path(output_dir) / "audit_log.json")
+    if not audit:
+        return []
+    turns = []
+    for entry in audit.get("transcript", []):
+        if entry.get("message_type") != "user" or not entry.get("value"):
+            continue
+        try:
+            timestamp = int(entry["timestamp"])
+        except (KeyError, TypeError, ValueError):
+            continue
+        turns.append((timestamp, entry["value"]))
+    return sorted(turns, key=lambda t: t[0])
+
+
+def _load_turn_fallback_timestamps(output_dir: str) -> list[int]:
+    """Return timestamps of turn-end fallback nudges recorded in audit_log.json's transcript, sorted.
+
+    A ``turn_fallback`` entry means the turn analyzer's own end-of-turn signal never fired within
+    the configured timeout, so the pipeline synthesized a nudge from partial STT text instead
+    (see fallback.py). Used only as evidence that a real, separate turn got stuck inside the
+    trailing part of a VAD-start window that an earlier, unrelated natural completion already
+    closed (see ``_classify_turns``) - such trailing gaps are ordinary inter-turn silence in the
+    overwhelming majority of windows, so they must not be flagged "stuck" without positive
+    evidence like this marker.
+    """
+    audit = load_audit_log(Path(output_dir) / "audit_log.json")
+    if not audit:
+        return []
+    timestamps = []
+    for entry in audit.get("transcript", []):
+        if entry.get("message_type") != "turn_fallback":
+            continue
+        try:
+            timestamps.append(int(entry["timestamp"]))
+        except (KeyError, TypeError, ValueError):
+            continue
+    return sorted(timestamps)
+
+
+def _transcript_for_vad_stop(dispatched_user_turns: list[tuple[int, str]], vad_stop: int) -> str | None:
+    """Return the text of the user turn that ``vad_stop`` closed, or None if none matches.
+
+    Each dispatched turn lands a handful of milliseconds after the user_stopped_speaking that
+    ended it, while consecutive vad_stops are seconds apart (2244ms at the closest observed).
+    ``_DISPATCH_VS_VAD_STOP_TOLERANCE_MS`` sits between those two scales, so the nearest turn
+    within tolerance is unambiguous - and a window whose utterance was never dispatched at all
+    matches nothing rather than borrowing a neighbour's text.
+    """
+    candidates = [
+        (abs(ts - vad_stop), text)
+        for ts, text in dispatched_user_turns
+        if abs(ts - vad_stop) <= _DISPATCH_VS_VAD_STOP_TOLERANCE_MS
+    ]
+    return min(candidates, key=lambda c: c[0])[1] if candidates else None
+
+
+def _load_turn_start_timestamps(output_dir: str) -> set[int]:
+    """Return the set of timestamps where pipecat's own TurnTrackingObserver started a turn.
+
+    Sourced from pipecat_logs.jsonl's "turn_start" entries (observers.py:_start_turn), on the
+    same shared epoch-ms clock as vad_starts/vad_stops. This is pipecat's own judgment of
+    whether a given user_started_speaking frame began a genuinely new turn - independent of
+    (and a check on) the turn-analyzer's is_complete/stop_secs signals used elsewhere in this
+    module. Empty when the run's pipeline doesn't emit turn_start at all, which callers must
+    treat as "signal unavailable" rather than "no turn ever started".
+    """
+    entries = load_jsonl(Path(output_dir) / "pipecat_logs.jsonl")
+    return {e["timestamp"] for e in entries if e.get("type") == "turn_start"}
+
+
+def _load_stop_secs_silence_ms(output_dir: str) -> list[float]:
+    """Return the 'Silence in ms' value from each pipecat stop_secs forced-completion line, in file order.
+
+    logs.log's local-time text timestamps are not safe to convert and compare against
+    pipecat_logs.jsonl/pipecat_metrics.jsonl's shared epoch (confirmed ~1-3s of slop,
+    machine/timezone-dependent) — only the count and self-contained silence value are used.
+    """
+    log_path = Path(output_dir) / "logs.log"
+    if not log_path.exists():
+        return []
+    return [float(m) for m in _STOP_SECS_RE.findall(log_path.read_text(errors="ignore"))]
+
+
+def _build_turn_windows(vad_starts: list[int]) -> list[tuple[int, int | None]]:
+    """Pair consecutive VAD-start timestamps into (window_start, window_end); last window is open-ended."""
+    return [(start, vad_starts[i + 1] if i + 1 < len(vad_starts) else None) for i, start in enumerate(vad_starts)]
+
+
+def _classify_turns(
+    vad_starts: list[int],
+    vad_stops: list[int],
+    turn_metrics_events: list[dict[str, Any]],
+    stop_secs_silence_ms: list[float],
+    dispatched_user_turns: list[tuple[int, str]] | None = None,
+    turn_start_timestamps: set[int] | None = None,
+    turn_fallback_timestamps: list[int] | None = None,
+) -> list[dict[str, Any]]:
+    """Classify each VAD-analyzer turn window as natural/forced/dispatched/early/stuck.
+
+    natural: one or more TurnMetricsData(is_complete=True) entries fall inside the window.
+        A raw VAD-start window can legitimately contain more than one - confirmed against real
+        Krisp data where a single window held two genuine completions for two separate real
+        user turns (Krisp's own "user_started_speaking" fired late/not-at-all for the second
+        one, so both landed in the first turn's window). Each is emitted as its own result
+        entry rather than keeping only the first and silently dropping the rest: the window is
+        split at each natural completion in turn order, with each sub-turn's open_duration_ms
+        measured from the previous split point (window_start for the first one). If a
+        ``turn_fallback_timestamps`` entry (see ``_load_turn_fallback_timestamps``) falls after
+        the last natural completion and before the window ends, a genuinely separate turn got
+        stuck in that trailing gap and needed a turn-end fallback nudge - emitted as its
+        own trailing "stuck" entry rather than silently dropped.
+    forced: no natural completion, AND this window has at least one user_stopped_speaking
+        event in its range (Smart Turn's stop_secs mechanism only ever fires after VAD has
+        registered a stop, so a window with no vad_stop cannot legitimately be the source of
+        a forced completion). Such a window consumes the next unused stop_secs silence value
+        in file order (Smart Turn only ever has one open turn waiting on stop_secs at a time,
+        so ordinal matching is safe among windows that actually have a vad_stop — see spec's
+        "Matching stop_secs lines to turn windows"). Close time = last vad_stop in the
+        window; the consumed silence value is only used to confirm the fallback fired, not
+        added to close_time — last vad_stop is emitted only after the turn analyzer has
+        already waited out that full silence duration internally (confirmed against real
+        data: it trails the LLM-dispatched turn by only 4-6ms), so adding it again would
+        double-count that wait. Also tagged with ``final_turn_flags`` (short /
+        acknowledgement / spelled_entity, via ``final_turn_input_flags``) computed on the
+        dispatched user turn that this window's closing vad_stop ended (see
+        ``_transcript_for_vad_stop``) — a candidate explanation for *why* the turn analyzer
+        needed the stop_secs fallback instead of closing naturally. Both the text and the flags
+        are omitted when no dispatched turn matches, so a flag is never computed on some other
+        turn's words.
+    dispatched: no natural completion AND no stop_secs-confirmed forced completion, but
+        audit_log.json shows a real LLM-dispatched user turn (see
+        ``_load_dispatched_user_turns``) landing inside this window's bounds. The user-simulator's audio bridge calls
+        stop() and signals "session_ended" mid-turn (a race with its own end_call tool),
+        which starved pipecat's VADController of further audio; it emitted
+        "user_stopped_speaking (strategy: None)" via its own "no audio received while
+        speaking, forcing speech stop" fallback - a third closure path that is neither the
+        turn analyzer's is_complete signal nor its stop_secs timeout, so neither the
+        natural nor the forced branch above ever sees it. Without this check the window
+        would wrongly read as "stuck" despite the turn genuinely completing and getting a
+        real assistant response. Close time = the last dispatched turn found in
+        the window (the nearest available proxy for when it actually closed, since there's
+        no vad_stop/stop_secs data to trust here). Tagged with ``transcript_text`` and
+        ``final_turn_flags`` the same way ``forced`` is, from the same matched dispatched
+        turn - both omitted if the match is otherwise unavailable, which cannot actually
+        happen here since the match is what triggered this branch.
+    vad_early: a natural or forced completion (see above) whose window is immediately
+        followed by another VAD-start window that (a) itself resolves to a natural, forced,
+        or dispatched completion - not "stuck" - and (b) opened with no "turn_start" logged
+        by pipecat's own TurnTrackingObserver (see ``_load_turn_start_timestamps``). Both
+        conditions matter:
+        (b) alone is not enough, since a missing turn_start also shows up ahead of an
+        ordinary Krisp VAD false-start/blip window that resolves "stuck" - confirmed against
+        real Krisp data, where that shape is benign noise already covered by stuck_rate, not
+        a wrongly-cut-short utterance. Condition (a) is what tells the two apart: only when
+        the *next* window turns out to hold a genuine completion of its own does the pair
+        read as one real utterance that got artificially split into two. Only reclassifies the
+        natural/forced result immediately bordering that missing turn_start (for a
+        multi-natural window, that's the last split segment) - earlier segments in the same
+        window, or the run's final window with no following window at all, are untouched.
+        Only checked when ``turn_start_timestamps`` is non-empty; an empty set means this
+        run's pipeline never emits turn_start at all, which must read as "signal
+        unavailable", not "every transition is early".
+    stuck: no natural completion, no stop_secs-confirmed forced completion, and no dispatched
+        user turn landed in this window either (see "dispatched" above - checked first, so a
+        window is only ever "stuck" once that escape hatch has already failed). This is
+        common, not a rare edge case (confirmed against real Krisp data: VAD false-starts and
+        silently-swallowed utterances both produce it) - a middle window's open_duration_ms
+        uses its own window_end (the next VAD start bounds it), never the conversation's
+        global last timestamp. Only the true last, open-ended window (window_end is None)
+        falls back to last_epoch as a conversation-end proxy.
+
+    ``turn_index`` on each result is a running count over the *output* list, not the raw
+    window index - windows that split into multiple natural completions push every later
+    entry's index forward accordingly.
+    """
+    windows = _build_turn_windows(vad_starts)
+    last_epoch = max([*vad_starts, *vad_stops, *(e["timestamp"] for e in turn_metrics_events)], default=0)
+    stop_secs_iter = iter(stop_secs_silence_ms)
+    dispatched_user_turns = dispatched_user_turns or []
+    turn_fallback_timestamps = turn_fallback_timestamps or []
+    # One entry list per raw VAD-start window (before turn_index is assigned), so the
+    # early pass below can look at "the next window's own entries" without having to
+    # re-derive window boundaries from the flattened, already-turn_indexed output.
+    per_window: list[list[dict[str, Any]]] = []
+
+    for window_start, window_end in windows:
+        in_window = [
+            e
+            for e in turn_metrics_events
+            if e["timestamp"] >= window_start and (window_end is None or e["timestamp"] < window_end)
+        ]
+        naturals = [e for e in in_window if e["is_complete"]]
+        if naturals:
+            entries = []
+            segment_start = window_start
+            for natural in naturals:
+                entries.append(
+                    {
+                        "open_duration_ms": natural["timestamp"] - segment_start,
+                        "vad_time_to_complete_ms": natural["e2e_processing_time_ms"],
+                        "completion": "natural",
+                    }
+                )
+                segment_start = natural["timestamp"]
+            trailing_fallbacks = [
+                ts for ts in turn_fallback_timestamps if ts >= segment_start and (window_end is None or ts < window_end)
+            ]
+            if trailing_fallbacks:
+                close_time = window_end if window_end is not None else max(trailing_fallbacks)
+                entries.append(
+                    {
+                        "open_duration_ms": close_time - segment_start,
+                        "vad_time_to_complete_ms": None,
+                        "completion": "stuck",
+                    }
+                )
+            per_window.append(entries)
+            continue
+
+        stops_in_window = [s for s in vad_stops if s >= window_start and (window_end is None or s < window_end)]
+        if stops_in_window:
+            silence_ms = next(stop_secs_iter, None)
+            if silence_ms is not None:
+                # silence_ms is only consumed here to confirm this window legitimately earned
+                # the stop_secs fallback (ordinal matching) - it is NOT added to close_time.
+                # last_stop (user_stopped_speaking) is emitted by
+                # TurnAnalyzerUserTurnStopStrategy only once the analyzer's append_audio has
+                # already waited out the full silence_ms internally (confirmed against real
+                # data: last_stop trails the audit-log dispatch by only 4-6ms). Adding
+                # silence_ms again here double-counted that wait, inflating open_duration_ms
+                # by roughly one extra stop_secs timeout per forced completion.
+                last_stop = max(stops_in_window)
+                close_time = last_stop
+                transcript_text = _transcript_for_vad_stop(dispatched_user_turns, last_stop)
+                result = {
+                    "open_duration_ms": close_time - window_start,
+                    "vad_time_to_complete_ms": None,
+                    "completion": "forced",
+                }
+                if transcript_text is not None:
+                    # Was stop_secs forced because the user's actual utterance is a shape the
+                    # turn analyzer struggles to close on its own (short / acknowledgement / spelled-out)?
+                    result["transcript_text"] = transcript_text
+                    result["final_turn_flags"] = final_turn_input_flags(transcript_text)
+                per_window.append([result])
+                continue
+
+        dispatched_in_window = [
+            (ts, text)
+            for ts, text in dispatched_user_turns
+            if ts >= window_start and (window_end is None or ts < window_end)
+        ]
+        if dispatched_in_window:
+            # Neither the turn analyzer's is_complete signal nor its stop_secs timeout ever
+            # fired for this window, but audit_log.json proves the turn was genuinely
+            # dispatched to the LLM anyway - e.g. pipecat's VADController force-stopping
+            # speech itself ("no audio received while speaking") when the user-simulator's
+            # audio bridge ends the session mid-turn. close_time uses the last dispatched
+            # turn's own timestamp, the best available proxy in the absence of a trustworthy
+            # vad_stop/stop_secs pairing.
+            close_time, transcript_text = max(dispatched_in_window, key=lambda t: t[0])
+            result = {
+                "open_duration_ms": close_time - window_start,
+                "vad_time_to_complete_ms": None,
+                "completion": "dispatched",
+                "transcript_text": transcript_text,
+                "final_turn_flags": final_turn_input_flags(transcript_text),
+            }
+            per_window.append([result])
+            continue
+
+        close_time = window_end if window_end is not None else last_epoch
+        per_window.append(
+            [
+                {
+                    "open_duration_ms": close_time - window_start,
+                    "vad_time_to_complete_ms": None,
+                    "completion": "stuck",
+                }
+            ]
+        )
+
+    # Early pass: a window's last entry (the one bordering the next window) gets
+    # reclassified when the next window itself resolved to a real completion (not "stuck")
+    # but opened with no turn_start - see "early" in this function's docstring for why
+    # both conditions are required.
+    if turn_start_timestamps:
+        for i, (_, window_end) in enumerate(windows[:-1]):
+            last_entry = per_window[i][-1]
+            next_window_resolved = per_window[i + 1][0]["completion"] in ("natural", "forced", "dispatched")
+            if (
+                last_entry["completion"] in ("natural", "forced", "dispatched")
+                and next_window_resolved
+                and window_end not in turn_start_timestamps
+            ):
+                last_entry["completion"] = "early"
+
+    results: list[dict[str, Any]] = []
+    for entries in per_window:
+        for entry in entries:
+            results.append({"turn_index": len(results), **entry})
+    return results
+
+
+# Tolerance for matching the real, ground-truth last user turn (from
+# context.audio_timestamps_user_turns, recorded by the user-simulator process) against the
+# turn-analyzer's own vad_starts (recorded by the pipecat server process). These two are on
+# different clocks with no fixed offset - confirmed against real Krisp/non-Krisp examples,
+# where legitimate skew for a turn VAD *did* register ranged ~0.4-6.8s, while turns VAD never
+# registered at all sat 10.8-24.4s away from the nearest vad_start. 8s sits between those two
+# clusters: generous enough to absorb real skew, tight enough to catch a genuine miss.
+_VAD_MISS_TOLERANCE_MS = 8000
+
+
+def _final_user_turn_never_reached_vad(
+    vad_starts: list[int], audio_timestamps_user_turns: dict[int, list[tuple[float, float]]] | None
+) -> bool:
+    """True if the turn analyzer's VAD never registered a start anywhere near the last real user turn.
+
+    Catches the case where the turn-analyzer's own "user_started_speaking" never fired at all
+    for the final utterance (confirmed against real Krisp examples: the conversation ends via
+    inactivity_timeout with the user having spoken last, but that last utterance produced no
+    VAD-start window whatsoever - so it never shows up in ``per_turn`` and
+    ``per_turn[-1]["completion"] == "stuck"`` can't catch it, since ``per_turn[-1]`` is really
+    describing an earlier, already-resolved turn).
+    """
+    if not audio_timestamps_user_turns or not vad_starts:
+        return False
+    last_key = max(audio_timestamps_user_turns)
+    segments = audio_timestamps_user_turns[last_key]
+    if not segments:
+        return False
+    last_start_ms = min(seg[0] for seg in segments) * 1000
+    return all(abs(vs - last_start_ms) > _VAD_MISS_TOLERANCE_MS for vs in vad_starts)
+
+
+def _turn_stop_strategy(output_dir: str) -> str | None:
+    """Return this run's configured model.turn_stop_strategy, or None if config.json wasn't found."""
+    config = _find_run_config(output_dir)
+    if config is None:
+        return None
+    model_cfg = config.get("model") or {}
+    return model_cfg.get("turn_stop_strategy", "turn_analyzer")
+
+
+def compute_vad_turn_sub_metrics(context: MetricContext) -> tuple[dict[str, MetricScore], list[dict[str, Any]]] | None:
+    """Compute the VAD-analyzer sub-metrics plus the per-turn diagnostic list.
+
+    Returns None when the run used no local turn-analyzer VAD (vad_turn_metrics_applicable
+    is False) or recorded no VAD-analyzer turns at all.
+    """
+    if not vad_turn_metrics_applicable(context.output_dir):
+        return None
+
+    vad_starts, vad_stops = _load_vad_events(context.output_dir)
+    if not vad_starts:
+        return None
+
+    turn_metrics_events = _load_turn_metrics_events(context.output_dir)
+    stop_secs_silence_ms = _load_stop_secs_silence_ms(context.output_dir)
+    dispatched_user_turns = _load_dispatched_user_turns(context.output_dir)
+    turn_start_timestamps = _load_turn_start_timestamps(context.output_dir)
+    turn_fallback_timestamps = _load_turn_fallback_timestamps(context.output_dir)
+    per_turn = _classify_turns(
+        vad_starts,
+        vad_stops,
+        turn_metrics_events,
+        stop_secs_silence_ms,
+        dispatched_user_turns,
+        turn_start_timestamps,
+        turn_fallback_timestamps,
+    )
+
+    def _wrap(key: str, value: float, normalized: bool) -> MetricScore:
+        return MetricScore(name=f"turn_taking.{key}", score=value, normalized_score=value if normalized else None)
+
+    # A user turn that never even got a VAD-start window (see
+    # _final_user_turn_never_reached_vad) is invisible to _classify_turns entirely - fold it
+    # in here as an explicit stuck entry so it's counted by stuck_rate below, rather than
+    # silently vanishing. Only added when it's also the reason the conversation died (matches
+    # the narrower signal this used to be gated on): Krisp has no silence-duration fallback,
+    # so a VAD-analyzer miss is what actually kills the call via inactivity_timeout.
+    if is_agent_timeout_on_user_turn(
+        context.conversation_ended_reason,
+        context.audio_timestamps_user_turns,
+        context.audio_timestamps_assistant_turns,
+    ) and _final_user_turn_never_reached_vad(vad_starts, context.audio_timestamps_user_turns):
+        per_turn = [
+            *per_turn,
+            {
+                "turn_index": len(per_turn),
+                "open_duration_ms": None,
+                "vad_time_to_complete_ms": None,
+                "completion": "stuck",
+                "vad_never_started": True,
+            },
+        ]
+
+    sub: dict[str, MetricScore] = {}
+
+    stuck_count = sum(1 for t in per_turn if t["completion"] == "stuck")
+    sub["vad_stuck_rate"] = _wrap("vad_stuck_rate", round(stuck_count / len(per_turn), 4), True)
+
+    natural_count = sum(1 for t in per_turn if t["completion"] == "natural")
+    forced_count = sum(1 for t in per_turn if t["completion"] == "forced")
+    dispatched_count = sum(1 for t in per_turn if t["completion"] == "dispatched")
+    early_count = sum(1 for t in per_turn if t["completion"] == "early")
+    strategy = _turn_stop_strategy(context.output_dir)
+    if strategy != "krisp_viva_turn" and (natural_count + forced_count) > 0:
+        sub["forced_completion_rate"] = _wrap(
+            "forced_completion_rate", round(forced_count / (natural_count + forced_count), 4), True
+        )
+
+    # Rate of turn-analyzer completions (natural, forced, or dispatched) that turned out to
+    # be early - the turn analyzer reported the user done, but the next VAD-start window was
+    # never promoted to a real pipecat turn (see "early" in _classify_turns' docstring),
+    # meaning the user's utterance actually continued and got wrongly cut short.
+    resolved_count = natural_count + forced_count + dispatched_count + early_count
+    if resolved_count > 0:
+        sub["vad_early_detection_rate"] = _wrap(
+            "vad_early_detection_rate", round(early_count / resolved_count, 4), True
+        )
+
+    # Among forced completions specifically: does the final utterance's shape (short /
+    # acknowledgement / spelled-out entity) explain why the turn analyzer needed the stop_secs
+    # fallback instead of closing naturally? Only emitted when there's at least one forced
+    # completion with a matched transcript, so cross-record aggregates reflect real evidence.
+    forced_flags = [t["final_turn_flags"] for t in per_turn if t["completion"] == "forced" and "final_turn_flags" in t]
+    if forced_flags:
+        for key in _FINAL_TURN_FLAG_KEYS:
+            rate = sum(1 for flags in forced_flags if flags[key]) / len(forced_flags)
+            sub[f"forced_completion_final_turn_{key}_rate"] = _wrap(
+                f"forced_completion_final_turn_{key}_rate", round(rate, 4), True
+            )
+
+    natural_times = [
+        t["vad_time_to_complete_ms"]
+        for t in per_turn
+        if t["completion"] == "natural" and t["vad_time_to_complete_ms"] is not None
+    ]
+    if natural_times:
+        sorted_times = sorted(natural_times)
+        n = len(sorted_times)
+
+        def _pct(p: float) -> float:
+            return sorted_times[min(n - 1, int(p * n))]
+
+        sub["mean_vad_time_to_complete_ms"] = _wrap(
+            "mean_vad_time_to_complete_ms", round(statistics.mean(natural_times), 3), False
+        )
+        sub["p50_vad_time_to_complete_ms"] = _wrap("p50_vad_time_to_complete_ms", round(_pct(0.50), 3), False)
+        sub["p90_vad_time_to_complete_ms"] = _wrap("p90_vad_time_to_complete_ms", round(_pct(0.90), 3), False)
+
+    return sub, per_turn
diff --git a/tests/fixtures/metric_signatures.json b/tests/fixtures/metric_signatures.json
index 12e02477..e15ecd1b 100644
--- a/tests/fixtures/metric_signatures.json
+++ b/tests/fixtures/metric_signatures.json
@@ -92,8 +92,8 @@
   "TurnTakingMetric": {
     "name": "turn_taking",
     "prompt_hash": null,
-    "source_hash": "5f654debef13",
-    "version": "v0.2"
+    "source_hash": "f3692f309294",
+    "version": "v0.3"
   },
   "UserBehavioralFidelityMetric": {
     "name": "user_behavioral_fidelity",
diff --git a/tests/unit/metrics/test_turn_taking.py b/tests/unit/metrics/test_turn_taking.py
index fbd9ce1d..02a8f9b6 100644
--- a/tests/unit/metrics/test_turn_taking.py
+++ b/tests/unit/metrics/test_turn_taking.py
@@ -1051,3 +1051,97 @@ async def test_sub_metric_populated_for_s2s(self, metric, tmp_path):
         result = await metric.compute(context)
         sub = result.sub_metrics
         assert sub["pretoolspeech_rate"].score == pytest.approx(1.0)
+
+
+def _write_jsonl(path, entries):
+    path.write_text("\n".join(json.dumps(e) for e in entries) + "\n")
+
+
+class TestVadTurnMetricsIntegration:
+    @pytest.mark.asyncio
+    async def test_vad_sub_metrics_populated_for_pipecat_run(self, metric, tmp_path):
+        (tmp_path / "config.json").write_text(
+            json.dumps({"framework": "pipecat", "model": {"turn_stop_strategy": "krisp_viva_turn"}})
+        )
+        _write_jsonl(tmp_path / "pipecat_logs.jsonl", [{"timestamp": 1000, "type": "user_started_speaking"}])
+        _write_jsonl(
+            tmp_path / "pipecat_metrics.jsonl",
+            [
+                {
+                    "timestamp": 1100,
+                    "type": "TurnMetricsData",
+                    "value": {"is_complete": True, "e2e_processing_time_ms": 100.0},
+                }
+            ],
+        )
+        ctx = make_metric_context(
+            output_dir=str(tmp_path),
+            audio_timestamps_user_turns={1: [(0.0, 1.0)]},
+            audio_timestamps_assistant_turns={1: [(1.5, 2.0)]},
+        )
+        result = await metric.compute(ctx)
+        assert result.sub_metrics["mean_vad_time_to_complete_ms"].score == 100.0
+        assert result.details["vad_turns"] == [
+            {"turn_index": 0, "open_duration_ms": 100, "vad_time_to_complete_ms": 100.0, "completion": "natural"}
+        ]
+
+    @pytest.mark.asyncio
+    async def test_vad_sub_metrics_absent_for_non_pipecat_run(self, metric, tmp_path):
+        (tmp_path / "config.json").write_text(json.dumps({"framework": "elevenlabs", "model": {}}))
+        ctx = make_metric_context(
+            output_dir=str(tmp_path),
+            audio_timestamps_user_turns={1: [(0.0, 1.0)]},
+            audio_timestamps_assistant_turns={1: [(1.5, 2.0)]},
+        )
+        result = await metric.compute(ctx)
+        assert "mean_vad_time_to_complete_ms" not in result.sub_metrics
+        assert "vad_turns" not in result.details
+
+    @pytest.mark.asyncio
+    async def test_vad_sub_metrics_absent_when_no_config_or_vad_files(self, metric, tmp_path):
+        ctx = make_metric_context(
+            output_dir=str(tmp_path),
+            audio_timestamps_user_turns={1: [(0.0, 1.0)]},
+            audio_timestamps_assistant_turns={1: [(1.5, 2.0)]},
+        )
+        result = await metric.compute(ctx)
+        assert "vad_stuck_rate" not in result.sub_metrics
+        assert "vad_turns" not in result.details
+
+
+class TestVadTurnMetricsOnEarlyReturn:
+    @pytest.mark.asyncio
+    async def test_vad_sub_metrics_emitted_when_no_turns_have_both_user_and_assistant_audio(self, metric, tmp_path):
+        """Regression test for the "hang on the first/only real turn" scenario:
+
+        the user speaks, the VAD turn analyzer never completes (no TurnMetricsData at
+        all), and inactivity_timeout kills the conversation before the agent ever
+        responds. There is no turn with both user AND assistant audio, so
+        _get_turn_ids_with_turn_taking returns [] and per_turn_score is empty -
+        compute() takes the early-return path. vad_stuck_rate exists specifically to catch
+        cases like this, so it must still be computed and attached on that early-return
+        MetricScore, not skipped.
+        """
+        (tmp_path / "config.json").write_text(
+            json.dumps({"framework": "pipecat", "model": {"turn_stop_strategy": "krisp_viva_turn"}})
+        )
+        # VAD fires (user_started_speaking) but never completes - no pipecat_metrics.jsonl
+        # TurnMetricsData at all, so the single window is "stuck".
+        _write_jsonl(tmp_path / "pipecat_logs.jsonl", [{"timestamp": 1000, "type": "user_started_speaking"}])
+        ctx = make_metric_context(
+            output_dir=str(tmp_path),
+            conversation_ended_reason="inactivity_timeout",
+            audio_timestamps_user_turns={1: [(0.0, 5.0)]},
+            audio_timestamps_assistant_turns={},
+        )
+        result = await metric.compute(ctx)
+
+        # Early-return path: no turns had both user and assistant audio.
+        assert result.details["per_turn_score"] == {}
+        assert result.score == 0.0
+
+        # But the VAD diagnostics that exist to catch exactly this scenario must still
+        # be present, not silently skipped.
+        assert result.sub_metrics["vad_stuck_rate"].score == 1.0
+        assert result.details["vad_turns"]
+        assert result.details["vad_turns"][0]["completion"] == "stuck"
diff --git a/tests/unit/metrics/test_vad_end_of_turn.py b/tests/unit/metrics/test_vad_end_of_turn.py
new file mode 100644
index 00000000..35293553
--- /dev/null
+++ b/tests/unit/metrics/test_vad_end_of_turn.py
@@ -0,0 +1,998 @@
+"""Tests for vad_turn_metrics: config gating, file parsing, turn classification, aggregation."""
+
+import json
+
+import pytest
+
+from eva.metrics.experience.vad_end_of_turn import (
+    _classify_turns,
+    _find_run_config,
+    _load_dispatched_user_turns,
+    _load_stop_secs_silence_ms,
+    _load_turn_fallback_timestamps,
+    _load_turn_metrics_events,
+    _load_turn_start_timestamps,
+    _load_vad_events,
+    _transcript_for_vad_stop,
+    compute_vad_turn_sub_metrics,
+    vad_turn_metrics_applicable,
+)
+from eva.utils.conversation_correctly_finished.final_turn import final_turn_input_flags
+
+from .conftest import make_metric_context
+
+
+class TestFindRunConfig:
+    def test_finds_config_in_output_dir(self, tmp_path):
+        (tmp_path / "config.json").write_text(json.dumps({"framework": "pipecat"}))
+        assert _find_run_config(str(tmp_path)) == {"framework": "pipecat"}
+
+    def test_finds_config_by_walking_up_from_nested_record_dir(self, tmp_path):
+        (tmp_path / "config.json").write_text(json.dumps({"framework": "pipecat"}))
+        record_dir = tmp_path / "records" / "1.2.3"
+        record_dir.mkdir(parents=True)
+        assert _find_run_config(str(record_dir)) == {"framework": "pipecat"}
+
+    def test_returns_none_when_no_config_json_found(self, tmp_path):
+        assert _find_run_config(str(tmp_path)) is None
+
+    def test_returns_none_on_malformed_json(self, tmp_path):
+        (tmp_path / "config.json").write_text("{not valid json")
+        assert _find_run_config(str(tmp_path)) is None
+
+    def test_returns_none_for_empty_output_dir(self):
+        assert _find_run_config("") is None
+
+
+class TestVadTurnMetricsApplicable:
+    def test_true_for_krisp_turn_stop_strategy(self, tmp_path):
+        (tmp_path / "config.json").write_text(
+            json.dumps({"framework": "pipecat", "model": {"turn_stop_strategy": "krisp_viva_turn"}})
+        )
+        assert vad_turn_metrics_applicable(str(tmp_path)) is True
+
+    def test_false_for_non_pipecat_framework(self, tmp_path):
+        (tmp_path / "config.json").write_text(json.dumps({"framework": "elevenlabs", "model": {}}))
+        assert vad_turn_metrics_applicable(str(tmp_path)) is False
+
+    def test_false_for_external_turn_stop_strategy(self, tmp_path):
+        (tmp_path / "config.json").write_text(
+            json.dumps({"framework": "pipecat", "model": {"turn_stop_strategy": "external", "vad": "none"}})
+        )
+        assert vad_turn_metrics_applicable(str(tmp_path)) is False
+
+    def test_false_when_no_config_found(self, tmp_path):
+        assert vad_turn_metrics_applicable(str(tmp_path)) is False
+
+    def test_defaults_to_turn_analyzer_when_field_absent(self, tmp_path):
+        (tmp_path / "config.json").write_text(json.dumps({"framework": "pipecat", "model": {}}))
+        assert vad_turn_metrics_applicable(str(tmp_path)) is True
+
+
+def _write_jsonl(path, entries):
+    path.write_text("\n".join(json.dumps(e) for e in entries) + "\n")
+
+
+class TestLoadTurnMetricsEvents:
+    def test_extracts_and_sorts_turn_metrics_data(self, tmp_path):
+        _write_jsonl(
+            tmp_path / "pipecat_metrics.jsonl",
+            [
+                {
+                    "timestamp": 200,
+                    "type": "TurnMetricsData",
+                    "value": {"is_complete": True, "e2e_processing_time_ms": 100.0},
+                },
+                {
+                    "timestamp": 100,
+                    "type": "TurnMetricsData",
+                    "value": {"is_complete": False, "e2e_processing_time_ms": 50.0},
+                },
+                {"timestamp": 150, "type": "TTFBMetricsData", "value": {"ttfb": 1.0}},
+            ],
+        )
+        events = _load_turn_metrics_events(str(tmp_path))
+        assert events == [
+            {"timestamp": 100, "is_complete": False, "e2e_processing_time_ms": 50.0},
+            {"timestamp": 200, "is_complete": True, "e2e_processing_time_ms": 100.0},
+        ]
+
+
+class TestLoadVadEvents:
+    def test_extracts_and_sorts_start_and_stop_events(self, tmp_path):
+        _write_jsonl(
+            tmp_path / "pipecat_logs.jsonl",
+            [
+                {"timestamp": 300, "type": "user_started_speaking"},
+                {"timestamp": 100, "type": "user_started_speaking"},
+                {"timestamp": 150, "type": "user_stopped_speaking"},
+                {"timestamp": 999, "type": "turn_start"},
+            ],
+        )
+        starts, stops = _load_vad_events(str(tmp_path))
+        assert starts == [100, 300]
+        assert stops == [150]
+
+    def test_empty_when_file_missing(self, tmp_path):
+        assert _load_vad_events(str(tmp_path)) == ([], [])
+
+
+class TestLoadTurnStartTimestamps:
+    def test_extracts_turn_start_timestamps(self, tmp_path):
+        _write_jsonl(
+            tmp_path / "pipecat_logs.jsonl",
+            [
+                {"timestamp": 1000, "type": "turn_start"},
+                {"timestamp": 1050, "type": "turn_end"},
+                {"timestamp": 5000, "type": "turn_start"},
+                {"timestamp": 5000, "type": "user_started_speaking"},
+            ],
+        )
+        assert _load_turn_start_timestamps(str(tmp_path)) == {1000, 5000}
+
+    def test_empty_when_file_missing(self, tmp_path):
+        assert _load_turn_start_timestamps(str(tmp_path)) == set()
+
+
+class TestLoadStopSecsSilenceMs:
+    def test_extracts_silence_values_in_file_order(self, tmp_path):
+        (tmp_path / "logs.log").write_text(
+            "2026-08-04 19:31:15,912 | DEBUG    | pipecat:133 | End of Turn complete due to stop_secs. "
+            "Silence in ms: 3032.0\n"
+            "2026-08-04 19:32:51,140 | DEBUG    | pipecat:133 | End of Turn complete due to stop_secs. "
+            "Silence in ms: 3105.5\n"
+        )
+        assert _load_stop_secs_silence_ms(str(tmp_path)) == [3032.0, 3105.5]
+
+    def test_empty_when_no_stop_secs_lines(self, tmp_path):
+        (tmp_path / "logs.log").write_text(
+            "2026-08-04 19:31:15,912 | DEBUG | pipecat:166 | End of Turn result: EndOfTurnState.COMPLETE\n"
+        )
+        assert _load_stop_secs_silence_ms(str(tmp_path)) == []
+
+
+def _write_audit_log(path, user_turns):
+    """Write an audit_log.json whose transcript holds the given (timestamp, text) user turns."""
+    path.write_text(
+        json.dumps(
+            {
+                "transcript": [
+                    {"value": text, "message_type": "user", "timestamp": str(ts), "isBotMessage": False}
+                    for ts, text in user_turns
+                ]
+            }
+        )
+    )
+
+
+class TestLoadDispatchedUserTurns:
+    def test_extracts_and_sorts_user_turns_with_string_timestamps(self, tmp_path):
+        _write_audit_log(tmp_path / "audit_log.json", [(5000, "second"), (1000, "first")])
+        assert _load_dispatched_user_turns(str(tmp_path)) == [(1000, "first"), (5000, "second")]
+
+    def test_ignores_assistant_entries(self, tmp_path):
+        (tmp_path / "audit_log.json").write_text(
+            json.dumps(
+                {
+                    "transcript": [
+                        {"value": "user text", "message_type": "user", "timestamp": "1000"},
+                        {"value": "bot text", "message_type": "assistant", "timestamp": "2000"},
+                    ]
+                }
+            )
+        )
+        assert _load_dispatched_user_turns(str(tmp_path)) == [(1000, "user text")]
+
+    def test_empty_when_audit_log_missing(self, tmp_path):
+        assert _load_dispatched_user_turns(str(tmp_path)) == []
+
+    def test_empty_when_audit_log_malformed(self, tmp_path):
+        (tmp_path / "audit_log.json").write_text("{not json")
+        assert _load_dispatched_user_turns(str(tmp_path)) == []
+
+    def test_skips_entries_with_unusable_timestamp_or_empty_text(self, tmp_path):
+        (tmp_path / "audit_log.json").write_text(
+            json.dumps(
+                {
+                    "transcript": [
+                        {"value": "", "message_type": "user", "timestamp": "1000"},
+                        {"value": "no timestamp", "message_type": "user"},
+                        {"value": "bad timestamp", "message_type": "user", "timestamp": "not-a-number"},
+                        {"value": "good", "message_type": "user", "timestamp": "4000"},
+                    ]
+                }
+            )
+        )
+        assert _load_dispatched_user_turns(str(tmp_path)) == [(4000, "good")]
+
+
+class TestLoadTurnFallbackTimestamps:
+    def test_extracts_and_sorts_fallback_timestamps(self, tmp_path):
+        (tmp_path / "audit_log.json").write_text(
+            json.dumps(
+                {
+                    "transcript": [
+                        {"value": "second fallback", "message_type": "turn_fallback", "timestamp": "5000"},
+                        {"value": "first fallback", "message_type": "turn_fallback", "timestamp": "1000"},
+                        {"value": "user text", "message_type": "user", "timestamp": "2000"},
+                    ]
+                }
+            )
+        )
+        assert _load_turn_fallback_timestamps(str(tmp_path)) == [1000, 5000]
+
+    def test_empty_when_audit_log_missing(self, tmp_path):
+        assert _load_turn_fallback_timestamps(str(tmp_path)) == []
+
+    def test_skips_entries_with_unusable_timestamp(self, tmp_path):
+        (tmp_path / "audit_log.json").write_text(
+            json.dumps(
+                {
+                    "transcript": [
+                        {"value": "bad", "message_type": "turn_fallback", "timestamp": "not-a-number"},
+                        {"value": "good", "message_type": "turn_fallback", "timestamp": "3000"},
+                    ]
+                }
+            )
+        )
+        assert _load_turn_fallback_timestamps(str(tmp_path)) == [3000]
+
+
+class TestTranscriptForVadStop:
+    def test_matches_turn_dispatched_just_after_the_stop(self):
+        # Measured against real fixtures: dispatch lands 3-8ms after the closing vad_stop.
+        turns = [(1056, "first turn"), (5104, "second turn")]
+        assert _transcript_for_vad_stop(turns, 1050) == "first turn"
+        assert _transcript_for_vad_stop(turns, 5100) == "second turn"
+
+    def test_returns_nearest_when_several_are_within_tolerance(self):
+        assert _transcript_for_vad_stop([(1400, "far"), (1005, "near")], 1000) == "near"
+
+    def test_none_when_no_turn_is_close_enough(self):
+        # A window whose utterance was never dispatched must not borrow a distant turn's text.
+        assert _transcript_for_vad_stop([(30000, "much later turn")], 1050) is None
+
+    def test_none_when_no_dispatched_turns(self):
+        assert _transcript_for_vad_stop([], 1050) is None
+
+
+class TestClassifyTurns:
+    def test_forced_turn_text_is_the_full_dispatched_turn_not_the_last_stt_fragment(self):
+        # Reproduces the nonkrisp_2 shape. Window 0 is [1000, 26000) and window 1 is [26000, None).
+        # The user's second utterance began before VAD registered it, so its STT fragments
+        # finalized at 24000/25000 -- inside window 0, whose own utterance ended back at 6000.
+        # Attribution must follow the vad_stop that closed each turn, not the fragment timestamps,
+        # and must report the whole dispatched turn rather than only its trailing fragment.
+        dispatched = [
+            (6005, "Hi. I need help resetting my VPN password."),
+            (32004, "My employee ID is e n p zero six six. The last four of my phone are four four two five."),
+        ]
+        result = _classify_turns(
+            vad_starts=[1000, 26000],
+            vad_stops=[6000, 32000],
+            turn_metrics_events=[],
+            stop_secs_silence_ms=[3000.0, 3000.0],
+            dispatched_user_turns=dispatched,
+        )
+        assert [r["completion"] for r in result] == ["forced", "forced"]
+        assert result[0]["transcript_text"] == "Hi. I need help resetting my VPN password."
+        assert result[1]["transcript_text"] == (
+            "My employee ID is e n p zero six six. The last four of my phone are four four two five."
+        )
+
+    def test_forced_turn_omits_text_when_no_dispatched_turn_matches(self):
+        result = _classify_turns(
+            vad_starts=[1000],
+            vad_stops=[1050],
+            turn_metrics_events=[],
+            stop_secs_silence_ms=[3000.0],
+            dispatched_user_turns=[(90000, "unrelated later turn")],
+        )
+        assert result[0]["completion"] == "forced"
+        assert "transcript_text" not in result[0]
+        assert "final_turn_flags" not in result[0]
+
+    def test_single_natural_completion(self):
+        # Window: [1000, None). TurnMetricsData at 1103 (is_complete=True) closes it naturally.
+        result = _classify_turns(
+            vad_starts=[1000],
+            vad_stops=[1050],
+            turn_metrics_events=[{"timestamp": 1103, "is_complete": True, "e2e_processing_time_ms": 103.2}],
+            stop_secs_silence_ms=[],
+        )
+        assert result == [
+            {"turn_index": 0, "open_duration_ms": 103, "vad_time_to_complete_ms": 103.2, "completion": "natural"}
+        ]
+
+    def test_incomplete_evaluations_before_natural_completion_are_ignored(self):
+        # Matches real Smart Turn data: several is_complete=False evaluations, then True.
+        result = _classify_turns(
+            vad_starts=[1000],
+            vad_stops=[1050],
+            turn_metrics_events=[
+                {"timestamp": 1100, "is_complete": False, "e2e_processing_time_ms": 105.0},
+                {"timestamp": 1200, "is_complete": False, "e2e_processing_time_ms": 110.0},
+                {"timestamp": 1300, "is_complete": True, "e2e_processing_time_ms": 108.0},
+            ],
+            stop_secs_silence_ms=[],
+        )
+        assert result == [
+            {"turn_index": 0, "open_duration_ms": 300, "vad_time_to_complete_ms": 108.0, "completion": "natural"}
+        ]
+
+    def test_forced_completion_via_stop_secs_ordinal_match(self):
+        # No natural completion in the window -> consumes the one stop_secs silence value
+        # to confirm the fallback fired, but close_time = last vad_stop in window (1050)
+        # directly - the silence value is NOT added again (it's already baked into when
+        # last_stop itself was emitted).
+        result = _classify_turns(
+            vad_starts=[1000],
+            vad_stops=[1050],
+            turn_metrics_events=[{"timestamp": 1100, "is_complete": False, "e2e_processing_time_ms": 105.0}],
+            stop_secs_silence_ms=[3032.0],
+        )
+        assert result == [
+            {"turn_index": 0, "open_duration_ms": 50, "vad_time_to_complete_ms": None, "completion": "forced"}
+        ]
+
+    def test_stuck_when_no_natural_or_forced_signal(self):
+        result = _classify_turns(
+            vad_starts=[1000],
+            vad_stops=[],
+            turn_metrics_events=[],
+            stop_secs_silence_ms=[],
+        )
+        assert result == [
+            {"turn_index": 0, "open_duration_ms": 0, "vad_time_to_complete_ms": None, "completion": "stuck"}
+        ]
+
+    def test_dispatched_when_a_real_llm_turn_landed_in_an_otherwise_signal_less_window(self):
+        # Reproduces example: the user-simulator's audio bridge ended the
+        # session mid-turn, so pipecat's VADController force-stopped speech itself ("no
+        # audio received while speaking") instead of the turn analyzer's is_complete or
+        # stop_secs mechanisms ever firing. Without checking audit_log.json this window
+        # would wrongly read as "stuck" even though the LLM genuinely got the turn and
+        # responded.
+        result = _classify_turns(
+            vad_starts=[1000],
+            vad_stops=[5908],
+            turn_metrics_events=[{"timestamp": 4892, "is_complete": False, "e2e_processing_time_ms": 104.0}],
+            stop_secs_silence_ms=[],
+            dispatched_user_turns=[(5912, "I do not need anything else today. Bye.")],
+        )
+        assert result[0]["completion"] == "dispatched"
+        assert result[0]["open_duration_ms"] == 5912 - 1000
+        assert result[0]["vad_time_to_complete_ms"] is None
+        assert result[0]["transcript_text"] == "I do not need anything else today. Bye."
+        assert result[0]["final_turn_flags"] == final_turn_input_flags("I do not need anything else today. Bye.")
+
+    def test_dispatched_not_used_when_forced_stop_secs_already_matched(self):
+        # A window that legitimately earns the stop_secs fallback must stay "forced", not
+        # get reclassified just because a dispatched turn also happens to land inside it.
+        result = _classify_turns(
+            vad_starts=[1000],
+            vad_stops=[1050],
+            turn_metrics_events=[],
+            stop_secs_silence_ms=[3032.0],
+            dispatched_user_turns=[(1056, "some turn")],
+        )
+        assert result[0]["completion"] == "forced"
+
+    def test_stuck_when_no_dispatched_turn_falls_inside_the_window_either(self):
+        result = _classify_turns(
+            vad_starts=[1000, 5000],
+            vad_stops=[],
+            turn_metrics_events=[],
+            stop_secs_silence_ms=[],
+            dispatched_user_turns=[(90000, "unrelated later turn")],
+        )
+        assert result[0]["completion"] == "stuck"
+
+    def test_multiple_turns_ordinal_stop_secs_matching(self):
+        # Turn 0: natural completion. Turn 1: no natural completion -> gets the stop_secs value.
+        result = _classify_turns(
+            vad_starts=[1000, 5000],
+            vad_stops=[1050, 5100],
+            turn_metrics_events=[
+                {"timestamp": 1103, "is_complete": True, "e2e_processing_time_ms": 103.0},
+            ],
+            stop_secs_silence_ms=[3000.0],
+        )
+        assert result == [
+            {"turn_index": 0, "open_duration_ms": 103, "vad_time_to_complete_ms": 103.0, "completion": "natural"},
+            {"turn_index": 1, "open_duration_ms": 100.0, "vad_time_to_complete_ms": None, "completion": "forced"},
+        ]
+
+    def test_stuck_last_window_open_duration_uses_last_observed_epoch(self):
+        # Turn 1 (last, open-ended window) is stuck; open_duration_ms measured to the
+        # latest timestamp seen anywhere (a proxy for conversation end), since there's
+        # no next VAD start to bound it.
+        result = _classify_turns(
+            vad_starts=[1000, 5000],
+            vad_stops=[1050, 5100],
+            turn_metrics_events=[
+                {"timestamp": 1103, "is_complete": True, "e2e_processing_time_ms": 103.0},
+            ],
+            stop_secs_silence_ms=[],
+        )
+        assert result[1] == {
+            "turn_index": 1,
+            "open_duration_ms": 5100 - 5000,
+            "vad_time_to_complete_ms": None,
+            "completion": "stuck",
+        }
+
+    def test_stuck_middle_window_bounded_by_next_vad_start_not_global_last_epoch(self):
+        # Turn 0 is stuck (no natural/forced signal) but has a KNOWN window_end (turn 1's
+        # start) - its open_duration_ms must use that bound, not the conversation's global
+        # last-observed timestamp (which is much later here, at turn 2's natural completion).
+        # A real Krisp example fixture showed this exact shape: a middle window with no
+        # natural completion, followed much later by other turns finishing normally.
+        result = _classify_turns(
+            vad_starts=[1000, 1050, 50000],
+            vad_stops=[1040],
+            turn_metrics_events=[
+                {"timestamp": 50103, "is_complete": True, "e2e_processing_time_ms": 103.0},
+            ],
+            stop_secs_silence_ms=[],
+        )
+        assert result[0] == {
+            "turn_index": 0,
+            "open_duration_ms": 1050 - 1000,
+            "vad_time_to_complete_ms": None,
+            "completion": "stuck",
+        }
+
+    def test_window_with_two_naturals_splits_into_two_turns_instead_of_dropping_the_second(self):
+        # Real Krisp data shape: a single raw VAD-start window contains two genuine
+        # completions for two separate real user turns (Krisp's own start event for the
+        # second turn fired late/not at all, so both completions landed in the first
+        # window). Both must survive as their own "natural" entries, not just the first.
+        result = _classify_turns(
+            vad_starts=[1000],
+            vad_stops=[],
+            turn_metrics_events=[
+                {"timestamp": 1001, "is_complete": True, "e2e_processing_time_ms": 188.9},
+                {"timestamp": 4680, "is_complete": True, "e2e_processing_time_ms": 268.7},
+            ],
+            stop_secs_silence_ms=[],
+        )
+        assert result == [
+            {"turn_index": 0, "open_duration_ms": 1, "vad_time_to_complete_ms": 188.9, "completion": "natural"},
+            {"turn_index": 1, "open_duration_ms": 3679, "vad_time_to_complete_ms": 268.7, "completion": "natural"},
+        ]
+
+    def test_turn_index_accounts_for_earlier_split_windows(self):
+        # A split in an earlier window must push every later window's turn_index forward.
+        result = _classify_turns(
+            vad_starts=[1000, 5000],
+            vad_stops=[],
+            turn_metrics_events=[
+                {"timestamp": 1001, "is_complete": True, "e2e_processing_time_ms": 100.0},
+                {"timestamp": 2000, "is_complete": True, "e2e_processing_time_ms": 100.0},
+                {"timestamp": 5100, "is_complete": True, "e2e_processing_time_ms": 100.0},
+            ],
+            stop_secs_silence_ms=[],
+        )
+        assert [r["turn_index"] for r in result] == [0, 1, 2]
+        assert result[2]["open_duration_ms"] == 100
+
+    def test_false_start_does_not_steal_stop_secs_from_later_genuinely_forced_window(self):
+        # Window 0: brief VAD false-start - a start immediately superseded by the next
+        # start, no user_stopped_speaking ever registered, no natural completion. Since
+        # Smart Turn's stop_secs mechanism only ever fires after VAD registers a stop, this
+        # window cannot legitimately be the source of a forced completion.
+        # Window 1 (last, open-ended): has a real user_stopped_speaking at 5100 and no
+        # natural completion - this is the genuinely forced window and should get the
+        # single available stop_secs silence value.
+        result = _classify_turns(
+            vad_starts=[1000, 5000],
+            vad_stops=[5100],
+            turn_metrics_events=[],
+            stop_secs_silence_ms=[3032.0],
+        )
+        assert result[0]["completion"] == "stuck"
+        assert result[1]["completion"] == "forced"
+        assert result[1]["open_duration_ms"] == 5100 - 5000
+        assert result[1]["vad_time_to_complete_ms"] is None
+
+    def test_forced_completion_does_not_double_count_silence_ms(self):
+        # Regression guard, values taken from an example (real Smart Turn run):
+        # vad_start=1787171562461, vad_stop=1787171568321, stop_secs silence=3032.0,
+        # and the LLM-dispatched turn in audit_log.json landed at 1787171568327 - only
+        # 6ms after vad_stop. That means vad_stop (user_stopped_speaking) already fires
+        # only once the analyzer's internal silence_ms wait is over, so close_time must
+        # be ~vad_stop, not ~vad_stop + silence_ms (which would overshoot the real
+        # dispatch time by ~3032ms - one whole extra stop_secs timeout).
+        result = _classify_turns(
+            vad_starts=[1787171562461],
+            vad_stops=[1787171568321],
+            turn_metrics_events=[],
+            stop_secs_silence_ms=[3032.0],
+        )
+        assert result[0]["completion"] == "forced"
+        assert result[0]["open_duration_ms"] == 1787171568321 - 1787171562461
+
+    def test_natural_completion_reclassified_early_when_next_turn_never_started(self):
+        # Reproduces example: the turn analyzer reported is_complete=True on
+        # "Sure. My employee ID is emp zero four eight two seven one." but the user was still
+        # mid-utterance. Real evidence of that: pipecat's own TurnTrackingObserver never
+        # logged a turn_start for the very next VAD-start window (5000) - it stayed inside
+        # the same ongoing turn rather than starting a fresh one - and that next window went
+        # on to complete naturally itself ("Last four of my phone number are seven two nine
+        # four."), confirming the pair is really one continued utterance.
+        result = _classify_turns(
+            vad_starts=[1000, 5000],
+            vad_stops=[],
+            turn_metrics_events=[
+                {"timestamp": 1103, "is_complete": True, "e2e_processing_time_ms": 103.0},
+                {"timestamp": 5103, "is_complete": True, "e2e_processing_time_ms": 100.0},
+            ],
+            stop_secs_silence_ms=[],
+            turn_start_timestamps={1000},
+        )
+        assert result[0]["completion"] == "early"
+        assert result[1]["completion"] == "natural"
+
+    def test_natural_completion_stays_natural_when_next_turn_started(self):
+        result = _classify_turns(
+            vad_starts=[1000, 5000],
+            vad_stops=[],
+            turn_metrics_events=[
+                {"timestamp": 1103, "is_complete": True, "e2e_processing_time_ms": 103.0},
+                {"timestamp": 5103, "is_complete": True, "e2e_processing_time_ms": 100.0},
+            ],
+            stop_secs_silence_ms=[],
+            turn_start_timestamps={1000, 5000},
+        )
+        assert result[0]["completion"] == "natural"
+
+    def test_natural_completion_stays_natural_when_next_window_is_stuck(self):
+        # A missing turn_start alone is not enough - it also shows up ahead of an ordinary
+        # Krisp VAD false-start/blip window that never resolves at all ("stuck"), which is
+        # already covered by vad_stuck_rate and must not be conflated with a early split.
+        result = _classify_turns(
+            vad_starts=[1000, 5000],
+            vad_stops=[],
+            turn_metrics_events=[{"timestamp": 1103, "is_complete": True, "e2e_processing_time_ms": 103.0}],
+            stop_secs_silence_ms=[],
+            turn_start_timestamps={1000},
+        )
+        assert result[0]["completion"] == "natural"
+        assert result[1]["completion"] == "stuck"
+
+    def test_forced_completion_reclassified_early_when_next_turn_never_started(self):
+        result = _classify_turns(
+            vad_starts=[1000, 5000],
+            vad_stops=[1050, 5100],
+            turn_metrics_events=[],
+            stop_secs_silence_ms=[3032.0, 3032.0],
+            turn_start_timestamps={1000},
+        )
+        assert result[0]["completion"] == "early"
+        assert result[1]["completion"] == "forced"
+
+    def test_last_window_never_reclassified_early_with_no_following_window(self):
+        # window_end is None for the final, open-ended window - there is no "next" VAD-start
+        # to check turn_start against, so the natural completion stands.
+        result = _classify_turns(
+            vad_starts=[1000],
+            vad_stops=[],
+            turn_metrics_events=[{"timestamp": 1103, "is_complete": True, "e2e_processing_time_ms": 103.0}],
+            stop_secs_silence_ms=[],
+            turn_start_timestamps={1000},
+        )
+        assert result[0]["completion"] == "natural"
+
+    def test_empty_turn_start_timestamps_disables_early_detection(self):
+        # An empty set means this run's pipeline never emits turn_start at all - that must
+        # read as "signal unavailable", not "every transition is early".
+        result = _classify_turns(
+            vad_starts=[1000, 5000],
+            vad_stops=[],
+            turn_metrics_events=[{"timestamp": 1103, "is_complete": True, "e2e_processing_time_ms": 103.0}],
+            stop_secs_silence_ms=[],
+            turn_start_timestamps=set(),
+        )
+        assert result[0]["completion"] == "natural"
+
+    def test_natural_completion_followed_by_stuck_fallback_is_not_dropped(self):
+        # Reproduces example: window 0 closes an earlier, unrelated turn
+        # naturally at 1103, but the turn analyzer's VAD never fires again until window 1's
+        # own vad_start at 60000 - a genuinely separate turn got stuck in between and only
+        # got resolved via the 20s turn-end fallback nudge (audit_log's turn_fallback marker,
+        # timestamped mid-gap at 45000). That stuck turn must not be silently swallowed just
+        # because the window already had an earlier, unrelated natural completion.
+        result = _classify_turns(
+            vad_starts=[1000, 60000],
+            vad_stops=[],
+            turn_metrics_events=[{"timestamp": 1103, "is_complete": True, "e2e_processing_time_ms": 103.0}],
+            stop_secs_silence_ms=[],
+            turn_fallback_timestamps=[45000],
+        )
+        assert result[0]["completion"] == "natural"
+        assert result[1] == {
+            "turn_index": 1,
+            "open_duration_ms": 60000 - 1103,
+            "vad_time_to_complete_ms": None,
+            "completion": "stuck",
+        }
+
+    def test_natural_completion_with_no_trailing_fallback_marker_stays_a_single_entry(self):
+        # Ordinary case: a natural completion followed only by normal inter-turn silence
+        # (assistant response + user think time) before the next window's vad_start - with
+        # no turn_fallback marker as evidence a turn got stuck in the gap, no trailing entry
+        # should be fabricated for window 0 itself. Otherwise every ordinary turn transition
+        # would be misread as containing a stuck turn. (Window 1, the next raw VAD-start
+        # window, legitimately resolves "stuck" on its own since it has no signal at all -
+        # that's unrelated to this fix.)
+        result = _classify_turns(
+            vad_starts=[1000, 60000],
+            vad_stops=[],
+            turn_metrics_events=[{"timestamp": 1103, "is_complete": True, "e2e_processing_time_ms": 103.0}],
+            stop_secs_silence_ms=[],
+            turn_fallback_timestamps=[],
+        )
+        assert result[0] == {
+            "turn_index": 0,
+            "open_duration_ms": 103,
+            "vad_time_to_complete_ms": 103.0,
+            "completion": "natural",
+        }
+        assert len(result) == 2
+        assert result[1]["completion"] == "stuck"
+
+    def test_natural_completion_followed_by_stuck_fallback_in_open_ended_window(self):
+        # Same shape as above but in the run's final, open-ended window (window_end=None):
+        # close_time must fall back to the fallback marker's own timestamp rather than a
+        # nonexistent next-window bound.
+        result = _classify_turns(
+            vad_starts=[1000],
+            vad_stops=[],
+            turn_metrics_events=[{"timestamp": 1103, "is_complete": True, "e2e_processing_time_ms": 103.0}],
+            stop_secs_silence_ms=[],
+            turn_fallback_timestamps=[45000],
+        )
+        assert result[1] == {
+            "turn_index": 1,
+            "open_duration_ms": 45000 - 1103,
+            "vad_time_to_complete_ms": None,
+            "completion": "stuck",
+        }
+
+    def test_only_last_split_segment_in_a_multi_natural_window_can_be_early(self):
+        # Two naturals split out of one raw window. Only the second (the one bordering the
+        # next VAD-start window) is eligible for reclassification - the first is followed by
+        # another natural completion in the same window, not a missing turn_start.
+        result = _classify_turns(
+            vad_starts=[1000, 5000],
+            vad_stops=[],
+            turn_metrics_events=[
+                {"timestamp": 1001, "is_complete": True, "e2e_processing_time_ms": 100.0},
+                {"timestamp": 2000, "is_complete": True, "e2e_processing_time_ms": 100.0},
+                {"timestamp": 5103, "is_complete": True, "e2e_processing_time_ms": 100.0},
+            ],
+            stop_secs_silence_ms=[],
+            turn_start_timestamps={1000},
+        )
+        assert result[0]["completion"] == "natural"
+        assert result[1]["completion"] == "early"
+        assert result[2]["completion"] == "natural"
+
+
+def _write_config(tmp_path, framework="pipecat", turn_stop_strategy="turn_analyzer"):
+    (tmp_path / "config.json").write_text(
+        json.dumps({"framework": framework, "model": {"turn_stop_strategy": turn_stop_strategy}})
+    )
+
+
+class TestComputeVadTurnSubMetrics:
+    def test_none_when_not_applicable(self, tmp_path):
+        _write_config(tmp_path, framework="elevenlabs")
+        ctx = make_metric_context(output_dir=str(tmp_path))
+        assert compute_vad_turn_sub_metrics(ctx) is None
+
+    def test_none_when_no_vad_starts_recorded(self, tmp_path):
+        _write_config(tmp_path)
+        ctx = make_metric_context(output_dir=str(tmp_path))
+        assert compute_vad_turn_sub_metrics(ctx) is None
+
+    def test_krisp_run_all_natural_omits_forced_completion_rate(self, tmp_path):
+        _write_config(tmp_path, turn_stop_strategy="krisp_viva_turn")
+        _write_jsonl(tmp_path / "pipecat_logs.jsonl", [{"timestamp": 1000, "type": "user_started_speaking"}])
+        _write_jsonl(
+            tmp_path / "pipecat_metrics.jsonl",
+            [
+                {
+                    "timestamp": 1100,
+                    "type": "TurnMetricsData",
+                    "value": {"is_complete": True, "e2e_processing_time_ms": 100.0},
+                }
+            ],
+        )
+        ctx = make_metric_context(output_dir=str(tmp_path))
+        result = compute_vad_turn_sub_metrics(ctx)
+        assert result is not None
+        sub_metrics, per_turn = result
+        assert "forced_completion_rate" not in sub_metrics
+        assert sub_metrics["mean_vad_time_to_complete_ms"].score == 100.0
+        assert per_turn == [
+            {"turn_index": 0, "open_duration_ms": 100, "vad_time_to_complete_ms": 100.0, "completion": "natural"}
+        ]
+
+    def test_smart_turn_run_all_natural_reports_zero_forced_completion_rate(self, tmp_path):
+        _write_config(tmp_path, turn_stop_strategy="turn_analyzer")
+        _write_jsonl(tmp_path / "pipecat_logs.jsonl", [{"timestamp": 1000, "type": "user_started_speaking"}])
+        _write_jsonl(
+            tmp_path / "pipecat_metrics.jsonl",
+            [
+                {
+                    "timestamp": 1100,
+                    "type": "TurnMetricsData",
+                    "value": {"is_complete": True, "e2e_processing_time_ms": 100.0},
+                }
+            ],
+        )
+        ctx = make_metric_context(output_dir=str(tmp_path))
+        result = compute_vad_turn_sub_metrics(ctx)
+        assert result is not None
+        sub_metrics, per_turn = result
+        assert "forced_completion_rate" in sub_metrics
+        assert sub_metrics["forced_completion_rate"].score == 0.0
+        assert sub_metrics["mean_vad_time_to_complete_ms"].score == 100.0
+        assert per_turn == [
+            {"turn_index": 0, "open_duration_ms": 100, "vad_time_to_complete_ms": 100.0, "completion": "natural"}
+        ]
+
+    def test_smart_turn_run_with_forced_completion(self, tmp_path):
+        _write_config(tmp_path, turn_stop_strategy="turn_analyzer")
+        _write_jsonl(
+            tmp_path / "pipecat_logs.jsonl",
+            [
+                {"timestamp": 1000, "type": "user_started_speaking"},
+                {"timestamp": 1050, "type": "user_stopped_speaking"},
+                {"timestamp": 5000, "type": "user_started_speaking"},
+                {"timestamp": 5103, "type": "user_stopped_speaking"},
+            ],
+        )
+        _write_jsonl(
+            tmp_path / "pipecat_metrics.jsonl",
+            [
+                {
+                    "timestamp": 5203,
+                    "type": "TurnMetricsData",
+                    "value": {"is_complete": True, "e2e_processing_time_ms": 100.0},
+                }
+            ],
+        )
+        (tmp_path / "logs.log").write_text(
+            "2026-08-04 19:31:15,912 | DEBUG | pipecat:133 | End of Turn complete due to stop_secs. Silence in ms: 3032.0\n"
+        )
+        ctx = make_metric_context(output_dir=str(tmp_path))
+        result = compute_vad_turn_sub_metrics(ctx)
+        assert result is not None
+        sub_metrics, per_turn = result
+        assert sub_metrics["forced_completion_rate"].score == 0.5
+        assert per_turn[0]["completion"] == "forced"
+        assert per_turn[1]["completion"] == "natural"
+        # Only the natural completion feeds mean_vad_time_to_complete_ms.
+        assert sub_metrics["mean_vad_time_to_complete_ms"].score == 100.0
+
+    def test_dispatched_turn_excluded_from_stuck_rate(self, tmp_path):
+        # Reproduces example: no TurnMetricsData is_complete=True and no
+        # stop_secs line for the final window, but audit_log.json proves the LLM was
+        # dispatched and responded - so this must not count toward stuck_rate.
+        _write_config(tmp_path, turn_stop_strategy="turn_analyzer")
+        _write_jsonl(tmp_path / "pipecat_logs.jsonl", [{"timestamp": 1000, "type": "user_started_speaking"}])
+        _write_jsonl(
+            tmp_path / "pipecat_metrics.jsonl",
+            [
+                {
+                    "timestamp": 4892,
+                    "type": "TurnMetricsData",
+                    "value": {"is_complete": False, "e2e_processing_time_ms": 104.0},
+                }
+            ],
+        )
+        _write_audit_log(tmp_path / "audit_log.json", [(5912, "I do not need anything else today. Bye.")])
+        ctx = make_metric_context(output_dir=str(tmp_path))
+        result = compute_vad_turn_sub_metrics(ctx)
+        assert result is not None
+        sub_metrics, per_turn = result
+        assert per_turn[0]["completion"] == "dispatched"
+        assert sub_metrics["vad_stuck_rate"].score == 0.0
+
+    def test_stuck_rate_counts_mid_conversation_hangs_regardless_of_outcome(self, tmp_path):
+        _write_config(tmp_path, turn_stop_strategy="krisp_viva_turn")
+        _write_jsonl(tmp_path / "pipecat_logs.jsonl", [{"timestamp": 1000, "type": "user_started_speaking"}])
+        ctx = make_metric_context(
+            output_dir=str(tmp_path),
+            conversation_ended_reason="goodbye",  # the call still ended cleanly
+            audio_timestamps_user_turns={1: [(0.0, 5.0)]},
+            audio_timestamps_assistant_turns={},
+        )
+        result = compute_vad_turn_sub_metrics(ctx)
+        assert result is not None
+        sub_metrics, per_turn = result
+        assert per_turn[0]["completion"] == "stuck"
+        assert sub_metrics["vad_stuck_rate"].score == 1.0
+
+    def test_stuck_rate_includes_final_turn_with_no_vad_start_at_all(self, tmp_path):
+        # Real Krisp shape: 5 real turns all completed naturally, but the 6th and final
+        # real user turn never got a user_started_speaking event from the turn analyzer at
+        # all - no window exists for it in per_turn, so it must be folded in explicitly to
+        # be counted, rather than silently vanishing from stuck_rate's denominator.
+        _write_config(tmp_path, turn_stop_strategy="krisp_viva_turn")
+        _write_jsonl(tmp_path / "pipecat_logs.jsonl", [{"timestamp": 1000, "type": "user_started_speaking"}])
+        _write_jsonl(
+            tmp_path / "pipecat_metrics.jsonl",
+            [
+                {
+                    "timestamp": 1100,
+                    "type": "TurnMetricsData",
+                    "value": {"is_complete": True, "e2e_processing_time_ms": 100.0},
+                }
+            ],
+        )
+        ctx = make_metric_context(
+            output_dir=str(tmp_path),
+            conversation_ended_reason="inactivity_timeout",
+            # Last real user turn started at 30s (30000ms) - over 28s away from the only
+            # vad_start (1000ms), far past legitimate clock skew.
+            audio_timestamps_user_turns={1: [(0.0, 0.5)], 2: [(30.0, 30.5)]},
+            audio_timestamps_assistant_turns={0: [(0.6, 1.0)]},
+        )
+        result = compute_vad_turn_sub_metrics(ctx)
+        assert result is not None
+        sub_metrics, per_turn = result
+        assert per_turn[0]["completion"] == "natural"
+        assert per_turn[-1] == {
+            "turn_index": 1,
+            "open_duration_ms": None,
+            "vad_time_to_complete_ms": None,
+            "completion": "stuck",
+            "vad_never_started": True,
+        }
+        # 1 stuck (the synthetic missing-turn entry) out of 2 total turns.
+        assert sub_metrics["vad_stuck_rate"].score == 0.5
+
+    def test_stuck_rate_does_not_add_synthetic_entry_within_clock_skew_tolerance(self, tmp_path):
+        # The only vad_start is within clock-skew tolerance of the last real user turn, so
+        # the synthetic missing-turn entry is NOT added - this call had exactly one real
+        # turn, and it's classified stuck for an unrelated reason (no completion signal),
+        # not because VAD never started at all.
+        _write_config(tmp_path, turn_stop_strategy="krisp_viva_turn")
+        _write_jsonl(tmp_path / "pipecat_logs.jsonl", [{"timestamp": 1000, "type": "user_started_speaking"}])
+        ctx = make_metric_context(
+            output_dir=str(tmp_path),
+            conversation_ended_reason="inactivity_timeout",
+            audio_timestamps_user_turns={1: [(0.0, 0.5)]},
+            audio_timestamps_assistant_turns={},
+        )
+        result = compute_vad_turn_sub_metrics(ctx)
+        assert result is not None
+        sub_metrics, per_turn = result
+        assert per_turn == [
+            {"turn_index": 0, "open_duration_ms": 0, "vad_time_to_complete_ms": None, "completion": "stuck"}
+        ]
+        assert sub_metrics["vad_stuck_rate"].score == 1.0
+
+    def test_early_detection_rate_reported_when_a_natural_completion_is_undone(self, tmp_path):
+        _write_config(tmp_path, turn_stop_strategy="turn_analyzer")
+        _write_jsonl(
+            tmp_path / "pipecat_logs.jsonl",
+            [
+                {"timestamp": 1000, "type": "turn_start"},
+                {"timestamp": 1000, "type": "user_started_speaking"},
+                {"timestamp": 5000, "type": "user_started_speaking"},
+                # No turn_start at 5000: pipecat kept the turn open through the split.
+                {"timestamp": 9000, "type": "turn_start"},
+                {"timestamp": 9000, "type": "user_started_speaking"},
+            ],
+        )
+        _write_jsonl(
+            tmp_path / "pipecat_metrics.jsonl",
+            [
+                {
+                    "timestamp": 1103,
+                    "type": "TurnMetricsData",
+                    "value": {"is_complete": True, "e2e_processing_time_ms": 103.0},
+                },
+                {
+                    "timestamp": 5103,
+                    "type": "TurnMetricsData",
+                    "value": {"is_complete": True, "e2e_processing_time_ms": 103.0},
+                },
+                {
+                    "timestamp": 9103,
+                    "type": "TurnMetricsData",
+                    "value": {"is_complete": True, "e2e_processing_time_ms": 103.0},
+                },
+            ],
+        )
+        ctx = make_metric_context(output_dir=str(tmp_path))
+        result = compute_vad_turn_sub_metrics(ctx)
+        assert result is not None
+        sub_metrics, per_turn = result
+        assert [t["completion"] for t in per_turn] == ["early", "natural", "natural"]
+        assert sub_metrics["vad_early_detection_rate"].score == pytest.approx(1 / 3, abs=1e-4)
+
+    def test_early_detection_rate_absent_when_no_turn_start_signal_available(self, tmp_path):
+        # No turn_start events recorded at all - early detection has no signal to work
+        # with, so it must not fabricate a rate off an empty turn_start set.
+        _write_config(tmp_path, turn_stop_strategy="krisp_viva_turn")
+        _write_jsonl(tmp_path / "pipecat_logs.jsonl", [{"timestamp": 1000, "type": "user_started_speaking"}])
+        _write_jsonl(
+            tmp_path / "pipecat_metrics.jsonl",
+            [
+                {
+                    "timestamp": 1100,
+                    "type": "TurnMetricsData",
+                    "value": {"is_complete": True, "e2e_processing_time_ms": 100.0},
+                }
+            ],
+        )
+        ctx = make_metric_context(output_dir=str(tmp_path))
+        result = compute_vad_turn_sub_metrics(ctx)
+        assert result is not None
+        sub_metrics, per_turn = result
+        assert per_turn[0]["completion"] == "natural"
+        assert sub_metrics["vad_early_detection_rate"].score == 0.0
+
+    def test_stuck_rate_counts_fallback_stuck_turn_hidden_behind_earlier_natural_completion(self, tmp_path):
+        # Reproduces example: window 0 has an earlier,
+        # unrelated natural completion, but the turn analyzer's VAD then went stuck on a
+        # separate turn that only got resolved via the 20s turn-end fallback nudge -
+        # recorded in audit_log.json as a turn_fallback entry, not a real "user" dispatch.
+        # Must count toward stuck_rate rather than vanishing because the window already
+        # resolved once.
+        _write_config(tmp_path, turn_stop_strategy="turn_analyzer")
+        _write_jsonl(tmp_path / "pipecat_logs.jsonl", [{"timestamp": 1000, "type": "user_started_speaking"}])
+        _write_jsonl(
+            tmp_path / "pipecat_metrics.jsonl",
+            [
+                {
+                    "timestamp": 1103,
+                    "type": "TurnMetricsData",
+                    "value": {"is_complete": True, "e2e_processing_time_ms": 103.0},
+                }
+            ],
+        )
+        (tmp_path / "audit_log.json").write_text(
+            json.dumps(
+                {
+                    "transcript": [
+                        {
+                            "value": "[TURN-END FALLBACK after 20s] partial user speech: 'Mac fourteen inch.'",
+                            "message_type": "turn_fallback",
+                            "timestamp": "45000",
+                        }
+                    ]
+                }
+            )
+        )
+        ctx = make_metric_context(output_dir=str(tmp_path))
+        result = compute_vad_turn_sub_metrics(ctx)
+        assert result is not None
+        sub_metrics, per_turn = result
+        assert [t["completion"] for t in per_turn] == ["natural", "stuck"]
+        assert sub_metrics["vad_stuck_rate"].score == 0.5
+
+    def test_stuck_rate_zero_on_clean_all_natural_ending(self, tmp_path):
+        _write_config(tmp_path, turn_stop_strategy="krisp_viva_turn")
+        _write_jsonl(tmp_path / "pipecat_logs.jsonl", [{"timestamp": 1000, "type": "user_started_speaking"}])
+        _write_jsonl(
+            tmp_path / "pipecat_metrics.jsonl",
+            [
+                {
+                    "timestamp": 1100,
+                    "type": "TurnMetricsData",
+                    "value": {"is_complete": True, "e2e_processing_time_ms": 100.0},
+                }
+            ],
+        )
+        ctx = make_metric_context(output_dir=str(tmp_path), conversation_ended_reason="goodbye")
+        result = compute_vad_turn_sub_metrics(ctx)
+        assert result is not None
+        sub_metrics, _ = result
+        assert sub_metrics["vad_stuck_rate"].score == 0.0