Merge #193: remote-command translation backend

Closes #193
This commit is contained in:
Miguel Palhas
2026-08-25 01:48:30 +01:00
6 changed files with 615 additions and 69 deletions
+4
View File
@@ -42,5 +42,9 @@ wiremock.workspace = true
name = "openai"
required-features = ["translate-openai"]
[[test]]
name = "command"
required-features = ["translate-command"]
[lints]
workspace = true
+292
View File
@@ -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());
}
}
+7 -4
View File
@@ -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};
+6 -65
View File
@@ -25,7 +25,10 @@ use serde::{Deserialize, Serialize};
use arr_core::Language;
use crate::translate::{Backend, BackendId, Batch, Error, Result, TranslateFuture, TranslatedCue};
use crate::translate::{
strip_code_fence, system_prompt, Backend, BackendId, Batch, Error, Result, TranslateFuture,
TranslatedCue,
};
const DEFAULT_BASE_URL: &str = "https://api.openai.com/v1/";
const BACKEND_NAME: &str = "openai";
@@ -277,69 +280,6 @@ fn truncate(body: &str) -> String {
body.chars().take(MAX_ERROR_BODY).collect()
}
/// Strip a ```` ```json ```` / ```` ``` ```` fence a model wrapped its reply
/// in despite being told not to. Cheaper than retrying the whole batch over a
/// habit some models never break.
fn strip_code_fence(content: &str) -> &str {
let trimmed = content.trim();
let trimmed = trimmed
.strip_prefix("```json")
.or_else(|| trimmed.strip_prefix("```"))
.unwrap_or(trimmed);
trimmed.strip_suffix("```").unwrap_or(trimmed).trim()
}
/// How a language reads in a prompt meant for an LLM, not a settings row.
fn language_name(language: &Language) -> String {
match language {
Language::PortuguesePortugal => "European Portuguese (pt-PT)".to_owned(),
Language::PortugueseBrazil => "Brazilian Portuguese (pt-BR)".to_owned(),
Language::PortugueseUnverified => "Portuguese".to_owned(),
Language::Other(tag) => tag.clone(),
}
}
/// A concrete nudge against the one mistake DESIGN.md §15 calls out by name:
/// answering the wrong Portuguese. Generic advice ("respect the variant")
/// is easy for a model to nod at and ignore; naming the words it must not
/// reach for is not.
fn variant_note(target: &Language) -> Option<&'static str> {
match target {
Language::PortuguesePortugal => Some(
"This is continental European Portuguese, not Brazilian Portuguese: write \"tu\", \
\"comboio\", \"casa de banho\" and European spelling and verb forms, never \
Brazilian ones.",
),
Language::PortugueseBrazil => Some(
"This is Brazilian Portuguese, not European Portuguese: write \"você\", \"trem\", \
\"banheiro\" and Brazilian spelling and verb forms, never European ones.",
),
_ => None,
}
}
fn system_prompt(source: &Language, target: &Language) -> String {
let mut prompt = format!(
"You translate subtitle cues for a film or TV episode from {} into {}. Write natural, \
spoken subtitle language a native speaker would actually use, not a stiff literal \
translation.",
language_name(source),
language_name(target)
);
if let Some(note) = variant_note(target) {
prompt.push(' ');
prompt.push_str(note);
}
prompt.push_str(
" Reply with nothing but a JSON array, one object per cue, each shaped \
{\"number\": <the same number you were given>, \"text\": \"<translation>\"}, in the \
order you received them. Every cue you were given must appear exactly once, under its \
original number: never add, drop, merge or renumber a cue. No prose, no markdown code \
fences, no explanation — JSON only.",
);
prompt
}
#[derive(Serialize)]
struct RequestCue<'a> {
number: usize,
@@ -386,8 +326,9 @@ struct ChatReplyMessage {
mod tests {
use arr_core::Language;
use super::{language_name, strip_code_fence, system_prompt, OpenAi};
use super::OpenAi;
use crate::translate::Backend;
use crate::translate::{language_name, strip_code_fence, system_prompt};
fn backend() -> OpenAi {
OpenAi::new("gpt-4o-mini", super::OpenAiConfig { api_key: None }).expect("client builds")
+66
View File
@@ -153,6 +153,72 @@ pub trait Backend: fmt::Debug + Send + Sync {
/// Result alias for translation.
pub type Result<T> = std::result::Result<T, Error>;
/// Strip a ```` ```json ```` / ```` ``` ```` fence a model wrapped its reply
/// in despite being told not to. Cheaper than retrying the whole batch over a
/// habit some models never break.
pub(crate) fn strip_code_fence(content: &str) -> &str {
let trimmed = content.trim();
let trimmed = trimmed
.strip_prefix("```json")
.or_else(|| trimmed.strip_prefix("```"))
.unwrap_or(trimmed);
trimmed.strip_suffix("```").unwrap_or(trimmed).trim()
}
/// How a language reads in a prompt meant for an LLM, not a settings row.
pub(crate) fn language_name(language: &Language) -> String {
match language {
Language::PortuguesePortugal => "European Portuguese (pt-PT)".to_owned(),
Language::PortugueseBrazil => "Brazilian Portuguese (pt-BR)".to_owned(),
Language::PortugueseUnverified => "Portuguese".to_owned(),
Language::Other(tag) => tag.clone(),
}
}
/// A concrete nudge against the one mistake DESIGN.md §15 calls out by name:
/// answering the wrong Portuguese. Generic advice ("respect the variant")
/// is easy for a model to nod at and ignore; naming the words it must not
/// reach for is not.
fn variant_note(target: &Language) -> Option<&'static str> {
match target {
Language::PortuguesePortugal => Some(
"This is continental European Portuguese, not Brazilian Portuguese: write \"tu\", \
\"comboio\", \"casa de banho\" and European spelling and verb forms, never \
Brazilian ones.",
),
Language::PortugueseBrazil => Some(
"This is Brazilian Portuguese, not European Portuguese: write \"você\", \"trem\", \
\"banheiro\" and Brazilian spelling and verb forms, never European ones.",
),
_ => None,
}
}
/// The instruction every LLM-shaped backend sends ahead of the batch. Shared
/// rather than per-backend so the reply contract — JSON only, every cue back
/// under its original number — cannot drift between them.
pub(crate) fn system_prompt(source: &Language, target: &Language) -> String {
let mut prompt = format!(
"You translate subtitle cues for a film or TV episode from {} into {}. Write natural, \
spoken subtitle language a native speaker would actually use, not a stiff literal \
translation.",
language_name(source),
language_name(target)
);
if let Some(note) = variant_note(target) {
prompt.push(' ');
prompt.push_str(note);
}
prompt.push_str(
" Reply with nothing but a JSON array, one object per cue, each shaped \
{\"number\": <the same number you were given>, \"text\": \"<translation>\"}, in the \
order you received them. Every cue you were given must appear exactly once, under its \
original number: never add, drop, merge or renumber a cue. No prose, no markdown code \
fences, no explanation — JSON only.",
);
prompt
}
/// Everything that can go wrong translating a subtitle.
///
/// Kept apart from the provider [`Error`](crate::Error): the two report to
+240
View File
@@ -0,0 +1,240 @@
//! Remote-command backend tests (`DESIGN.md` §15, issue #193). Everything
//! runs against stub shell scripts in a tempdir — no remote host is touched.
//!
//! Built only with the `translate-command` feature (`required-features` in
//! `Cargo.toml`), same as the module itself: with the feature off, cargo
//! skips this target rather than building an empty one.
// Same per-target quirk as the crate's own tests: an integration test links
// the library's dependencies without using them all directly.
use {
arr_parse as _, chardetng as _, encoding_rs as _, reqwest as _, serde as _, serde_json as _,
thiserror as _, tracing as _, wiremock as _, zip as _,
};
use std::{
fs,
os::unix::fs::PermissionsExt,
path::{Path, PathBuf},
time::Duration,
};
use arr_core::Language;
use arr_subs::translate::{translate, Error};
use arr_subs::{Command, CommandConfig};
fn cues(texts: &[&str]) -> Vec<arr_subs::Cue> {
texts
.iter()
.enumerate()
.map(|(index, text)| arr_subs::Cue {
start: Duration::from_secs(index as u64),
end: Duration::from_secs(index as u64 + 1),
text: (*text).to_owned(),
})
.collect()
}
/// A stub command in `dir`, executable, with `body` as its shell script.
fn stub(dir: &Path, name: &str, body: &str) -> PathBuf {
let path = dir.join(name);
fs::write(&path, format!("#!/bin/sh\n{body}")).expect("stub writes");
fs::set_permissions(&path, fs::Permissions::from_mode(0o755)).expect("chmod");
path
}
fn config(template: &str, timeout: Duration) -> CommandConfig {
CommandConfig {
template: template.to_owned(),
timeout,
}
}
#[tokio::test]
async fn a_batch_round_trips_through_the_command() {
let dir = tempfile::tempdir().expect("tempdir");
// Save stdin to $1 and echo everything after the prompt line straight
// back — a valid reply, since numbers and count then match by
// construction.
let script = stub(dir.path(), "echo.sh", "tee \"$1\" | tail -n +2\n");
let backend = Command::new(config(
&format!(
"{} {}/received.json",
script.display(),
dir.path().display()
),
DEFAULT,
))
.expect("backend constructs");
let translated = translate(
&backend,
&cues(&["hello", "world"]),
&Language::Other("en".to_owned()),
&Language::PortuguesePortugal,
)
.await
.expect("the echo stub answers");
assert_eq!(translated.len(), 2);
assert_eq!(translated[0].text, "hello");
assert_eq!(translated[1].end, cues(&["x", "y"])[1].end);
// What reached the command's stdin: prompt first, batch after.
let sent = fs::read_to_string(dir.path().join("received.json")).expect("stub saved stdin");
assert!(sent.contains("European Portuguese"));
assert!(sent.contains(r#""number":2"#));
}
#[tokio::test]
async fn template_placeholders_reach_the_command() {
let dir = tempfile::tempdir().expect("tempdir");
let script = stub(dir.path(), "dump.sh", "cat > \"$1\"\nprintf '[]'\n");
// The dump file is named after the {target} placeholder, so a successful
// write proves substitution happened before spawn.
let backend = Command::new(config(
&format!(
"{} {}/{{target}}.json",
script.display(),
dir.path().display()
),
DEFAULT,
))
.expect("backend constructs");
translate(
&backend,
&cues(&["hi"]),
&Language::Other("en".to_owned()),
&Language::PortuguesePortugal,
)
.await
.expect_err("an empty reply is malformed");
assert!(dir.path().join("pt-PT.json").exists());
}
#[tokio::test]
async fn a_fenced_reply_is_still_accepted() {
let dir = tempfile::tempdir().expect("tempdir");
let script = stub(
dir.path(),
"fence.sh",
"cat >/dev/null\nprintf '```json\\n[{\"number\":1,\"text\":\"ola\"}]\\n```\\n'\n",
);
let backend =
Command::new(config(&script.display().to_string(), DEFAULT)).expect("backend constructs");
let translated = translate(
&backend,
&cues(&["hello"]),
&Language::Other("en".to_owned()),
&Language::PortuguesePortugal,
)
.await
.expect("a fenced reply unwraps");
assert_eq!(translated[0].text, "ola");
}
#[tokio::test]
async fn a_non_zero_exit_is_a_transport_error() {
let dir = tempfile::tempdir().expect("tempdir");
let script = stub(
dir.path(),
"fail.sh",
"echo connection refused >&2\nexit 7\n",
);
let backend =
Command::new(config(&script.display().to_string(), DEFAULT)).expect("backend constructs");
let error = translate(
&backend,
&cues(&["hello"]),
&Language::Other("en".to_owned()),
&Language::PortuguesePortugal,
)
.await
.expect_err("exit 7 must fail");
match error {
Error::Transport { ref source, .. } => {
assert!(source.to_string().contains("connection refused"));
assert!(error.is_transient());
}
other => panic!("expected Transport, got {other:?}"),
}
}
#[tokio::test]
async fn garbage_output_is_malformed() {
let dir = tempfile::tempdir().expect("tempdir");
let script = stub(dir.path(), "garbage.sh", "cat >/dev/null\necho I refuse\n");
let backend =
Command::new(config(&script.display().to_string(), DEFAULT)).expect("backend constructs");
let error = translate(
&backend,
&cues(&["hello"]),
&Language::Other("en".to_owned()),
&Language::PortuguesePortugal,
)
.await
.expect_err("prose is not a reply");
assert!(matches!(error, Error::Malformed { .. }));
assert!(!error.is_transient());
}
#[tokio::test]
async fn a_short_reply_fails_validation() {
let dir = tempfile::tempdir().expect("tempdir");
let script = stub(
dir.path(),
"short.sh",
"cat >/dev/null\nprintf '[{\"number\":1,\"text\":\"so\"}]'\n",
);
let backend =
Command::new(config(&script.display().to_string(), DEFAULT)).expect("backend constructs");
let error = translate(
&backend,
&cues(&["one", "two"]),
&Language::Other("en".to_owned()),
&Language::PortuguesePortugal,
)
.await
.expect_err("two cues went out, one came back");
assert!(matches!(error, Error::CueMismatch { .. }));
}
#[tokio::test]
async fn a_hanging_command_is_killed_at_the_timeout() {
let dir = tempfile::tempdir().expect("tempdir");
let script = stub(dir.path(), "hang.sh", "sleep 30\n");
let backend = Command::new(config(
&script.display().to_string(),
Duration::from_millis(150),
))
.expect("backend constructs");
let error = translate(
&backend,
&cues(&["hello"]),
&Language::Other("en".to_owned()),
&Language::PortuguesePortugal,
)
.await
.expect_err("a hung command must not wait forever");
match error {
Error::Transport { ref source, .. } => {
assert!(source.to_string().contains("timed out"));
assert!(error.is_transient());
}
other => panic!("expected Transport, got {other:?}"),
}
}
const DEFAULT: Duration = arr_subs::COMMAND_DEFAULT_TIMEOUT;