asimov_runner/
jsonl.rs

1// This is free and unencumbered software released into the public domain.
2
3//! Batched JSONL transport for graph programs, with line framing underneath.
4//!
5//! # Connecting graph programs
6//!
7//! `GraphInput::Jsonl` below connects user-space batch streams. For a supervised
8//! process chain using native OS pipes, use [`crate::Pipeline`].
9//!
10//! ```no_run
11//! use asimov_runner::{Fetcher, GraphInput, GraphOutput, Matcher, StreamExt};
12//!
13//! # async fn example() -> Result<(), asimov_runner::ExecutorError> {
14//! let source = Fetcher::new(
15//!     "asimov-example-fetcher",
16//!     "https://example.com/resource",
17//!     GraphOutput::Captured,
18//!     Default::default(),
19//! ).execute().await?;
20//! let mut matches = Matcher::new(
21//!     "asimov-example-matcher",
22//!     GraphInput::Jsonl(source),
23//!     GraphOutput::Captured,
24//!     Default::default(),
25//! ).execute().await?;
26//! while let Some(batch) = matches.next().await {
27//!     for bytes in batch?.lines() {
28//!         // Process a line, or submit the whole batch to a network service.
29//!     }
30//!     // Continue to EOF to observe eventual failures.
31//! }
32//! # Ok(())
33//! # }
34//! ```
35
36use crate::batch::{FrameStream, FramedLine, batch_frames};
37use crate::{
38    BatchOptions, BatchStream, Executor, ExecutorError, Input, LineStream, Output, StreamExt,
39};
40use alloc::boxed::Box;
41use tokio::io::AsyncRead;
42
43/// A fallible stream of [`crate::JsonlBatch`] values, without JSON parsing or UTF-8 validation.
44///
45/// This alias is also usable for reader adapters and caller-supplied streams;
46/// the type alone imposes no process lifecycle or record-validation behavior.
47///
48/// Streams returned by [`Executor::execute_jsonl`] and
49/// [`Executor::execute_jsonl_with_input`] retain output LF/CRLF terminators and
50/// a final unterminated line. Execution returns after spawning; polling drives
51/// input, stdout, and stderr concurrently, with backpressure. No background task
52/// drains the pipes while the stream is idle. Consume to completion to check
53/// exit status: buffered complete lines are flushed in a partial batch before a
54/// subsequent error is yielded as the final item. [`BatchOptions`] bounds batch
55/// count/size and collection delay. Stderr and individual lines have no configured
56/// size bound. Dropping a process
57/// stream requests termination under the executor's default kill-on-drop policy;
58/// overriding that policy through [`Executor::command`] also affects streaming.
59/// [`Executor::execute_jsonl_with_io`] additionally supports forwarding instead
60/// of capture: those streams yield no payload batches but must still be consumed
61/// to drive I/O and observe completion. Errors use
62/// [`crate::ExecutionCompletion::into_result`] precedence.
63pub type JsonlStream = BatchStream<ExecutorError>;
64
65/// Frames an asynchronous byte reader into validated shared lines, preserving bytes.
66///
67/// Splits at LF, retaining LF/CRLF endings and a final unterminated line. Blank
68/// lines are yielded and neither JSON nor UTF-8 is validated. Reading is driven
69/// by polling, with no maximum line length. A read error is yielded once and
70/// ends the stream; bytes in a partially read line are not yielded on error.
71/// Dropping this stream drops its reader, without checking any process status.
72/// Reads use shared backing buffers, not a separate payload allocation per line.
73/// Retained lines can keep a larger read allocation alive; compact sparse,
74/// long-lived selections using [`crate::JsonlLine::into_compact`] or [`crate::JsonlBatch::into_compact`].
75pub fn jsonl_lines(reader: impl AsyncRead + Send + Unpin + 'static) -> LineStream {
76    Box::pin(jsonl_frames(reader).map(|line| line.map(FramedLine::into_line)))
77}
78
79/// Internal line framing with enough read-buffer provenance to build contiguous
80/// batches. Public individual lines contain only their sliced Bytes views.
81pub(crate) fn jsonl_frames(reader: impl AsyncRead + Send + Unpin + 'static) -> FrameStream {
82    Box::pin(asimov_flow::jsonl::jsonl_frames(reader).map(|line| line.map_err(Into::into)))
83}
84
85/// Reads and batches JSONL without changing bytes or line endings. Reading is
86/// lazy; see [`crate::batch_lines`] for thresholds, timer, EOF, and error behavior.
87/// The framer and batch builder retain contiguous backing metadata; extracting
88/// individual `JsonlLine` values yields ordinary sliced `Bytes` views.
89pub fn jsonl_batches(
90    reader: impl AsyncRead + Send + Unpin + 'static,
91    options: BatchOptions,
92) -> JsonlStream {
93    batch_frames(jsonl_frames(reader), options)
94}
95
96#[cfg(test)]
97mod framing_tests {
98    use super::*;
99    use crate::{Bytes, JsonlBatch, JsonlLine};
100    use alloc::{vec, vec::Vec};
101    use std::io::Cursor;
102    use tokio::io::AsyncWriteExt;
103
104    #[tokio::test]
105    async fn frames_a_read_buffer_into_shared_lines_without_payload_copies() {
106        let mut stream = jsonl_lines(Cursor::new(b"a\nb\r\nc\nlast"));
107        let mut lines = Vec::new();
108        for expected in [b"a\n".as_slice(), b"b\r\n", b"c\n"] {
109            let line = stream.next().await.unwrap().unwrap();
110            assert!(matches!(line, JsonlLine::Shared(_)));
111            assert_eq!(line.as_bytes(), expected);
112            lines.push(line);
113        }
114        assert_eq!(
115            lines[1].as_bytes().as_ptr(),
116            lines[0].as_bytes().as_ptr().wrapping_add(2)
117        );
118        let batch = JsonlBatch::new(lines);
119        // Detached Bytes views do not retain batch-level coalescing metadata.
120        assert!(batch.as_contiguous_bytes().is_none());
121        let last = stream.next().await.unwrap().unwrap();
122        assert_eq!(last.as_bytes(), b"last");
123        assert!(!last.is_terminated());
124        assert!(stream.next().await.is_none());
125        // Retained lines remain valid after both reader and framing state are gone.
126        drop(stream);
127        assert_eq!(
128            batch.lines().collect::<Vec<_>>(),
129            [b"a\n".as_slice(), b"b\r\n", b"c\n"]
130        );
131    }
132
133    #[tokio::test]
134    async fn reader_batches_preserve_contiguous_metadata_across_batch_boundaries() {
135        let mut batches = jsonl_batches(
136            Cursor::new(b"a\nb\nc\nd\ntail"),
137            BatchOptions::new(2, 1024, core::time::Duration::from_secs(1)).unwrap(),
138        );
139        let first = batches.next().await.unwrap().unwrap();
140        let second = batches.next().await.unwrap().unwrap();
141        let tail = batches.next().await.unwrap().unwrap();
142        assert!(batches.next().await.is_none());
143        drop(batches);
144        assert_eq!(first.as_contiguous_bytes(), Some(b"a\nb\n".as_slice()));
145        assert_eq!(second.as_contiguous_bytes(), Some(b"c\nd\n".as_slice()));
146        assert_eq!(
147            second.as_contiguous_bytes().unwrap().as_ptr(),
148            first
149                .as_contiguous_bytes()
150                .unwrap()
151                .as_ptr()
152                .wrapping_add(4)
153        );
154        let pointer = first.as_contiguous_bytes().unwrap().as_ptr();
155        let lines = first.into_lines();
156        assert_eq!(lines[0].as_bytes().as_ptr(), pointer);
157        assert_eq!(lines[1].as_bytes().as_ptr(), pointer.wrapping_add(2));
158        assert_eq!(tail.lines().collect::<Vec<_>>(), [b"tail".as_slice()]);
159        assert!(tail.as_contiguous_bytes().is_none());
160    }
161
162    #[tokio::test]
163    async fn byte_threshold_lookahead_preserves_batch_views() {
164        let mut batches = jsonl_batches(
165            Cursor::new(b"a\nb\nc\n"),
166            BatchOptions::new(10, 3, core::time::Duration::from_secs(1)).unwrap(),
167        );
168        let first = batches.next().await.unwrap().unwrap();
169        let second = batches.next().await.unwrap().unwrap();
170        let third = batches.next().await.unwrap().unwrap();
171        assert!(batches.next().await.is_none());
172        assert_eq!(first.as_contiguous_bytes(), Some(b"a\n".as_slice()));
173        assert_eq!(second.as_contiguous_bytes(), Some(b"b\n".as_slice()));
174        assert_eq!(third.as_contiguous_bytes(), Some(b"c\n".as_slice()));
175        let first_pointer = first.as_contiguous_bytes().unwrap().as_ptr();
176        assert_eq!(
177            second.as_contiguous_bytes().unwrap().as_ptr(),
178            first_pointer.wrapping_add(2)
179        );
180        assert_eq!(
181            third.as_contiguous_bytes().unwrap().as_ptr(),
182            first_pointer.wrapping_add(4)
183        );
184    }
185
186    #[tokio::test]
187    async fn frames_split_crlf_long_lines_and_unterminated_tails() {
188        let (reader, mut writer) = tokio::io::duplex(7);
189        let mut expected = vec![b'x'; 128 * 1024];
190        expected.extend_from_slice(b"\r\n");
191        let data = expected.clone();
192        let writing = tokio::spawn(async move {
193            writer.write_all(&data[..data.len() - 1]).await.unwrap();
194            tokio::task::yield_now().await;
195            writer.write_all(b"\ntail").await.unwrap();
196        });
197        let mut lines = jsonl_lines(reader);
198        assert_eq!(lines.next().await.unwrap().unwrap().as_bytes(), expected);
199        assert_eq!(lines.next().await.unwrap().unwrap().as_bytes(), b"tail");
200        assert!(lines.next().await.is_none());
201        writing.await.unwrap();
202    }
203
204    #[tokio::test]
205    async fn every_chunk_boundary_preserves_framing() {
206        let data = Bytes::from_static(b"\n{}\r\n\xff\nend");
207        for width in 1..=data.len() {
208            let (reader, mut writer) = tokio::io::duplex(width);
209            let source = data.clone();
210            let writing = tokio::spawn(async move {
211                for chunk in source.chunks(width) {
212                    writer.write_all(chunk).await.unwrap();
213                    tokio::task::yield_now().await;
214                }
215            });
216            let mut lines = jsonl_lines(reader);
217            for expected in [b"\n".as_slice(), b"{}\r\n", b"\xff\n", b"end"] {
218                let line = lines.next().await.unwrap().unwrap();
219                assert_eq!(line.as_bytes(), expected);
220                assert!(JsonlLine::shared(line.into_bytes()).is_ok());
221            }
222            assert!(lines.next().await.is_none());
223            writing.await.unwrap();
224        }
225    }
226}
227
228impl Executor {
229    /// Spawns a program and streams its stdout as JSONL batches.
230    ///
231    /// Uses the configured standard streams; stdout must be piped to yield batches.
232    /// No input is written. Any piped stdin is closed when the stream is polled.
233    /// Even with ignored or inherited stdout, consume the stream to completion
234    /// to check process success. See [`JsonlStream`] for polling and drop behavior.
235    ///
236    /// # Errors
237    ///
238    /// Spawn errors are returned directly; read, wait, and exit errors are stream items.
239    pub async fn execute_jsonl(&mut self) -> Result<JsonlStream, ExecutorError> {
240        self.execute_jsonl_with_input(&mut Input::Ignored).await
241    }
242
243    /// Spawns a program, feeding input concurrently with streaming its stdout.
244    ///
245    /// After a successful spawn, ownership of `input` moves into the returned
246    /// stream and it is replaced with [`Input::Ignored`]. Repeated execution does
247    /// not replay input. No input is consumed on spawn failure. Configure the
248    /// command's stdin with [`Input::as_stdio`] before calling this method.
249    /// Stdout and stderr handling also use the existing command configuration.
250    /// [`Input::AsyncRead`] copies raw bytes; [`Input::Jsonl`] writes framed lines.
251    /// This method does not automatically adapt byte input into JSONL.
252    ///
253    /// Polling drives I/O; see [`JsonlStream`] for buffering and drop behavior.
254    /// Early child completion cancels the input feed and reports
255    /// [`ExecutorError::IncompleteInput`] if the child otherwise succeeded.
256    /// Errors follow [`crate::ExecutionCompletion::into_result`] precedence.
257    ///
258    /// # Errors
259    ///
260    /// Spawn errors are returned directly; input, read, wait, and exit errors are
261    /// stream items. Non-ignored input requires piped stdin; otherwise feeding
262    /// it reports an I/O `InvalidInput` error through the stream.
263    pub async fn execute_jsonl_with_input(
264        &mut self,
265        input: &mut Input,
266    ) -> Result<JsonlStream, ExecutorError> {
267        self.execute_jsonl_with_io(input, &mut Output::Captured)
268            .await
269    }
270
271    /// Spawns a graph producer with the supplied stdout policy and no input.
272    /// Configure stdout with [`Output::as_stdio`] first. Forwarded output is
273    /// written while the returned stream is polled and yields no payload items.
274    /// Spawn errors are returned directly; subsequent failures are stream items.
275    pub async fn execute_jsonl_with_output(
276        &mut self,
277        output: &mut Output,
278    ) -> Result<JsonlStream, ExecutorError> {
279        self.execute_jsonl_with_io(&mut Input::Ignored, output)
280            .await
281    }
282
283    /// Spawns a graph program with concurrent input and stdout routing.
284    ///
285    /// Configure the command using the policies' `as_stdio` methods first.
286    /// Successful spawning transfers input and any output writer into the stream.
287    /// Captured stdout yields batches using [`Self::batch_options`]; other modes
288    /// yield only eventual errors. Complete lines buffered before an error are
289    /// delivered as a partial batch first.
290    /// A transferred writer is flushed at EOF, not shut down, and the wrapper's
291    /// output policy becomes [`Output::Ignored`] for subsequent executions.
292    /// Non-writer output policies remain reusable. Spawn failure consumes neither.
293    ///
294    /// # Errors
295    ///
296    /// Spawn errors are returned directly. Input, transport, writer, wait, and
297    /// exit failures are stream items. Success requires complete input delivery
298    /// and zero exit status; see [`crate::ExecutionCompletion::into_result`].
299    pub async fn execute_jsonl_with_io(
300        &mut self,
301        input: &mut Input,
302        output: &mut Output,
303    ) -> Result<JsonlStream, ExecutorError> {
304        let options = self.batch_options();
305        let frames = self.execute_jsonl_frames_with_io(input, output).await?;
306        Ok(batch_frames(frames, options))
307    }
308
309    /// The single process-lifecycle implementation. Batching and local line caps
310    /// are layered above this primitive rather than duplicating execution logic.
311    pub(crate) async fn execute_jsonl_frames_with_io(
312        &mut self,
313        input: &mut Input,
314        output: &mut Output,
315    ) -> Result<FrameStream, ExecutorError> {
316        let mut process = self.spawn().await?;
317        let stdout = if matches!(output, Output::Captured) {
318            process.stdout.take()
319        } else {
320            None
321        };
322        let mut input = core::mem::replace(input, Input::Ignored);
323        let mut destination = output.take_for_stream();
324        Ok(Box::pin(async_stream::try_stream! {
325            let completion = async move {
326                crate::executor::communicate(process, &mut input, &mut destination).await
327            };
328            tokio::pin!(completion);
329            let mut output = None;
330            if let Some(stdout) = stdout {
331                let mut lines = jsonl_frames(stdout);
332                loop {
333                    let line = tokio::select! {
334                        result = &mut completion, if output.is_none() => {
335                            output = Some(result);
336                            continue;
337                        },
338                        line = lines.next() => line,
339                    };
340                    match line {
341                        Some(line) => yield line?,
342                        None => break,
343                    }
344                }
345            }
346            let output = match output {
347                Some(output) => output?,
348                None => completion.await?,
349            };
350            output.into_result()?;
351        }))
352    }
353}
354
355#[cfg(all(test, unix))]
356mod tests {
357    use super::*;
358    use crate::*;
359    use alloc::{string::ToString, vec, vec::Vec};
360    use std::{io::Cursor, process::Stdio, time::Duration};
361    use tokio::time::timeout;
362
363    #[tokio::test]
364    async fn graph_producers_stream_before_exit() {
365        let script = "printf '{}\\n'; exec sleep 30";
366        let args = vec!["-c".to_string(), script.to_string()];
367        macro_rules! check {
368            ($runner:expr) => {{
369                let mut stream = timeout(Duration::from_secs(5), $runner.execute())
370                    .await
371                    .expect("spawn must not wait for exit")
372                    .unwrap();
373                let line = timeout(Duration::from_secs(5), stream.next())
374                    .await
375                    .expect("output must not wait for exit");
376                assert_eq!(
377                    line.unwrap().unwrap().lines().collect::<Vec<_>>(),
378                    vec![b"{}\n".to_vec()]
379                );
380            }};
381        }
382        check!(Adapter::new(
383            "/bin/sh",
384            Input::Ignored,
385            GraphOutput::Captured,
386            AdapterOptions {
387                other: args.clone(),
388                ..Default::default()
389            },
390        ));
391        check!(Emitter::new(
392            "/bin/sh",
393            GraphOutput::Captured,
394            EmitterOptions {
395                other: args.clone(),
396                ..Default::default()
397            },
398        ));
399        check!(Fetcher::new(
400            "/bin/sh",
401            script,
402            GraphOutput::Captured,
403            FetcherOptions {
404                other: vec!["-c".into()],
405                ..Default::default()
406            },
407        ));
408        check!(Reader::new(
409            "/bin/sh",
410            Input::Ignored,
411            GraphOutput::Captured,
412            ReaderOptions {
413                other: args.clone(),
414                ..Default::default()
415            },
416        ));
417        check!(Matcher::new(
418            "/bin/sh",
419            Input::Ignored,
420            GraphOutput::Captured,
421            MatcherOptions {
422                other: args.clone(),
423                ..Default::default()
424            },
425        ));
426        check!(Reasoner::new(
427            "/bin/sh",
428            Input::Ignored,
429            GraphOutput::Captured,
430            ReasonerOptions {
431                other: args,
432                ..Default::default()
433            },
434        ));
435    }
436
437    #[tokio::test]
438    async fn output_is_available_while_input_is_still_open() {
439        timeout(Duration::from_secs(5), async {
440            let source = Box::pin(async_stream::try_stream! {
441                yield JsonlBatch::try_from(vec![b"{}".to_vec()]).unwrap();
442                core::future::pending::<()>().await;
443            });
444            let mut stream = Matcher::new(
445                "/bin/cat",
446                GraphInput::Jsonl(source),
447                GraphOutput::Captured,
448                MatcherOptions::default(),
449            )
450            .execute()
451            .await
452            .unwrap();
453            assert_eq!(
454                stream
455                    .next()
456                    .await
457                    .unwrap()
458                    .unwrap()
459                    .lines()
460                    .collect::<Vec<_>>(),
461                vec![b"{}\n".to_vec()]
462            );
463        })
464        .await
465        .expect("output must not wait for input EOF");
466    }
467
468    #[tokio::test]
469    async fn dropping_stream_terminates_child() {
470        let mut stream = Emitter::new(
471            "/bin/sh",
472            GraphOutput::Captured,
473            EmitterOptions::builder()
474                .other("-c")
475                .other("printf '%s\\n' $$; exec sleep 30")
476                .build(),
477        )
478        .execute()
479        .await
480        .unwrap();
481        let pid = timeout(Duration::from_secs(5), stream.next())
482            .await
483            .unwrap()
484            .unwrap()
485            .unwrap();
486        let pid = std::str::from_utf8(pid.lines().next().unwrap())
487            .unwrap()
488            .trim();
489        drop(stream);
490        timeout(Duration::from_secs(5), async {
491            while tokio::process::Command::new("/bin/kill")
492                .args(["-0", pid])
493                .stdout(Stdio::null())
494                .stderr(Stdio::null())
495                .status()
496                .await
497                .unwrap()
498                .success()
499            {
500                tokio::time::sleep(Duration::from_millis(10)).await;
501            }
502        })
503        .await
504        .expect("dropping the stream must terminate the child");
505    }
506
507    #[tokio::test]
508    async fn feeds_large_graph_while_draining_both_output_pipes() {
509        timeout(Duration::from_secs(10), async {
510            let line = [b"\"".as_slice(), &vec![b'x'; 1024], b"\"\n"].concat();
511            let source_line = line.clone();
512            let source = Box::pin(async_stream::try_stream! {
513                for _ in 0..64 {
514                    yield JsonlBatch::try_from(vec![source_line.clone(); 64]).unwrap();
515                }
516            });
517            let mut stream = Reasoner::new(
518                "/bin/sh",
519                GraphInput::Jsonl(source),
520                GraphOutput::Captured,
521                ReasonerOptions::builder().other("-c").other(
522                    "i=0; while [ $i -lt 10000 ]; do printf 'diagnostic\\n' >&2; i=$((i + 1)); done; cat",
523                ).build(),
524            ).execute().await.unwrap();
525            let mut count = 0;
526            while let Some(actual) = stream.next().await {
527                let batch = actual.unwrap();
528                for actual in batch.lines() {
529                    assert_eq!(actual, line);
530                }
531                count += batch.len();
532            }
533            assert_eq!(count, 4096);
534        }).await.expect("full-duplex graph transport must not deadlock");
535    }
536
537    #[tokio::test]
538    async fn composes_graph_producers_and_consumers() {
539        let source = Emitter::new(
540            "/bin/sh",
541            GraphOutput::Captured,
542            EmitterOptions::builder()
543                .other("-c")
544                .other("printf '{}\\r\\n{\"last\":true}'")
545                .build(),
546        )
547        .execute()
548        .await
549        .unwrap();
550        let mut stream = Matcher::new(
551            "/bin/cat",
552            GraphInput::Jsonl(source),
553            GraphOutput::Captured,
554            MatcherOptions::default(),
555        )
556        .execute()
557        .await
558        .map(flatten_batches)
559        .unwrap();
560        assert_eq!(stream.next().await.unwrap().unwrap().as_bytes(), b"{}\r\n");
561        assert_eq!(
562            stream.next().await.unwrap().unwrap().as_bytes(),
563            b"{\"last\":true}\n"
564        );
565        assert!(stream.next().await.is_none());
566    }
567
568    #[tokio::test]
569    async fn propagates_upstream_failure() {
570        timeout(Duration::from_secs(5), async {
571            let source = Emitter::new(
572                "/bin/sh",
573                GraphOutput::Captured,
574                EmitterOptions::builder()
575                    .other("-c")
576                    .other("printf 'upstream failed' >&2; exit 65")
577                    .build(),
578            )
579            .execute()
580            .await
581            .unwrap();
582            let mut stream = Matcher::new(
583                "/bin/cat",
584                GraphInput::Jsonl(source),
585                GraphOutput::Captured,
586                MatcherOptions::default(),
587            )
588            .execute()
589            .await
590            .unwrap();
591            match stream.next().await.unwrap().unwrap_err() {
592                ExecutorError::Failure(code, Some(stderr)) => {
593                    assert_eq!(code.code(), Some(65));
594                    assert_eq!(stderr, "upstream failed");
595                },
596                error => panic!("unexpected error: {error}"),
597            }
598            assert!(stream.next().await.is_none());
599        })
600        .await
601        .expect("source failure must terminate the downstream child");
602    }
603
604    #[tokio::test]
605    async fn early_exit_cancels_idle_input() {
606        timeout(Duration::from_secs(5), async {
607            let input = Input::Jsonl(Box::pin(crate::stream::pending()));
608            let mut stream = Matcher::new(
609                "/bin/sh",
610                input,
611                GraphOutput::Captured,
612                MatcherOptions::builder()
613                    .other("-c")
614                    .other("exit 65")
615                    .build(),
616            )
617            .execute()
618            .await
619            .unwrap();
620            assert!(matches!(
621                stream.next().await,
622                Some(Err(ExecutorError::Failure(_, _)))
623            ));
624            assert!(stream.next().await.is_none());
625        })
626        .await
627        .expect("child exit must not wait for an idle input stream");
628    }
629
630    #[tokio::test]
631    async fn writer_and_indexer_consume_jsonl() {
632        let input = || {
633            GraphInput::Jsonl(Box::pin(crate::stream::iter([
634                Ok(JsonlBatch::default()),
635                Ok(JsonlBatch::try_from(vec![b"{}".to_vec(), b"[]\r\n".to_vec()]).unwrap()),
636            ])))
637        };
638        let script = "test \"$(cat)\" = \"$(printf '{}\\n[]\\r')\" || exit 65";
639        Indexer::new(
640            "/bin/sh",
641            input(),
642            IndexerOptions::builder().other("-c").other(script).build(),
643        )
644        .execute()
645        .await
646        .unwrap();
647        let output = Writer::new(
648            "/bin/sh",
649            input(),
650            AnyOutput::Captured,
651            WriterOptions::builder()
652                .other("-c")
653                .other(alloc::format!("{script}; printf '\\000\\377'"))
654                .build(),
655        )
656        .execute()
657        .await
658        .unwrap();
659        assert_eq!(output.into_inner(), b"\x00\xff");
660    }
661
662    #[tokio::test]
663    async fn byte_graph_input_is_framed_and_spawn_failure_preserves_input() {
664        let mut input = GraphInput::AsyncRead(Box::new(Cursor::new(b"{}".to_vec()))).into_jsonl();
665        let mut missing = Executor::new("/this-jsonl-program-does-not-exist");
666        missing.command().stdin(input.as_stdio());
667        assert!(matches!(
668            missing.execute_jsonl_with_input(&mut input).await,
669            Err(ExecutorError::MissingProgram(_))
670        ));
671        let mut executor = Executor::new("/bin/cat");
672        executor
673            .command()
674            .stdin(input.as_stdio())
675            .stdout(Stdio::piped());
676        let mut stream = executor.execute_jsonl_with_input(&mut input).await.unwrap();
677        assert!(matches!(input, Input::Ignored));
678        assert_eq!(
679            stream
680                .next()
681                .await
682                .unwrap()
683                .unwrap()
684                .lines()
685                .collect::<Vec<_>>(),
686            vec![b"{}\n".to_vec()]
687        );
688        assert!(stream.next().await.is_none());
689    }
690}
asimov_runner/jsonl.rs

asimov_runner/
jsonl.rs
