Instrument request latency by verification stage

app.py

@@ -126,6 +126,11 @@ from glyph_hybrid_adapter import (
     HardenedGlyphHybridAdapter,
 )
 from glyph_hybrid_evidence import GatePolicy as GlyphEvidenceGatePolicy
+from latency_timing import (
+    latency_stage,
+    timed_latency_request,
+    timed_latency_stage,
+)
 from production import (
     StopHysteresisController,
     GenerationChunkSpec,
@@ -532,25 +537,30 @@ def _get_ecapa_encoder():
     global _ECAPA_ENCODER
     with _ECAPA_LOCK:
         if _ECAPA_ENCODER is None:
-            import torchaudio
-
-            # SpeechBrain 1.0.3 still probes this API during import, while the
-            # ZeroGPU torchaudio build has removed it. Audio loading below is
-            # handled by librosa, so an empty compatibility result is correct.
-            if not hasattr(torchaudio, "list_audio_backends"):
-                torchaudio.list_audio_backends = lambda: []
-            from speechbrain.inference.speaker import EncoderClassifier
-
-            cache_dir = os.path.join(os.environ.get("HF_HOME", "/tmp"), "speechbrain", "ecapa")
-            _ECAPA_ENCODER = EncoderClassifier.from_hparams(
-                source=ECAPA_DIR,
-                # The upstream hyperparams otherwise points back to the repo
-                # and SpeechBrain 1.0.3 uses a removed Hub keyword.  Keep every
-                # weight fetch inside the already pinned local snapshot.
-                overrides={"pretrained_path": ECAPA_DIR},
-                savedir=cache_dir,
-                run_opts={"device": "cpu"},
-            )
+            with latency_stage("ecapa"):
+                import torchaudio
+
+                # SpeechBrain 1.0.3 still probes this API during import, while the
+                # ZeroGPU torchaudio build has removed it. Audio loading below is
+                # handled by librosa, so an empty compatibility result is correct.
+                if not hasattr(torchaudio, "list_audio_backends"):
+                    torchaudio.list_audio_backends = lambda: []
+                from speechbrain.inference.speaker import EncoderClassifier
+
+                cache_dir = os.path.join(
+                    os.environ.get("HF_HOME", "/tmp"),
+                    "speechbrain",
+                    "ecapa",
+                )
+                _ECAPA_ENCODER = EncoderClassifier.from_hparams(
+                    source=ECAPA_DIR,
+                    # The upstream hyperparams otherwise points back to the repo
+                    # and SpeechBrain 1.0.3 uses a removed Hub keyword. Keep every
+                    # weight fetch inside the already pinned local snapshot.
+                    overrides={"pretrained_path": ECAPA_DIR},
+                    savedir=cache_dir,
+                    run_opts={"device": "cpu"},
+                )
     return _ECAPA_ENCODER
 
 
@@ -756,11 +766,15 @@ def _generate_chunk(
             expected_steps=expected_steps,
             hard_stop_steps=hard_stop_steps,
         )
-    try:
-        audio = model.generate(**kwargs)
-    finally:
-        if _STOP_CONTROLLER is not None:
-            _STOP_CONTROLLER.end()
+    with latency_stage("generation"):
+        try:
+            audio = model.generate(**kwargs)
+        finally:
+            if _STOP_CONTROLLER is not None:
+                _STOP_CONTROLLER.end()
+        # Copying to CPU closes the asynchronous CUDA work inside this stage,
+        # so the reported wall time is not shifted into later verification.
+        audio = audio.detach().float().cpu().numpy().reshape(-1)
     stop_reason = (
         _STOP_CONTROLLER.last_stop_reason
         if _STOP_CONTROLLER is not None
@@ -787,8 +801,6 @@ def _generate_chunk(
         f"min_len={min_len} "
         f"short_headroom_floor_applied={short_headroom_floor_applied}"
     )
-    audio = audio.detach().float().cpu().numpy().reshape(-1)
-
     def outcome(waveform: np.ndarray) -> _ChunkGenerationOutcome:
         normalized = np.asarray(waveform, dtype=np.float32).reshape(-1)
         return _ChunkGenerationOutcome(
@@ -1569,6 +1581,7 @@ def _local_endpoint_role_rows(
     )
 
 
+@timed_latency_stage("assemble")
 def _assemble_trajectory_audio(
     trajectory: tuple[np.ndarray, ...],
     chunks: tuple[str, ...],
@@ -3428,14 +3441,15 @@ def _synthesize_glyph_hybrid(
         immutable_segments,
         semantic_commitment=commitment,
     )
-    assembly = assemble_hybrid_pcm16(
-        profile_state.bundle,
-        immutable_segments,
-        semantic_commitment=commitment,
-        expected_plan_sha256=assembly_plan_sha256,
-        speed=1.0,
-        post_gain=1.0,
-    )
+    with latency_stage("assemble"):
+        assembly = assemble_hybrid_pcm16(
+            profile_state.bundle,
+            immutable_segments,
+            semantic_commitment=commitment,
+            expected_plan_sha256=assembly_plan_sha256,
+            speed=1.0,
+            post_gain=1.0,
+        )
     verify_hybrid_assembly(
         assembly,
         trusted_bundle=profile_state.bundle,
@@ -3551,6 +3565,7 @@ def _maybe_synthesize_glyph_speaker_request(
 
 
 @gpu
+@timed_latency_request
 def tts_speaker(
     text: str,
     speaker: str = DEFAULT_SPEAKER,
@@ -3579,6 +3594,7 @@ def tts_speaker(
 
 
 @gpu
+@timed_latency_request
 def tts_reference(
     text: str,
     reference_wav: str,
@@ -3589,16 +3605,17 @@ def tts_reference(
     if not reference_wav:
         raise gr.Error("請先錄音或上傳參考音檔。")
     try:
-        centroid = extract_windowed_speaker_embedding(
-            reference_wav,
-            _get_ecapa_encoder(),
-            device="cpu",
-            min_duration_seconds=3.0,
-            window_seconds=3.0,
-            hop_seconds=1.5,
-            max_windows=12,
-            full_clip_max_seconds=12.0,
-        )
+        with latency_stage("ecapa"):
+            centroid = extract_windowed_speaker_embedding(
+                reference_wav,
+                _get_ecapa_encoder(),
+                device="cpu",
+                min_duration_seconds=3.0,
+                window_seconds=3.0,
+                hop_seconds=1.5,
+                max_windows=12,
+                full_clip_max_seconds=12.0,
+            )
     except ValueError as error:
         raise gr.Error(str(error)) from error
     return pcm16_audio_output(
@@ -3607,6 +3624,7 @@ def tts_reference(
 
 
 @gpu
+@timed_latency_request
 def tts_longform(
     text: str,
     speaker: str = DEFAULT_SPEAKER,

audio_artifact_gate.py

@@ -24,6 +24,8 @@ from typing import Any
 
 import numpy as np
 
+from latency_timing import timed_latency_stage
+
 
 @dataclass(frozen=True)
 class EchoSmearingGateConfig:
@@ -268,6 +270,7 @@ def _window_cepstrum(
     return np.asarray(cepstrum[: max_delay_samples + 1], dtype=np.float64)
 
 
+@timed_latency_stage("echo")
 def evaluate_echo_smearing(
     audio: Any,
     sample_rate: Any,

latency_timing.py

@@ -0,0 +1,149 @@
+"""Content-free, request-local latency instrumentation.
+
+The production Space emits one fixed-schema latency line per public synthesis
+request.  Only stage names, elapsed seconds, and call counts are retained; no
+request text, waveform, transcript, seed, or exception data enters this module.
+"""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+from contextvars import ContextVar
+from dataclasses import dataclass, field
+from functools import wraps
+from time import perf_counter
+from typing import Callable, Iterator
+
+
+LATENCY_STAGE_NAMES = (
+    "request_total",
+    "generation",
+    "turbo_asr",
+    "large_v3_asr",
+    "ecapa",
+    "squim",
+    "echo",
+    "f0",
+    "assemble",
+)
+_TIMED_STAGE_NAMES = frozenset(LATENCY_STAGE_NAMES[1:])
+LATENCY_LOG_PREFIX = "[BlueMagpie] latency "
+
+
+@dataclass
+class RequestLatency:
+    """Mutable request-local totals with a fixed, content-free schema."""
+
+    seconds: dict[str, float] = field(
+        default_factory=lambda: {name: 0.0 for name in LATENCY_STAGE_NAMES}
+    )
+    calls: dict[str, int] = field(
+        default_factory=lambda: {name: 0 for name in LATENCY_STAGE_NAMES}
+    )
+    active_depths: dict[str, int] = field(default_factory=dict)
+
+    def add(self, stage: str, elapsed_seconds: float) -> None:
+        if stage not in LATENCY_STAGE_NAMES:
+            raise ValueError("unknown latency stage")
+        self.seconds[stage] += max(0.0, float(elapsed_seconds))
+        self.calls[stage] += 1
+
+
+_CURRENT_REQUEST: ContextVar[RequestLatency | None] = ContextVar(
+    "bluemagpie_current_request_latency",
+    default=None,
+)
+
+
+def format_latency_log(totals: RequestLatency) -> str:
+    """Return the fixed-schema log line without accepting request content."""
+
+    fields: list[str] = []
+    for stage in LATENCY_STAGE_NAMES:
+        fields.append(f"{stage}_seconds={totals.seconds[stage]:.6f}")
+        fields.append(f"{stage}_calls={totals.calls[stage]}")
+    return LATENCY_LOG_PREFIX + " ".join(fields)
+
+
+@contextmanager
+def latency_request(
+    *,
+    emit: Callable[[str], object] | None = None,
+) -> Iterator[RequestLatency]:
+    """Open one request timer and emit exactly once at the outermost boundary."""
+
+    existing = _CURRENT_REQUEST.get()
+    if existing is not None:
+        yield existing
+        return
+
+    totals = RequestLatency()
+    token = _CURRENT_REQUEST.set(totals)
+    started = perf_counter()
+    try:
+        yield totals
+    finally:
+        totals.add("request_total", perf_counter() - started)
+        _CURRENT_REQUEST.reset(token)
+        (emit or print)(format_latency_log(totals))
+
+
+@contextmanager
+def latency_stage(stage: str) -> Iterator[None]:
+    """Accumulate one stage call when a public request timer is active.
+
+    Re-entering the same stage is deliberately folded into its outer call.  It
+    keeps a high-level ECAPA operation from double-counting a nested lazy load
+    or lower-level speaker measurement.
+    """
+
+    if stage not in _TIMED_STAGE_NAMES:
+        raise ValueError("unknown latency stage")
+    totals = _CURRENT_REQUEST.get()
+    if totals is None:
+        yield
+        return
+
+    depth = totals.active_depths.get(stage, 0)
+    totals.active_depths[stage] = depth + 1
+    if depth:
+        try:
+            yield
+        finally:
+            totals.active_depths[stage] -= 1
+        return
+
+    started = perf_counter()
+    try:
+        yield
+    finally:
+        totals.add(stage, perf_counter() - started)
+        totals.active_depths.pop(stage, None)
+
+
+def timed_latency_stage(stage: str):
+    """Decorate a synchronous function as one fixed latency stage."""
+
+    if stage not in _TIMED_STAGE_NAMES:
+        raise ValueError("unknown latency stage")
+
+    def decorate(function):
+        @wraps(function)
+        def timed(*args, **kwargs):
+            with latency_stage(stage):
+                return function(*args, **kwargs)
+
+        return timed
+
+    return decorate
+
+
+def timed_latency_request(function):
+    """Decorate one public synchronous synthesis entry point."""
+
+    @wraps(function)
+    def timed(*args, **kwargs):
+        with latency_request():
+            return function(*args, **kwargs)
+
+    return timed

quality_runtime.py

@@ -22,6 +22,7 @@ import numpy as np
 import torch
 import torch.nn.functional as torch_functional
 
+from latency_timing import timed_latency_stage
 from production import (
     AsrComparison,
     CandidateSequenceSelection,
@@ -499,6 +500,7 @@ def _bounded_squim_windows(waveform: np.ndarray) -> tuple[np.ndarray, ...]:
 
 
 @torch.inference_mode()
+@timed_latency_stage("squim")
 def squim_objective_evidence_from_audio(
     audio: np.ndarray | Sequence[float],
     sample_rate: int,
@@ -1078,7 +1080,7 @@ def _split_whisper_audio(
 
 
 @torch.inference_mode()
-def transcribe_whisper(
+def _transcribe_whisper(
     audio: np.ndarray | Sequence[float],
     sample_rate: int,
     *,
@@ -1089,7 +1091,7 @@ def transcribe_whisper(
     max_new_tokens: int = 128,
     max_verification_segments: int = WHISPER_MAX_VERIFICATION_SEGMENTS,
 ) -> str:
-    """Transcribe ndarray audio using deterministic decoding.
+    """Transcribe ndarray audio using the shared deterministic decoder.
 
     ``runtime`` and ``lazy_asr`` are mutually exclusive injection points.  An
     empty decoded string is returned as-is; the semantic verifier will reject
@@ -1144,6 +1146,33 @@ def transcribe_whisper(
     return " ".join(text for text in decoded_segments if text)
 
 
+@timed_latency_stage("turbo_asr")
+def transcribe_whisper(
+    audio: np.ndarray | Sequence[float],
+    sample_rate: int,
+    *,
+    lazy_asr: LazyWhisperASR | None = None,
+    runtime: WhisperRuntime | None = None,
+    language: str = "zh",
+    task: str = "transcribe",
+    max_new_tokens: int = 128,
+    max_verification_segments: int = WHISPER_MAX_VERIFICATION_SEGMENTS,
+) -> str:
+    """Transcribe with the pinned Turbo candidate-screening ASR."""
+
+    return _transcribe_whisper(
+        audio,
+        sample_rate,
+        lazy_asr=lazy_asr,
+        runtime=runtime,
+        language=language,
+        task=task,
+        max_new_tokens=max_new_tokens,
+        max_verification_segments=max_verification_segments,
+    )
+
+
+@timed_latency_stage("large_v3_asr")
 def transcribe_verification_whisper(
     audio: np.ndarray | Sequence[float],
     sample_rate: int,
@@ -1160,7 +1189,7 @@ def transcribe_verification_whisper(
     selected_lazy = lazy_asr
     if runtime is None and selected_lazy is None:
         selected_lazy = _DEFAULT_VERIFICATION_WHISPER
-    return transcribe_whisper(
+    return _transcribe_whisper(
         audio,
         sample_rate,
         lazy_asr=selected_lazy,
@@ -1303,6 +1332,7 @@ class SpeakerEvidence:
     active_rms_db: float
 
 
+@timed_latency_stage("ecapa")
 def speaker_evidence_from_audio(
     audio: np.ndarray | Sequence[float],
     sample_rate: int,
@@ -1378,6 +1408,7 @@ def speaker_evidence_from_audio(
     )
 
 
+@timed_latency_stage("ecapa")
 def release_speaker_evidence_from_audio(
     audio: np.ndarray | Sequence[float],
     sample_rate: int,
@@ -1509,6 +1540,7 @@ def endpoint_tail_energy_ratio(
     return float(np.clip(tail_rms / peak, 0.0, 1.0))
 
 
+@timed_latency_stage("f0")
 def active_audio_median_f0_hz(
     audio: np.ndarray | Sequence[float],
     sample_rate: int,

tests/test_glyph_app_integration.py

@@ -18,6 +18,7 @@ from glyph_assets import (
     compute_ordered_attempt_ledger_sha256,
 )
 from glyph_hybrid_plan import GlyphHybridPlanError
+from latency_timing import latency_stage
 from production import count_speech_units, pcm16_audio_output
 from quality_runtime import CandidateGenerationContext
 
@@ -31,6 +32,7 @@ class FakeGradioError(Exception):
 
 
 def _isolated_nodes(names: tuple[str, ...], namespace: dict[str, object]):
+    namespace.setdefault("latency_stage", latency_stage)
     tree = ast.parse(APP_PATH.read_text(encoding="utf-8"))
     selected = [
         node

tests/test_inference_cleanup.py

@@ -7,6 +7,7 @@ import numpy as np
 import pytest
 import torch
 
+from latency_timing import latency_stage, timed_latency_stage
 from production import count_speech_units
 from quality_runtime import (
     CandidateObservation,
@@ -21,6 +22,8 @@ ROOT = Path(__file__).resolve().parents[1]
 
 
 def _isolated_app_function(name, namespace):
+    namespace.setdefault("latency_stage", latency_stage)
+    namespace.setdefault("timed_latency_stage", timed_latency_stage)
     app_path = ROOT / "app.py"
     tree = ast.parse(app_path.read_text(encoding="utf-8"))
     function = next(

tests/test_latency_timing.py

@@ -0,0 +1,208 @@
+import ast
+from pathlib import Path
+from types import SimpleNamespace
+
+import numpy as np
+import pytest
+import torch
+
+import latency_timing
+from quality_runtime import (
+    WhisperRuntime,
+    transcribe_verification_whisper,
+    transcribe_whisper,
+)
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _clock(monkeypatch, values):
+    ticks = iter(values)
+    monkeypatch.setattr(latency_timing, "perf_counter", lambda: next(ticks))
+
+
+def _fields(line):
+    assert line.startswith(latency_timing.LATENCY_LOG_PREFIX)
+    return dict(
+        field.split("=", 1)
+        for field in line.removeprefix(latency_timing.LATENCY_LOG_PREFIX).split()
+    )
+
+
+def test_request_timer_emits_fixed_content_free_cumulative_schema(monkeypatch):
+    _clock(monkeypatch, (10.0, 11.0, 13.5, 14.0, 15.0, 20.0))
+    emitted = []
+
+    with latency_timing.latency_request(emit=emitted.append) as totals:
+        with latency_timing.latency_stage("generation"):
+            pass
+        with latency_timing.latency_stage("turbo_asr"):
+            pass
+
+    assert len(emitted) == 1
+    fields = _fields(emitted[0])
+    assert tuple(fields) == tuple(
+        item
+        for stage in latency_timing.LATENCY_STAGE_NAMES
+        for item in (f"{stage}_seconds", f"{stage}_calls")
+    )
+    assert fields["request_total_seconds"] == "10.000000"
+    assert fields["request_total_calls"] == "1"
+    assert fields["generation_seconds"] == "2.500000"
+    assert fields["generation_calls"] == "1"
+    assert fields["turbo_asr_seconds"] == "1.000000"
+    assert fields["turbo_asr_calls"] == "1"
+    assert fields["large_v3_asr_seconds"] == "0.000000"
+    assert fields["large_v3_asr_calls"] == "0"
+    assert totals.active_depths == {}
+
+
+def test_nested_request_and_same_stage_emit_and_count_only_once(monkeypatch):
+    _clock(monkeypatch, (0.0, 1.0, 5.0, 6.0))
+    emitted = []
+
+    with latency_timing.latency_request(emit=emitted.append):
+        with latency_timing.latency_request(emit=emitted.append):
+            with latency_timing.latency_stage("ecapa"):
+                with latency_timing.latency_stage("ecapa"):
+                    pass
+
+    fields = _fields(emitted[0])
+    assert len(emitted) == 1
+    assert fields["request_total_seconds"] == "6.000000"
+    assert fields["request_total_calls"] == "1"
+    assert fields["ecapa_seconds"] == "4.000000"
+    assert fields["ecapa_calls"] == "1"
+
+
+def test_public_decorator_logs_on_failure_without_request_arguments(
+    monkeypatch,
+    capsys,
+):
+    _clock(monkeypatch, (2.0, 3.0))
+
+    @latency_timing.timed_latency_request
+    def fail(private_value):
+        raise RuntimeError(private_value)
+
+    with pytest.raises(RuntimeError, match="private-payload"):
+        fail("private-payload")
+
+    line = capsys.readouterr().out.strip()
+    assert line.startswith(latency_timing.LATENCY_LOG_PREFIX)
+    assert "private-payload" not in line
+    assert _fields(line)["request_total_seconds"] == "1.000000"
+
+
+class _Processor:
+    def __call__(self, audio, **_kwargs):
+        rows = len(audio) if isinstance(audio, list) else 1
+        return SimpleNamespace(
+            input_features=torch.ones((rows, 4, 4)),
+            attention_mask=torch.ones((rows, 4), dtype=torch.long),
+        )
+
+    def batch_decode(self, token_ids, **_kwargs):
+        return ["內容完整"] * int(token_ids.shape[0])
+
+
+class _Model:
+    def generate(self, features, **_kwargs):
+        return torch.ones((features.shape[0], 2), dtype=torch.long)
+
+
+def test_dual_asr_wrappers_accumulate_under_distinct_stage_names():
+    runtime = WhisperRuntime(
+        processor=_Processor(),
+        model=_Model(),
+        device=torch.device("cpu"),
+        dtype=torch.float32,
+    )
+    emitted = []
+    waveform = np.ones(4_000, dtype=np.float32)
+
+    with latency_timing.latency_request(emit=emitted.append):
+        assert transcribe_whisper(waveform, 16_000, runtime=runtime) == "內容完整"
+        assert (
+            transcribe_verification_whisper(
+                waveform,
+                16_000,
+                runtime=runtime,
+            )
+            == "內容完整"
+        )
+
+    fields = _fields(emitted[0])
+    assert fields["turbo_asr_calls"] == "1"
+    assert fields["large_v3_asr_calls"] == "1"
+    assert float(fields["turbo_asr_seconds"]) >= 0.0
+    assert float(fields["large_v3_asr_seconds"]) >= 0.0
+
+
+def _decorator_stage(node):
+    for decorator in node.decorator_list:
+        if (
+            isinstance(decorator, ast.Call)
+            and isinstance(decorator.func, ast.Name)
+            and decorator.func.id == "timed_latency_stage"
+            and len(decorator.args) == 1
+            and isinstance(decorator.args[0], ast.Constant)
+        ):
+            return decorator.args[0].value
+    return None
+
+
+def test_space_wires_every_requested_latency_stage_without_gate_changes():
+    app_tree = ast.parse((ROOT / "app.py").read_text(encoding="utf-8"))
+    quality_tree = ast.parse(
+        (ROOT / "quality_runtime.py").read_text(encoding="utf-8")
+    )
+    echo_tree = ast.parse(
+        (ROOT / "audio_artifact_gate.py").read_text(encoding="utf-8")
+    )
+    app_functions = {
+        node.name: node for node in app_tree.body if isinstance(node, ast.FunctionDef)
+    }
+    quality_functions = {
+        node.name: node
+        for node in quality_tree.body
+        if isinstance(node, ast.FunctionDef)
+    }
+    echo_functions = {
+        node.name: node
+        for node in echo_tree.body
+        if isinstance(node, ast.FunctionDef)
+    }
+
+    assert _decorator_stage(app_functions["_assemble_trajectory_audio"]) == "assemble"
+    generate_source = ast.get_source_segment(
+        (ROOT / "app.py").read_text(encoding="utf-8"),
+        app_functions["_generate_chunk"],
+    )
+    assert generate_source is not None
+    assert 'with latency_stage("generation"):' in generate_source
+    assert _decorator_stage(quality_functions["transcribe_whisper"]) == "turbo_asr"
+    assert (
+        _decorator_stage(quality_functions["transcribe_verification_whisper"])
+        == "large_v3_asr"
+    )
+    assert _decorator_stage(quality_functions["speaker_evidence_from_audio"]) == "ecapa"
+    assert (
+        _decorator_stage(quality_functions["release_speaker_evidence_from_audio"])
+        == "ecapa"
+    )
+    assert (
+        _decorator_stage(quality_functions["squim_objective_evidence_from_audio"])
+        == "squim"
+    )
+    assert _decorator_stage(quality_functions["active_audio_median_f0_hz"]) == "f0"
+    assert _decorator_stage(echo_functions["evaluate_echo_smearing"]) == "echo"
+
+    for name in ("tts_speaker", "tts_reference", "tts_longform"):
+        decorators = {
+            decorator.id
+            for decorator in app_functions[name].decorator_list
+            if isinstance(decorator, ast.Name)
+        }
+        assert decorators == {"gpu", "timed_latency_request"}

tests/test_release_pins.py

@@ -17,6 +17,7 @@ from production import (
     punctuation_pause_seconds,
     split_quality_text_for_tts,
 )
+from latency_timing import timed_latency_stage
 from quality_runtime import LocalIndependentGateEvidence
 
 
@@ -115,6 +116,7 @@ def _isolated_assemble_trajectory_audio():
         "count_speech_units": count_speech_units,
         "finish_audio": lambda waveform, _sample_rate, **_kwargs: waveform,
         "pcm16_verification_waveform": pcm16_verification_waveform,
+        "timed_latency_stage": timed_latency_stage,
     }
     exec(compile(module, str(app_path), "exec"), namespace)
     return namespace["_assemble_trajectory_audio"]