Add recovery for missing oneliners
This commit introduces a recovery mechanism to regenerate `oneliner.txt` files for cases where transcription was successful but the oneliner generation failed due to a crash between writing the transcript and the oneliner. The `transcribe::recovery::regenerate_missing_oneliners` function is added and called at startup after `scan_and_enqueue`. This function identifies cases with non-empty transcripts but missing oneliners and triggers their regeneration. The `transcribe::worker::update_oneliner` function is refactored to always regenerate the oneliner from all available transcripts in a case, ensuring that later recordings can correct earlier ones. This replaces the previous `ensure_oneliner` logic which only generated the oneliner if it didn't exist. New helper functions `cases_needing_oneliner` and `all_transcripts_joined` are introduced to support these changes. Comprehensive unit tests are included for the new recovery and file processing logic.
This commit is contained in:
@@ -77,8 +77,13 @@ async fn main() {
|
|||||||
{
|
{
|
||||||
let tx = transcribe_tx.clone();
|
let tx = transcribe_tx.clone();
|
||||||
let data_path = config.data_path.clone();
|
let data_path = config.data_path.clone();
|
||||||
|
let client = http_client.clone();
|
||||||
|
let cfg = config.clone();
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
transcribe::recovery::scan_and_enqueue(&data_path, &tx).await;
|
transcribe::recovery::scan_and_enqueue(&data_path, &tx).await;
|
||||||
|
// Then catch up oneliners for cases that crashed between
|
||||||
|
// transcript-write and oneliner-write in a previous run.
|
||||||
|
transcribe::recovery::regenerate_missing_oneliners(&data_path, &client, &cfg).await;
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,8 +1,9 @@
|
|||||||
use std::path::Path;
|
use std::path::{Path, PathBuf};
|
||||||
|
|
||||||
use tracing::{info, warn};
|
use tracing::{info, warn};
|
||||||
|
|
||||||
use super::{TranscribeJob, TranscribeSender};
|
use super::{TranscribeJob, TranscribeSender};
|
||||||
|
use crate::config::Config;
|
||||||
|
|
||||||
/// Walk `data_path/*/` and enqueue every recording pending transcription for
|
/// Walk `data_path/*/` and enqueue every recording pending transcription for
|
||||||
/// every user. Intended for startup; the same primitive backs the per-user
|
/// every user. Intended for startup; the same primitive backs the per-user
|
||||||
@@ -79,3 +80,137 @@ pub async fn enqueue_pending_for_user(
|
|||||||
}
|
}
|
||||||
sent
|
sent
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Walk `data_path/*/*` and return every case directory that has at least
|
||||||
|
/// one non-empty `*.transcript.txt` but no `oneliner.txt`. Pure filesystem
|
||||||
|
/// scan, no LLM calls — separated from the regeneration wrapper so it can
|
||||||
|
/// be unit-tested without mocking Ollama.
|
||||||
|
pub(crate) async fn cases_needing_oneliner(data_path: &Path) -> Vec<PathBuf> {
|
||||||
|
let mut out: Vec<PathBuf> = Vec::new();
|
||||||
|
let Ok(mut users) = tokio::fs::read_dir(data_path).await else {
|
||||||
|
return out;
|
||||||
|
};
|
||||||
|
while let Ok(Some(user_entry)) = users.next_entry().await {
|
||||||
|
let user_path = user_entry.path();
|
||||||
|
if !user_path.is_dir() {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
let Ok(mut cases) = tokio::fs::read_dir(&user_path).await else {
|
||||||
|
continue;
|
||||||
|
};
|
||||||
|
while let Ok(Some(case_entry)) = cases.next_entry().await {
|
||||||
|
let case_dir = case_entry.path();
|
||||||
|
if !case_dir.is_dir() || crate::paths::is_deleted(&case_dir).await {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
if case_dir.join("oneliner.txt").exists() {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
if has_non_empty_transcript(&case_dir).await {
|
||||||
|
out.push(case_dir);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
out.sort();
|
||||||
|
out
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn has_non_empty_transcript(case_dir: &Path) -> bool {
|
||||||
|
let Ok(mut files) = tokio::fs::read_dir(case_dir).await else {
|
||||||
|
return false;
|
||||||
|
};
|
||||||
|
while let Ok(Some(f)) = files.next_entry().await {
|
||||||
|
let path = f.path();
|
||||||
|
let Some(name) = path.file_name().and_then(|s| s.to_str()) else {
|
||||||
|
continue;
|
||||||
|
};
|
||||||
|
if !name.ends_with(".transcript.txt") {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
if let Ok(text) = tokio::fs::read_to_string(&path).await
|
||||||
|
&& !text.trim().is_empty()
|
||||||
|
{
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
false
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Regenerate `oneliner.txt` for every case that has transcripts but no
|
||||||
|
/// oneliner (crash between transcript-write and oneliner-write). Called at
|
||||||
|
/// startup after `scan_and_enqueue`. Sequential on purpose — the oneliner
|
||||||
|
/// model is light, but spamming it in parallel is not worth it.
|
||||||
|
pub async fn regenerate_missing_oneliners(
|
||||||
|
data_path: &Path,
|
||||||
|
client: &reqwest::Client,
|
||||||
|
config: &Config,
|
||||||
|
) {
|
||||||
|
let cases = cases_needing_oneliner(data_path).await;
|
||||||
|
if cases.is_empty() {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
info!(count = cases.len(), "Regenerating missing oneliners");
|
||||||
|
for case_dir in cases {
|
||||||
|
super::worker::update_oneliner(&case_dir, client, config).await;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use tempfile::tempdir;
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn cases_needing_oneliner_finds_crash_survivor() {
|
||||||
|
let data = tempdir().unwrap();
|
||||||
|
let case = data.path().join("user").join("case1");
|
||||||
|
tokio::fs::create_dir_all(&case).await.unwrap();
|
||||||
|
tokio::fs::write(case.join("2026-04-16T10-00-00Z.transcript.txt"), "content")
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let got = cases_needing_oneliner(data.path()).await;
|
||||||
|
assert_eq!(got, vec![case]);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn cases_needing_oneliner_skips_cases_with_oneliner() {
|
||||||
|
let data = tempdir().unwrap();
|
||||||
|
let case = data.path().join("user").join("case1");
|
||||||
|
tokio::fs::create_dir_all(&case).await.unwrap();
|
||||||
|
tokio::fs::write(case.join("2026-04-16T10-00-00Z.transcript.txt"), "content")
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
tokio::fs::write(case.join("oneliner.txt"), "Existing")
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
assert!(cases_needing_oneliner(data.path()).await.is_empty());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn cases_needing_oneliner_skips_cases_with_only_empty_transcripts() {
|
||||||
|
let data = tempdir().unwrap();
|
||||||
|
let case = data.path().join("user").join("case1");
|
||||||
|
tokio::fs::create_dir_all(&case).await.unwrap();
|
||||||
|
tokio::fs::write(case.join("2026-04-16T10-00-00Z.transcript.txt"), " ")
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
assert!(cases_needing_oneliner(data.path()).await.is_empty());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn cases_needing_oneliner_skips_deleted_cases() {
|
||||||
|
let data = tempdir().unwrap();
|
||||||
|
let case = data.path().join("user").join("case1");
|
||||||
|
tokio::fs::create_dir_all(&case).await.unwrap();
|
||||||
|
tokio::fs::write(case.join("2026-04-16T10-00-00Z.transcript.txt"), "content")
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
// Empty JSON object is a valid delete marker (see paths::is_deleted).
|
||||||
|
tokio::fs::write(case.join(".deleted"), "{}").await.unwrap();
|
||||||
|
|
||||||
|
assert!(cases_needing_oneliner(data.path()).await.is_empty());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
+127
-15
@@ -71,12 +71,12 @@ pub async fn run(
|
|||||||
"Transcript written"
|
"Transcript written"
|
||||||
);
|
);
|
||||||
|
|
||||||
// Silent recording: Whisper returned an empty string. Skip the
|
// Regenerate the oneliner from all transcripts of the case. Later
|
||||||
// oneliner step — otherwise the LLM echoes the system prompt back.
|
// recordings override earlier ones — symmetric to the analyze-LLM
|
||||||
if !text.trim().is_empty()
|
// rule "Spätere Aufnahmen haben Vorrang". Silent / empty cases are
|
||||||
&& let Some(case_dir) = audio_path.parent()
|
// no-ops; LLM errors leave any existing oneliner untouched.
|
||||||
{
|
if let Some(case_dir) = audio_path.parent() {
|
||||||
ensure_oneliner(case_dir, &text, &client, &config).await;
|
update_oneliner(case_dir, &client, &config).await;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -106,25 +106,31 @@ async fn mark_failed(audio_path: &Path) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Generate `case_dir/oneliner.txt` from the given transcript iff the file
|
/// Regenerate `case_dir/oneliner.txt` from **all** non-empty transcripts in
|
||||||
/// does not already exist. Errors are non-fatal: logged and swallowed.
|
/// the case, joined chronologically. Called after every successful transcript
|
||||||
async fn ensure_oneliner(
|
/// write so later recordings can correct earlier ones (mirror of the
|
||||||
|
/// analyze-LLM rule "later recordings override"). Overwrites any existing
|
||||||
|
/// oneliner on success; LLM errors are non-fatal and leave the previous
|
||||||
|
/// oneliner (if any) untouched. Silent case (no non-empty transcript) is a
|
||||||
|
/// no-op — next transcript write retries.
|
||||||
|
pub(crate) async fn update_oneliner(
|
||||||
case_dir: &Path,
|
case_dir: &Path,
|
||||||
transcript: &str,
|
|
||||||
client: &reqwest::Client,
|
client: &reqwest::Client,
|
||||||
config: &Config,
|
config: &Config,
|
||||||
) {
|
) {
|
||||||
let path = case_dir.join("oneliner.txt");
|
let path = case_dir.join("oneliner.txt");
|
||||||
if path.exists() {
|
|
||||||
return;
|
let transcript = match all_transcripts_joined(case_dir).await {
|
||||||
}
|
Some(t) => t,
|
||||||
|
None => return,
|
||||||
|
};
|
||||||
|
|
||||||
match ollama::generate_oneliner(
|
match ollama::generate_oneliner(
|
||||||
client,
|
client,
|
||||||
&config.ollama_url,
|
&config.ollama_url,
|
||||||
&config.ollama_model,
|
&config.ollama_model,
|
||||||
config.ollama_keep_alive,
|
config.ollama_keep_alive,
|
||||||
transcript,
|
&transcript,
|
||||||
ONELINER_TIMEOUT,
|
ONELINER_TIMEOUT,
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
@@ -133,7 +139,7 @@ async fn ensure_oneliner(
|
|||||||
if let Err(e) = tokio::fs::write(&path, &line).await {
|
if let Err(e) = tokio::fs::write(&path, &line).await {
|
||||||
error!(path = %path.display(), error = %e, "writing oneliner failed");
|
error!(path = %path.display(), error = %e, "writing oneliner failed");
|
||||||
} else {
|
} else {
|
||||||
info!(path = %path.display(), chars = line.chars().count(), "Oneliner written");
|
info!(path = %path.display(), chars = line.chars().count(), "Oneliner updated");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
@@ -141,3 +147,109 @@ async fn ensure_oneliner(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Read every non-empty `*.transcript.txt` in `case_dir`, ordered
|
||||||
|
/// chronologically (ISO-8601 filename sort equals chronological order), and
|
||||||
|
/// join them with `\n\n---\n\n`. Empty / whitespace-only transcripts (silent
|
||||||
|
/// recordings) are skipped. Returns None if no non-empty transcript exists.
|
||||||
|
pub(crate) async fn all_transcripts_joined(case_dir: &Path) -> Option<String> {
|
||||||
|
let mut entries = tokio::fs::read_dir(case_dir).await.ok()?;
|
||||||
|
let mut files: Vec<std::path::PathBuf> = Vec::new();
|
||||||
|
while let Ok(Some(entry)) = entries.next_entry().await {
|
||||||
|
let name = entry.file_name();
|
||||||
|
let Some(s) = name.to_str() else { continue };
|
||||||
|
if s.ends_with(".transcript.txt") {
|
||||||
|
files.push(entry.path());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
files.sort();
|
||||||
|
|
||||||
|
let mut parts: Vec<String> = Vec::with_capacity(files.len());
|
||||||
|
for path in files {
|
||||||
|
if let Ok(text) = tokio::fs::read_to_string(&path).await {
|
||||||
|
let trimmed = text.trim();
|
||||||
|
if !trimmed.is_empty() {
|
||||||
|
parts.push(trimmed.to_string());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if parts.is_empty() {
|
||||||
|
return None;
|
||||||
|
}
|
||||||
|
Some(parts.join("\n\n---\n\n"))
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use tempfile::tempdir;
|
||||||
|
|
||||||
|
/// Regression guard for the Womax bug: the joined transcript must
|
||||||
|
/// include ALL recordings in chronological order, so the LLM sees
|
||||||
|
/// both the initial term and any later correction.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn all_transcripts_joined_includes_all_in_order() {
|
||||||
|
let dir = tempdir().unwrap();
|
||||||
|
tokio::fs::write(
|
||||||
|
dir.path().join("2026-04-16T10-05-00Z.transcript.txt"),
|
||||||
|
"Korrektur. Das Mittel heißt Vomex.",
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
tokio::fs::write(
|
||||||
|
dir.path().join("2026-04-16T10-00-00Z.transcript.txt"),
|
||||||
|
"Jetzt weiß ich es wieder. Es heißt Womax.",
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let got = all_transcripts_joined(dir.path()).await.expect("Some");
|
||||||
|
assert_eq!(
|
||||||
|
got,
|
||||||
|
"Jetzt weiß ich es wieder. Es heißt Womax.\n\n---\n\nKorrektur. Das Mittel heißt Vomex."
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Silent recordings are skipped, not treated as end-of-stream — a
|
||||||
|
/// later non-empty recording still contributes.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn all_transcripts_joined_skips_empty_recordings() {
|
||||||
|
let dir = tempdir().unwrap();
|
||||||
|
tokio::fs::write(
|
||||||
|
dir.path().join("2026-04-16T10-00-00Z.transcript.txt"),
|
||||||
|
"",
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
tokio::fs::write(
|
||||||
|
dir.path().join("2026-04-16T10-05-00Z.transcript.txt"),
|
||||||
|
"non-empty content",
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let got = all_transcripts_joined(dir.path()).await.expect("Some");
|
||||||
|
assert_eq!(got, "non-empty content");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn all_transcripts_joined_is_none_when_all_empty() {
|
||||||
|
let dir = tempdir().unwrap();
|
||||||
|
tokio::fs::write(
|
||||||
|
dir.path().join("2026-04-16T10-00-00Z.transcript.txt"),
|
||||||
|
" \n ",
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert!(all_transcripts_joined(dir.path()).await.is_none());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn all_transcripts_joined_is_none_when_no_transcripts() {
|
||||||
|
let dir = tempdir().unwrap();
|
||||||
|
tokio::fs::write(dir.path().join("2026-04-16T10-00-00Z.m4a"), b"audio")
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert!(all_transcripts_joined(dir.path()).await.is_none());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user