feat(arr): remote-command translation backend

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

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<Self> {
+        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<Vec<TranslatedCue>> {
+        let request_cues: Vec<WireCue> = 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<Vec<TranslatedCue>> {
+        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<WireCue> = 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<String> {
+    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<Vec<u8>> {
+    // 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());
+    }
+}

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};