diff --git a/crates/arr-subs/Cargo.toml b/crates/arr-subs/Cargo.toml index 9420ea7..19b4313 100644 --- a/crates/arr-subs/Cargo.toml +++ b/crates/arr-subs/Cargo.toml @@ -42,5 +42,9 @@ wiremock.workspace = true name = "openai" required-features = ["translate-openai"] +[[test]] +name = "command" +required-features = ["translate-command"] + [lints] workspace = true diff --git a/crates/arr-subs/src/command.rs b/crates/arr-subs/src/command.rs new file mode 100644 index 0000000..07384d9 --- /dev/null +++ b/crates/arr-subs/src/command.rs @@ -0,0 +1,292 @@ +//! The remote-command translation backend (`DESIGN.md` §15, issue #193). +//! +//! A configured command template — `ssh box claude -p` is one instance, a +//! local CLI or a container exec another — is spawned per batch: the prompt +//! and the cue batch go to its stdin as JSON, the translated batch comes back +//! on stdout in the same `{number, text}` shape every LLM-shaped backend +//! speaks. Nothing is resident; `kill_on_drop` means a command that outlives +//! its call does not either. No assumption about ssh is made here, which is +//! the point: it reaches an agent CLI on another host with no API key in +//! arr's config at all. +//! +//! The template may carry `{source}` and `{target}` placeholders, replaced +//! with the batch's language tags before spawning, so a wrapper script can +//! route on them. Tokens split on whitespace — an ssh-style template has no +//! quoted arguments to worry about. +//! +//! [`translate::translate`] still owns chunking and validation: non-zero +//! exit, timeout and unparseable output fail loudly here, and a reply with +//! the wrong cue count or numbering never becomes a subtitle. + +use std::{io::ErrorKind, process::Stdio, time::Duration}; + +use serde::{Deserialize, Serialize}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + process::Command as Process, + time, +}; + +use arr_core::Language; + +use crate::translate::{ + strip_code_fence, system_prompt, Backend, BackendId, Batch, Error, Result, TranslateFuture, + TranslatedCue, +}; + +const BACKEND_NAME: &str = "command"; + +/// How a language reads in a command argument: a short tag a wrapper script +/// can route on — `pt-PT`, not "European Portuguese (pt-PT)". +fn language_tag(language: &Language) -> &str { + match language { + Language::PortuguesePortugal => "pt-PT", + Language::PortugueseBrazil => "pt-BR", + Language::PortugueseUnverified => "pt", + Language::Other(tag) => tag, + } +} +/// How long one command may run when the configuration does not say. +pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(120); + +/// Settings for the remote-command backend (DESIGN.md §10, issue #198). +/// +/// The template is not a credential, but it names a host — it arrives from +/// bootstrap config like the API keys do, never from the database. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct CommandConfig { + /// The command to run per batch, e.g. `ssh box claude -p`. May carry + /// `{source}` and `{target}` placeholders. + pub template: String, + /// How long one command may run before it is killed and the batch fails. + pub timeout: Duration, +} + +/// The remote-command backend behind the [`Backend`] trait. +#[derive(Clone, Debug)] +pub struct Command { + config: CommandConfig, + id: BackendId, +} + +impl Command { + /// A backend running `config.template` once per batch. + /// + /// # Errors + /// + /// [`Error::Malformed`] when the template splits into no tokens at all. + pub fn new(config: CommandConfig) -> Result { + if config.template.trim().is_empty() { + return Err(Error::Malformed { + backend: BackendId::new(BACKEND_NAME), + detail: "empty command template".to_owned(), + }); + } + Ok(Self { + config, + id: BackendId::new(BACKEND_NAME), + }) + } + + async fn translate_inner(&self, batch: &Batch) -> Result> { + let request_cues: Vec = batch + .cues + .iter() + .map(|cue| WireCue { + number: cue.number, + text: cue.text.clone(), + }) + .collect(); + let batch_json = serde_json::to_string(&request_cues).map_err(|err| Error::Malformed { + backend: self.id.clone(), + detail: format!("could not encode batch: {err}"), + })?; + let payload = format!( + "{}\n{batch_json}\n", + system_prompt(&batch.source, &batch.target) + ); + + let mut tokens = argv(&self.config.template); + for token in &mut tokens { + *token = token + .replace("{source}", language_tag(&batch.source)) + .replace("{target}", language_tag(&batch.target)); + } + let (program, args) = tokens.split_first().expect("the constructor rejects empty"); + + let work = run(program, args, payload, self.id.clone()); + + // Dropping `work` on timeout drops the child, whose `kill_on_drop` + // ends the command — nothing is left resident (DESIGN.md §15). + match time::timeout(self.config.timeout, work).await { + Err(_) => Err(Error::Transport { + backend: self.id.clone(), + source: format!("timed out after {:?}", self.config.timeout).into(), + }), + Ok(Err(err)) => Err(err), + Ok(Ok(out)) => Self::parse_reply(&self.id, &out), + } + } + + fn parse_reply(id: &BackendId, out: &[u8]) -> Result> { + let malformed = |detail: String| Error::Malformed { + backend: id.clone(), + detail, + }; + + let text = std::str::from_utf8(out) + .map_err(|err| malformed(format!("output was not UTF-8: {err}")))?; + if text.trim().is_empty() { + return Err(malformed("produced no output".to_owned())); + } + + let cues: Vec = serde_json::from_str(strip_code_fence(text)) + .map_err(|err| malformed(format!("reply was not the expected JSON: {err}")))?; + + Ok(cues + .into_iter() + .map(|cue| TranslatedCue { + number: cue.number, + text: cue.text, + }) + .collect()) + } +} + +/// Split a template into program and arguments on whitespace. +fn argv(template: &str) -> Vec { + template.split_whitespace().map(str::to_owned).collect() +} + +/// The batch shape on both wires: out on stdin, back on stdout. +#[derive(Serialize, Deserialize)] +struct WireCue { + number: usize, + text: String, +} + +impl Backend for Command { + fn id(&self) -> BackendId { + self.id.clone() + } + + /// True for every target except [`Language::PortugueseUnverified`], same + /// rule as the `OpenAI` backend: the template usually drives an agent CLI, + /// which can be told pt-PT from pt-BR but cannot aim at an unverified + /// variant that names no exact target. + fn supports(&self, target: &Language) -> bool { + !matches!(target, Language::PortugueseUnverified) + } + + fn translate<'a>(&'a self, batch: &'a Batch) -> TranslateFuture<'a> { + Box::pin(async move { self.translate_inner(batch).await }) + } +} + +/// Spawn the command once, feed it `payload`, and collect its stdout. +/// +/// The child is killed when this future is dropped — including on timeout. +async fn run(program: &str, args: &[String], payload: String, id: BackendId) -> Result> { + // Overlay filesystems (Docker, some CI) can report ETXTBSY for a moment + // after an executable is written; retry briefly instead of failing a + // batch over it. + let mut child = loop { + match Process::new(program) + .args(args) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true) + .spawn() + { + Ok(child) => break child, + Err(err) if err.kind() == ErrorKind::ExecutableFileBusy => { + time::sleep(Duration::from_millis(10)).await; + } + Err(err) => { + return Err(Error::Transport { + backend: id.clone(), + source: Box::new(err), + }) + } + } + }; + + let mut stdout = child.stdout.take().expect("stdout was piped"); + let mut stderr = child.stderr.take().expect("stderr was piped"); + + // Drained concurrently: a chatty stderr must not be able to fill its + // pipe and deadlock the child while we wait on stdout. + let stderr_task = tokio::spawn(async move { + let mut buf = Vec::new(); + let _ignored = stderr.read_to_end(&mut buf).await; + buf + }); + + { + let mut stdin = child.stdin.take().expect("stdin was piped"); + stdin + .write_all(payload.as_bytes()) + .await + .map_err(|err| Error::Transport { + backend: id.clone(), + source: Box::new(err), + })?; + stdin.flush().await.map_err(|err| Error::Transport { + backend: id.clone(), + source: Box::new(err), + })?; + // Dropped here rather than merely shut down: the child sees EOF on + // stdin only when our end of the pipe actually closes. + } + + let transport = |err: std::io::Error| Error::Transport { + backend: id.clone(), + source: Box::new(err), + }; + + let mut out = Vec::new(); + stdout.read_to_end(&mut out).await.map_err(transport)?; + let status = child.wait().await.map_err(transport)?; + let stderr = stderr_task + .await + .unwrap_or_else(|_| b"stderr lost".to_vec()); + + if !status.success() { + return Err(Error::Transport { + backend: id, + source: format!( + "exited {}: {}", + status, + String::from_utf8_lossy(&stderr).trim() + ) + .into(), + }); + } + + Ok(out) +} + +#[cfg(test)] +mod tests { + use super::{argv, Command, CommandConfig}; + use crate::translate::Error; + + #[test] + fn an_empty_template_is_refused() { + assert!(matches!( + Command::new(CommandConfig { + template: " ".to_owned(), + timeout: super::DEFAULT_TIMEOUT, + }), + Err(Error::Malformed { .. }) + )); + } + + #[test] + fn the_template_splits_on_whitespace() { + assert_eq!(argv("ssh box claude -p"), ["ssh", "box", "claude", "-p"]); + assert_eq!(argv(" claude "), ["claude"]); + assert!(argv(" ").is_empty()); + } +} diff --git a/crates/arr-subs/src/lib.rs b/crates/arr-subs/src/lib.rs index 4518d32..eee71a8 100644 --- a/crates/arr-subs/src/lib.rs +++ b/crates/arr-subs/src/lib.rs @@ -12,16 +12,17 @@ //! //! Translation backends each sit behind their own cargo feature — //! `translate-openai`, `translate-deepl`, `translate-google`, -//! `translate-command`. The features are declared; the `openai` and -//! `command` backends are later issues (#191, #193), while `DeepL` -//! (`deepl`) and Google Translate (`google`) live here behind -//! their flags. What they share — the [`translate::Backend`] +//! `translate-command`. The features are declared; the backends live here +//! behind their flags (`openai` #191, `deepl`/`google` #192, `command` +//! #193). What they share — the [`translate::Backend`] //! trait, cue chunking, reply validation — lives in [`translate`], and the SRT //! cues it works in live in [`srt`]. Which compiled-in backend runs is a //! database setting, so switching engines never needs a rebuild. use std::{fmt, future::Future, pin::Pin}; +#[cfg(feature = "translate-command")] +pub mod command; pub mod convert; pub mod decode; #[cfg(feature = "translate-deepl")] @@ -38,6 +39,8 @@ pub mod srt; pub mod sync; pub mod translate; +#[cfg(feature = "translate-command")] +pub use command::{Command, CommandConfig, DEFAULT_TIMEOUT as COMMAND_DEFAULT_TIMEOUT}; #[cfg(feature = "translate-deepl")] pub use deepl::{DeepL, DeepLConfig}; pub use error::{Error, Result};