From 74c2a180eecb8795240cdb1cca8223bbc2eb4bea Mon Sep 17 00:00:00 2001 From: GodsQuantum <193283917+GodsQuantum@users.noreply.github.com> Date: Wed, 16 Sep 2026 11:10:07 +0200 Subject: [PATCH 1/6] feat: add transcription chain model --- src/api/providers.rs | 14 +-- src/api/quick.rs | 1 + src/domain.rs | 51 +++++++++ src/jobs.rs | 28 ++--- src/lib.rs | 1 + src/quick.rs | 34 +++--- src/quick/tests.rs | 1 + src/transcription_chain.rs | 217 +++++++++++++++++++++++++++++++++++++ src/workflows.rs | 22 ++-- 9 files changed, 323 insertions(+), 46 deletions(-) create mode 100644 src/transcription_chain.rs diff --git a/src/api/providers.rs b/src/api/providers.rs index 17465d9..d799c24 100644 --- a/src/api/providers.rs +++ b/src/api/providers.rs @@ -113,13 +113,13 @@ pub async fn delete( State(state): State, Path(id): Path, ) -> AppResult { - if state - .workflows - .read() - .await - .iter() - .any(|workflow| workflow.provider_id == id) - { + if state.workflows.read().await.iter().any(|workflow| { + workflow.provider_id == id + || workflow + .transcription_chain + .iter() + .any(|route| route.provider_id == id) + }) { return Err(AppError::Conflict( "provider is still used by a workflow".into(), )); diff --git a/src/api/quick.rs b/src/api/quick.rs index a98774f..e695ba7 100644 --- a/src/api/quick.rs +++ b/src/api/quick.rs @@ -176,6 +176,7 @@ pub async fn upload( .filter(|value| !value.trim().is_empty()) .ok_or_else(|| AppError::BadRequest("providerId is required".into()))?, model: model.unwrap_or_default(), + transcription_chain: Vec::new(), language, output_kind: parse_output_kind(output_kind)?, output_dir, diff --git a/src/domain.rs b/src/domain.rs index 6161e1b..efc2d12 100644 --- a/src/domain.rs +++ b/src/domain.rs @@ -49,6 +49,39 @@ impl From<&Provider> for ProviderView { } } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct TranscriptionRoute { + pub provider_id: String, + pub model: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub fallback_after_seconds: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum TranscriptionAttemptOutcome { + Success, + Failed, + TimedOut, + Cancelled, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct TranscriptionAttempt { + pub run: u32, + pub route_index: usize, + pub provider_id: String, + pub provider_name: String, + pub model: String, + pub started_at_ms: u128, + pub finished_at_ms: u128, + pub outcome: TranscriptionAttemptOutcome, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "camelCase")] pub struct MarkdownOptions { @@ -82,9 +115,12 @@ pub struct Workflow { pub archive_dir: String, #[serde(default)] pub tags: Vec, + #[serde(default)] pub provider_id: String, #[serde(default)] pub model: String, + #[serde(default)] + pub transcription_chain: Vec, #[serde(default, skip_serializing_if = "Option::is_none")] pub language: Option, #[serde(default)] @@ -162,6 +198,16 @@ pub struct Job { #[serde(default, skip_serializing_if = "Option::is_none")] pub quick: Option, pub provider_id: String, + #[serde(default)] + pub transcription_chain: Vec, + #[serde(default)] + pub transcription_attempts: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub used_provider_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub used_provider_name: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub used_model: Option, pub original_name: String, pub source_path: PathBuf, pub source_size: u64, @@ -212,12 +258,17 @@ mod tests { let workflow: Workflow = serde_json::from_str(workflow_json).unwrap(); assert_eq!(workflow.output_dir, None); assert!(workflow.tags.is_empty()); + assert!(workflow.transcription_chain.is_empty()); let job_json = r#"{"id":"j","workflowId":"w","providerId":"p","originalName":"a.m4a","sourcePath":"/w/a.m4a","sourceSize":1,"sourceMtimeNs":1,"model":"m","status":"done","attempts":1,"markdownPublished":true,"createdAtMs":1,"updatedAtMs":1}"#; let job: Job = serde_json::from_str(job_json).unwrap(); assert_eq!(job.kind, JobKind::Workflow); assert_eq!(job.workflow_id.as_deref(), Some("w")); assert!(job.quick.is_none()); + assert!(job.transcription_chain.is_empty()); + assert!(job.transcription_attempts.is_empty()); + assert!(job.used_provider_id.is_none()); + assert!(job.used_model.is_none()); } #[test] diff --git a/src/jobs.rs b/src/jobs.rs index 4bfaa31..a9022ad 100644 --- a/src/jobs.rs +++ b/src/jobs.rs @@ -53,19 +53,13 @@ pub async fn create_job( source_size: u64, source_mtime_ns: i128, ) -> Result { - let provider = state - .providers - .read() - .await - .iter() - .find(|provider| provider.id == workflow.provider_id) + let providers = state.providers.read().await.clone(); + let transcription_chain = + crate::transcription_chain::normalize_workflow_chain(workflow, &providers)?; + let primary = transcription_chain + .first() .cloned() - .ok_or_else(|| anyhow!("provider not found: {}", workflow.provider_id))?; - let model = if workflow.model.trim().is_empty() { - provider.model - } else { - workflow.model.clone() - }; + .ok_or_else(|| anyhow!("transcription chain is empty"))?; let original_name = source_path .file_name() .and_then(|v| v.to_str()) @@ -77,12 +71,17 @@ pub async fn create_job( kind: crate::domain::JobKind::Workflow, workflow_id: Some(workflow.id.clone()), quick: None, - provider_id: workflow.provider_id.clone(), + provider_id: primary.provider_id.clone(), + transcription_chain, + transcription_attempts: Vec::new(), + used_provider_id: None, + used_provider_name: None, + used_model: None, original_name, source_path, source_size, source_mtime_ns, - model, + model: primary.model, language: workflow.language.clone(), status: JobStatus::Pending, attempts: 1, @@ -255,6 +254,7 @@ mod tests { tags: Vec::new(), provider_id: "provider".into(), model: String::new(), + transcription_chain: Vec::new(), language: None, markdown: MarkdownOptions::default(), enabled: true, diff --git a/src/lib.rs b/src/lib.rs index df115ee..7ab4969 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -10,4 +10,5 @@ mod pipeline_quick; pub mod provider; pub mod quick; pub mod state; +pub mod transcription_chain; pub mod workflows; diff --git a/src/quick.rs b/src/quick.rs index 2c716ab..4430146 100644 --- a/src/quick.rs +++ b/src/quick.rs @@ -1,5 +1,7 @@ use crate::{ - domain::{Job, JobKind, JobStatus, QuickJobMeta, QuickOutputKind, QuickSourceKind}, + domain::{ + Job, JobKind, JobStatus, QuickJobMeta, QuickOutputKind, QuickSourceKind, TranscriptionRoute, + }, jobs::{now_ms, persist_job}, state::AppState, workflows::is_audio_candidate, @@ -16,10 +18,13 @@ use uuid::Uuid; #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct QuickOptions { + #[serde(default)] pub provider_id: String, #[serde(default)] pub model: String, #[serde(default)] + pub transcription_chain: Vec, + #[serde(default)] pub language: Option, pub output_kind: QuickOutputKind, #[serde(default)] @@ -50,19 +55,13 @@ async fn normalized_job( if !is_audio_candidate(Path::new(&original_name)) { bail!("unsupported audio file: {original_name}"); } - let provider = state - .providers - .read() - .await - .iter() - .find(|provider| provider.id == options.provider_id) + let providers = state.providers.read().await.clone(); + let transcription_chain = + crate::transcription_chain::normalize_quick_chain(&options, &providers)?; + let primary = transcription_chain + .first() .cloned() - .ok_or_else(|| anyhow!("provider not found: {}", options.provider_id))?; - let model = if options.model.trim().is_empty() { - provider.model.clone() - } else { - options.model.trim().to_owned() - }; + .ok_or_else(|| anyhow!("transcription chain is empty"))?; options.language = options .language .as_deref() @@ -103,12 +102,17 @@ async fn normalized_job( result_name: None, frontmatter: options.frontmatter, }), - provider_id: provider.id, + provider_id: primary.provider_id.clone(), + transcription_chain, + transcription_attempts: Vec::new(), + used_provider_id: None, + used_provider_name: None, + used_model: None, original_name, source_path, source_size: metadata.len(), source_mtime_ns: mtime_ns(&metadata), - model, + model: primary.model, language: options.language, status: JobStatus::Pending, attempts: 1, diff --git a/src/quick/tests.rs b/src/quick/tests.rs index c54342a..4f52de9 100644 --- a/src/quick/tests.rs +++ b/src/quick/tests.rs @@ -77,6 +77,7 @@ fn options(output_kind: QuickOutputKind, output_dir: Option) -> QuickOpt QuickOptions { provider_id: "provider".into(), model: String::new(), + transcription_chain: Vec::new(), language: None, output_kind, output_dir, diff --git a/src/transcription_chain.rs b/src/transcription_chain.rs new file mode 100644 index 0000000..fb31601 --- /dev/null +++ b/src/transcription_chain.rs @@ -0,0 +1,217 @@ +use crate::{ + domain::{Job, Provider, TranscriptionRoute, Workflow}, + quick::QuickOptions, +}; +use anyhow::{Result, anyhow, bail}; + +fn normalize_explicit_routes( + routes: &[TranscriptionRoute], + providers: &[Provider], +) -> Result> { + let mut normalized = Vec::with_capacity(routes.len()); + for (index, route) in routes.iter().enumerate() { + let provider_id = route.provider_id.trim(); + let model = route.model.trim(); + if provider_id.is_empty() { + bail!("transcription route {} providerId is required", index + 1); + } + if model.is_empty() { + bail!("transcription route {} model is required", index + 1); + } + if route.fallback_after_seconds == Some(0) { + bail!( + "transcription route {} fallback timeout must be greater than zero", + index + 1 + ); + } + if !providers.iter().any(|provider| provider.id == provider_id) { + bail!( + "unknown providerId in transcription route {}: {provider_id}", + index + 1 + ); + } + normalized.push(TranscriptionRoute { + provider_id: provider_id.to_owned(), + model: model.to_owned(), + fallback_after_seconds: route.fallback_after_seconds, + }); + } + Ok(normalized) +} + +fn normalize_legacy_route( + provider_id: &str, + model: &str, + providers: &[Provider], +) -> Result> { + let provider_id = provider_id.trim(); + if provider_id.is_empty() { + bail!("providerId is required"); + } + let provider = providers + .iter() + .find(|provider| provider.id == provider_id) + .ok_or_else(|| anyhow!("unknown providerId: {provider_id}"))?; + let model = if model.trim().is_empty() { + provider.model.trim() + } else { + model.trim() + }; + if model.is_empty() { + bail!("model is required"); + } + Ok(vec![TranscriptionRoute { + provider_id: provider_id.to_owned(), + model: model.to_owned(), + fallback_after_seconds: None, + }]) +} + +pub fn normalize_workflow_chain( + workflow: &Workflow, + providers: &[Provider], +) -> Result> { + if workflow.transcription_chain.is_empty() { + normalize_legacy_route(&workflow.provider_id, &workflow.model, providers) + } else { + normalize_explicit_routes(&workflow.transcription_chain, providers) + } +} + +pub fn normalize_quick_chain( + options: &QuickOptions, + providers: &[Provider], +) -> Result> { + if options.transcription_chain.is_empty() { + normalize_legacy_route(&options.provider_id, &options.model, providers) + } else { + normalize_explicit_routes(&options.transcription_chain, providers) + } +} + +pub fn effective_job_chain(job: &Job) -> Result> { + if !job.transcription_chain.is_empty() { + let mut normalized = Vec::with_capacity(job.transcription_chain.len()); + for (index, route) in job.transcription_chain.iter().enumerate() { + let provider_id = route.provider_id.trim(); + let model = route.model.trim(); + if provider_id.is_empty() || model.is_empty() { + bail!("persisted transcription route {} is invalid", index + 1); + } + if route.fallback_after_seconds == Some(0) { + bail!( + "persisted transcription route {} has invalid timeout", + index + 1 + ); + } + normalized.push(TranscriptionRoute { + provider_id: provider_id.to_owned(), + model: model.to_owned(), + fallback_after_seconds: route.fallback_after_seconds, + }); + } + return Ok(normalized); + } + let provider_id = job.provider_id.trim(); + let model = job.model.trim(); + if provider_id.is_empty() || model.is_empty() { + bail!("legacy persisted job has no usable provider/model"); + } + Ok(vec![TranscriptionRoute { + provider_id: provider_id.to_owned(), + model: model.to_owned(), + fallback_after_seconds: None, + }]) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::domain::{MarkdownOptions, Provider, TranscriptionRoute, Workflow}; + + fn provider(id: &str, model: &str) -> Provider { + Provider { + id: id.into(), + name: id.into(), + transcription_url: "http://localhost/v1/audio/transcriptions".into(), + model: model.into(), + api_key: String::new(), + timeout_seconds: 600, + enabled: true, + } + } + + fn route(provider_id: &str, model: &str) -> TranscriptionRoute { + TranscriptionRoute { + provider_id: provider_id.into(), + model: model.into(), + fallback_after_seconds: None, + } + } + + fn workflow_with_chain(transcription_chain: Vec) -> Workflow { + Workflow { + id: "w".into(), + name: "test".into(), + watch_dir: "/w".into(), + output_dir: None, + archive_dir: "/a".into(), + tags: vec![], + provider_id: "p".into(), + model: String::new(), + transcription_chain, + language: None, + markdown: MarkdownOptions::default(), + enabled: true, + } + } + + #[test] + fn legacy_workflow_becomes_one_route_with_provider_default_model() { + let workflow: Workflow = serde_json::from_str( + r#"{ + "id":"w","name":"old","watchDir":"/w","archiveDir":"/a", + "providerId":"p","model":"","enabled":true + }"#, + ) + .unwrap(); + let providers = vec![provider("p", "whisper-large-v3")]; + let chain = normalize_workflow_chain(&workflow, &providers).unwrap(); + assert_eq!( + chain, + vec![TranscriptionRoute { + provider_id: "p".into(), + model: "whisper-large-v3".into(), + fallback_after_seconds: None, + }] + ); + } + + #[test] + fn same_provider_can_appear_twice_with_different_models() { + let routes = vec![route("p", "large-v3"), route("p", "distil-large-v3")]; + let workflow = workflow_with_chain(routes.clone()); + assert_eq!( + normalize_workflow_chain(&workflow, &[provider("p", "default")]).unwrap(), + routes + ); + } + + #[test] + fn explicit_routes_are_strict_but_allow_disabled_provider() { + let mut p = provider("p", "default"); + p.enabled = false; + let mut workflow = workflow_with_chain(vec![route(" p ", " large-v3 ")]); + workflow.transcription_chain[0].fallback_after_seconds = Some(30); + assert_eq!( + normalize_workflow_chain(&workflow, &[p]).unwrap(), + vec![TranscriptionRoute { + provider_id: "p".into(), + model: "large-v3".into(), + fallback_after_seconds: Some(30), + }] + ); + workflow.transcription_chain[0].fallback_after_seconds = Some(0); + assert!(normalize_workflow_chain(&workflow, &[provider("p", "default")]).is_err()); + } +} diff --git a/src/workflows.rs b/src/workflows.rs index e65e0c4..a593b4c 100644 --- a/src/workflows.rs +++ b/src/workflows.rs @@ -317,20 +317,19 @@ pub async fn normalize_definition(state: &AppState, mut workflow: Workflow) -> R if archive == watch || archive.starts_with(&watch) { anyhow::bail!("archive directory must be outside the watch directory"); } - if !state - .providers - .read() - .await - .iter() - .any(|provider| provider.id == workflow.provider_id) - { - anyhow::bail!("unknown providerId"); - } + let providers = state.providers.read().await.clone(); + let chain = crate::transcription_chain::normalize_workflow_chain(&workflow, &providers)?; + let primary = chain + .first() + .cloned() + .ok_or_else(|| anyhow::anyhow!("transcription chain is empty"))?; + workflow.transcription_chain = chain; + workflow.provider_id = primary.provider_id; + workflow.model = primary.model; workflow.watch_dir = watch.to_string_lossy().into_owned(); workflow.output_dir = output.map(|path| path.to_string_lossy().into_owned()); workflow.archive_dir = archive.to_string_lossy().into_owned(); workflow.tags = crate::markdown::normalize_custom_tags(&workflow.tags); - workflow.model = workflow.model.trim().to_owned(); workflow.language = workflow .language .as_deref() @@ -417,6 +416,7 @@ mod validation_tests { tags: Vec::new(), provider_id: "provider".into(), model: String::new(), + transcription_chain: Vec::new(), language: Some("auto".into()), markdown: MarkdownOptions::default(), enabled: true, @@ -462,6 +462,7 @@ mod validation_tests { tags: Vec::new(), provider_id: "provider".into(), model: String::new(), + transcription_chain: Vec::new(), language: None, markdown: MarkdownOptions::default(), enabled: true, @@ -494,6 +495,7 @@ mod validation_tests { tags: Vec::new(), provider_id: "provider".into(), model: String::new(), + transcription_chain: Vec::new(), language: None, markdown: MarkdownOptions::default(), enabled: true, From 328f51dc1dc0a76a14abb7d2ae269a418e254d30 Mon Sep 17 00:00:00 2001 From: GodsQuantum <193283917+GodsQuantum@users.noreply.github.com> Date: Wed, 16 Sep 2026 11:13:03 +0200 Subject: [PATCH 2/6] feat: execute resilient transcription fallbacks --- src/provider.rs | 7 +- src/transcription_chain.rs | 567 ++++++++++++++++++++++++++++++++++++- 2 files changed, 567 insertions(+), 7 deletions(-) diff --git a/src/provider.rs b/src/provider.rs index 4df931d..1f989bb 100644 --- a/src/provider.rs +++ b/src/provider.rs @@ -51,12 +51,7 @@ pub async fn transcribe( form = form.text("language", language.to_owned()); } - let mut request = client - .post(&provider.transcription_url) - .timeout(std::time::Duration::from_secs( - provider.timeout_seconds.max(1), - )) - .multipart(form); + let mut request = client.post(&provider.transcription_url).multipart(form); if !provider.api_key.trim().is_empty() { request = request.bearer_auth(&provider.api_key); } diff --git a/src/transcription_chain.rs b/src/transcription_chain.rs index fb31601..3ea8493 100644 --- a/src/transcription_chain.rs +++ b/src/transcription_chain.rs @@ -1,8 +1,16 @@ use crate::{ - domain::{Job, Provider, TranscriptionRoute, Workflow}, + domain::{ + Job, JobStatus, Provider, TranscriptionAttempt, TranscriptionAttemptOutcome, + TranscriptionRoute, Workflow, + }, + jobs::{get_job, now_ms, update_job}, + provider::{self, TranscriptionError}, quick::QuickOptions, + state::AppState, }; use anyhow::{Result, anyhow, bail}; +use std::{path::Path, time::Duration}; +use tokio_util::sync::CancellationToken; fn normalize_explicit_routes( routes: &[TranscriptionRoute], @@ -124,6 +132,245 @@ pub fn effective_job_chain(job: &Job) -> Result> { }]) } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TranscriptionSuccess { + pub transcript: String, + pub provider_id: String, + pub provider_name: String, + pub model: String, +} + +fn safe_attempt_error(value: impl ToString) -> String { + let flattened = value.to_string().replace(['\r', '\n'], " "); + flattened.chars().take(300).collect() +} + +fn append_attempt(state: &AppState, job_id: &str, attempt: TranscriptionAttempt) -> Result<()> { + update_job(state, job_id, |job| { + job.transcription_attempts.push(attempt) + })?; + Ok(()) +} + +fn finished_attempt( + run: u32, + route_index: usize, + route: &TranscriptionRoute, + provider_name: &str, + started_at_ms: u128, + outcome: TranscriptionAttemptOutcome, + error: Option, +) -> TranscriptionAttempt { + TranscriptionAttempt { + run, + route_index, + provider_id: route.provider_id.clone(), + provider_name: provider_name.to_owned(), + model: route.model.clone(), + started_at_ms, + finished_at_ms: now_ms(), + outcome, + error, + } +} + +pub async fn transcribe_job_chain( + state: &AppState, + job_id: &str, + source: &Path, + language: Option<&str>, + token: &CancellationToken, +) -> Result { + let job = get_job(state, job_id)?; + let routes = effective_job_chain(&job)?; + let run = job.attempts.max(1); + let mut failures = Vec::new(); + + for (route_index, route) in routes.iter().enumerate() { + if token.is_cancelled() { + bail!("cancelled"); + } + + let provider = state + .providers + .read() + .await + .iter() + .find(|provider| provider.id == route.provider_id) + .cloned(); + let started_at_ms = now_ms(); + let Some(provider) = provider else { + let message = "provider is no longer configured".to_owned(); + append_attempt( + state, + job_id, + finished_attempt( + run, + route_index, + route, + &route.provider_id, + started_at_ms, + TranscriptionAttemptOutcome::Failed, + Some(message.clone()), + ), + )?; + failures.push(format!("{}/{}: {message}", route.provider_id, route.model)); + continue; + }; + if !provider.enabled { + let message = "provider is disabled".to_owned(); + append_attempt( + state, + job_id, + finished_attempt( + run, + route_index, + route, + &provider.name, + started_at_ms, + TranscriptionAttemptOutcome::Failed, + Some(message.clone()), + ), + )?; + failures.push(format!("{}/{}: {message}", provider.name, route.model)); + continue; + } + + update_job(state, job_id, |job| { + job.status = JobStatus::Transcribing; + job.error = None; + })?; + + let request = provider::transcribe( + &provider, + &route.model, + language, + source, + &state.http, + token, + ); + let result = if let Some(seconds) = route.fallback_after_seconds { + match tokio::time::timeout(Duration::from_secs(seconds), request).await { + Ok(result) => result, + Err(_) => { + let message = format!("fallback timeout after {seconds}s"); + append_attempt( + state, + job_id, + finished_attempt( + run, + route_index, + route, + &provider.name, + started_at_ms, + TranscriptionAttemptOutcome::TimedOut, + Some(message.clone()), + ), + )?; + failures.push(format!("{}/{}: {message}", provider.name, route.model)); + continue; + } + } + } else { + request.await + }; + + match result { + Ok(transcript) => { + let attempt = finished_attempt( + run, + route_index, + route, + &provider.name, + started_at_ms, + TranscriptionAttemptOutcome::Success, + None, + ); + update_job(state, job_id, |job| { + job.transcription_attempts.push(attempt); + job.used_provider_id = Some(provider.id.clone()); + job.used_provider_name = Some(provider.name.clone()); + job.used_model = Some(route.model.clone()); + })?; + return Ok(TranscriptionSuccess { + transcript, + provider_id: provider.id, + provider_name: provider.name, + model: route.model.clone(), + }); + } + Err(TranscriptionError::Cancelled) => { + append_attempt( + state, + job_id, + finished_attempt( + run, + route_index, + route, + &provider.name, + started_at_ms, + TranscriptionAttemptOutcome::Cancelled, + Some("cancelled".into()), + ), + )?; + bail!("cancelled"); + } + Err(TranscriptionError::Io(error)) => { + let message = safe_attempt_error(&error); + append_attempt( + state, + job_id, + finished_attempt( + run, + route_index, + route, + &provider.name, + started_at_ms, + TranscriptionAttemptOutcome::Failed, + Some(message.clone()), + ), + )?; + return Err(error.into()); + } + Err(error) => { + if token.is_cancelled() { + append_attempt( + state, + job_id, + finished_attempt( + run, + route_index, + route, + &provider.name, + started_at_ms, + TranscriptionAttemptOutcome::Cancelled, + Some("cancelled".into()), + ), + )?; + bail!("cancelled"); + } + let message = safe_attempt_error(&error); + append_attempt( + state, + job_id, + finished_attempt( + run, + route_index, + route, + &provider.name, + started_at_ms, + TranscriptionAttemptOutcome::Failed, + Some(message.clone()), + ), + )?; + failures.push(format!("{}/{}: {message}", provider.name, route.model)); + } + } + } + + bail!("all transcription routes failed: {}", failures.join("; ")) +} + #[cfg(test)] mod tests { use super::*; @@ -214,4 +461,322 @@ mod tests { workflow.transcription_chain[0].fallback_after_seconds = Some(0); assert!(normalize_workflow_chain(&workflow, &[provider("p", "default")]).is_err()); } + + use crate::{ + config::Config, + domain::{Job, JobKind, JobStatus}, + jobs::{get_job, persist_job}, + state::AppState, + }; + use axum::{Json, Router, body::Bytes, http::StatusCode, routing::post}; + use serde_json::json; + use std::{ + path::{Path, PathBuf}, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + time::Duration, + }; + use tokio_util::sync::CancellationToken; + + async fn mock_provider( + status: StatusCode, + text: &'static str, + delay_ms: u64, + calls: Arc, + ) -> String { + let app = Router::new().route("/v1/audio/transcriptions", post(move |_body: Bytes| { + let calls = calls.clone(); + async move { + calls.fetch_add(1, Ordering::SeqCst); + if delay_ms > 0 { tokio::time::sleep(Duration::from_millis(delay_ms)).await; } + (status, Json(json!({"text": text, "error": if status.is_success() { "" } else { text }}))) + } + })); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + format!("http://{address}/v1/audio/transcriptions") + } + + fn test_config(root: &Path) -> Config { + Config { + host: "127.0.0.1".into(), + port: 0, + config_dir: root.join("config"), + dist_dir: root.join("dist"), + data_dir: root.join("data"), + allowed_roots: vec![root.to_path_buf()], + scan_seconds: 1, + file_stability_ms: 20, + max_transcription_jobs: 1, + max_upload_bytes: 2_147_483_648, + quick_result_retention_hours: 24, + } + } + + async fn state_with_job( + root: &Path, + providers: Vec, + routes: Vec, + ) -> (AppState, Job, PathBuf) { + let state = AppState::load(test_config(root)).await.unwrap(); + for provider in &providers { + state.db.upsert("provider", &provider.id, provider).unwrap(); + } + *state.providers.write().await = providers; + let source = root.join("audio.m4a"); + std::fs::write(&source, b"audio").unwrap(); + let primary = routes.first().unwrap().clone(); + let now = crate::jobs::now_ms(); + let job = Job { + id: "chain-job".into(), + kind: JobKind::Quick, + workflow_id: None, + quick: None, + provider_id: primary.provider_id.clone(), + transcription_chain: routes, + transcription_attempts: vec![], + used_provider_id: None, + used_provider_name: None, + used_model: None, + original_name: "audio.m4a".into(), + source_path: source.clone(), + source_size: 5, + source_mtime_ns: 1, + model: primary.model, + language: None, + status: JobStatus::Pending, + attempts: 1, + error: None, + markdown_path: None, + archive_path: None, + markdown_published: false, + created_at_ms: now, + updated_at_ms: now, + }; + state.jobs.insert(job.id.clone(), job.clone()); + persist_job(&state, &job).unwrap(); + (state, job, source) + } + + fn test_provider(id: &str, url: String, enabled: bool) -> Provider { + Provider { + id: id.into(), + name: id.into(), + transcription_url: url, + model: "default".into(), + api_key: String::new(), + timeout_seconds: 1, + enabled, + } + } + + #[tokio::test] + async fn primary_failure_then_fallback_success_records_both_attempts() { + let temp = tempfile::tempdir().unwrap(); + let c1 = Arc::new(AtomicUsize::new(0)); + let c2 = Arc::new(AtomicUsize::new(0)); + let p1 = test_provider( + "p1", + mock_provider(StatusCode::BAD_GATEWAY, "down", 0, c1.clone()).await, + true, + ); + let p2 = test_provider( + "p2", + mock_provider(StatusCode::OK, "fallback works", 0, c2.clone()).await, + true, + ); + let routes = vec![route("p1", "m1"), route("p2", "m2")]; + let (state, job, source) = state_with_job(temp.path(), vec![p1, p2], routes).await; + let result = + transcribe_job_chain(&state, &job.id, &source, None, &CancellationToken::new()) + .await + .unwrap(); + assert_eq!(result.transcript, "fallback works"); + assert_eq!(result.provider_id, "p2"); + assert_eq!(result.model, "m2"); + let saved = get_job(&state, &job.id).unwrap(); + assert_eq!(saved.transcription_attempts.len(), 2); + assert_eq!( + saved.transcription_attempts[0].outcome, + crate::domain::TranscriptionAttemptOutcome::Failed + ); + assert_eq!( + saved.transcription_attempts[1].outcome, + crate::domain::TranscriptionAttemptOutcome::Success + ); + assert_eq!(c1.load(Ordering::SeqCst), 1); + assert_eq!(c2.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn disabled_or_missing_provider_falls_through() { + let temp = tempfile::tempdir().unwrap(); + let calls = Arc::new(AtomicUsize::new(0)); + let disabled = test_provider( + "disabled", + "http://127.0.0.1:1/v1/audio/transcriptions".into(), + false, + ); + let good = test_provider( + "good", + mock_provider(StatusCode::OK, "ok", 0, calls.clone()).await, + true, + ); + let routes = vec![ + route("missing", "m0"), + route("disabled", "m1"), + route("good", "m2"), + ]; + let (state, job, source) = state_with_job(temp.path(), vec![disabled, good], routes).await; + let result = + transcribe_job_chain(&state, &job.id, &source, None, &CancellationToken::new()) + .await + .unwrap(); + assert_eq!(result.provider_id, "good"); + assert_eq!(calls.load(Ordering::SeqCst), 1); + assert_eq!( + get_job(&state, &job.id) + .unwrap() + .transcription_attempts + .len(), + 3 + ); + } + + #[tokio::test] + async fn route_timeout_falls_through() { + let temp = tempfile::tempdir().unwrap(); + let slow = Arc::new(AtomicUsize::new(0)); + let fast = Arc::new(AtomicUsize::new(0)); + let p1 = test_provider( + "slow", + mock_provider(StatusCode::OK, "too late", 1500, slow.clone()).await, + true, + ); + let p2 = test_provider( + "fast", + mock_provider(StatusCode::OK, "fast", 0, fast.clone()).await, + true, + ); + let mut first = route("slow", "m1"); + first.fallback_after_seconds = Some(1); + let (state, job, source) = + state_with_job(temp.path(), vec![p1, p2], vec![first, route("fast", "m2")]).await; + let result = + transcribe_job_chain(&state, &job.id, &source, None, &CancellationToken::new()) + .await + .unwrap(); + assert_eq!(result.transcript, "fast"); + let saved = get_job(&state, &job.id).unwrap(); + assert_eq!( + saved.transcription_attempts[0].outcome, + crate::domain::TranscriptionAttemptOutcome::TimedOut + ); + assert_eq!(slow.load(Ordering::SeqCst), 1); + assert_eq!(fast.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn no_route_timeout_waits_for_slow_success() { + let temp = tempfile::tempdir().unwrap(); + let calls = Arc::new(AtomicUsize::new(0)); + let p = test_provider( + "slow", + mock_provider(StatusCode::OK, "eventually", 150, calls.clone()).await, + true, + ); + let (state, job, source) = + state_with_job(temp.path(), vec![p], vec![route("slow", "m1")]).await; + let result = + transcribe_job_chain(&state, &job.id, &source, None, &CancellationToken::new()) + .await + .unwrap(); + assert_eq!(result.transcript, "eventually"); + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn cancellation_stops_chain_before_second_provider() { + let temp = tempfile::tempdir().unwrap(); + let slow = Arc::new(AtomicUsize::new(0)); + let second = Arc::new(AtomicUsize::new(0)); + let p1 = test_provider( + "slow", + mock_provider(StatusCode::OK, "late", 1000, slow.clone()).await, + true, + ); + let p2 = test_provider( + "second", + mock_provider(StatusCode::OK, "should not run", 0, second.clone()).await, + true, + ); + let (state, job, source) = state_with_job( + temp.path(), + vec![p1, p2], + vec![route("slow", "m1"), route("second", "m2")], + ) + .await; + let token = CancellationToken::new(); + let cancel = token.clone(); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(50)).await; + cancel.cancel(); + }); + assert!( + transcribe_job_chain(&state, &job.id, &source, None, &token) + .await + .is_err() + ); + assert_eq!(second.load(Ordering::SeqCst), 0); + assert_eq!( + get_job(&state, &job.id) + .unwrap() + .transcription_attempts + .last() + .unwrap() + .outcome, + crate::domain::TranscriptionAttemptOutcome::Cancelled + ); + } + + #[tokio::test] + async fn all_routes_fail_returns_aggregate_error_and_history() { + let temp = tempfile::tempdir().unwrap(); + let c1 = Arc::new(AtomicUsize::new(0)); + let c2 = Arc::new(AtomicUsize::new(0)); + let p1 = test_provider( + "p1", + mock_provider(StatusCode::BAD_GATEWAY, "first down", 0, c1).await, + true, + ); + let p2 = test_provider( + "p2", + mock_provider(StatusCode::TOO_MANY_REQUESTS, "rate limited", 0, c2).await, + true, + ); + let (state, job, source) = state_with_job( + temp.path(), + vec![p1, p2], + vec![route("p1", "m1"), route("p2", "m2")], + ) + .await; + let error = transcribe_job_chain(&state, &job.id, &source, None, &CancellationToken::new()) + .await + .unwrap_err() + .to_string(); + assert!(error.contains("p1/m1")); + assert!(error.contains("p2/m2")); + assert_eq!( + get_job(&state, &job.id) + .unwrap() + .transcription_attempts + .len(), + 2 + ); + } } From 9411ed68f7b9d20ead0c3866684f54b7f08d58c8 Mon Sep 17 00:00:00 2001 From: GodsQuantum <193283917+GodsQuantum@users.noreply.github.com> Date: Wed, 16 Sep 2026 11:50:40 +0200 Subject: [PATCH 3/6] feat: wire fallback chains into transcription jobs --- src/jobs.rs | 151 +++++++++++++++++++++++++++++++++++++++++- src/pipeline.rs | 32 +++------ src/pipeline_quick.rs | 31 +++------ 3 files changed, 165 insertions(+), 49 deletions(-) diff --git a/src/jobs.rs b/src/jobs.rs index a9022ad..e33cd24 100644 --- a/src/jobs.rs +++ b/src/jobs.rs @@ -174,11 +174,18 @@ mod tests { use crate::pipeline::process_job; use crate::{ config::Config, - domain::{MarkdownOptions, Provider}, + domain::{MarkdownOptions, Provider, TranscriptionAttemptOutcome, TranscriptionRoute}, }; use axum::{Json, Router, body::Bytes, http::StatusCode, routing::post}; use serde_json::json; - use std::{path::Path, time::Duration}; + use std::{ + path::Path, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + time::Duration, + }; async fn success_provider() -> String { let app = Router::new().route( @@ -200,6 +207,37 @@ mod tests { ); spawn_server(app).await } + + async fn counted_provider( + status: StatusCode, + text: &'static str, + ) -> (String, Arc) { + let calls = Arc::new(AtomicUsize::new(0)); + let seen = calls.clone(); + let app = Router::new().route( + "/v1/audio/transcriptions", + post(move |_body: Bytes| { + let seen = seen.clone(); + async move { + seen.fetch_add(1, Ordering::SeqCst); + if status.is_success() { + (status, Json(json!({"text": text}))) + } else { + (status, Json(json!({"error": text}))) + } + } + }), + ); + (spawn_server(app).await, calls) + } + + fn route(provider_id: &str, model: &str) -> TranscriptionRoute { + TranscriptionRoute { + provider_id: provider_id.into(), + model: model.into(), + fallback_after_seconds: None, + } + } async fn spawn_server(app: Router) -> String { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); @@ -412,6 +450,102 @@ mod tests { .is_file() ); } + #[tokio::test] + async fn markdown_uses_provider_and_model_that_actually_succeeded() { + let temp = tempfile::tempdir().unwrap(); + let (fail_url, _) = counted_provider(StatusCode::BAD_GATEWAY, "primary down").await; + let (success_url, success_calls) = + counted_provider(StatusCode::OK, "fallback transcript").await; + let (state, mut workflow) = setup(temp.path(), fail_url).await; + state.providers.write().await.push(Provider { + id: "fallback".into(), + name: "Fallback Provider".into(), + transcription_url: success_url, + model: "fallback-default".into(), + api_key: String::new(), + timeout_seconds: 5, + enabled: true, + }); + workflow.transcription_chain = vec![ + route("provider", "primary-model"), + route("fallback", "fallback-model"), + ]; + let job = make_job(&state, &workflow, "winner.m4a").await; + process_job(&state, &job.id, &CancellationToken::new()) + .await + .unwrap(); + assert_eq!(success_calls.load(Ordering::SeqCst), 1); + let finished = get_job(&state, &job.id).unwrap(); + assert_eq!(finished.used_provider_id.as_deref(), Some("fallback")); + assert_eq!(finished.used_model.as_deref(), Some("fallback-model")); + let markdown = std::fs::read_to_string(finished.markdown_path.unwrap()).unwrap(); + assert!(markdown.contains("provider: \"Fallback Provider\"")); + assert!(markdown.contains("model: \"fallback-model\"")); + assert!(markdown.contains("fallback transcript")); + } + + #[tokio::test] + async fn publication_failure_does_not_contact_another_route() { + let temp = tempfile::tempdir().unwrap(); + let (primary_url, primary_calls) = counted_provider(StatusCode::OK, "publish me").await; + let (fallback_url, fallback_calls) = + counted_provider(StatusCode::OK, "should not run").await; + let (state, mut workflow) = setup(temp.path(), primary_url).await; + state.providers.write().await.push(Provider { + id: "fallback".into(), + name: "Fallback".into(), + transcription_url: fallback_url, + model: "fallback-model".into(), + api_key: String::new(), + timeout_seconds: 5, + enabled: true, + }); + let notes = temp.path().join("notes"); + std::fs::create_dir_all(¬es).unwrap(); + workflow.output_dir = Some(notes.to_string_lossy().into_owned()); + workflow.transcription_chain = vec![ + route("provider", "primary-model"), + route("fallback", "fallback-model"), + ]; + state.workflows.write().await[0] = workflow.clone(); + state + .db + .upsert("workflow", &workflow.id, &workflow) + .unwrap(); + let job = make_job(&state, &workflow, "publish-fail.m4a").await; + std::fs::remove_dir(¬es).unwrap(); + assert!( + process_job(&state, &job.id, &CancellationToken::new()) + .await + .is_err() + ); + assert_eq!(primary_calls.load(Ordering::SeqCst), 1); + assert_eq!(fallback_calls.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn archive_failure_does_not_retranscribe() { + let temp = tempfile::tempdir().unwrap(); + let (primary_url, primary_calls) = counted_provider(StatusCode::OK, "archive me").await; + let (state, mut workflow) = setup(temp.path(), primary_url).await; + workflow.transcription_chain = vec![route("provider", "primary-model")]; + let job = make_job(&state, &workflow, "archive-fail.m4a").await; + std::fs::remove_dir(&workflow.archive_dir).unwrap(); + assert!( + process_job(&state, &job.id, &CancellationToken::new()) + .await + .is_err() + ); + assert_eq!(primary_calls.load(Ordering::SeqCst), 1); + assert_eq!( + get_job(&state, &job.id) + .unwrap() + .transcription_attempts + .len(), + 1 + ); + } + #[tokio::test] async fn retry_reuses_failed_job_and_completes() { let temp = tempfile::tempdir().unwrap(); @@ -427,6 +561,19 @@ mod tests { let finished = wait_for_terminal(&state, &job.id).await; assert_eq!(finished.status, JobStatus::Done); assert_eq!(finished.attempts, 2); + assert_eq!(finished.transcription_attempts.len(), 2); + assert_eq!(finished.transcription_attempts[0].run, 1); + assert_eq!(finished.transcription_attempts[0].route_index, 0); + assert_eq!( + finished.transcription_attempts[0].outcome, + TranscriptionAttemptOutcome::Failed + ); + assert_eq!(finished.transcription_attempts[1].run, 2); + assert_eq!(finished.transcription_attempts[1].route_index, 0); + assert_eq!( + finished.transcription_attempts[1].outcome, + TranscriptionAttemptOutcome::Success + ); assert!( Path::new(&workflow.watch_dir) .join("hello from provider.md") diff --git a/src/pipeline.rs b/src/pipeline.rs index 38fd376..a7a52f9 100644 --- a/src/pipeline.rs +++ b/src/pipeline.rs @@ -2,8 +2,8 @@ use crate::{ domain::{JobKind, JobStatus, Workflow}, jobs::{get_job, update_job}, markdown::{self, NoteContext}, - provider, state::AppState, + transcription_chain, }; use anyhow::{Context, Result, anyhow, bail}; use std::{ @@ -38,18 +38,6 @@ async fn process_workflow_job( .find(|workflow| workflow.id == workflow_id) .cloned() .ok_or_else(|| anyhow!("workflow not found: {workflow_id}"))?; - let provider = state - .providers - .read() - .await - .iter() - .find(|provider| provider.id == job.provider_id) - .cloned() - .ok_or_else(|| anyhow!("provider not found: {}", job.provider_id))?; - if !provider.enabled { - bail!("provider '{}' is disabled", provider.name); - } - let source = state.config.resolve_allowed_file(&job.source_path)?; let watch_dir = state .config @@ -71,20 +59,16 @@ async fn process_workflow_job( return finish_archive(state, &job.id, &workflow, &source, token).await; } - update_job(state, &job.id, |job| { - job.status = JobStatus::Transcribing; - job.error = None; - })?; - let transcript = provider::transcribe( - &provider, - &job.model, - job.language.as_deref(), + let success = transcription_chain::transcribe_job_chain( + state, + &job.id, &source, - &state.http, + job.language.as_deref(), token, ) .await .context("transcription")?; + let transcript = success.transcript; if token.is_cancelled() { bail!("cancelled"); } @@ -99,8 +83,8 @@ async fn process_workflow_job( title: &title, source_name: &job.original_name, workflow_name: Some(&workflow.name), - provider_name: &provider.name, - model: &job.model, + provider_name: &success.provider_name, + model: &success.model, language: job.language.as_deref(), tags: &workflow.tags, frontmatter: workflow.markdown.frontmatter, diff --git a/src/pipeline_quick.rs b/src/pipeline_quick.rs index 54cdb19..f216081 100644 --- a/src/pipeline_quick.rs +++ b/src/pipeline_quick.rs @@ -2,8 +2,8 @@ use crate::{ domain::{Job, JobStatus, QuickJobMeta, QuickOutputKind, QuickSourceKind}, jobs::update_job, markdown::{self, NoteContext}, - provider, state::AppState, + transcription_chain, }; use anyhow::{Context, Result, anyhow, bail}; use std::path::Path; @@ -45,34 +45,19 @@ async fn process_quick_inner( source: &Path, token: &CancellationToken, ) -> Result<()> { - let provider = state - .providers - .read() - .await - .iter() - .find(|provider| provider.id == job.provider_id) - .cloned() - .ok_or_else(|| anyhow!("provider not found: {}", job.provider_id))?; - if !provider.enabled { - bail!("provider '{}' is disabled", provider.name); - } if token.is_cancelled() { bail!("cancelled"); } - update_job(state, &job.id, |job| { - job.status = JobStatus::Transcribing; - job.error = None; - })?; - let transcript = provider::transcribe( - &provider, - &job.model, - job.language.as_deref(), + let success = transcription_chain::transcribe_job_chain( + state, + &job.id, source, - &state.http, + job.language.as_deref(), token, ) .await .context("transcription")?; + let transcript = success.transcript; if token.is_cancelled() { bail!("cancelled"); } @@ -82,8 +67,8 @@ async fn process_quick_inner( title: &title, source_name: &job.original_name, workflow_name: None, - provider_name: &provider.name, - model: &job.model, + provider_name: &success.provider_name, + model: &success.model, language: job.language.as_deref(), tags: &[], frontmatter: meta.frontmatter, From c7c349f34b92a19f042a03dc8ada3bfd12927f74 Mon Sep 17 00:00:00 2001 From: GodsQuantum <193283917+GodsQuantum@users.noreply.github.com> Date: Wed, 16 Sep 2026 11:54:43 +0200 Subject: [PATCH 4/6] feat: expose transcription chains through the API --- frontend/src/lib/api.ts | 5 +- frontend/src/lib/types.ts | 29 ++++++++- frontend/tests/scribewatch-surface.test.mjs | 11 ++++ src/api/quick.rs | 66 +++++++++++++++++++-- 4 files changed, 103 insertions(+), 8 deletions(-) diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 83d2e73..9968e48 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -25,8 +25,9 @@ async function request(path: string, init?: RequestInit): Promise { } function appendQuickOptions(form: FormData, options: QuickOptions) { - form.append('providerId', options.providerId); - form.append('model', options.model ?? ''); + if (options.providerId) form.append('providerId', options.providerId); + if (options.model) form.append('model', options.model); + if (options.transcriptionChain) form.append('transcriptionChain', JSON.stringify(options.transcriptionChain)); form.append('language', options.language ?? ''); form.append('outputKind', options.outputKind); form.append('outputDir', options.outputDir ?? ''); diff --git a/frontend/src/lib/types.ts b/frontend/src/lib/types.ts index f5d1c0b..6225b29 100644 --- a/frontend/src/lib/types.ts +++ b/frontend/src/lib/types.ts @@ -5,6 +5,26 @@ export type JobKind = 'workflow' | 'quick'; export type QuickSourceKind = 'server' | 'upload'; export type QuickOutputKind = 'server' | 'client'; +export interface TranscriptionRoute { + providerId: string; + model: string; + fallbackAfterSeconds?: number; +} + +export type TranscriptionAttemptOutcome = 'success' | 'failed' | 'timed_out' | 'cancelled'; + +export interface TranscriptionAttempt { + run: number; + routeIndex: number; + providerId: string; + providerName: string; + model: string; + startedAtMs: number; + finishedAtMs: number; + outcome: TranscriptionAttemptOutcome; + error?: string; +} + export interface Provider { id: string; name: string; @@ -38,6 +58,7 @@ export interface Workflow { tags: string[]; providerId: string; model: string; + transcriptionChain?: TranscriptionRoute[]; language?: string; markdown: MarkdownOptions; enabled: boolean; @@ -57,6 +78,11 @@ export interface Job { workflowId?: string; quick?: QuickJobMeta; providerId: string; + transcriptionChain: TranscriptionRoute[]; + transcriptionAttempts: TranscriptionAttempt[]; + usedProviderId?: string; + usedProviderName?: string; + usedModel?: string; originalName: string; sourcePath: string; sourceSize: number; @@ -74,8 +100,9 @@ export interface Job { } export interface QuickOptions { - providerId: string; + providerId?: string; model?: string; + transcriptionChain?: TranscriptionRoute[]; language?: string; outputKind: QuickOutputKind; outputDir?: string; diff --git a/frontend/tests/scribewatch-surface.test.mjs b/frontend/tests/scribewatch-surface.test.mjs index f801fee..7d08c6f 100644 --- a/frontend/tests/scribewatch-surface.test.mjs +++ b/frontend/tests/scribewatch-surface.test.mjs @@ -40,3 +40,14 @@ test('product promise is visible on Home', async () => { const view = await read('lib/views/DashboardView.svelte'); assert.ok(view.includes('Turn voice notes into Markdown before you forget them.')); }); + + +test('quick transport exposes transcription chains', async () => { + const api = await read('lib/api.ts'); + const types = await read('lib/types.ts'); + assert.ok(api.includes("form.append('transcriptionChain'")); + assert.ok(types.includes('interface TranscriptionRoute')); + assert.ok(types.includes('transcriptionChain')); + assert.ok(types.includes('transcriptionAttempts')); + assert.ok(types.includes('usedProviderId')); +}); diff --git a/src/api/quick.rs b/src/api/quick.rs index e695ba7..29ccdae 100644 --- a/src/api/quick.rs +++ b/src/api/quick.rs @@ -1,5 +1,5 @@ use crate::{ - domain::{Job, QuickOutputKind}, + domain::{Job, QuickOutputKind, TranscriptionRoute}, error::{AppError, AppResult}, jobs, quick::{self, QuickOptions}, @@ -44,6 +44,27 @@ fn text_bool(value: Option, default: bool) -> Result { } } +fn parse_transcription_chain(value: Option) -> Result, AppError> { + let Some(value) = value else { + return Ok(Vec::new()); + }; + let value = value.trim(); + if value.is_empty() { + return Ok(Vec::new()); + } + let routes: Vec = serde_json::from_str(value) + .map_err(|error| AppError::BadRequest(format!("invalid transcriptionChain: {error}")))?; + if routes + .iter() + .any(|route| route.fallback_after_seconds == Some(0)) + { + return Err(AppError::BadRequest( + "fallbackAfterSeconds must be greater than zero".into(), + )); + } + Ok(routes) +} + fn parse_output_kind(value: Option) -> Result { match value.as_deref().map(str::trim) { Some("server") => Ok(QuickOutputKind::Server), @@ -67,6 +88,7 @@ pub async fn upload( let mut original_name: Option = None; let mut provider_id = None; let mut model = None; + let mut transcription_chain = None; let mut language = None; let mut output_kind = None; let mut output_dir = None; @@ -159,6 +181,7 @@ pub async fn upload( match name.as_str() { "providerId" => provider_id = Some(value), "model" => model = Some(value), + "transcriptionChain" => transcription_chain = Some(value), "language" => language = Some(value), "outputKind" => output_kind = Some(value), "outputDir" => output_dir = Some(value), @@ -171,12 +194,15 @@ pub async fn upload( staged.ok_or_else(|| AppError::BadRequest("audio file is required".into()))?; let original_name = original_name.unwrap_or_else(|| "audio".into()); let options_result: AppResult = (|| { + let transcription_chain = parse_transcription_chain(transcription_chain)?; + let provider_id = provider_id.unwrap_or_default(); + if transcription_chain.is_empty() && provider_id.trim().is_empty() { + return Err(AppError::BadRequest("providerId is required".into())); + } Ok(QuickOptions { - provider_id: provider_id - .filter(|value| !value.trim().is_empty()) - .ok_or_else(|| AppError::BadRequest("providerId is required".into()))?, + provider_id, model: model.unwrap_or_default(), - transcription_chain: Vec::new(), + transcription_chain, language, output_kind: parse_output_kind(output_kind)?, output_dir, @@ -231,4 +257,34 @@ mod tests { assert!(text_bool(Some("1".into()), false).unwrap()); assert!(text_bool(Some("sometimes".into()), true).is_err()); } + + #[test] + fn transcription_chain_parser_accepts_ordered_routes() { + let chain = parse_transcription_chain(Some( + r#"[ + {"providerId":"p","model":"large-v3"}, + {"providerId":"p","model":"distil-large-v3","fallbackAfterSeconds":1800} + ]"# + .into(), + )) + .unwrap(); + assert_eq!(chain.len(), 2); + assert_eq!(chain[0].provider_id, "p"); + assert_eq!(chain[0].model, "large-v3"); + assert_eq!(chain[0].fallback_after_seconds, None); + assert_eq!(chain[1].model, "distil-large-v3"); + assert_eq!(chain[1].fallback_after_seconds, Some(1800)); + } + + #[test] + fn transcription_chain_parser_rejects_bad_json_and_zero_timeout() { + assert!(parse_transcription_chain(Some("not-json".into())).is_err()); + assert!( + parse_transcription_chain(Some( + r#"[{"providerId":"p","model":"m","fallbackAfterSeconds":0}]"#.into() + )) + .is_err() + ); + assert!(parse_transcription_chain(None).unwrap().is_empty()); + } } From b0b7703fb86c1969730df2470a16c0849cd32e2c Mon Sep 17 00:00:00 2001 From: GodsQuantum <193283917+GodsQuantum@users.noreply.github.com> Date: Wed, 16 Sep 2026 12:01:28 +0200 Subject: [PATCH 5/6] feat: add fallback chain editor --- frontend/src/app.css | 5 + .../TranscriptionChainEditor.svelte | 127 ++++++++++++++++++ frontend/src/lib/transcription-chain.js | 30 +++++ frontend/src/lib/views/JobsView.svelte | 24 +++- frontend/src/lib/views/ProvidersView.svelte | 2 +- .../src/lib/views/QuickTranscribeView.svelte | 29 ++-- frontend/src/lib/views/WorkflowsView.svelte | 29 ++-- frontend/tests/scribewatch-surface.test.mjs | 17 +++ frontend/tests/transcription-chain.test.mjs | 32 +++++ 9 files changed, 269 insertions(+), 26 deletions(-) create mode 100644 frontend/src/lib/components/TranscriptionChainEditor.svelte create mode 100644 frontend/src/lib/transcription-chain.js create mode 100644 frontend/tests/transcription-chain.test.mjs diff --git a/frontend/src/app.css b/frontend/src/app.css index b474b5b..bd92bd0 100644 --- a/frontend/src/app.css +++ b/frontend/src/app.css @@ -47,3 +47,8 @@ button,input,select,textarea{font:inherit} button{color:inherit} a{color:inherit @media(max-width:1080px){.home-hero{grid-template-columns:1fr}.hero-visual{display:none}.quick-layout{grid-template-columns:1fr}.result-card{position:static;min-height:300px}.result-body{min-height:230px}.home-kpis{grid-template-columns:repeat(2,minmax(0,1fr))}} @media(max-width:680px){.home-hero{padding:27px 20px;border-radius:18px}.home-hero h1{font-size:39px}.hero-actions{flex-direction:column}.hero-actions .btn{width:100%}.hero-proof{gap:10px;flex-direction:column}.quick-card .card-body{padding:17px}.source-tabs{width:100%}.bottom-nav{grid-template-columns:repeat(5,1fr)}.bottom-nav button{font-size:9px}.folder-step{padding-left:45px}.step-no{left:11px}.home-kpis{grid-template-columns:repeat(2,minmax(0,1fr))}} @media(prefers-reduced-motion:reduce){*,*:before,*:after{scroll-behavior:auto!important;animation-duration:.001ms!important;animation-iteration-count:1!important;transition-duration:.001ms!important}.drop-zone:hover{transform:none}} + +/* resilient transcription chains */ +.chain-editor{display:flex;flex-direction:column;gap:10px;padding:14px;border:1px solid var(--line);border-radius:14px;background:#0c1513}.chain-head{display:flex;justify-content:space-between;gap:12px}.chain-head strong{font-size:13px}.chain-head p{margin:3px 0 0}.chain-route{display:grid;grid-template-columns:100px minmax(150px,.85fr) minmax(180px,1.15fr) minmax(130px,.55fr) auto;gap:10px;align-items:end;padding:12px;border:1px solid var(--line);border-radius:12px;background:#0e1816}.chain-route-label{align-self:center;display:flex;flex-direction:column;gap:3px}.chain-route-label strong{font-size:11px;color:var(--accent)}.chain-route-label span{font-size:9px;color:var(--muted-2)}.chain-actions{display:flex;gap:5px;align-items:end;padding-bottom:1px}.icon-btn{width:38px;min-width:38px;padding:0}.chain-add{align-self:flex-start}.chain-timeout .timeout-custom{display:grid;grid-template-columns:minmax(0,1fr) auto;gap:6px;align-items:center}.timeout-custom span{font-size:10px;color:var(--muted)}.error-inline{color:var(--danger)}.link-button{padding:0;border:0;background:transparent;color:var(--accent);font-weight:800;cursor:pointer}.attempt-details{margin:10px 0;border:1px solid var(--line);border-radius:10px;background:#0b1412}.attempt-details summary{cursor:pointer;padding:9px 11px;color:var(--muted);font-size:11px;font-weight:800}.attempt-list{display:flex;flex-direction:column;border-top:1px solid var(--line)}.attempt-row{display:grid;grid-template-columns:90px minmax(0,1fr);gap:7px 10px;padding:9px 11px;border-bottom:1px solid rgba(255,255,255,.04);font-size:10px;color:var(--muted)}.attempt-row:last-child{border-bottom:0}.attempt-outcome{font-weight:900;text-transform:uppercase;font-size:9px;letter-spacing:.05em}.attempt-outcome.success{color:var(--success)}.attempt-outcome.failed,.attempt-outcome.timed_out{color:var(--danger)}.attempt-outcome.cancelled{color:var(--warning)}.attempt-error{grid-column:2;color:var(--danger);overflow-wrap:anywhere} +@media(max-width:1080px){.chain-route{grid-template-columns:90px repeat(2,minmax(0,1fr));}.chain-timeout{grid-column:2}.chain-actions{grid-column:3;justify-content:flex-end}} +@media(max-width:680px){.chain-route{grid-template-columns:1fr}.chain-route-label,.chain-timeout,.chain-actions{grid-column:auto}.chain-actions{justify-content:flex-start}.chain-route-label{flex-direction:row;align-items:center;justify-content:space-between}.attempt-row{grid-template-columns:1fr}.attempt-error{grid-column:1}} diff --git a/frontend/src/lib/components/TranscriptionChainEditor.svelte b/frontend/src/lib/components/TranscriptionChainEditor.svelte new file mode 100644 index 0000000..faa2a57 --- /dev/null +++ b/frontend/src/lib/components/TranscriptionChainEditor.svelte @@ -0,0 +1,127 @@ + + +
+
+
Transcription chain

Choose the exact provider + model order. If one route fails, ScribeWatch tries the next.

+
+ {#each value as route,index (index)} +
+
{index===0?'Primary':`Fallback ${index}`}#{index+1}
+
+ + +
+
+ + + {#if errorsByProvider[route.providerId]} + Model discovery unavailable. + {/if} +
+
+ + + {#if timeoutChoice(route)==='custom'} +
customTimeoutChanged(index,event)} />min
+ {/if} +
+
+ + + +
+
+ {/each} + +
diff --git a/frontend/src/lib/transcription-chain.js b/frontend/src/lib/transcription-chain.js new file mode 100644 index 0000000..24beb2b --- /dev/null +++ b/frontend/src/lib/transcription-chain.js @@ -0,0 +1,30 @@ +// @ts-check +/** @typedef {import('./types').TranscriptionRoute} TranscriptionRoute */ + +/** @param {TranscriptionRoute[]} routes @returns {TranscriptionRoute[]} */ +export function addRoute(routes) { + return [...routes, { providerId: '', model: '' }]; +} + +/** @param {TranscriptionRoute[]} routes @param {number} index @returns {TranscriptionRoute[]} */ +export function removeRoute(routes, index) { + if (routes.length <= 1 || index < 0 || index >= routes.length) return [...routes]; + return routes.filter((_, current) => current !== index); +} + +/** @param {TranscriptionRoute[]} routes @param {number} from @param {number} to @returns {TranscriptionRoute[]} */ +export function moveRoute(routes, from, to) { + if (from < 0 || from >= routes.length || to < 0 || to >= routes.length || from === to) { + return [...routes]; + } + const next = [...routes]; + const [route] = next.splice(from, 1); + next.splice(to, 0, route); + return next; +} + +/** @param {TranscriptionRoute[]} routes @param {number} index @param {Partial} patch @returns {TranscriptionRoute[]} */ +export function updateRoute(routes, index, patch) { + if (index < 0 || index >= routes.length) return [...routes]; + return routes.map((route, current) => current === index ? { ...route, ...patch } : route); +} diff --git a/frontend/src/lib/views/JobsView.svelte b/frontend/src/lib/views/JobsView.svelte index 3339304..1c7ee40 100644 --- a/frontend/src/lib/views/JobsView.svelte +++ b/frontend/src/lib/views/JobsView.svelte @@ -1,7 +1,7 @@
-

HISTORY

Jobs

Every automatic and one-off transcription keeps a durable lifecycle, including retries and restart recovery.

+

HISTORY

Jobs

Every automatic and one-off transcription keeps a durable lifecycle, including retries, fallback attempts and restart recovery.

{#if visible.length===0}
No jobs in this view.
{/if} {#each visible as job} @@ -35,8 +39,22 @@ {#if job.archivePath}
Archive{job.archivePath}
{/if}
{#if job.error}
Processing error{job.error}
{/if} + {#if (job.transcriptionAttempts??[]).length>0} +
+ Transcription attempts ({job.transcriptionAttempts.length}) +
+ {#each job.transcriptionAttempts as attempt} +
+ {attempt.outcome.replace('_',' ')} + Run {attempt.run} · {attempt.providerName} / {attempt.model} · {elapsed(attempt)} + {#if attempt.error}{attempt.error}{/if} +
+ {/each} +
+
+ {/if}
- Attempt {job.attempts} · model {job.model}{job.language?` · ${job.language}`:' · auto language'} + {job.usedProviderName&&job.usedModel?`Run ${job.attempts} · used ${job.usedProviderName} / ${job.usedModel}`:`Run ${job.attempts} · model ${job.model}`}{job.language?` · ${job.language}`:' · auto language'}
{#if job.kind==='quick'&&job.status==='done'&&job.quick?.outputKind==='client'}{/if} {#if job.status==='error'||job.status==='interrupted'||job.status==='cancelled'}{/if} diff --git a/frontend/src/lib/views/ProvidersView.svelte b/frontend/src/lib/views/ProvidersView.svelte index 7059c87..d1988ee 100644 --- a/frontend/src/lib/views/ProvidersView.svelte +++ b/frontend/src/lib/views/ProvidersView.svelte @@ -57,7 +57,7 @@
Use the complete OpenAI-compatible transcription endpoint. Nothing is hardcoded to a specific service.
The stored value is never returned to this browser.
-
+
This only bounds model discovery. Transcription has no provider-wide processing timeout; optional fallback timeouts are configured per route.
{#if models.length>0}
Models reported by provider
{#each models as model}{/each}
{/if} diff --git a/frontend/src/lib/views/QuickTranscribeView.svelte b/frontend/src/lib/views/QuickTranscribeView.svelte index 63b13fd..0213396 100644 --- a/frontend/src/lib/views/QuickTranscribeView.svelte +++ b/frontend/src/lib/views/QuickTranscribeView.svelte @@ -2,7 +2,8 @@ import { api } from '$lib/api'; import { canSaveToDirectory, saveMarkdownLocally } from '$lib/local-save'; import PathPicker from '$lib/components/PathPicker.svelte'; - import type { Job, Provider, QuickOptions } from '$lib/types'; + import TranscriptionChainEditor from '$lib/components/TranscriptionChainEditor.svelte'; + import type { Job, Provider, QuickOptions, TranscriptionRoute } from '$lib/types'; export let providers:Provider[]=[]; export let jobs:Job[]=[]; export let refresh:()=>Promise=async()=>{}; @@ -13,8 +14,8 @@ let localFile:File|null=null; let serverFile=''; let outputDir=''; - let providerId=''; - let model=''; + let transcriptionChain:TranscriptionRoute[]=[{providerId:'',model:''}]; + let chainInitialized=false; let language=''; let frontmatter=true; let advanced=false; @@ -23,11 +24,15 @@ let submitted:Job|undefined; let drag=false; - $: if(!providerId && providers.length) providerId=providers.find((p)=>p.enabled)?.id??providers[0].id; + $: if(!chainInitialized&&providers.length){ + transcriptionChain=[{providerId:providers.find((provider)=>provider.enabled)?.id??providers[0].id,model:''}]; + chainInitialized=true; + } $: current=submitted ? (jobs.find((job)=>job.id===submitted?.id)??submitted) : undefined; $: sourceReady=sourceMode==='computer' ? !!localFile : !!serverFile; $: outputReady=outputMode==='computer' || !!outputDir; - $: canSubmit=sourceReady&&outputReady&&!!providerId&&!busy; + $: chainReady=transcriptionChain.length>0&&transcriptionChain.every((route)=>!!route.providerId&&!!route.model); + $: canSubmit=sourceReady&&outputReady&&chainReady&&!busy; $: localDirectoryAvailable=typeof window!=='undefined'&&canSaveToDirectory(window); function chooseFile(file:File|undefined){ if(file){localFile=file;submitted=undefined;} } @@ -35,8 +40,10 @@ async function submit(){ if(!canSubmit)return; busy=true; submitted=undefined; + const primary=transcriptionChain[0]; const options:QuickOptions={ - providerId, model:model.trim()||undefined, language:language.trim()||undefined, + providerId:primary?.providerId, model:primary?.model, transcriptionChain, + language:language.trim()||undefined, outputKind:outputMode==='server'?'server':'client', outputDir:outputMode==='server'?outputDir:undefined, frontmatter }; @@ -82,13 +89,11 @@
Only explicitly mounted and allowed server roots are visible.
{/if} -
02
Transcription

Use any configured OpenAI-compatible transcription provider.

-
-
-
-
+
02
Transcription

Pick an exact provider + model chain. Failed routes automatically fall through.

+ transcriptionChain=routes} /> +
- {#if advanced}
{/if} + {#if advanced}
{/if}
03
Save the note

Publish on the server or bring the Markdown back to this computer.

diff --git a/frontend/src/lib/views/WorkflowsView.svelte b/frontend/src/lib/views/WorkflowsView.svelte index d479a5a..2dc794a 100644 --- a/frontend/src/lib/views/WorkflowsView.svelte +++ b/frontend/src/lib/views/WorkflowsView.svelte @@ -2,15 +2,17 @@ import { api } from '$lib/api'; import PathPicker from '$lib/components/PathPicker.svelte'; import TagEditor from '$lib/components/TagEditor.svelte'; - import type { Provider, Workflow } from '$lib/types'; + import TranscriptionChainEditor from '$lib/components/TranscriptionChainEditor.svelte'; + import type { Provider, TranscriptionRoute, Workflow } from '$lib/types'; export let workflows:Workflow[]=[]; export let providers:Provider[]=[]; export let refresh:()=>Promise=async()=>{}; export let notify:(type:'success'|'error',message:string)=>void=()=>{}; + const firstProvider=()=>providers.find((provider)=>provider.enabled)?.id??providers[0]?.id??''; const blank=():Workflow=>({ id:'', name:'', watchDir:'', outputDir:undefined, archiveDir:'', tags:[], - providerId:providers[0]?.id||'', model:'', language:undefined, + providerId:'', model:'', transcriptionChain:[{providerId:firstProvider(),model:''}], language:undefined, markdown:{frontmatter:true,transcriptHeading:'Transcript'}, enabled:true }); let selectedId=''; @@ -18,8 +20,14 @@ let busy=false; let picker:''|'watch'|'output'|'archive'=''; + function chainFor(workflow:Workflow):TranscriptionRoute[] { + if(workflow.transcriptionChain?.length) return structuredClone(workflow.transcriptionChain); + const providerId=workflow.providerId??''; + const fallbackModel=providers.find((provider)=>provider.id===providerId)?.model??''; + return [{providerId,model:(workflow.model||fallbackModel).trim()}]; + } function createNew(){ selectedId=''; draft=blank(); picker=''; } - function select(workflow:Workflow){ selectedId=workflow.id; draft=structuredClone({...workflow,tags:workflow.tags??[]}); picker=''; } + function select(workflow:Workflow){ selectedId=workflow.id; draft=structuredClone({...workflow,tags:workflow.tags??[],transcriptionChain:chainFor(workflow)}); picker=''; } function choosePath(path:string){ if(picker==='watch') draft.watchDir=path; if(picker==='output') draft.outputDir=path; @@ -27,9 +35,12 @@ picker=''; } async function save(){ + const chain=draft.transcriptionChain??[]; + if(!chain.length||chain.some((route)=>!route.providerId||!route.model)){notify('error','Choose a provider and model for every transcription route.');return;} busy=true; try { - const payload={...draft,outputDir:draft.outputDir?.trim()||undefined,tags:draft.tags??[]}; + const primary=chain[0]; + const payload={...draft,providerId:primary.providerId,model:primary.model,transcriptionChain:chain,outputDir:draft.outputDir?.trim()||undefined,tags:draft.tags??[]}; const saved=await api.saveWorkflow(payload); await refresh(); select(saved); notify('success','Workflow saved.'); } catch(e){ notify('error',e instanceof Error?e.message:String(e)); } finally{ busy=false; } @@ -40,6 +51,7 @@ try{await api.deleteWorkflow(selectedId);createNew();await refresh();notify('success','Workflow deleted.');} catch(e){notify('error',e instanceof Error?e.message:String(e));} } + $: chainValid=(draft.transcriptionChain??[]).length>0&&(draft.transcriptionChain??[]).every((route)=>!!route.providerId&&!!route.model); $: pickerTitle=picker==='watch'?'Choose Watch folder':picker==='output'?'Choose Markdown folder':'Choose Audio archive'; $: pickerInitial=picker==='watch'?draft.watchDir:picker==='output'?(draft.outputDir??draft.watchDir):draft.archiveDir; @@ -64,8 +76,9 @@
-
+
Empty or “auto” lets the provider detect it.
+ draft={...draft,transcriptionChain:routes}} />
1
Stable audio files detected here are queued automatically.
↓
@@ -73,10 +86,6 @@
↓
3
The source moves here only after the Markdown note is safely published.
-
-
Leave empty to inherit the provider model.
-
Empty or “auto” lets the provider detect it.
-
Obsidian / Markdown

Every note gets a transcript-derived title. YAML properties stay optional.

@@ -85,7 +94,7 @@
{#if selectedId}{/if} - +
diff --git a/frontend/tests/scribewatch-surface.test.mjs b/frontend/tests/scribewatch-surface.test.mjs index 7d08c6f..344e243 100644 --- a/frontend/tests/scribewatch-surface.test.mjs +++ b/frontend/tests/scribewatch-surface.test.mjs @@ -51,3 +51,20 @@ test('quick transport exposes transcription chains', async () => { assert.ok(types.includes('transcriptionAttempts')); assert.ok(types.includes('usedProviderId')); }); + + +test('workflow and quick share the fallback chain editor with live model discovery', async () => { + const editor = await read('lib/components/TranscriptionChainEditor.svelte'); + const workflows = await read('lib/views/WorkflowsView.svelte'); + const quick = await read('lib/views/QuickTranscribeView.svelte'); + const providers = await read('lib/views/ProvidersView.svelte'); + assert.ok(editor.includes('api.models')); + assert.ok(editor.includes('+ Add fallback')); + assert.ok(editor.includes('Primary')); + assert.ok(editor.includes('Fallback')); + assert.ok(workflows.includes('TranscriptionChainEditor')); + assert.ok(quick.includes('TranscriptionChainEditor')); + assert.ok(!workflows.includes('Model override')); + assert.ok(!quick.includes('Model override')); + assert.ok(providers.includes('Model discovery timeout')); +}); diff --git a/frontend/tests/transcription-chain.test.mjs b/frontend/tests/transcription-chain.test.mjs new file mode 100644 index 0000000..83fda92 --- /dev/null +++ b/frontend/tests/transcription-chain.test.mjs @@ -0,0 +1,32 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { addRoute, removeRoute, moveRoute, updateRoute } from '../src/lib/transcription-chain.js'; + +test('fallback routes preserve complete tuples while reordering', () => { + const a = { providerId: 'p', model: 'large' }; + const b = { providerId: 'p', model: 'distil', fallbackAfterSeconds: 1800 }; + const source = [a, b]; + const moved = moveRoute(source, 1, 0); + assert.deepEqual(moved, [b, a]); + assert.deepEqual(source, [a, b]); + assert.notEqual(moved, source); +}); + +test('add and remove are immutable and never leave an empty chain', () => { + const source = [{ providerId: 'p', model: 'large' }]; + const added = addRoute(source); + assert.deepEqual(added, [source[0], { providerId: '', model: '' }]); + assert.deepEqual(source, [{ providerId: 'p', model: 'large' }]); + assert.deepEqual(removeRoute(source, 0), source); + assert.deepEqual(removeRoute(added, 1), source); +}); + +test('updateRoute changes exactly one tuple without mutating source', () => { + const source = [ + { providerId: 'a', model: 'm1' }, + { providerId: 'b', model: 'm2', fallbackAfterSeconds: 600 }, + ]; + const updated = updateRoute(source, 1, { model: 'm3' }); + assert.deepEqual(updated, [source[0], { providerId: 'b', model: 'm3', fallbackAfterSeconds: 600 }]); + assert.equal(source[1].model, 'm2'); +}); From 5436c3186888324504932b585cf67e77028d3222 Mon Sep 17 00:00:00 2001 From: GodsQuantum <193283917+GodsQuantum@users.noreply.github.com> Date: Wed, 16 Sep 2026 12:08:04 +0200 Subject: [PATCH 6/6] docs: promote resilient transcription chains --- CHANGELOG.md | 11 ++++++ README.fr.md | 19 +++++++++ README.md | 19 +++++++++ scripts/e2e-v02.sh | 48 +++++++++++++++++++++- scripts/mock-transcription-server.py | 59 ++++++++++++++++++++++------ 5 files changed, 142 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 42d6ac6..ee8ada0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,17 @@ All notable ScribeWatch changes are documented here. +## [Unreleased] + +### Added +- Ordered provider + model fallback chains for Workflows and Quick Transcribe, including multiple models from the same provider. +- Live model discovery per provider, per-route optional fallback timeouts, and durable per-route attempt history. +- Jobs now record the provider/model that actually succeeded and expose fallback attempts in the UI. + +### Changed +- Transcription processing no longer inherits a provider-wide fixed request timeout; long audio waits by default unless a route explicitly defines a fallback timeout. +- Legacy `providerId + model` workflow and Quick Transcribe payloads remain accepted during the v0.2.x migration window. + ## [0.2.0] - 2026-09-16 ### Added diff --git a/README.fr.md b/README.fr.md index 03a378d..805b824 100644 --- a/README.fr.md +++ b/README.fr.md @@ -37,6 +37,7 @@ Le cas d’usage naturel est **notes vocales → Obsidian**, mais rien n’est v - **Watch folders** — automatisez les nouveaux enregistrements d’un serveur, NAS ou dossier synchronisé. - **Prêt pour Obsidian** — noms propres, titre H1, propriétés YAML et tags. - **Votre moteur de transcription** — endpoint STT compatible OpenAI, notamment Speaches/Whisper. +- **Chaînes de fallback résilientes** — mélangez providers et modèles dans l’ordre voulu. Si une route échoue, ScribeWatch essaie automatiquement la suivante — y compris un autre modèle du même provider. - **Aucun LLM obligatoire** — titres et formatage sont déterministes ; la transcription reste la source de vérité. - **Publication sûre** — le Markdown est publié avant l’archivage de l’audio, sans écraser une note existante. - **Petite stack auto-hébergée** — backend Rust/Axum, UI SvelteKit, SQLite, un conteneur. @@ -56,6 +57,22 @@ Le cas d’usage naturel est **notes vocales → Obsidian**, mais rien n’est v Dossiers d’un workflow ScribeWatch

+## 🔁 Fallbacks provider + modèle + +Chaque route de transcription est un couple explicite **provider + modèle**. ScribeWatch aspire la liste des modèles exposés par chaque provider configuré : vous choisissez donc exactement la chaîne voulue, y compris plusieurs modèles d’un même provider. + +```text +Speaches / whisper-large-v3 + ↓ échec +Speaches / distil-whisper-large-v3 + ↓ échec +OpenAI / gpt-4o-transcribe + ↓ +Markdown +``` + +Il n’y a **aucun timeout de traitement par défaut**, ce qui évite de pénaliser les longs enregistrements. Chaque route peut cependant définir son propre délai de fallback (`Jamais`, 10/30/60 minutes ou personnalisé). Une panne réseau, erreur provider, rate limit ou réponse invalide peut passer à la route suivante ; une annulation utilisateur stoppe toute la chaîne. L’historique conserve chaque tentative et le provider/modèle réellement utilisé. + ## 🚀 Installation rapide Prérequis : Docker Engine + Compose et un endpoint de transcription compatible OpenAI. @@ -92,6 +109,8 @@ tags: - transcription - scribewatch source: "Enregistrement 42.m4a" +provider: "Speaches local" +model: "whisper-large-v3" language: "fr" --- diff --git a/README.md b/README.md index 18c3392..8ee5cc7 100644 --- a/README.md +++ b/README.md @@ -37,6 +37,7 @@ It is especially useful as a **voice notes → Obsidian** bridge, but nothing is - **Watch folders** — automatically process new recordings from a server, NAS or mounted sync folder. - **Obsidian-friendly by default** — clean filenames, H1 titles, YAML properties and tags. - **Bring your own transcription engine** — works with OpenAI-compatible STT endpoints such as Speaches/Whisper services. +- **Resilient transcription chains** — mix providers and models in any order. If one route fails, ScribeWatch automatically tries the next — even another model from the same provider. - **No LLM required** — titles and Markdown formatting are deterministic; the transcript stays authoritative. - **Safe publication** — Markdown is written before workflow audio is archived, and existing notes are never silently overwritten. - **Small self-hosted stack** — Rust/Axum backend, SvelteKit UI, SQLite state, one container. @@ -56,6 +57,22 @@ It is especially useful as a **voice notes → Obsidian** bridge, but nothing is ScribeWatch workflow folders

+## 🔁 Provider + model fallbacks + +Every transcription route is an explicit **provider + model** pair. ScribeWatch discovers the models exposed by each configured provider, so you choose the exact chain — including multiple models from the same provider. + +```text +Speaches / whisper-large-v3 + ↓ failed +Speaches / distil-whisper-large-v3 + ↓ failed +OpenAI / gpt-4o-transcribe + ↓ +Markdown +``` + +There is **no processing timeout by default**, which matters for long recordings. Each route can optionally define its own fallback timeout (`Never`, 10/30/60 minutes or custom). Connection failures, provider errors, rate limits and invalid responses can fall through; explicit user cancellation stops the whole chain. Jobs keep the attempt history and record the provider/model that actually succeeded. + ## 🚀 Quick start Requirements: Docker Engine + Compose and an OpenAI-compatible transcription endpoint. @@ -92,6 +109,8 @@ tags: - transcription - scribewatch source: "Recording 42.m4a" +provider: "Local Speaches" +model: "whisper-large-v3" language: "en" --- diff --git a/scripts/e2e-v02.sh b/scripts/e2e-v02.sh index b46b0cb..0c4fc06 100755 --- a/scripts/e2e-v02.sh +++ b/scripts/e2e-v02.sh @@ -12,7 +12,8 @@ cleanup(){ rm -rf "$TMP" } trap cleanup EXIT -mkdir -p "$TMP/config" "$TMP/data" "$TMP/media/watch" "$TMP/media/notes" "$TMP/media/archive" "$TMP/media/quick" +mkdir -p "$TMP/config" "$TMP/data" "$TMP/media/watch" "$TMP/media/notes" "$TMP/media/archive" "$TMP/media/quick" \ + "$TMP/media/fallback-watch" "$TMP/media/fallback-notes" "$TMP/media/fallback-archive" PORT="$MOCK_PORT" python3 "$ROOT/scripts/mock-transcription-server.py" >"$TMP/mock.log" 2>&1 & MOCK_PID=$! ( @@ -62,6 +63,51 @@ grep -q '^title: "Remember to book the train tomorrow\."' "$workflow_note" grep -q '^ - ideas$' "$workflow_note" grep -q '^# Remember to book the train tomorrow\.$' "$workflow_note" +# Fallback chain: same provider, first model fails, second model succeeds. +fallback_workflow_json=$(python3 - "$TMP" <<'PY' +import json,sys +r=sys.argv[1] +print(json.dumps({ + "id":"fallback-workflow","name":"Fallback proof", + "watchDir":r+"/media/fallback-watch","outputDir":r+"/media/fallback-notes","archiveDir":r+"/media/fallback-archive", + "tags":["fallback"],"providerId":"provider","model":"fail-model", + "transcriptionChain":[ + {"providerId":"provider","model":"fail-model"}, + {"providerId":"provider","model":"fallback-model"} + ], + "language":"en","markdown":{"frontmatter":True,"transcriptHeading":"Transcript"},"enabled":True +})) +PY +) +curl -fsS -X POST "$BASE/workflows" -H 'content-type: application/json' -d "$fallback_workflow_json" >/dev/null +printf 'fallback workflow audio' >"$TMP/media/fallback-watch/fallback.m4a" +curl -fsS -X PUT "$BASE/workflows/fallback-workflow/scan" >/dev/null +fallback_id="" +for _ in $(seq 1 80); do + fallback_id="$(curl -fsS "$BASE/jobs" | python3 -c 'import json,sys; x=json.load(sys.stdin); print(next((j["id"] for j in x if j.get("workflowId")=="fallback-workflow"), ""))')" + [[ -n "$fallback_id" ]] && break + sleep .1 +done +[[ -n "$fallback_id" ]] +wait_job "$fallback_id" +fallback_job="$(curl -fsS "$BASE/jobs/$fallback_id")" +printf '%s' "$fallback_job" | python3 -c ' +import json,sys +j=json.load(sys.stdin); a=j["transcriptionAttempts"] +assert len(a)==2, a +assert (a[0]["model"],a[0]["outcome"])==("fail-model","failed"), a +assert (a[1]["model"],a[1]["outcome"])==("fallback-model","success"), a +assert j["usedProviderId"]=="provider", j +assert j["usedModel"]=="fallback-model", j +' +[[ ! -e "$TMP/media/fallback-watch/fallback.m4a" ]] +[[ -f "$TMP/media/fallback-archive/fallback.m4a" ]] +fallback_note="$(find "$TMP/media/fallback-notes" -maxdepth 1 -name '*.md' -print -quit)" +[[ -n "$fallback_note" && -f "$fallback_note" ]] +grep -q '^provider: "Mock Provider"$' "$fallback_note" +grep -q '^model: "fallback-model"$' "$fallback_note" +grep -q 'Fallback route worked\.' "$fallback_note" + # Quick server file: publish another note, preserve source. printf 'server quick audio' >"$TMP/media/quick/server.m4a" server_payload=$(python3 - "$TMP" <<'PY' diff --git a/scripts/mock-transcription-server.py b/scripts/mock-transcription-server.py index a98923f..8947252 100755 --- a/scripts/mock-transcription-server.py +++ b/scripts/mock-transcription-server.py @@ -1,13 +1,17 @@ #!/usr/bin/env python3 import json import os +import re +import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer TEXT = os.environ.get("SCRIBEWATCH_MOCK_TEXT", "Remember to book the train tomorrow.") +MODELS = ["test-model", "fail-model", "slow-model", "fallback-model"] class Handler(BaseHTTPRequestHandler): - def _drain_body(self): + def _read_body(self): if self.headers.get("Transfer-Encoding", "").lower() == "chunked": + chunks = [] while True: size_line = self.rfile.readline().strip().split(b";", 1)[0] if not size_line: @@ -16,25 +20,54 @@ def _drain_body(self): if size == 0: self.rfile.readline() break - self.rfile.read(size) + chunks.append(self.rfile.read(size)) self.rfile.read(2) - else: - length = int(self.headers.get("Content-Length", "0")) - if length: - self.rfile.read(length) + return b"".join(chunks) + length = int(self.headers.get("Content-Length", "0")) + return self.rfile.read(length) if length else b"" - def do_POST(self): - if self.path != "/v1/audio/transcriptions": - self.send_error(404) - return - self._drain_body() - body = json.dumps({"text": TEXT}).encode() - self.send_response(200) + def _multipart_field(self, body, name): + content_type = self.headers.get("Content-Type", "") + match = re.search(r"boundary=([^;]+)", content_type) + if not match: + return "" + boundary = match.group(1).strip().strip('"').encode() + marker = b'--' + boundary + needle = f'name="{name}"'.encode() + for part in body.split(marker): + headers, separator, payload = part.partition(b"\r\n\r\n") + if separator and needle in headers: + return payload.rstrip(b"\r\n-").decode("utf-8", "replace").strip() + return "" + + def _json(self, status, payload): + body = json.dumps(payload).encode() + self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) + def do_GET(self): + if self.path == "/v1/models": + self._json(200, {"data": [{"id": model} for model in MODELS]}) + return + self.send_error(404) + + def do_POST(self): + if self.path != "/v1/audio/transcriptions": + self.send_error(404) + return + body = self._read_body() + model = self._multipart_field(body, "model") + if model == "fail-model": + self._json(503, {"error": "intentional model failure"}) + return + if model == "slow-model": + time.sleep(float(os.environ.get("SCRIBEWATCH_MOCK_SLOW_SECONDS", "2"))) + text = "Fallback route worked." if model == "fallback-model" else TEXT + self._json(200, {"text": text}) + def log_message(self, fmt, *args): pass