feat(realtime): adapt semantic endpointing controls

CHANGELOG.md

@@ -53,6 +53,16 @@ named rather than smoothed.
   if they cannot be proven; `--json` prints it instead. The bug-report issue
   template now asks for the bundle first. Ported idea from
   bielcarpi/hermes-live-voice's `diagnostics` (MIT) — idea only, no code.
+- Added a deterministic, content-free endpointing trace evaluator that reports
+  nearest-rank endpoint/playback latency and classified cutoff, split, timeout,
+  and false-activation counts without retaining audio or transcripts.
+- Provider-neutral semantic endpointing controls for the core realtime
+  contract: `RealtimeTurnDetection` with native/server/semantic modes and
+  `RealtimeSemanticEagerness` are mirrored through the setup and live-update
+  paths, Grok advertises native/server support and Gemini native-only with
+  fail-fast refusals on unsupported modes, and the semantic names are probed
+  as an optional capability so pre-semantic core heads keep the full lane
+  with provider-native turn detection instead of losing it.
 - hermes-talk's three realtime lanes now register on the Hermes core
   `RealtimeVoiceProvider` contract (`agent/realtime_voice_provider.py`, API v2 —
   NousResearch/hermes-agent#101808) as `hermes-talk/openai`,

pyproject.toml

@@ -52,6 +52,7 @@ py-modules = [
   "talk_grok_realtime",
   "talk_gemini_realtime",
   "talk_core_realtime",
+  "talk_endpointing_benchmark",
   "talk_core_provider",
   "talk_core_session",
   "talk_discord",
@@ -101,6 +102,7 @@ known-first-party = [
   "talk_diagnostics",
   "talk_discord",
   "talk_doctor",
+  "talk_endpointing_benchmark",
   "talk_gemini_realtime",
   "talk_grok_auth",
   "talk_grok_realtime",

talk_core_provider.py

@@ -108,6 +108,37 @@
     # poison hermes-talk's own imports. Every released Hermes lands here.
     _CORE_IMPORT_ERROR = exc
 
+_TURN_DETECTION_IMPORT_ERROR: BaseException | None = None
+try:
+    from agent.realtime_voice_provider import (
+        RealtimeSemanticEagerness,
+        RealtimeTurnDetection,
+        RealtimeTurnDetectionMode,
+    )
+except Exception as exc:  # noqa: BLE001 - pre-semantic core heads lack these names
+    _TURN_DETECTION_IMPORT_ERROR = exc
+    RealtimeSemanticEagerness = None  # type: ignore[assignment]
+    RealtimeTurnDetection = None  # type: ignore[assignment]
+    RealtimeTurnDetectionMode = None  # type: ignore[assignment]
+
+
+def turn_detection_available() -> bool:
+    """True when the host contract carries the semantic turn-detection names.
+
+    A pre-semantic core head (today's #101808) still gets the full core lane;
+    only turn-detection controls degrade to provider-native. Never raises.
+    """
+
+    return _CORE_IMPORT_ERROR is None and _TURN_DETECTION_IMPORT_ERROR is None
+
+
+def _contract_turn_modes(*names: str) -> frozenset:
+    """Contract turn-detection modes by member name; empty on pre-semantic heads."""
+
+    if not turn_detection_available():
+        return frozenset()
+    return frozenset(getattr(RealtimeTurnDetectionMode, name) for name in names)
+
 
 def core_contract_available() -> bool:
     """True when this host exposes the exact core realtime contract we target."""
@@ -117,10 +148,10 @@ def core_contract_available() -> bool:
 
 def core_contract_diagnostic() -> dict[str, Any]:
     """Read-only receipt for ``talk doctor`` / ``talk_status``. Never raises."""
-
     available = core_contract_available()
     return {
         "contract_available": available,
+        "turn_detection_available": turn_detection_available(),
         "provider_names": list(PROVIDER_NAMES),
         "detail": "" if available else type(_CORE_IMPORT_ERROR).__name__,
     }
@@ -355,6 +386,8 @@ class _TalkCoreProvider(RealtimeVoiceProvider):
         input_audio = PCM16_24K
         output_audio = PCM16_24K
 
+        supported_turn_detection_modes = _contract_turn_modes("PROVIDER_NATIVE")
+
         def __init__(
             self,
             *,
@@ -390,6 +423,34 @@ def get_setup_schema(self) -> Mapping[str, Any]:
                 ),
             }
 
+        def _talk_turn_detection(self, turn_detection: Any) -> rt.RealtimeTurnDetection:
+            if turn_detection is None:
+                # Pre-semantic host: the setup carries no turn_detection at all.
+                return rt.RealtimeTurnDetection()
+            if turn_detection.mode not in self.supported_turn_detection_modes:
+                raise ValueError(
+                    f"{self.display_name} does not support turn detection mode "
+                    f"{turn_detection.mode.value}"
+                )
+            modes = {
+                RealtimeTurnDetectionMode.PROVIDER_NATIVE: (
+                    rt.RealtimeTurnDetectionMode.PROVIDER_NATIVE
+                ),
+                RealtimeTurnDetectionMode.SERVER_VAD: (rt.RealtimeTurnDetectionMode.SERVER_VAD),
+                RealtimeTurnDetectionMode.SEMANTIC_VAD: (rt.RealtimeTurnDetectionMode.SEMANTIC_VAD),
+            }
+            eagerness = {
+                None: None,
+                RealtimeSemanticEagerness.AUTO: rt.RealtimeSemanticEagerness.AUTO,
+                RealtimeSemanticEagerness.LOW: rt.RealtimeSemanticEagerness.LOW,
+                RealtimeSemanticEagerness.MEDIUM: rt.RealtimeSemanticEagerness.MEDIUM,
+                RealtimeSemanticEagerness.HIGH: rt.RealtimeSemanticEagerness.HIGH,
+            }
+            return rt.RealtimeTurnDetection(
+                mode=modes[turn_detection.mode],
+                semantic_eagerness=eagerness[turn_detection.semantic_eagerness],
+            )
+
         def _talk_setup(self, setup: RealtimeVoiceSetup) -> Any:
             for label, requested, expected in (
                 ("input", setup.input_audio, self.input_audio),
@@ -412,12 +473,13 @@ def _talk_setup(self, setup: RealtimeVoiceSetup) -> Any:
                     )
                     for tool in setup.tools
                 ),
-                automatic_response=setup.automatic_response,
+                turn_detection=self._talk_turn_detection(getattr(setup, "turn_detection", None)),
             )
 
         async def open_session(self, setup: RealtimeVoiceSetup) -> RealtimeVoiceSession:
             # Shape first: an unusable setup is refused before any credential is
             # resolved and before a socket is opened.
+            self.validate_setup(setup)
             talk_setup = self._talk_setup(setup)
             resolver = self._auth_resolver or self._resolve_auth
             auth = resolver()
@@ -466,6 +528,9 @@ class TalkOpenAICoreProvider(_TalkCoreProvider):
         provider_tag = "gpt-realtime speech-to-speech"
         env_key = "OPENAI_API_KEY"
         env_url = "https://platform.openai.com/api-keys"
+        supported_turn_detection_modes = _contract_turn_modes(
+            "PROVIDER_NATIVE", "SERVER_VAD", "SEMANTIC_VAD"
+        )
 
         def default_model(self) -> str | None:
             return talk_config.talk_model() or talk_config.DEFAULT_TALK_MODEL
@@ -527,6 +592,9 @@ class TalkGrokCoreProvider(_TalkCoreProvider):
         provider_tag = "grok-voice speech-to-speech"
         env_key = "XAI_API_KEY"
         env_url = "https://console.x.ai"
+        supported_turn_detection_modes = _contract_turn_modes(
+            "PROVIDER_NATIVE", "SERVER_VAD"
+        )
 
         def default_model(self) -> str | None:
             return talk_config.talk_grok_model() or talk_config.DEFAULT_GROK_MODEL
@@ -684,8 +752,8 @@ def build_providers() -> tuple[Any, ...]:
     "TalkGrokCoreProvider",
     "TalkOpenAICoreProvider",
     "build_providers",
-    "core_contract_available",
     "core_contract_diagnostic",
     "redact",
     "translate_event",
+    "turn_detection_available",
 ]

talk_endpointing_benchmark.py

@@ -0,0 +1,174 @@
+"""Deterministic, content-free metrics for endpointing trace fixtures.
+
+Times are monotonic integer milliseconds chosen by the caller.  This module has
+no clock, provider, audio, or transcript dependency and deliberately stores no
+media or text.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from dataclasses import dataclass
+from math import ceil
+
+
+@dataclass(frozen=True, slots=True)
+class EndpointingTrace:
+    """Receipts for one annotated opportunity to activate.
+
+    ``annotated_utterance_end_ms=None`` describes a period in which no utterance
+    was annotated; any provider or playback receipt then counts as one false
+    activation.  A timeout is terminal but is not itself an activation.  A
+    speech-stop before the annotation is a premature cutoff, and a turn-end
+    before it is a false split.  ``timed_out_at_ms`` is an explicit terminal
+    receipt, not a timeout inferred from missing data.
+    """
+
+    trace_id: str
+    annotated_utterance_end_ms: int | None = None
+    provider_speech_stop_ms: int | None = None
+    provider_turn_end_ms: int | None = None
+    first_audio_ms: int | None = None
+    local_playback_ms: int | None = None
+    timed_out_at_ms: int | None = None
+
+    def __post_init__(self) -> None:
+        if not isinstance(self.trace_id, str) or not self.trace_id.strip():
+            raise ValueError("trace_id must be a non-empty string")
+
+        names = (
+            "annotated_utterance_end_ms",
+            "provider_speech_stop_ms",
+            "provider_turn_end_ms",
+            "first_audio_ms",
+            "local_playback_ms",
+            "timed_out_at_ms",
+        )
+        for name in names:
+            value = getattr(self, name)
+            if value is not None and (
+                isinstance(value, bool) or not isinstance(value, int) or value < 0
+            ):
+                raise ValueError(f"{name} must be a non-negative integer or None")
+
+        stop = self.provider_speech_stop_ms
+        turn = self.provider_turn_end_ms
+        audio = self.first_audio_ms
+        playback = self.local_playback_ms
+        timeout = self.timed_out_at_ms
+        annotated = self.annotated_utterance_end_ms
+
+        if stop is not None and turn is not None and stop > turn:
+            raise ValueError("provider speech-stop cannot follow provider turn-end")
+        if audio is not None and turn is None:
+            raise ValueError("first-audio requires a provider turn-end receipt")
+        if turn is not None and audio is not None and turn > audio:
+            raise ValueError("provider turn-end cannot follow first-audio")
+        if playback is not None and audio is None:
+            raise ValueError("local-playback requires a first-audio receipt")
+        if audio is not None and playback is not None and audio > playback:
+            raise ValueError("first-audio cannot follow local-playback")
+        if stop is not None and timeout is not None and stop > timeout:
+            raise ValueError("provider speech-stop cannot follow timeout")
+        if timeout is not None and any(value is not None for value in (turn, audio, playback)):
+            raise ValueError("timeout and completion receipts are mutually exclusive")
+        if timeout is not None and annotated is not None and timeout < annotated:
+            raise ValueError("timeout cannot precede the annotated utterance end")
+
+
+@dataclass(frozen=True, slots=True)
+class LatencyDistribution:
+    """Nearest-rank latency summary.
+
+    For ``n`` sorted samples, percentile ``p`` selects the 1-based item at
+    ``ceil(p * n)``.  Empty distributions report ``None`` rather than inventing
+    zero latency.
+    """
+
+    count: int
+    p50_ms: int | None
+    p95_ms: int | None
+    max_ms: int | None
+
+
+@dataclass(frozen=True, slots=True)
+class EndpointingSummary:
+    trace_count: int
+    endpoint_latency: LatencyDistribution
+    playback_latency: LatencyDistribution
+    premature_cutoff_count: int
+    false_split_count: int
+    timeout_count: int
+    false_activation_count: int
+
+
+def _distribution(samples: Iterable[int]) -> LatencyDistribution:
+    ordered = sorted(samples)
+    if not ordered:
+        return LatencyDistribution(count=0, p50_ms=None, p95_ms=None, max_ms=None)
+
+    def nearest_rank(percentile: float) -> int:
+        return ordered[ceil(percentile * len(ordered)) - 1]
+
+    return LatencyDistribution(
+        count=len(ordered),
+        p50_ms=nearest_rank(0.50),
+        p95_ms=nearest_rank(0.95),
+        max_ms=ordered[-1],
+    )
+
+
+def evaluate_endpointing_traces(traces: Iterable[EndpointingTrace]) -> EndpointingSummary:
+    """Evaluate immutable receipts without consulting a live provider.
+
+    Endpoint latency runs from annotated utterance end to provider turn-end.
+    Premature (negative) turn-ends are classified as false splits and excluded
+    from latency.  Playback latency runs from first-audio receipt to the local
+    playback receipt, isolating local delivery from model response time.
+    """
+
+    materialized = tuple(traces)
+    if any(not isinstance(trace, EndpointingTrace) for trace in materialized):
+        raise TypeError("traces must contain EndpointingTrace records")
+    ids = tuple(trace.trace_id for trace in materialized)
+    if len(ids) != len(set(ids)):
+        raise ValueError("trace_id values must be unique")
+
+    endpoint_latencies: list[int] = []
+    playback_latencies: list[int] = []
+    premature_cutoffs = 0
+    false_splits = 0
+    timeouts = 0
+    false_activations = 0
+
+    for trace in materialized:
+        annotated = trace.annotated_utterance_end_ms
+        activation_receipts = (
+            trace.provider_speech_stop_ms,
+            trace.provider_turn_end_ms,
+            trace.first_audio_ms,
+            trace.local_playback_ms,
+        )
+        if annotated is None:
+            false_activations += int(any(receipt is not None for receipt in activation_receipts))
+        else:
+            stop = trace.provider_speech_stop_ms
+            turn = trace.provider_turn_end_ms
+            premature_cutoffs += int(stop is not None and stop < annotated)
+            false_splits += int(turn is not None and turn < annotated)
+            if turn is not None and turn >= annotated:
+                endpoint_latencies.append(turn - annotated)
+
+        if trace.first_audio_ms is not None and trace.local_playback_ms is not None:
+            playback_latencies.append(trace.local_playback_ms - trace.first_audio_ms)
+        timeouts += int(trace.timed_out_at_ms is not None)
+
+    return EndpointingSummary(
+        trace_count=len(materialized),
+        endpoint_latency=_distribution(endpoint_latencies),
+        playback_latency=_distribution(playback_latencies),
+        premature_cutoff_count=premature_cutoffs,
+        false_split_count=false_splits,
+        timeout_count=timeouts,
+        false_activation_count=false_activations,
+    )

talk_gemini_realtime.py

@@ -212,6 +212,15 @@ def _tool_wire(tool: rt.ToolDefinition) -> dict[str, Any]:
     }
 
 
+def _validate_turn_detection(setup: rt.SessionSetup) -> None:
+    turn_detection = setup.turn_detection
+    if turn_detection.mode is not rt.RealtimeTurnDetectionMode.PROVIDER_NATIVE:
+        raise rt.RealtimeSessionError(
+            "Gemini Live supports only provider-native turn detection; "
+            f"{turn_detection.mode.value} is refused rather than degraded"
+        )
+
+
 def build_setup_message(setup: rt.SessionSetup) -> dict[str, Any]:
     """Map neutral setup to the Live ``setup`` message.
 
@@ -221,6 +230,7 @@ def build_setup_message(setup: rt.SessionSetup) -> dict[str, Any]:
     deliberately absent because touching it can only narrow detection, never
     reproduce the OpenAI lane's ``create_response`` gating.
     """
+    _validate_turn_detection(setup)
 
     payload: dict[str, Any] = {
         "model": _wire_model(setup.model),
@@ -520,6 +530,11 @@ def __init__(self, *, auth_token: str, auth_source: str, aiohttp_module=None) ->
     async def connect(self, setup: rt.SessionSetup) -> None:
         if self.state is not rt.SessionState.NEW:
             raise rt.RealtimeSessionError("Realtime session connect may only run once")
+        try:
+            _validate_turn_detection(setup)
+        except rt.RealtimeSessionError:
+            self.state = rt.SessionState.FAILED
+            raise
         self.state = rt.SessionState.CONNECTING
         if not setup.automatic_response:
             # The gated-response flow (Discord's authorization ledger) needs