text_mirror/convert/
subprocess.rs

1//! Subprocess adapters: converters that run behind the sandboxed
2//! runner instead of in this process.
3//!
4//! Each adapter speaks the framed stdin/stdout protocol to a worker
5//! mode and renders the response to the outcome shape every converter
6//! emits. The PDF adapter is the production occupant: it carries the
7//! same outcome, warnings, and failure reasons as the in-process PDF
8//! path did, so its manifest records differ only in converter id. The
9//! OCR, ASR, and video adapters are the protocol half of their
10//! conversions. Their engines are not pinned yet, so the registry
11//! routes their formats to `unsupported` with reason `engine-unpinned`
12//! until a rules bump pins an engine, and the adapters are exercised
13//! against fake engines under test.
14
15use serde::de::DeserializeOwned;
16
17use crate::segments::Segment;
18
19use super::{ConvertError, Converter, Outcome, normalize_text};
20
21/// Registry id of the subprocess PDF adapter.
22pub const PDF_SUBPROCESS_ID: &str = "pdf-subprocess";
23/// Version of the subprocess PDF adapter.
24pub const PDF_SUBPROCESS_VERSION: &str = "1.0.0";
25
26/// Registry id of the OCR adapter.
27pub const OCR_ADAPTER_ID: &str = "ocr-adapter";
28/// Version of the OCR adapter.
29pub const OCR_ADAPTER_VERSION: &str = "1.0.0";
30
31/// Registry id of the speech transcription adapter.
32pub const ASR_ADAPTER_ID: &str = "asr-adapter";
33/// Version of the speech transcription adapter.
34pub const ASR_ADAPTER_VERSION: &str = "1.0.0";
35
36/// Registry id of the video adapter.
37pub const VIDEO_ADAPTER_ID: &str = "video-adapter";
38/// Version of the video adapter.
39pub const VIDEO_ADAPTER_VERSION: &str = "1.0.0";
40
41/// OCR spans below this confidence carry a warning.
42pub const OCR_CONFIDENCE_WARNING: f64 = 0.5;
43
44/// Renders whole seconds as `HH:MM:SS` for transcript timecodes.
45/// The input must be finite and non-negative.
46pub fn format_timecode(seconds: f64) -> Result<String, String> {
47    if !seconds.is_finite() || seconds < 0.0 {
48        return Err(format!("timestamp {seconds} is not a non-negative time"));
49    }
50    let whole = seconds as u64;
51    Ok(format!(
52        "{:02}:{:02}:{:02}",
53        whole / 3600,
54        (whole % 3600) / 60,
55        whole % 60
56    ))
57}
58
59fn single_line(text: &str) -> String {
60    text.replace(['\n', '\r'], " ").trim().to_string()
61}
62
63#[cfg(unix)]
64mod imp {
65    use std::sync::OnceLock;
66
67    use super::*;
68    use crate::runner::protocol::bodies::{
69        AsrOk, AsrRequest, OcrInput, OcrOk, OcrRequest, PdfOk, PdfRequest, VideoOk, VideoRequest,
70    };
71    use crate::runner::{Runner, RunnerError};
72
73    /// Maps a wire reason to the static reason vocabulary. Codes the
74    /// in-process converters already use pass through unchanged, so a
75    /// record's failure reason does not depend on which side of the
76    /// process boundary produced it.
77    fn static_code(code: &str) -> &'static str {
78        const KNOWN: &[&str] = &[
79            "pdf_no_text_layer",
80            "malformed",
81            "encrypted",
82            "resourceLimit",
83            "missingPart",
84            "unsupported",
85            "io",
86            "empty_output",
87            "invalid_utf8",
88            "converter_panic",
89            "unclaimed_format",
90            "resource_limit",
91        ];
92        KNOWN
93            .iter()
94            .find(|known| **known == code)
95            .copied()
96            .unwrap_or("adapter_error")
97    }
98
99    fn runner_failure(error: RunnerError) -> ConvertError {
100        ConvertError {
101            code: error.code,
102            message: error.message,
103        }
104    }
105
106    /// Runs one adapter mode and returns the parsed success body.
107    fn call<Req: serde::Serialize, Ok: DeserializeOwned>(
108        runner: &Runner,
109        mode: &str,
110        request: &Req,
111        files: &[(&str, &[u8])],
112    ) -> Result<Ok, ConvertError> {
113        let payload = serde_json::to_value(request).map_err(|e| ConvertError {
114            code: "adapter_error",
115            message: format!("cannot encode the adapter request: {e}"),
116        })?;
117        let response = runner.run(mode, payload, files).map_err(runner_failure)?;
118        if let Some(error) = response.error {
119            return Err(ConvertError {
120                code: static_code(&error.code),
121                message: error.message,
122            });
123        }
124        let ok = response.ok.expect("validated response has a body");
125        serde_json::from_value(ok).map_err(|e| ConvertError {
126            code: "adapter_error",
127            message: format!("adapter success body did not parse: {e}"),
128        })
129    }
130
131    fn shared_runner() -> Result<&'static Runner, ConvertError> {
132        static RUNNER: OnceLock<Result<Runner, RunnerError>> = OnceLock::new();
133        match RUNNER.get_or_init(Runner::with_platform_defaults) {
134            Ok(runner) => Ok(runner),
135            Err(error) => Err(runner_failure(error.clone())),
136        }
137    }
138
139    /// The production PDF conversion behind the sandbox runner.
140    ///
141    /// The worker links the same anydoc path the in-process document
142    /// adapter used, so text, warnings such as `pdf_partial_text`, and
143    /// failure reasons such as `pdf_no_text_layer` are unchanged.
144    pub struct PdfSubprocess;
145
146    impl Converter for PdfSubprocess {
147        fn id(&self) -> &'static str {
148            PDF_SUBPROCESS_ID
149        }
150
151        fn version(&self) -> &'static str {
152            PDF_SUBPROCESS_VERSION
153        }
154
155        fn convert(
156            &self,
157            source: &[u8],
158            detected_format: &str,
159        ) -> std::result::Result<Outcome, ConvertError> {
160            if detected_format != "pdf" {
161                return Err(ConvertError {
162                    code: "unclaimed_format",
163                    message: format!(
164                        "the subprocess PDF adapter does not handle {detected_format}"
165                    ),
166                });
167            }
168            let body: PdfOk = call(
169                shared_runner()?,
170                "pdf",
171                &PdfRequest {
172                    input: "input.pdf".to_string(),
173                },
174                &[("input.pdf", source)],
175            )?;
176            Ok(Outcome {
177                converter_id: PDF_SUBPROCESS_ID.to_string(),
178                converter_version: PDF_SUBPROCESS_VERSION.to_string(),
179                detected_format: detected_format.to_string(),
180                text: body.text,
181                warnings: body.warnings,
182                segments: body.segments,
183            })
184        }
185    }
186
187    /// The OCR protocol adapter. Engine-unpinned: reachable only
188    /// through an explicit runner until a rules bump pins an engine.
189    pub struct OcrAdapter {
190        runner: Runner,
191    }
192
193    impl OcrAdapter {
194        /// An adapter over an explicit runner.
195        pub fn new(runner: Runner) -> OcrAdapter {
196            OcrAdapter { runner }
197        }
198    }
199
200    impl Converter for OcrAdapter {
201        fn id(&self) -> &'static str {
202            OCR_ADAPTER_ID
203        }
204
205        fn version(&self) -> &'static str {
206            OCR_ADAPTER_VERSION
207        }
208
209        fn convert(
210            &self,
211            source: &[u8],
212            detected_format: &str,
213        ) -> std::result::Result<Outcome, ConvertError> {
214            if !matches!(detected_format, "png" | "jpeg" | "tiff") {
215                return Err(ConvertError {
216                    code: "unclaimed_format",
217                    message: format!("the OCR adapter does not handle {detected_format}"),
218                });
219            }
220            let input = format!("input.{detected_format}");
221            let body: OcrOk = call(
222                &self.runner,
223                "ocr",
224                &OcrRequest {
225                    input: input.clone(),
226                    kind: OcrInput::Image,
227                },
228                &[(&input, source)],
229            )?;
230            let mut warnings = Vec::new();
231            let mut lines = Vec::new();
232            for (index, span) in body.spans.iter().enumerate() {
233                if !(0.0..=1.0).contains(&span.confidence) {
234                    return Err(ConvertError {
235                        code: "adapter_error",
236                        message: format!(
237                            "span {index} confidence {} is outside 0 to 1",
238                            span.confidence
239                        ),
240                    });
241                }
242                if span.confidence < OCR_CONFIDENCE_WARNING {
243                    warnings.push(format!(
244                        "ocr_low_confidence: span {index} at {:.2}",
245                        span.confidence
246                    ));
247                }
248                lines.push(single_line(&span.text));
249            }
250            let text = normalize_text(&finish_lines(lines));
251            if !source.is_empty() && text.is_empty() {
252                return Err(empty_output(source.len()));
253            }
254            Ok(Outcome {
255                converter_id: OCR_ADAPTER_ID.to_string(),
256                converter_version: OCR_ADAPTER_VERSION.to_string(),
257                detected_format: detected_format.to_string(),
258                segments: vec![Segment::span(0, text.len(), "document")],
259                text,
260                warnings,
261            })
262        }
263    }
264
265    /// The speech transcription protocol adapter. Engine-unpinned.
266    pub struct AsrAdapter {
267        runner: Runner,
268    }
269
270    impl AsrAdapter {
271        /// An adapter over an explicit runner.
272        pub fn new(runner: Runner) -> AsrAdapter {
273            AsrAdapter { runner }
274        }
275    }
276
277    impl Converter for AsrAdapter {
278        fn id(&self) -> &'static str {
279            ASR_ADAPTER_ID
280        }
281
282        fn version(&self) -> &'static str {
283            ASR_ADAPTER_VERSION
284        }
285
286        fn convert(
287            &self,
288            source: &[u8],
289            detected_format: &str,
290        ) -> std::result::Result<Outcome, ConvertError> {
291            if !matches!(detected_format, "mp3" | "wav" | "m4a") {
292                return Err(ConvertError {
293                    code: "unclaimed_format",
294                    message: format!("the ASR adapter does not handle {detected_format}"),
295                });
296            }
297            let input = format!("input.{detected_format}");
298            let body: AsrOk = call(
299                &self.runner,
300                "asr",
301                &AsrRequest {
302                    input: input.clone(),
303                    language: None,
304                },
305                &[(&input, source)],
306            )?;
307            let text = normalize_text(&render_transcript(&body.segments).map_err(|detail| {
308                ConvertError {
309                    code: "adapter_error",
310                    message: detail,
311                }
312            })?);
313            if !source.is_empty() && text.is_empty() {
314                return Err(empty_output(source.len()));
315            }
316            Ok(Outcome {
317                converter_id: ASR_ADAPTER_ID.to_string(),
318                converter_version: ASR_ADAPTER_VERSION.to_string(),
319                detected_format: detected_format.to_string(),
320                segments: vec![Segment::span(0, text.len(), "document")],
321                text,
322                warnings: Vec::new(),
323            })
324        }
325    }
326
327    /// The video protocol adapter. Engine-unpinned.
328    pub struct VideoAdapter {
329        runner: Runner,
330    }
331
332    impl VideoAdapter {
333        /// An adapter over an explicit runner.
334        pub fn new(runner: Runner) -> VideoAdapter {
335            VideoAdapter { runner }
336        }
337    }
338
339    impl Converter for VideoAdapter {
340        fn id(&self) -> &'static str {
341            VIDEO_ADAPTER_ID
342        }
343
344        fn version(&self) -> &'static str {
345            VIDEO_ADAPTER_VERSION
346        }
347
348        fn convert(
349            &self,
350            source: &[u8],
351            detected_format: &str,
352        ) -> std::result::Result<Outcome, ConvertError> {
353            if !matches!(detected_format, "mp4" | "mov") {
354                return Err(ConvertError {
355                    code: "unclaimed_format",
356                    message: format!("the video adapter does not handle {detected_format}"),
357                });
358            }
359            let input = format!("input.{detected_format}");
360            let body: VideoOk = call(
361                &self.runner,
362                "video",
363                &VideoRequest {
364                    input: input.clone(),
365                    sample_interval_seconds: 1.0,
366                    max_frames: 10_000,
367                },
368                &[(&input, source)],
369            )?;
370            let text = normalize_text(&render_screen_states(&body.states).map_err(|detail| {
371                ConvertError {
372                    code: "adapter_error",
373                    message: detail,
374                }
375            })?);
376            if !source.is_empty() && text.is_empty() {
377                return Err(empty_output(source.len()));
378            }
379            Ok(Outcome {
380                converter_id: VIDEO_ADAPTER_ID.to_string(),
381                converter_version: VIDEO_ADAPTER_VERSION.to_string(),
382                detected_format: detected_format.to_string(),
383                segments: vec![Segment::span(0, text.len(), "document")],
384                text,
385                warnings: Vec::new(),
386            })
387        }
388    }
389
390    fn empty_output(source_len: usize) -> ConvertError {
391        ConvertError {
392            code: "empty_output",
393            message: format!("source is {source_len} bytes but conversion produced no text"),
394        }
395    }
396}
397
398#[cfg(unix)]
399pub use imp::{AsrAdapter, OcrAdapter, PdfSubprocess, VideoAdapter};
400
401#[cfg(not(unix))]
402mod imp {
403    use super::*;
404
405    /// The subprocess PDF adapter on a platform without a jail
406    /// backend. It refuses every run.
407    pub struct PdfSubprocess;
408
409    impl Converter for PdfSubprocess {
410        fn id(&self) -> &'static str {
411            PDF_SUBPROCESS_ID
412        }
413
414        fn version(&self) -> &'static str {
415            PDF_SUBPROCESS_VERSION
416        }
417
418        fn convert(
419            &self,
420            _source: &[u8],
421            _detected_format: &str,
422        ) -> std::result::Result<Outcome, ConvertError> {
423            Err(ConvertError {
424                code: "sandbox_unavailable",
425                message: "no jail backend exists for this platform, refusing to run adapters"
426                    .to_string(),
427            })
428        }
429    }
430}
431
432#[cfg(not(unix))]
433pub use imp::PdfSubprocess;
434
435/// Renders speech segments to the transcript format:
436/// `[HH:MM:SS -> HH:MM:SS] Speaker N: text`, one line per segment,
437/// ordered by start time.
438pub fn render_transcript(segments: &[crate::runner_bodies::AsrSegment]) -> Result<String, String> {
439    let mut ordered: Vec<_> = segments.iter().collect();
440    ordered.sort_by(|a, b| {
441        a.start_seconds
442            .total_cmp(&b.start_seconds)
443            .then(a.end_seconds.total_cmp(&b.end_seconds))
444            .then(a.speaker.cmp(&b.speaker))
445    });
446    let mut lines = Vec::new();
447    for segment in ordered {
448        if segment.end_seconds < segment.start_seconds {
449            return Err(format!(
450                "segment ends at {} before it starts at {}",
451                segment.end_seconds, segment.start_seconds
452            ));
453        }
454        lines.push(format!(
455            "[{} -> {}] Speaker {}: {}",
456            format_timecode(segment.start_seconds)?,
457            format_timecode(segment.end_seconds)?,
458            segment.speaker,
459            single_line(&segment.text)
460        ));
461    }
462    Ok(finish_lines(lines))
463}
464
465/// Renders deduplicated screen states to the transcript format:
466/// `[on-screen @ HH:MM:SS] text`, ordered by first appearance, with
467/// consecutive identical text collapsed.
468pub fn render_screen_states(
469    states: &[crate::runner_bodies::ScreenState],
470) -> Result<String, String> {
471    let mut ordered: Vec<_> = states.iter().collect();
472    ordered.sort_by(|a, b| {
473        a.first_seen_seconds
474            .total_cmp(&b.first_seen_seconds)
475            .then(a.text.cmp(&b.text))
476    });
477    let mut lines = Vec::new();
478    let mut previous: Option<String> = None;
479    for state in ordered {
480        let text = single_line(&state.text);
481        if previous.as_deref() == Some(text.as_str()) {
482            continue;
483        }
484        lines.push(format!(
485            "[on-screen @ {}] {}",
486            format_timecode(state.first_seen_seconds)?,
487            text
488        ));
489        previous = Some(text);
490    }
491    Ok(finish_lines(lines))
492}
493
494fn finish_lines(lines: Vec<String>) -> String {
495    if lines.is_empty() {
496        String::new()
497    } else {
498        let mut text = lines.join("\n");
499        text.push('\n');
500        text
501    }
502}
503
504#[cfg(test)]
505mod tests {
506    use super::*;
507    use crate::runner_bodies::{AsrSegment, ScreenState};
508
509    #[test]
510    fn timecodes_render_as_hours_minutes_seconds() {
511        assert_eq!(format_timecode(0.0).unwrap(), "00:00:00");
512        assert_eq!(format_timecode(12.9).unwrap(), "00:00:12");
513        assert_eq!(format_timecode(182.0).unwrap(), "00:03:02");
514        assert_eq!(format_timecode(3661.0).unwrap(), "01:01:01");
515        assert!(format_timecode(-1.0).is_err());
516        assert!(format_timecode(f64::NAN).is_err());
517    }
518
519    #[test]
520    fn transcripts_render_one_ordered_line_per_segment() {
521        let segments = vec![
522            AsrSegment {
523                start_seconds: 19.0,
524                end_seconds: 24.0,
525                speaker: 2,
526                text: "second\nline".to_string(),
527            },
528            AsrSegment {
529                start_seconds: 12.0,
530                end_seconds: 19.0,
531                speaker: 1,
532                text: "first remark".to_string(),
533            },
534        ];
535        assert_eq!(
536            render_transcript(&segments).unwrap(),
537            "[00:00:12 -> 00:00:19] Speaker 1: first remark\n\
538             [00:00:19 -> 00:00:24] Speaker 2: second line\n"
539        );
540    }
541
542    #[test]
543    fn transcripts_reject_a_segment_that_ends_before_it_starts() {
544        let segments = vec![AsrSegment {
545            start_seconds: 5.0,
546            end_seconds: 4.0,
547            speaker: 1,
548            text: "impossible".to_string(),
549        }];
550        assert!(render_transcript(&segments).is_err());
551    }
552
553    #[test]
554    fn screen_states_render_deduplicated_in_order() {
555        let states = vec![
556            ScreenState {
557                first_seen_seconds: 182.0,
558                text: "Quarterly revenue".to_string(),
559            },
560            ScreenState {
561                first_seen_seconds: 10.0,
562                text: "Agenda".to_string(),
563            },
564            ScreenState {
565                first_seen_seconds: 200.0,
566                text: "Quarterly revenue".to_string(),
567            },
568        ];
569        assert_eq!(
570            render_screen_states(&states).unwrap(),
571            "[on-screen @ 00:00:10] Agenda\n[on-screen @ 00:03:02] Quarterly revenue\n"
572        );
573    }
574}
text_mirror/convert/subprocess.rs

text_mirror/convert/
subprocess.rs
