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

text_mirror/convert/
subprocess.rs
