fix(local): decode Ogg recordings under any path and clear a stopped recording's audio

Review of the step 5 backend half (553048c02).

- An Ogg Vorbis recording whose path held a character outside the Windows
  ANSI code page (a file named 会議.ogg, or a data folder under an account
  named José) was refused for good as "This recording could not be decoded.":
  miniaudio's vorbis_get_file_info opens the path with fopen, while its stream
  reader opens a wide one. The decode child now reads an Ogg file's length from
  its own pages, the last page's granule position over the rate its first
  packet names; it matched miniaudio's reading on every fixture.
- A speech model chosen midway restarted the recording under the old language
  intent, so the header read "Japanese" and the checkpoint kept "ja" although
  the new model, whose settings start empty, detected the language. The job now
  asks the status route again and starts over under what it reports.
- A cancel of a recording queued again after giving way left its decoded audio,
  115 MB an hour, with no run left to clear it. The cancel route now drops it
  and progress.json, keeping the plan and windows for a retry, as a run's own
  end does (recordings/ended_run.py, shared with the worker, best-effort for a
  file another process holds). A cancel that lands just as a job would give
  way now ends it, instead of pause_job refusing quietly with the audio kept.
- build-audiocpp.yml's path filter takes modules/documents/recordings/, which
  the end-to-end gate reads too.
This commit is contained in:
DESKTOP-RTLN3BA\$punk 2026-10-14 08:58:40 -07:00
parent 553048c02e
commit 0de99e8a7f
11 changed files with 226 additions and 30 deletions

View file

@ -24,6 +24,7 @@ on:
- surfsense_local/backend/modules/llm/transcription/**
- surfsense_local/backend/worker/ingestion/audio/**
- surfsense_local/backend/worker/audio_decode/**
- surfsense_local/backend/modules/documents/recordings/**
- surfsense_local/backend/tests/packaging/test_audiocpp_voicing.py
- surfsense_local/backend/tests/packaging/test_audiocpp_transcription.py
- surfsense_local/backend/tests/packaging/test_audiocpp_transcription_config.py

View file

@ -173,10 +173,10 @@ A recording becomes a source through the speech model chosen for `speech_to_text
Everything the job keeps lives under the document's `transcript/` folder, since the original must stay the only file beside it ([`recordings/layout.py`](../../surfsense_local/backend/modules/documents/recordings/layout.py)): `transcript/work/` holds `audio16k.raw`, `plan.json`, `windows.jsonl` and `progress.json`, and `transcript/segments.json` is written when the transcript is done.
1. **Ready?** The job asks `GET /llm/transcription/status`. With no speech model it stops before decoding ([below](#no-speech-model-yet)). With the server unavailable, as in the seconds after a model is chosen or a crash, it asks again every 5 s for up to 2 minutes, giving the queue up as a window does, then fails with "The speech recognition engine stopped." and Huey tries again.
2. **Decode** ([`worker/audio_decode/`](../../surfsense_local/backend/worker/audio_decode/)). The worker's own binary, started as `worker --decode-audio <in> <out>` ahead of the consumer's imports, as `--run-document-script` is, reads the file's length from its headers and checks the data volume for room at 115.2 MB an hour plus 64 MiB, then streams it through pyminiaudio (MIT; WAV, MP3, FLAC, Ogg Vorbis) to 16 kHz mono 16-bit PCM in `audio16k.raw.part`, renamed to `audio16k.raw` once every sample is written. It prints its outcome on one line and exits when the worker's end of its stdin closes. A malformed file kills the child, not the consumer, and the job checks for a cancel every second while it waits. An hour of MP3 decodes in 3.6 s. miniaudio's linear resampler cost 0.08 points of word error over scipy's `resample_poly` on an 8.6-minute 44.1 kHz file (5.91% against 5.83%, under the proposal's 2% relative), so it stays.
2. **Decode** ([`worker/audio_decode/`](../../surfsense_local/backend/worker/audio_decode/)). The worker's own binary, started as `worker --decode-audio <in> <out>` ahead of the consumer's imports, as `--run-document-script` is, reads the file's length from its headers (an Ogg file's from its last page, since miniaudio's Vorbis reader opens a path with `fopen`, which on Windows cannot open a name outside the ANSI code page) and checks the data volume for room at 115.2 MB an hour plus 64 MiB, then streams it through pyminiaudio (MIT; WAV, MP3, FLAC, Ogg Vorbis) to 16 kHz mono 16-bit PCM in `audio16k.raw.part`, renamed to `audio16k.raw` once every sample is written. It prints its outcome on one line and exits when the worker's end of its stdin closes. A malformed file kills the child, not the consumer, and the job checks for a cancel every second while it waits. An hour of MP3 decodes in 3.6 s. miniaudio's linear resampler cost 0.08 points of word error over scipy's `resample_poly` on an 8.6-minute 44.1 kHz file (5.91% against 5.83%, under the proposal's 2% relative), so it stays.
3. **Speech** ([`speech_regions.py`](../../surfsense_local/backend/worker/ingestion/audio/speech_regions.py)). Silero VAD, through the API's own reader ([`voice_activity/silero.py`](../../surfsense_local/backend/modules/llm/transcription/voice_activity/silero.py)) fed a minute at a time off disk, finds speech by its rule (0.5 threshold, at least 250 ms); each span is padded by 200 ms and joined to the next across gaps under 300 ms. It ran at 207 times real time on one thread. Without a staged Silero, or should it fail, all of the recording counts as speech, as a dictated phrase does.
4. **Windows** ([`windows.py`](../../surfsense_local/backend/worker/ingestion/audio/windows.py)). Speech packs into windows of at most 28 s, which bounds a request's memory, ending at the longest pause once a window holds 20 s, the one nearest 24 s among equals. A single run past 28 s is cut at its quietest 100 ms, by loudness, between 20 and 28 s. Silence between windows is never sent, and every window keeps its place in the recording. The plan goes to `plan.json` once; a job that finds it skips the speech pass, and decodes again only while windows remain to send.
5. **The loop** ([`window_loop.py`](../../surfsense_local/backend/worker/ingestion/audio/window_loop.py)). For each window not yet in `windows.jsonl`: check for cancel, send it as an in-memory WAV to `POST /internal/transcription/windows` (`last` when it is the final window and no other recording is pending or processing), append `{n, start_s, end_s, text, language, model_id, intent}`, write `progress.json` and send the window's `progress` event. A dropped connection or `503 transcription_unavailable` is asked again after 1, 2, 4, 8 and 16 s, then fails with "The speech recognition engine stopped." `503 waiting` is asked again after the cancel and yield checks: a hold or a dictation has no bound, and memory waits are summed in `progress.json` across yields, failing past 10 minutes with the API's sentence. A window answered by another model than the job began with starts the recording over, so one model writes the whole transcript.
5. **The loop** ([`window_loop.py`](../../surfsense_local/backend/worker/ingestion/audio/window_loop.py)). For each window not yet in `windows.jsonl`: check for cancel, send it as an in-memory WAV to `POST /internal/transcription/windows` (`last` when it is the final window and no other recording is pending or processing), append `{n, start_s, end_s, text, language, model_id, intent}`, write `progress.json` and send the window's `progress` event. A dropped connection or `503 transcription_unavailable` is asked again after 1, 2, 4, 8 and 16 s, then fails with "The speech recognition engine stopped." `503 waiting` is asked again after the cancel and yield checks: a hold or a dictation has no bound, and memory waits are summed in `progress.json` across yields, failing past 10 minutes with the API's sentence. A window answered by another model than the job began with starts the recording over under the status the API reports then, so one model writes the whole transcript and the language set for the last model is not carried to it.
6. **Finish.** The transcript becomes `content` through `finish_job`. `transcript/segments.json` keeps `{model_id, language, duration_s, windows: [{line, start_s, end_s, language}]}`, the line being where the window's paragraph is, counted from 1 as chunks count theirs. The row's metadata gains `transcription: {language, duration_s, model}`, written with `merge_metadata`, which reads and writes the one key in a transaction so every other key stands. `transcript/work/` is deleted.
On the measuring machine ([proposal](../proposals/local-speech-to-text.md#measurements)), an 8.6-minute 16 kHz file went from `pending` to `ready` in 60 s in 22 windows, at 5.83% word error.
@ -201,7 +201,7 @@ The ingest queue has one thread, and an hour at the measured 11.5 times real tim
### Cancel and retry
Cancel is checked between windows without holding the write lock, and lands within one window; the window in flight finishes. A yielded job's queued copy is revoked as any is. No speech model, a memory wait past its limit, no room on disk ("Transcribing needs about 2.5 GB of free disk space."), no speech ("No speech was found in this recording.", from Silero or from a model that wrote nothing), no audio ("This file has no audio.") and a file the decoder cannot read ("This recording could not be decoded.") fail without retry: `RecordingRefusedError` becomes `CancelExecution(retry=False)` in [`tasks.py`](../../surfsense_local/backend/modules/documents/tasks.py). Anything else uses Huey's two retries, and every retry, Huey's or the user's, resumes from `windows.jsonl` when the model id and the language intent match; another model or language starts over. A run that ends the job, a cancel included, deletes `audio16k.raw` and `progress.json` and keeps the plan and the windows.
Cancel is checked between windows without holding the write lock, and lands within one window; the window in flight finishes. A yielded job's queued copy is revoked as any is. No speech model, a memory wait past its limit, no room on disk ("Transcribing needs about 2.5 GB of free disk space."), no speech ("No speech was found in this recording.", from Silero or from a model that wrote nothing), no audio ("This file has no audio.") and a file the decoder cannot read ("This recording could not be decoded.") fail without retry: `RecordingRefusedError` becomes `CancelExecution(retry=False)` in [`tasks.py`](../../surfsense_local/backend/modules/documents/tasks.py). Anything else uses Huey's two retries, and every retry, Huey's or the user's, resumes from `windows.jsonl` when the model id and the language intent match; another model or language starts over. A run that ends the job, a cancel included, deletes `audio16k.raw` and `progress.json` and keeps the plan and the windows; so does a cancel of a recording waiting in the queue after giving way, which has no run left to do it ([`recordings/ended_run.py`](../../surfsense_local/backend/modules/documents/recordings/ended_run.py)).
### No speech model yet

View file

@ -0,0 +1,31 @@
"""What a recording's work keeps once its job stops short of a transcript, a
cancel included: the plan and the windows written, which a retry carries on
from, but not the decoded audio, 115 MB an hour and seconds to make again."""
import logging
from modules.documents.models import Document, DocumentStatus
from modules.documents.recordings.kind import is_recording
from modules.documents.recordings.layout import WorkFiles, work_files
from shared.config import get_storage_settings
__all__ = ["drop_run_files", "drop_when_cancelled_in_queue"]
logger = logging.getLogger(__name__)
def drop_run_files(files: WorkFiles) -> None:
"""Best-effort: a file another process holds open on Windows stays."""
for path in (files.audio, files.progress):
try:
path.unlink(missing_ok=True)
except OSError as error:
logger.info("transcription: %s stays: %s", path, error)
def drop_when_cancelled_in_queue(document: Document, was: DocumentStatus) -> None:
"""A recording that gave the queue up has no run left to clear its own; a
running one clears them as it stops."""
if was is DocumentStatus.PENDING and is_recording(document.mime_type):
folder = get_storage_settings().document_dir(document.workspace_id, document.id)
drop_run_files(work_files(folder))

View file

@ -26,6 +26,7 @@ from modules.chunks.models import Chunk
from modules.documents.dependencies import DocumentDep
from modules.documents.models import Document, DocumentStatus, DocumentType
from modules.documents.original_file import original_path, stored_name
from modules.documents.recordings.ended_run import drop_when_cancelled_in_queue
from modules.documents.schemas import (
DocumentByChunkRead,
DocumentDetail,
@ -448,8 +449,10 @@ def retry_document(
def cancel_ingest(
document: DocumentDep, session: SessionDep, broker: EventBrokerDep
) -> Document:
was = document.status
if not cancel_ingest_job(session, document):
raise HTTPException(status.HTTP_409_CONFLICT, "nothing is running")
drop_when_cancelled_in_queue(document, was)
notify_document_updates(broker, document)
return document

View file

@ -417,6 +417,32 @@ async def test_a_pending_document_can_be_cancelled(
assert ingest_queue.is_revoked(task)
async def test_cancelling_a_recording_that_gave_way_drops_its_decoded_audio(
client: AsyncClient, workspace_id: int
) -> None:
"""No run of its job is left to clear the 115 MB an hour it decoded; the
windows written stay, so a retry carries on from them."""
tones = Path(__file__).resolve().parents[2] / "fixtures" / "audio"
uploaded = await client.post(
f"/workspaces/{workspace_id}/documents/upload",
files={"files": ("lecture.flac", (tones / "tone.flac").read_bytes(), "")},
)
recording = uploaded.json()["created"][0]["id"]
work = work_files(get_storage_settings().document_dir(workspace_id, recording))
work.folder.mkdir(parents=True)
work.audio.write_bytes(b"\0" * 64)
work.windows.write_text('{"n": 0}\n', encoding="utf-8")
write_progress(work.progress, TranscriptionProgress(done_s=28.0, total_s=3480.0))
response = await client.post(
f"/workspaces/{workspace_id}/documents/{recording}/cancel"
)
assert response.status_code == 200
assert (work.audio.exists(), work.progress.exists()) == (False, False)
assert work.windows.exists()
async def test_a_ready_document_cannot_be_cancelled(
client: AsyncClient, workspace_id: int, engine: Engine
) -> None:

View file

@ -349,6 +349,40 @@ def test_a_cancel_during_a_window_stops_the_recording_after_it(
assert not work_files(folder_of(recording)).audio.exists()
def test_a_cancel_that_lands_as_the_recording_would_give_way_still_stops_it(
session: Session,
stub_model: None,
speech_api: SpeechApi,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Cancelled, not queued again, and its decoded audio gone with it."""
monkeypatch.setattr(queue_turn, "YIELD_AFTER_SECONDS", 0.0)
recording = add_recording(session)
session.add(
Document(
workspace_id=recording.workspace_id,
title="Note",
document_type=DocumentType.NOTE,
content="text",
)
)
session.commit()
def cancel(n: int) -> None:
with create_session_factory(session.get_bind())() as api:
api.get(Document, recording.id).status = DocumentStatus.CANCELLED
api.commit()
speech_api.during_window = cancel
run(recording.id)
session.expire_all()
assert recording.status is DocumentStatus.CANCELLED
assert ingest_queue.pending() == []
assert not work_files(folder_of(recording)).audio.exists()
# Failures
@ -501,19 +535,31 @@ def test_a_disk_without_room_fails_with_what_it_needs(
def test_another_model_chosen_midway_writes_the_whole_transcript(
session: Session, stub_model: None, speech_api: SpeechApi
) -> None:
"""One model's text is never stitched to another's."""
"""One model's text is never stitched to another's. The new model starts
without the language set for the last, so the header says what it heard."""
speech_api.status["language_intent"] = "ja"
other = {"text": "Other words.", "language": "en", "model_id": "other-model"}
speech_api.replies = [
(200, speech_api.written(0)),
(200, other),
(200, other),
(200, other),
]
def choose_another(n: int) -> None:
if n == 1:
speech_api.status = {
**speech_api.status,
"model": "other-model",
"language_intent": "auto",
}
speech_api.during_window = choose_another
speech_api.replies = [(200, speech_api.written(0))] + [(200, other)] * 3
recording = add_recording(session)
run(recording.id)
session.expire_all()
assert recording.document_metadata["transcription"]["model"] == "other-model"
assert recording.document_metadata["transcription"] == {
"language": "en",
"duration_s": pytest.approx(36.65, abs=0.01),
"model": "other-model",
}
assert "Words of window 0." not in recording.content
assert " · English (detected) · other-model." in recording.content.split("\n")[0]
assert len(speech_api.windows) == 4

View file

@ -97,20 +97,45 @@ def test_bytes_the_decoder_cannot_read_are_undecodable(tmp_path: Path) -> None:
assert list(tmp_path.iterdir()) == [source]
@pytest.mark.parametrize("name", ["tone.mp3", "tone.ogg"])
def test_a_disk_without_room_for_the_decoded_audio_is_refused_first(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, name: str
) -> None:
"""Checked from the file's own length before a byte is written."""
usage = namedtuple("usage", "total used free")
monkeypatch.setattr(shutil, "disk_usage", lambda _path: usage(1, 1, 10_000))
out = tmp_path / "audio16k.raw"
outcome = decode_to_pcm(AUDIO / "tone.mp3", out)
outcome = decode_to_pcm(AUDIO / name, out)
assert (outcome.kind, outcome.needed_bytes) == ("no_room", needed_bytes(1.0))
assert not out.exists()
@pytest.mark.parametrize(
("fixture", "name"),
[
("tone.mp3", "réunion.mp3"),
("tone.flac", "会議.flac"),
("tone.ogg", "会議.ogg"),
("tone.ogg", "Besprechung für Jürgen.oga"),
],
)
def test_a_recording_named_in_any_script_decodes(
tmp_path: Path, fixture: str, name: str
) -> None:
"""A file called 会議.ogg, or a data folder under a Windows account named
José: no format may be refused for its path."""
folder = tmp_path / "José"
folder.mkdir()
source = folder / name
source.write_bytes((AUDIO / fixture).read_bytes())
outcome = decode_to_pcm(source, folder / "audio16k.raw")
assert outcome.kind == "decoded", outcome
def test_an_hour_needs_its_pcm_and_some_room_to_spare() -> None:
"""115.2 MB an hour at 16 kHz 16-bit mono, plus 64 MiB."""
assert needed_bytes(3600) == 115_200_000 + 64 * 2**20

View file

@ -9,6 +9,7 @@ from pathlib import Path
import miniaudio
from worker.audio_decode.ogg_length import vorbis_seconds
from worker.audio_decode.outcome import DecodeOutcome
from worker.audio_decode.room import needed_bytes
@ -25,7 +26,7 @@ def decode_to_pcm(source: Path, out: Path) -> DecodeOutcome:
part = out.with_name(f"{out.name}.part")
try:
needed = needed_bytes(_header_seconds(source))
except miniaudio.MiniaudioError as error:
except (miniaudio.MiniaudioError, ValueError) as error:
return DecodeOutcome("undecodable", detail=str(error))
if shutil.disk_usage(out.parent).free < needed:
return DecodeOutcome("no_room", needed_bytes=needed)
@ -47,9 +48,9 @@ def decode_to_pcm(source: Path, out: Path) -> DecodeOutcome:
def _header_seconds(source: Path) -> float:
# miniaudio picks the reader by extension and knows .oga by no name.
if source.suffix.lower() == ".oga":
return miniaudio.vorbis_get_file_info(str(source)).duration
# miniaudio's Vorbis reader opens by a narrow path; the others by a wide one.
if source.suffix.lower() in (".ogg", ".oga"):
return vorbis_seconds(source)
return miniaudio.get_file_info(str(source)).duration

View file

@ -0,0 +1,66 @@
"""An Ogg Vorbis recording's length read off its own pages: the last page's
granule position over the rate its first packet names. stb_vorbis opens a path
with fopen, which on Windows fails for any name outside the ANSI code page."""
import os
import struct
from pathlib import Path
__all__ = ["vorbis_seconds"]
# OggS, version, flags, granule position, serial, sequence, CRC, segment count.
_PAGE = struct.Struct("<4sBBqIIIB")
# A page holds at most 255 segments of 255 bytes: its whole size bounds both
# reads.
_MAX_PAGE = _PAGE.size + 255 + 255 * 255
_VORBIS_ID = b"\x01vorbis"
# The identification packet's rate, after its type, name, version and channels.
_RATE_AT = 12
# No packet ends on such a page, so it has no position.
_NO_GRANULE = -1
def vorbis_seconds(source: Path) -> float:
"""Raises ValueError when no Vorbis stream can be measured in it."""
with source.open("rb") as recording:
head = recording.read(_MAX_PAGE)
size = recording.seek(0, os.SEEK_END)
recording.seek(max(0, size - 2 * _MAX_PAGE))
tail = recording.read()
serial, rate = _vorbis_stream(head)
return _last_granule(tail, serial) / rate
def _vorbis_stream(head: bytes) -> tuple[int, int]:
if len(head) < _PAGE.size or not head.startswith(b"OggS"):
raise ValueError("not an Ogg file")
*_, serial, _, _, segments = _PAGE.unpack_from(head)
packet = head[_PAGE.size + segments :]
if not packet.startswith(_VORBIS_ID) or len(packet) < _RATE_AT + 4:
raise ValueError("not an Ogg Vorbis file")
(rate,) = struct.unpack_from("<I", packet, _RATE_AT)
if rate == 0:
raise ValueError("an Ogg Vorbis file with no sample rate")
return serial, rate
def _last_granule(tail: bytes, serial: int) -> int:
"""The last whole page of the stream that ends a packet."""
at = tail.rfind(b"OggS")
while at >= 0:
if len(tail) - at >= _PAGE.size:
_, version, _, granule, page_serial, _, _, segments = _PAGE.unpack_from(
tail, at
)
table = tail[at + _PAGE.size : at + _PAGE.size + segments]
whole = at + _PAGE.size + segments + sum(table) <= len(tail)
if (
version == 0
and page_serial == serial
and granule != _NO_GRANULE
and len(table) == segments
and whole
):
return granule
at = tail.rfind(b"OggS", 0, at)
raise ValueError("an Ogg Vorbis file with no audio page")

View file

@ -32,4 +32,6 @@ class QueueTurn:
if time.monotonic() - self._started < YIELD_AFTER_SECONDS:
return
if other_sources_pending(self._session):
# A cancel landed meanwhile ends the job, files and all, instead.
self.check_cancelled()
raise JobYieldedError

View file

@ -17,6 +17,7 @@ from pathlib import Path
from sqlalchemy.orm import Session, object_session
from modules.documents.models import Document
from modules.documents.recordings.ended_run import drop_run_files
from modules.documents.recordings.layout import WorkFiles, segments_file, work_files
from modules.documents.recordings.metadata import AWAITING, KEY, is_awaiting
from modules.llm.model_route.api_address import api_url
@ -79,10 +80,10 @@ def recording_markdown(document: Document, original: Path) -> str:
raise
except AwaitingSpeechModelError:
merge_metadata(session, document, KEY, {AWAITING: True})
_end_of_run(files)
drop_run_files(files)
raise
except BaseException:
_end_of_run(files)
drop_run_files(files)
raise
@ -95,10 +96,10 @@ def _transcribe(
folder: Path,
status: SpeechStatus,
) -> str:
assert status.model is not None, "a ready status names its model"
files = work_files(folder)
model_id, intent = status.model, status.language_intent
while True:
assert status.model is not None, "a ready status names its model"
model_id, intent = status.model, status.language_intent
plan, done = _planned(original, files, turn, model_id, intent)
with _full_disk_refused(plan.seconds):
try:
@ -108,9 +109,10 @@ def _transcribe(
),
done,
)
except ModelChangedError as changed:
except ModelChangedError:
files.windows.unlink(missing_ok=True)
model_id = changed.model_id
# A model chosen anew starts without the last one's language.
status = speech_recognition_ready(api, turn)
continue
return _finished(session, document, folder, plan, done, model_id, status)
@ -194,13 +196,6 @@ def _finished(
return markdown
def _end_of_run(files: WorkFiles) -> None:
"""A run that ends the job keeps its windows and plan for the next one, not
the decoded audio, which decoding again costs seconds to make."""
files.audio.unlink(missing_ok=True)
files.progress.unlink(missing_ok=True)
@contextmanager
def _full_disk_refused(seconds: float) -> Iterator[None]:
try: