solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Content-free stage telemetry on the observe.transcribed event.
5
6See solstone/observe/transcribe/failure-and-telemetry.md for the field contract.
7"""
8
9from __future__ import annotations
10
11import argparse
12import importlib
13import json
14from pathlib import Path
15from unittest.mock import MagicMock, patch
16
17import numpy as np
18import pytest
19
20from solstone.observe.transcribe.overlap import (
21 OverlapInferenceResult,
22 SpeakerWindowStats,
23)
24from solstone.observe.utils import SAMPLE_RATE
25from solstone.observe.vad import VadResult
26from solstone.think.providers.parakeet_server import ParakeetServerNotReady
27from tests.helpers.module_mocks import module_mock
28
29# A string that exists nowhere but in the (mocked) transcript. If it shows up in a
30# serialized event, transcript content leaked into telemetry.
31TRANSCRIPT_SENTINEL = "zzq-secret-utterance-do-not-leak"
32NO_SPEECH_STATS = (SpeakerWindowStats(0, 0, 0),)
33
34
35def _overlap_result() -> OverlapInferenceResult:
36 return OverlapInferenceResult(
37 0.0,
38 np.zeros((589, 7), dtype=np.float32),
39 NO_SPEECH_STATS,
40 )
41
42
43@pytest.fixture
44def raw_path(tmp_path: Path) -> Path:
45 path = tmp_path / "chronicle" / "20260416" / "default" / "120000_300" / "audio.m4a"
46 path.parent.mkdir(parents=True)
47 path.write_bytes(b"audio")
48 return path
49
50
51@pytest.fixture
52def audio_buffer() -> np.ndarray:
53 return np.zeros(10 * SAMPLE_RATE, dtype=np.float32)
54
55
56@pytest.fixture
57def vad_result() -> VadResult:
58 return VadResult(
59 duration=10.0,
60 speech_duration=5.0,
61 has_speech=True,
62 speech_segments=[(1.0, 6.0)],
63 )
64
65
66def _backend_module() -> MagicMock:
67 backend_module = MagicMock()
68 backend_module.get_model_info.return_value = {
69 "model": "parakeet-v3-q8_0.gguf",
70 "device": "gpu",
71 "compute_type": "q8_0",
72 }
73 return backend_module
74
75
76def _run_success(
77 raw_path: Path,
78 audio_buffer,
79 vad_result,
80 backend_module: MagicMock | None = None,
81) -> dict:
82 """Run a successful process_audio and return the emitted event kwargs."""
83 from solstone.observe.transcribe.main import process_audio
84
85 statements = [
86 {"id": 0, "start": 0.0, "end": 1.0, "text": f"{TRANSCRIPT_SENTINEL} hello"}
87 ]
88
89 with (
90 patch(
91 "solstone.observe.transcribe.main.get_config",
92 return_value={"transcribe": {"preserve_all": False}},
93 ),
94 patch(
95 "solstone.observe.transcribe.main.get_journal",
96 return_value=str(raw_path.parents[4]),
97 ),
98 patch(
99 "solstone.observe.transcribe.main.stt_transcribe", return_value=statements
100 ),
101 patch(
102 "solstone.observe.transcribe.main.get_backend",
103 return_value=backend_module or _backend_module(),
104 ),
105 patch("solstone.observe.transcribe.main._embed_statements", return_value=None),
106 patch(
107 "solstone.observe.transcribe.overlap.compute_overlap_and_logprobs",
108 return_value=_overlap_result(),
109 ),
110 patch("solstone.observe.transcribe.main.callosum_send") as mock_send,
111 ):
112 process_audio(raw_path, audio_buffer, vad_result, {}, backend="parakeet-cpp")
113
114 assert mock_send.call_args.args[:2] == ("observe", "transcribed")
115 return mock_send.call_args.kwargs
116
117
118def _run_parakeet_cpp_process_one_event(
119 monkeypatch: pytest.MonkeyPatch,
120 raw_path: Path,
121 audio_buffer,
122 vad_result,
123 *,
124 stt_error: Exception,
125 expected_exit: int,
126 placement: str | None = None,
127 configured_device: str | None = "auto",
128) -> dict:
129 from solstone.observe.transcribe import _parakeet_cpp as parakeet_cpp
130 from solstone.observe.transcribe.main import _process_one
131
132 journal_root = raw_path.parents[4]
133 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_root))
134 if placement is not None:
135 parakeet_cpp.parakeet_server.write_parakeet_placement(placement)
136 parakeet_cpp_config = {}
137 if configured_device is not None:
138 parakeet_cpp_config["device"] = configured_device
139 args = argparse.Namespace(backend=None, cpu=False, model=None, redo=False)
140
141 with (
142 patch(
143 "solstone.observe.transcribe.main.get_journal",
144 return_value=str(journal_root),
145 ),
146 patch("solstone.observe.transcribe.main.load_audio", return_value=audio_buffer),
147 patch("solstone.observe.vad.run_vad", return_value=vad_result),
148 patch("solstone.observe.vad.reduce_audio", return_value=(None, None)),
149 patch("solstone.observe.transcribe.main.tag_audio", return_value={"tags": {}}),
150 patch(
151 "solstone.observe.transcribe.main.stt_transcribe",
152 side_effect=stt_error,
153 ),
154 patch("solstone.observe.transcribe.main.callosum_send") as mock_send,
155 ):
156 with pytest.raises(SystemExit) as exc_info:
157 _process_one(
158 raw_path,
159 args,
160 {"backend": "parakeet-cpp", "parakeet-cpp": parakeet_cpp_config},
161 "parakeet-cpp",
162 )
163
164 assert exc_info.value.code == expected_exit
165 assert mock_send.call_args.args[:2] == ("observe", "transcribed")
166 return mock_send.call_args.kwargs
167
168
169def test_success_event_carries_stage_timings_and_envelope(
170 raw_path: Path, audio_buffer: np.ndarray, vad_result: VadResult
171) -> None:
172 kwargs = _run_success(raw_path, audio_buffer, vad_result)
173
174 assert kwargs["outcome"] == "transcribed"
175 timings = kwargs["timings"]
176 # Stages that ran inside process_audio. decode/vad/reduce are measured in
177 # _process_one, which this test calls past; no speech-bearing evidence
178 # resolves the speaker decision to none, so diarization is skipped.
179 assert {"asr_ms", "embed_ms", "overlap_ms", "write_ms"} <= set(timings)
180 assert "diarize_ms" not in timings
181 assert all(isinstance(v, int) and v >= 0 for v in timings.values())
182
183 assert kwargs["backend"] == "parakeet-cpp"
184 assert kwargs["device"] == "gpu"
185 assert kwargs["model"] == "parakeet-v3-q8_0.gguf"
186 header = json.loads(raw_path.with_suffix(".jsonl").read_text().splitlines()[0])
187 assert header["device"] == "gpu"
188 assert kwargs["audio_seconds"] == 10.0
189 assert isinstance(kwargs["peak_rss_mib"], int)
190 assert kwargs["peak_rss_mib"] > 0
191
192
193def test_success_event_is_content_free(
194 raw_path: Path, audio_buffer: np.ndarray, vad_result: VadResult
195) -> None:
196 """No transcript text may appear anywhere in the serialized event."""
197 kwargs = _run_success(raw_path, audio_buffer, vad_result)
198
199 serialized = json.dumps(kwargs, default=str)
200 assert TRANSCRIPT_SENTINEL not in serialized
201
202 # Content fields stay out of the event envelope.
203 for banned in ("text", "words", "statements", "topics", "setting", "emotions"):
204 assert banned not in kwargs
205
206
207@pytest.mark.parametrize("placement", ["gpu", "cpu"])
208def test_success_event_and_header_use_parakeet_placement_record(
209 monkeypatch: pytest.MonkeyPatch,
210 raw_path: Path,
211 audio_buffer: np.ndarray,
212 vad_result: VadResult,
213 placement: str,
214) -> None:
215 from solstone.observe.transcribe import _parakeet_cpp as parakeet_cpp
216
217 monkeypatch.setattr(parakeet_cpp.sys, "platform", "linux")
218 monkeypatch.setenv("SOLSTONE_JOURNAL", str(raw_path.parents[4]))
219 parakeet_cpp.parakeet_server.write_parakeet_placement(placement)
220 backend_module = MagicMock()
221 backend_module.get_model_info.side_effect = parakeet_cpp.get_model_info
222
223 kwargs = _run_success(raw_path, audio_buffer, vad_result, backend_module)
224
225 assert kwargs["device"] == placement
226 assert kwargs["device"] != "auto"
227 header = json.loads(raw_path.with_suffix(".jsonl").read_text().splitlines()[0])
228 assert header["device"] == placement
229 assert header["device"] != "auto"
230
231
232@pytest.mark.parametrize("placement", ["cpu", "gpu"])
233def test_deferred_event_uses_parakeet_placement_record(
234 monkeypatch: pytest.MonkeyPatch,
235 raw_path: Path,
236 audio_buffer: np.ndarray,
237 vad_result: VadResult,
238 placement: str,
239) -> None:
240 from solstone.observe.exit_codes import EXIT_PROVIDER_BLOCKED
241
242 kwargs = _run_parakeet_cpp_process_one_event(
243 monkeypatch,
244 raw_path,
245 audio_buffer,
246 vad_result,
247 stt_error=ParakeetServerNotReady("warming", retry_reason="no_port"),
248 expected_exit=EXIT_PROVIDER_BLOCKED,
249 placement=placement,
250 configured_device="auto",
251 )
252
253 assert kwargs["outcome"] == "deferred"
254 assert kwargs["backend"] == "parakeet-cpp"
255 assert kwargs["device"] == placement
256 assert kwargs["device"] != "auto"
257 assert "model" not in kwargs
258
259
260def test_deferred_event_uses_configured_device_without_parakeet_placement_record(
261 monkeypatch: pytest.MonkeyPatch,
262 raw_path: Path,
263 audio_buffer: np.ndarray,
264 vad_result: VadResult,
265) -> None:
266 from solstone.observe.exit_codes import EXIT_PROVIDER_BLOCKED
267
268 kwargs = _run_parakeet_cpp_process_one_event(
269 monkeypatch,
270 raw_path,
271 audio_buffer,
272 vad_result,
273 stt_error=ParakeetServerNotReady("warming", retry_reason="no_port"),
274 expected_exit=EXIT_PROVIDER_BLOCKED,
275 placement=None,
276 configured_device="auto",
277 )
278
279 assert kwargs["outcome"] == "deferred"
280 assert kwargs["device"] == "auto"
281
282
283def test_deferred_event_reports_parakeet_placement_when_config_has_no_device(
284 monkeypatch: pytest.MonkeyPatch,
285 raw_path: Path,
286 audio_buffer: np.ndarray,
287 vad_result: VadResult,
288) -> None:
289 from solstone.observe.exit_codes import EXIT_PROVIDER_BLOCKED
290
291 kwargs = _run_parakeet_cpp_process_one_event(
292 monkeypatch,
293 raw_path,
294 audio_buffer,
295 vad_result,
296 stt_error=ParakeetServerNotReady("warming", retry_reason="no_port"),
297 expected_exit=EXIT_PROVIDER_BLOCKED,
298 placement="cpu",
299 configured_device=None,
300 )
301
302 assert kwargs["outcome"] == "deferred"
303 assert kwargs["device"] == "cpu"
304
305
306@pytest.mark.parametrize("placement", ["cpu", "gpu"])
307def test_failed_event_uses_parakeet_placement_record(
308 monkeypatch: pytest.MonkeyPatch,
309 raw_path: Path,
310 audio_buffer: np.ndarray,
311 vad_result: VadResult,
312 placement: str,
313) -> None:
314 kwargs = _run_parakeet_cpp_process_one_event(
315 monkeypatch,
316 raw_path,
317 audio_buffer,
318 vad_result,
319 stt_error=RuntimeError("boom"),
320 expected_exit=1,
321 placement=placement,
322 configured_device="auto",
323 )
324
325 assert kwargs["outcome"] == "failed"
326 assert kwargs["backend"] == "parakeet-cpp"
327 assert kwargs["device"] == placement
328 assert kwargs["device"] != "auto"
329
330
331def test_failed_event_is_content_free_even_when_the_exception_message_is_not(
332 raw_path: Path, audio_buffer: np.ndarray, vad_result: VadResult
333) -> None:
334 """The failed path must not put an exception *message* on the bus.
335
336 Real exception messages can embed model output from provider wrappers. So the
337 event carries the exception *type*, never the message.
338 """
339 from solstone.observe.transcribe.main import process_audio
340
341 leaky = RuntimeError(f"Provider response included preview={TRANSCRIPT_SENTINEL!r}")
342
343 with (
344 patch("solstone.observe.transcribe.main.stt_transcribe", side_effect=leaky),
345 patch(
346 "solstone.observe.transcribe.main.get_journal",
347 return_value=str(raw_path.parents[4]),
348 ),
349 patch("solstone.observe.transcribe.main.callosum_send") as mock_send,
350 ):
351 with pytest.raises(SystemExit) as exc_info:
352 process_audio(raw_path, audio_buffer, vad_result, {}, backend="parakeet")
353
354 assert exc_info.value.code == 1
355 kwargs = mock_send.call_args.kwargs
356 assert kwargs["outcome"] == "failed"
357 assert kwargs["reason"] == "RuntimeError"
358 assert kwargs["error"] == "RuntimeError"
359 assert TRANSCRIPT_SENTINEL not in json.dumps(kwargs, default=str)
360
361
362def test_rtfx_derived_from_asr_time(
363 raw_path: Path, audio_buffer: np.ndarray, vad_result: VadResult
364) -> None:
365 kwargs = _run_success(raw_path, audio_buffer, vad_result)
366
367 asr_ms = kwargs["timings"]["asr_ms"]
368 if asr_ms:
369 assert kwargs["rtfx"] == pytest.approx(
370 kwargs["audio_seconds"] / (asr_ms / 1000), rel=0.01
371 )
372 else:
373 # A sub-millisecond mocked ASR cannot produce an honest ratio, so none is
374 # fabricated.
375 assert "rtfx" not in kwargs
376
377
378def test_queue_wait_is_read_from_env(
379 monkeypatch: pytest.MonkeyPatch, tmp_path: Path
380) -> None:
381 from solstone.observe.transcribe.main import _read_queue_wait_ms
382
383 monkeypatch.setenv("SOL_QUEUE_WAIT_MS", "4200")
384 assert _read_queue_wait_ms() == 4200
385
386 monkeypatch.setenv("SOL_QUEUE_WAIT_MS", "not-a-number")
387 assert _read_queue_wait_ms() is None
388
389 monkeypatch.delenv("SOL_QUEUE_WAIT_MS")
390 assert _read_queue_wait_ms() is None
391
392
393def test_stage_timings_accumulate_repeated_stages(
394 monkeypatch: pytest.MonkeyPatch,
395) -> None:
396 """write_ms covers the jsonl AND the npz, so a second entry must sum, not clobber."""
397 # NB: `solstone.observe.transcribe.main` as an attribute path resolves to the
398 # re-exported main() *function*, not the module -- import it explicitly.
399 transcribe_main = importlib.import_module("solstone.observe.transcribe.main")
400
401 # Drive perf_counter so the two blocks have distinct, known durations.
402 # Values are binary-exact so the int() truncation is not off by a millisecond.
403 monkeypatch.setattr(
404 transcribe_main,
405 "time",
406 module_mock(
407 transcribe_main.time,
408 perf_counter=MagicMock(side_effect=[0.0, 0.25, 1.0, 1.5]),
409 ),
410 )
411
412 timings = transcribe_main._StageTimings()
413 assert timings.as_dict() == {}
414
415 with timings.time("write"): # 250 ms
416 pass
417 assert timings.get_ms("write") == 250
418
419 with timings.time("write"): # 500 ms
420 pass
421 assert timings.get_ms("write") == 750 # summed, not clobbered to 500
422 assert set(timings.as_dict()) == {"write_ms"}
423
424
425def test_stage_timing_recorded_even_when_the_stage_raises() -> None:
426 """A server that dies mid-ASR must still report how long ASR ran before dying."""
427 from solstone.observe.transcribe.main import _StageTimings
428
429 timings = _StageTimings()
430 with pytest.raises(RuntimeError):
431 with timings.time("asr"):
432 raise RuntimeError("server died")
433
434 assert timings.get_ms("asr") is not None