text_mirror/runner/
protocol.rs

1//! The framed stdin/stdout protocol between the runner and a worker.
2//!
3//! One frame is a four-byte big-endian length prefix followed by
4//! exactly that many bytes of one UTF-8 JSON value. The same frame
5//! format runs both directions. The declared length is checked
6//! against a limit before any allocation, so an oversized declaration
7//! is refused without buying its buffer. EOF inside the header or the
8//! payload is a truncated frame, the footprint of a killed writer,
9//! and never parses.
10//!
11//! A response frame alone is not success. The runner also requires a
12//! clean child exit and no trailing stdout bytes, so a worker that
13//! crashes after writing plausible JSON, or logs past its frame, is
14//! refused.
15
16use std::io::{self, Read, Write};
17
18use serde::de::Error as _;
19use serde::{Deserialize, Deserializer, Serialize, Serializer};
20
21use crate::segments::Segment;
22
23/// The schema identifier carried by every adapter request.
24pub const REQUEST_SCHEMA: &str = "text-mirror/adapter-request@1";
25
26/// The schema identifier carried by every adapter response.
27pub const RESPONSE_SCHEMA: &str = "text-mirror/adapter-response@1";
28
29/// Ceiling on request frame payload bytes, enforced by the worker
30/// before allocation. Requests carry paths and options, never file
31/// content, so they stay small.
32pub const MAX_REQUEST_BYTES: u32 = 1024 * 1024;
33
34macro_rules! literal_schema {
35    ($name:ident, $value:expr) => {
36        /// A schema field pinned to one literal value. Serializes to
37        /// that literal and rejects every other value on read.
38        #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
39        pub struct $name;
40
41        impl Serialize for $name {
42            fn serialize<S: Serializer>(
43                &self,
44                serializer: S,
45            ) -> std::result::Result<S::Ok, S::Error> {
46                serializer.serialize_str($value)
47            }
48        }
49
50        impl<'de> Deserialize<'de> for $name {
51            fn deserialize<D: Deserializer<'de>>(
52                deserializer: D,
53            ) -> std::result::Result<Self, D::Error> {
54                let value = String::deserialize(deserializer)?;
55                if value == $value {
56                    Ok($name)
57                } else {
58                    Err(D::Error::custom(format!(
59                        "unrecognized schema {value:?}, expected {:?}",
60                        $value
61                    )))
62                }
63            }
64        }
65    };
66}
67
68literal_schema!(RequestSchema, REQUEST_SCHEMA);
69literal_schema!(ResponseSchema, RESPONSE_SCHEMA);
70
71/// One request frame from the runner to a worker.
72#[derive(Debug, Clone, Serialize, Deserialize)]
73#[serde(deny_unknown_fields)]
74pub struct Request {
75    /// The literal string `text-mirror/adapter-request@1`.
76    pub schema: RequestSchema,
77    /// The adapter mode this request addresses.
78    pub adapter: String,
79    /// The adapter-specific request body.
80    pub payload: serde_json::Value,
81}
82
83/// One response frame from a worker to the runner.
84///
85/// Exactly one of `ok` and `error` is present. [`Response::validate`]
86/// enforces that, and the runner refuses any frame that fails it.
87#[derive(Debug, Clone, Serialize, Deserialize)]
88#[serde(deny_unknown_fields)]
89pub struct Response {
90    /// The literal string `text-mirror/adapter-response@1`.
91    pub schema: ResponseSchema,
92    /// The adapter-specific success body.
93    #[serde(skip_serializing_if = "Option::is_none", default)]
94    pub ok: Option<serde_json::Value>,
95    /// The failure body.
96    #[serde(skip_serializing_if = "Option::is_none", default)]
97    pub error: Option<WireError>,
98}
99
100impl Response {
101    /// A success response.
102    pub fn ok(value: serde_json::Value) -> Response {
103        Response {
104            schema: ResponseSchema,
105            ok: Some(value),
106            error: None,
107        }
108    }
109
110    /// A failure response.
111    pub fn err(code: &str, message: String) -> Response {
112        Response {
113            schema: ResponseSchema,
114            ok: None,
115            error: Some(WireError {
116                code: code.to_string(),
117                message,
118            }),
119        }
120    }
121
122    /// Checks that exactly one of `ok` and `error` is present.
123    pub fn validate(&self) -> std::result::Result<(), String> {
124        match (&self.ok, &self.error) {
125            (Some(_), None) | (None, Some(_)) => Ok(()),
126            (Some(_), Some(_)) => Err("response carries both ok and error".to_string()),
127            (None, None) => Err("response carries neither ok nor error".to_string()),
128        }
129    }
130}
131
132/// A machine-readable failure carried over the wire.
133#[derive(Debug, Clone, Serialize, Deserialize)]
134#[serde(deny_unknown_fields)]
135pub struct WireError {
136    /// Stable reason code, such as `pdf_no_text_layer`.
137    pub code: String,
138    /// Detail for a human reading the manifest.
139    pub message: String,
140}
141
142/// How reading one frame ended.
143#[derive(Debug)]
144pub enum FrameRead {
145    /// A complete frame payload.
146    Complete(Vec<u8>),
147    /// EOF before any header byte. The writer produced nothing.
148    Empty,
149    /// EOF inside the header or the payload. The writer died mid-frame.
150    Truncated,
151    /// The header declared more bytes than the limit allows. Nothing
152    /// was allocated or read past the header.
153    Oversized {
154        /// The declared payload length.
155        declared: u64,
156    },
157}
158
159/// Reads one length-prefixed frame, checking the declared length
160/// against `max_bytes` before allocating.
161pub fn read_frame(reader: &mut impl Read, max_bytes: u32) -> io::Result<FrameRead> {
162    let mut header = [0u8; 4];
163    match read_exact_or_eof(reader, &mut header)? {
164        Fill::Empty => return Ok(FrameRead::Empty),
165        Fill::Partial => return Ok(FrameRead::Truncated),
166        Fill::Full => {}
167    }
168    let declared = u32::from_be_bytes(header);
169    if declared > max_bytes {
170        return Ok(FrameRead::Oversized {
171            declared: u64::from(declared),
172        });
173    }
174    let mut payload = vec![0u8; declared as usize];
175    match read_exact_or_eof(reader, &mut payload)? {
176        Fill::Full => Ok(FrameRead::Complete(payload)),
177        Fill::Empty | Fill::Partial => Ok(FrameRead::Truncated),
178    }
179}
180
181enum Fill {
182    Full,
183    Partial,
184    Empty,
185}
186
187fn read_exact_or_eof(reader: &mut impl Read, buffer: &mut [u8]) -> io::Result<Fill> {
188    let mut filled = 0;
189    while filled < buffer.len() {
190        match reader.read(&mut buffer[filled..]) {
191            Ok(0) => {
192                return Ok(if filled == 0 {
193                    Fill::Empty
194                } else {
195                    Fill::Partial
196                });
197            }
198            Ok(n) => filled += n,
199            Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
200            Err(e) => return Err(e),
201        }
202    }
203    Ok(Fill::Full)
204}
205
206/// Writes one value as one frame and flushes.
207pub fn write_frame<T: Serialize>(writer: &mut impl Write, value: &T) -> io::Result<()> {
208    let payload = serde_json::to_vec(value).map_err(io::Error::other)?;
209    let length = u32::try_from(payload.len())
210        .map_err(|_| io::Error::other("frame payload exceeds the u32 length prefix"))?;
211    writer.write_all(&length.to_be_bytes())?;
212    writer.write_all(&payload)?;
213    writer.flush()
214}
215
216#[cfg(test)]
217mod tests {
218    use super::*;
219
220    #[test]
221    fn frames_roundtrip() {
222        let mut buffer = Vec::new();
223        let request = Request {
224            schema: RequestSchema,
225            adapter: "pdf".to_string(),
226            payload: serde_json::json!({"input": "input.pdf"}),
227        };
228        write_frame(&mut buffer, &request).unwrap();
229        let mut cursor = std::io::Cursor::new(buffer);
230        let FrameRead::Complete(payload) = read_frame(&mut cursor, MAX_REQUEST_BYTES).unwrap()
231        else {
232            panic!("expected a complete frame");
233        };
234        let back: Request = serde_json::from_slice(&payload).unwrap();
235        assert_eq!(back.adapter, "pdf");
236    }
237
238    #[test]
239    fn oversized_declarations_are_refused_before_allocation() {
240        let mut bytes = u32::MAX.to_be_bytes().to_vec();
241        bytes.extend_from_slice(b"junk");
242        let mut cursor = std::io::Cursor::new(bytes);
243        let FrameRead::Oversized { declared } = read_frame(&mut cursor, 1024).unwrap() else {
244            panic!("expected an oversized refusal");
245        };
246        assert_eq!(declared, u64::from(u32::MAX));
247    }
248
249    #[test]
250    fn truncation_is_detected_in_header_and_payload() {
251        let mut cursor = std::io::Cursor::new(vec![0u8, 0, 0]);
252        assert!(matches!(
253            read_frame(&mut cursor, 1024).unwrap(),
254            FrameRead::Truncated
255        ));
256
257        let mut bytes = 10u32.to_be_bytes().to_vec();
258        bytes.extend_from_slice(b"half");
259        let mut cursor = std::io::Cursor::new(bytes);
260        assert!(matches!(
261            read_frame(&mut cursor, 1024).unwrap(),
262            FrameRead::Truncated
263        ));
264
265        let mut cursor = std::io::Cursor::new(Vec::<u8>::new());
266        assert!(matches!(
267            read_frame(&mut cursor, 1024).unwrap(),
268            FrameRead::Empty
269        ));
270    }
271
272    #[test]
273    fn responses_require_exactly_one_body() {
274        assert!(Response::ok(serde_json::json!({})).validate().is_ok());
275        assert!(Response::err("io", "detail".to_string()).validate().is_ok());
276        let both = Response {
277            schema: ResponseSchema,
278            ok: Some(serde_json::json!({})),
279            error: Some(WireError {
280                code: "io".to_string(),
281                message: "detail".to_string(),
282            }),
283        };
284        assert!(both.validate().is_err());
285        let neither = Response {
286            schema: ResponseSchema,
287            ok: None,
288            error: None,
289        };
290        assert!(neither.validate().is_err());
291    }
292
293    #[test]
294    fn wrong_schema_values_are_rejected() {
295        let json = serde_json::to_string(&Response::ok(serde_json::json!({}))).unwrap();
296        let bumped = json.replace("adapter-response@1", "adapter-response@2");
297        assert!(serde_json::from_str::<Response>(&bumped).is_err());
298    }
299}
300
301/// Typed request and response bodies for the shipped adapters. Each
302/// adapter's `payload` and `ok` values are these shapes.
303pub mod bodies {
304    use super::*;
305
306    /// Request body for the `pdf` adapter.
307    #[derive(Debug, Clone, Serialize, Deserialize)]
308    #[serde(deny_unknown_fields)]
309    pub struct PdfRequest {
310        /// Input file name inside the jail.
311        pub input: String,
312    }
313
314    /// Success body for the `pdf` adapter.
315    #[derive(Debug, Clone, Serialize, Deserialize)]
316    #[serde(deny_unknown_fields)]
317    pub struct PdfOk {
318        /// The converted text, UTF-8, NFC, LF line endings.
319        pub text: String,
320        /// Non-fatal notes about the conversion.
321        pub warnings: Vec<String>,
322        /// Structure spans over the text.
323        pub segments: Vec<Segment>,
324    }
325
326    /// Request body for the `records` adapter, which reads parquet,
327    /// avro, and sqlite behind the jail.
328    ///
329    /// The ceilings travel in the request because the worker, not the
330    /// parent, enforces them, and they are rules data the parent reads
331    /// from the registry. A source over any ceiling fails closed with
332    /// reason `record-limit-exceeded`.
333    #[derive(Debug, Clone, Serialize, Deserialize)]
334    #[serde(deny_unknown_fields)]
335    pub struct RecordsRequest {
336        /// Input file name inside the jail.
337        pub input: String,
338        /// The detected format: `parquet`, `avro`, or `sqlite`.
339        pub format: String,
340        /// Ceiling on records per parquet or avro file, or per sqlite
341        /// table.
342        pub max_records: u64,
343        /// Ceiling on tables per sqlite database.
344        pub max_tables: u64,
345        /// Ceiling on rendered output bytes.
346        pub max_output_bytes: u64,
347    }
348
349    /// Success body for the `records` adapter.
350    #[derive(Debug, Clone, Serialize, Deserialize)]
351    #[serde(deny_unknown_fields)]
352    pub struct RecordsOk {
353        /// The rendered text, UTF-8, NFC, LF line endings.
354        pub text: String,
355        /// Non-fatal notes about the conversion.
356        pub warnings: Vec<String>,
357        /// Structure spans over the text. Sqlite carries a sheet
358        /// boundary per table.
359        pub segments: Vec<Segment>,
360    }
361
362    /// What an OCR request asks the engine to read.
363    #[derive(Debug, Clone, Serialize, Deserialize)]
364    #[serde(rename_all = "snake_case", deny_unknown_fields)]
365    pub enum OcrInput {
366        /// A standalone raster image.
367        Image,
368        /// One page of a paginated source rendered to raster.
369        PageRender {
370            /// Zero-based page index.
371            page: u32,
372            /// Render resolution in dots per inch.
373            dpi: u32,
374        },
375    }
376
377    /// Request body for the `ocr` adapter.
378    #[derive(Debug, Clone, Serialize, Deserialize)]
379    #[serde(deny_unknown_fields)]
380    pub struct OcrRequest {
381        /// Input file name inside the jail.
382        pub input: String,
383        /// What the engine should read.
384        pub kind: OcrInput,
385    }
386
387    /// One recognized text span with its confidence.
388    #[derive(Debug, Clone, Serialize, Deserialize)]
389    #[serde(deny_unknown_fields)]
390    pub struct OcrSpan {
391        /// The recognized text.
392        pub text: String,
393        /// Engine confidence in the closed range 0 to 1.
394        pub confidence: f64,
395    }
396
397    /// Success body for the `ocr` adapter.
398    #[derive(Debug, Clone, Serialize, Deserialize)]
399    #[serde(deny_unknown_fields)]
400    pub struct OcrOk {
401        /// Recognized spans in reading order.
402        pub spans: Vec<OcrSpan>,
403    }
404
405    /// Request body for the `asr` adapter.
406    #[derive(Debug, Clone, Serialize, Deserialize)]
407    #[serde(deny_unknown_fields)]
408    pub struct AsrRequest {
409        /// Input file name inside the jail.
410        pub input: String,
411        /// Language hint, when the caller has one.
412        #[serde(skip_serializing_if = "Option::is_none", default)]
413        pub language: Option<String>,
414    }
415
416    /// One transcribed speech segment.
417    #[derive(Debug, Clone, Serialize, Deserialize)]
418    #[serde(deny_unknown_fields)]
419    pub struct AsrSegment {
420        /// Segment start in seconds from the recording start.
421        pub start_seconds: f64,
422        /// Segment end in seconds.
423        pub end_seconds: f64,
424        /// One-based speaker index from diarization.
425        pub speaker: u32,
426        /// The transcribed text.
427        pub text: String,
428    }
429
430    /// Success body for the `asr` adapter.
431    #[derive(Debug, Clone, Serialize, Deserialize)]
432    #[serde(deny_unknown_fields)]
433    pub struct AsrOk {
434        /// Speech segments ordered by start time.
435        pub segments: Vec<AsrSegment>,
436        /// Detected language tag, when the engine reports one.
437        #[serde(skip_serializing_if = "Option::is_none", default)]
438        pub language: Option<String>,
439        /// Source duration in seconds, when the engine reports it.
440        #[serde(skip_serializing_if = "Option::is_none", default)]
441        pub duration_seconds: Option<f64>,
442    }
443
444    /// Request body for the `video` adapter.
445    #[derive(Debug, Clone, Serialize, Deserialize)]
446    #[serde(deny_unknown_fields)]
447    pub struct VideoRequest {
448        /// Input file name inside the jail.
449        pub input: String,
450        /// Seconds between sampled frames.
451        pub sample_interval_seconds: f64,
452        /// Ceiling on sampled frames.
453        pub max_frames: u32,
454    }
455
456    /// One deduplicated on-screen text state.
457    #[derive(Debug, Clone, Serialize, Deserialize)]
458    #[serde(deny_unknown_fields)]
459    pub struct ScreenState {
460        /// When this state first appeared, in seconds.
461        pub first_seen_seconds: f64,
462        /// The on-screen text.
463        pub text: String,
464    }
465
466    /// Success body for the `video` adapter.
467    #[derive(Debug, Clone, Serialize, Deserialize)]
468    #[serde(deny_unknown_fields)]
469    pub struct VideoOk {
470        /// Screen states deduplicated by content, ordered by first
471        /// appearance.
472        pub states: Vec<ScreenState>,
473        /// Source duration in seconds, when the engine reports it.
474        #[serde(skip_serializing_if = "Option::is_none", default)]
475        pub duration_seconds: Option<f64>,
476    }
477
478    /// Success body for the `probe-net` mode.
479    #[derive(Debug, Clone, Serialize, Deserialize)]
480    #[serde(deny_unknown_fields)]
481    pub struct ProbeReport {
482        /// One entry per attempted egress path.
483        pub attempts: Vec<ProbeAttempt>,
484    }
485
486    /// One egress attempt and how it ended.
487    #[derive(Debug, Clone, Serialize, Deserialize)]
488    #[serde(deny_unknown_fields)]
489    pub struct ProbeAttempt {
490        /// Which path was attempted, such as `ipv4_tcp`.
491        pub name: String,
492        /// `denied` for an immediate error, `connected` for success,
493        /// `timeout` when the attempt hung. Only `denied` proves the
494        /// jail.
495        pub outcome: String,
496        /// The underlying error or address detail.
497        pub detail: String,
498    }
499
500    /// Request body for the `probe-net` mode.
501    ///
502    /// Every target is one the parent controls and holds reachable
503    /// while the probe runs, so a connection that fails from inside
504    /// the jail is evidence of the jail and not of an unreachable
505    /// address. From an unjailed process every leg would succeed.
506    #[derive(Debug, Clone, Serialize, Deserialize)]
507    #[serde(deny_unknown_fields)]
508    pub struct ProbeNetRequest {
509        /// A Unix socket target the parent listens on outside the
510        /// jail. An abstract name on Linux, a filesystem path
511        /// elsewhere.
512        pub unix_target: String,
513        /// An IPv4 TCP address the parent listens on, such as
514        /// `127.0.0.1:49152`.
515        pub tcp4_target: String,
516        /// An IPv6 TCP address the parent listens on, such as
517        /// `[::1]:49152`.
518        pub tcp6_target: String,
519    }
520}
text_mirror/runner/protocol.rs

text_mirror/runner/
protocol.rs
