feat(stream): streaming Markdown conversion, default in the CLI

Add an incremental Markdown path so the converter can be embedded in apps
that want output as it is produced rather than one buffered string.

- core: `MarkdownStreamer` renders finalized, in-order node batches into
  Markdown chunks whose concatenation is byte-identical to the buffered
  `export_to_markdown_images` (placeholder + embedded image modes). Strict-mode
  recovered links are applied per chunk with a carry-forward queue.
- pdf: `Pipeline::convert_streaming` emits each page's nodes in document order
  as the parallel worker pool finishes them. A `StreamAssembler` holds back only
  the one-page tail that might still merge across a page break, so cross-page
  paragraph merging is preserved; an in-order reorder buffer on the calling
  thread feeds `emit` (so it needn't be Send) with channel backpressure.
- fleischwolf: `DocumentConverter::convert_streaming` /
  `convert_streaming_images` return a `MarkdownStream` iterator of chunks,
  produced on a background thread (dropping it cancels). PDF streams page by
  page; other backends build their document then stream it through the same API.
  JSON and the referenced image mode stay on the buffered path.
- cli: stream Markdown to stdout by default; `--no-stream` opts back into
  buffering.

Tests: streamer byte-identity unit tests; `StreamAssembler` equivalence to
one-shot `merge_continuations` over random page splits; a regression check that
streamed Markdown equals the buffered export for every non-PDF corpus source.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01H3AUFjF71HP2YY9cD32wx8

Author: claude

README.md

@@ -118,6 +118,40 @@ strict:  Foo ***both***.    |   ```rust (lang kept)  |   Name: ___
 `result.document.export_to_markdown_with(strict)` overrides the mode per call.
 Python docling has no such switch.
 
+### Streaming Markdown
+
+For embedding in real apps, `convert_streaming` returns the document's Markdown
+as an iterator of chunks instead of one big string — handy for piping a long
+document straight to stdout, an HTTP response, or a socket as it is produced:
+
+```rust
+use std::io::Write;
+use fleischwolf::{DocumentConverter, SourceDocument};
+
+let source = SourceDocument::from_file("input.pdf").unwrap();
+let mut out = std::io::stdout();
+for chunk in DocumentConverter::new().convert_streaming(source).unwrap() {
+    out.write_all(chunk.unwrap().as_bytes()).unwrap();
+}
+```
+
+The headline win is PDF. The ML pipeline already processes pages **in parallel**;
+streaming emits each page's Markdown **in document order, as soon as it is ready**
+(with a one-page look-ahead so paragraphs that wrap across a page break still
+merge), so output starts flowing before the last page is done. The conversion
+runs on a background thread and the chunk iterator applies backpressure; dropping
+it cancels the work. Concatenating every chunk is **byte-identical** to the
+buffered `export_to_markdown()`.
+
+Streaming is Markdown-only — JSON serializes docling-core's reference-based tree
+and needs every node up front. Picture placeholders and `embedded` data-URI
+images stream; the `referenced` mode writes sidecar files, so it stays on the
+buffered `export_to_markdown_with_images` path. Use
+`convert_streaming_images(source, ImageMode::Embedded)` to pick the image mode.
+
+The CLI streams Markdown by default (`--no-stream` opts back into buffering;
+`--to json` and `--images referenced` always buffer).
+
 ## Testing
 
 All commands run from the `fleischwolf/` workspace root.
@@ -171,8 +205,13 @@ cargo run -p fleischwolf-cli -- --to json crates/fleischwolf/sample.html > out.j
 cargo run -p fleischwolf-cli -- --images embedded   document.pdf
 cargo run -p fleischwolf-cli -- --images referenced document.pdf > out.md
 
-# or via the example
+# stream Markdown to stdout page by page (the CLI's default; --no-stream to buffer)
+cargo run -p fleischwolf-cli -- document.pdf
+cargo run -p fleischwolf-cli -- --no-stream document.pdf
+
+# or via the examples
 cargo run -p fleischwolf --example convert -- crates/fleischwolf/sample.md
+cargo run -p fleischwolf --example stream  -- crates/fleischwolf/sample.md
 
 # score HTML output against the latest published docling (installed from PyPI)
 scripts/conformance.sh html

crates/fleischwolf-cli/src/main.rs

@@ -3,7 +3,7 @@
 //! A stand-in for `docling.cli.main`; the full Typer-style CLI (batch mode,
 //! pipeline options) is a later phase.
 //!
-//! Usage: fleischwolf [--strict] [--to md|json] [--images MODE] [--fetch-images] <input-file>
+//! Usage: fleischwolf [--strict] [--to md|json] [--images MODE] [--fetch-images] [--no-stream] <input-file>
 //!   --to md|json       output format (default: md). `json` emits docling-core's
 //!                      native DoclingDocument JSON (export_to_dict).
 //!   --images MODE      picture handling for Markdown (mirrors docling's
@@ -15,7 +15,12 @@
 //!                      the bytes. Off by default; fetches over the network.
 //!   --strict           cleaner, more conformant Markdown instead of byte-for-byte
 //!                      docling-legacy output (Markdown only).
+//!   --no-stream        build the whole document before printing Markdown instead
+//!                      of streaming it page by page. Streaming is the default for
+//!                      Markdown (placeholder/embedded images); JSON and referenced
+//!                      images always use the buffered path.
 
+use std::io::{self, Write};
 use std::path::Path;
 use std::process::ExitCode;
 
@@ -26,13 +31,15 @@ fn main() -> ExitCode {
     let mut to = "md".to_string();
     let mut images = "placeholder".to_string();
     let mut fetch_images = false;
+    let mut no_stream = false;
     let mut bench_warm: Option<usize> = None;
     let mut path: Option<String> = None;
     let mut args = std::env::args().skip(1);
     while let Some(arg) = args.next() {
         match arg.as_str() {
             "--strict" => strict = true,
             "--fetch-images" => fetch_images = true,
+            "--no-stream" => no_stream = true,
             "--to" => to = args.next().unwrap_or_default(),
             "--images" => images = args.next().unwrap_or_default(),
             // Hidden benchmarking aid: load the PDF/image pipeline once, then time
@@ -67,7 +74,7 @@ fn main() -> ExitCode {
     };
 
     let Some(path) = path else {
-        eprintln!("usage: fleischwolf [--strict] [--to md|json] [--images MODE] [--fetch-images] <input-file>");
+        eprintln!("usage: fleischwolf [--strict] [--to md|json] [--images MODE] [--fetch-images] [--no-stream] <input-file>");
         return ExitCode::from(2);
     };
 
@@ -97,11 +104,48 @@ fn main() -> ExitCode {
         };
     }
 
-    let document = match DocumentConverter::new()
+    let converter = DocumentConverter::new()
         .strict(strict)
-        .fetch_images(fetch_images)
-        .convert(source)
-    {
+        .fetch_images(fetch_images);
+
+    // Stream Markdown by default: print each chunk as the converter produces it
+    // (page by page for PDF). JSON needs the whole tree, and the referenced image
+    // mode writes sidecar files, so both keep the buffered path. `--no-stream` opts
+    // back into buffering for the streamable cases too.
+    let is_markdown = matches!(to.as_str(), "md" | "markdown");
+    if is_markdown && image_mode != ImageMode::Referenced && !no_stream {
+        let stream = match converter.convert_streaming_images(source, image_mode) {
+            Ok(s) => s,
+            Err(e) => {
+                eprintln!("error: {e}");
+                return ExitCode::FAILURE;
+            }
+        };
+        let stdout = io::stdout();
+        let mut out = io::BufWriter::new(stdout.lock());
+        for chunk in stream {
+            match chunk {
+                Ok(s) => {
+                    if let Err(e) = out.write_all(s.as_bytes()) {
+                        eprintln!("error: writing output: {e}");
+                        return ExitCode::FAILURE;
+                    }
+                }
+                Err(e) => {
+                    let _ = out.flush();
+                    eprintln!("error: {e}");
+                    return ExitCode::FAILURE;
+                }
+            }
+        }
+        if let Err(e) = out.flush() {
+            eprintln!("error: writing output: {e}");
+            return ExitCode::FAILURE;
+        }
+        return ExitCode::SUCCESS;
+    }
+
+    let document = match converter.convert(source) {
         Ok(result) => result.document,
         Err(e) => {
             eprintln!("error: {e}");

crates/fleischwolf-core/src/lib.rs

@@ -18,4 +18,4 @@ mod markdown;
 
 pub use document::{DoclingDocument, Node, PictureImage, Table};
 pub use labels::DocItemLabel;
-pub use markdown::ImageMode;
+pub use markdown::{ImageMode, MarkdownStreamer};

crates/fleischwolf-core/src/markdown.rs

@@ -99,6 +99,128 @@ fn apply_links(body: &str, links: &[(String, String)]) -> String {
     out
 }
 
+/// Like [`apply_links`] but over a single chunk, consuming from a shared queue so
+/// the same `[anchor](href)` rewriting can be applied incrementally as Markdown is
+/// streamed out. Each queued link is matched (in document order) against `chunk`
+/// and rewritten in place; a link whose anchor is not in this chunk is carried
+/// forward in the queue for a later chunk. Anchors are recovered in document
+/// order and a chunk is always a contiguous run of whole blocks, so this
+/// reproduces [`apply_links`]' single moving cursor: the link lands in whichever
+/// chunk contains its anchor, identically to the buffered path. (A link whose
+/// anchor never appears is carried to the end and dropped — the same no-op
+/// `apply_links` performs for an unlocatable anchor.)
+fn apply_links_chunk(chunk: &str, queue: &mut Vec<(String, String)>) -> String {
+    let mut out = chunk.to_string();
+    let mut cursor = 0usize;
+    let mut carried: Vec<(String, String)> = Vec::new();
+    for (anchor_raw, href) in std::mem::take(queue) {
+        let anchor = anchor_raw
+            .replace('&', "&amp;")
+            .replace('<', "&lt;")
+            .replace('>', "&gt;");
+        if anchor.is_empty() {
+            continue;
+        }
+        if let Some(rel) = out[cursor..].find(&anchor) {
+            let at = cursor + rel;
+            let replacement = format!("[{anchor}]({href})");
+            out.replace_range(at..at + anchor.len(), &replacement);
+            cursor = at + replacement.len();
+        } else {
+            // Not in this chunk; try again when its block is flushed.
+            carried.push((anchor_raw, href));
+        }
+    }
+    *queue = carried;
+    out
+}
+
+/// Incremental Markdown serializer: feed finalized, in-document-order batches of
+/// [`Node`]s and receive Markdown chunks whose concatenation is **byte-identical**
+/// to [`to_markdown_images`] over the same nodes. This is the streaming
+/// counterpart of the buffered serializer — used to emit a document's Markdown in
+/// chunks (e.g. page by page, as the parallel PDF pipeline finishes pages) instead
+/// of building the whole string up front.
+///
+/// Only [`ImageMode::Placeholder`] and [`ImageMode::Embedded`] are streamable:
+/// [`ImageMode::Referenced`] needs a side-channel for the image bytes, which only
+/// the buffered [`to_markdown_images`] provides.
+///
+/// Each [`push`](Self::push) must contain whole blocks in reading order: a caller
+/// must not split a run of list items across two pushes (the run would render as
+/// two separate lists). Finalized PDF page batches already satisfy this.
+pub struct MarkdownStreamer {
+    strict: bool,
+    images: ImageMode,
+    compact_tables: bool,
+    /// Whether any non-empty chunk has been emitted yet (drives `\n\n` joins and
+    /// the trailing newline).
+    emitted_any: bool,
+    /// Recovered links not yet placed (strict mode), consumed in document order.
+    links: Vec<(String, String)>,
+}
+
+impl MarkdownStreamer {
+    /// Create a streamer. `compact_tables` mirrors [`DoclingDocument::compact_tables`].
+    pub fn new(strict: bool, images: ImageMode, compact_tables: bool) -> Self {
+        debug_assert!(
+            images != ImageMode::Referenced,
+            "referenced image mode is not streamable; use to_markdown_images"
+        );
+        Self {
+            strict,
+            images,
+            compact_tables,
+            emitted_any: false,
+            links: Vec::new(),
+        }
+    }
+
+    /// Render one finalized batch of nodes (plus any links recovered from the same
+    /// span, in document order) into the next Markdown chunk. Returns an empty
+    /// string when the batch produces no output (e.g. empty tables/pictures), in
+    /// which case nothing should be written.
+    pub fn push(&mut self, nodes: &[Node], links: &[(String, String)]) -> String {
+        self.links.extend(links.iter().cloned());
+        let mut ctx = Ctx {
+            strict: self.strict,
+            compact_tables: self.compact_tables,
+            images: self.images,
+            // Referenced mode is rejected at construction, so the artifact sink is
+            // never touched.
+            artifacts_dir: String::new(),
+            artifacts: Vec::new(),
+            pic_index: 0,
+        };
+        let mut blocks: Vec<String> = Vec::new();
+        render(nodes, &mut blocks, &mut ctx);
+        if blocks.is_empty() {
+            return String::new();
+        }
+        let mut body = blocks.join("\n\n");
+        if self.strict && !self.links.is_empty() {
+            body = apply_links_chunk(&body, &mut self.links);
+        }
+        let chunk = if self.emitted_any {
+            format!("\n\n{body}")
+        } else {
+            body
+        };
+        self.emitted_any = true;
+        chunk
+    }
+
+    /// Emit the trailing newline that finishes the document (empty if no content
+    /// was produced). Call exactly once, after the final [`push`](Self::push).
+    pub fn finish(self) -> String {
+        if self.emitted_any {
+            "\n".to_string()
+        } else {
+            String::new()
+        }
+    }
+}
+
 /// In `strict` mode, rewrite inline text for readability rather than byte-for-byte
 /// docling fidelity: undo the legacy `\_` underscore escaping, and tighten stray
 /// spaces around punctuation (`[ 37 , 36 ]` → `[37, 36]`, `( x )` → `(x)`). This
@@ -476,6 +598,101 @@ mod tests {
         assert_eq!(doc.export_to_markdown_with(true), "# a_b\n\nx_y\n\n- i_j\n");
     }
 
+    /// Drive a document's nodes through [`MarkdownStreamer`] in the given page
+    /// splits and assert the concatenated chunks equal the buffered serializer.
+    fn assert_stream_matches(
+        doc: &DoclingDocument,
+        strict: bool,
+        images: ImageMode,
+        splits: &[usize],
+    ) {
+        let want = to_markdown_images(doc, strict, images, "artifacts").0;
+        let mut streamer = MarkdownStreamer::new(strict, images, doc.compact_tables);
+        let mut got = String::new();
+        let mut start = 0;
+        for &end in splits {
+            // Links only matter in strict mode; feed them all with the first batch
+            // that has content (document order is preserved by the queue).
+            let links = if start == 0 {
+                doc.links.as_slice()
+            } else {
+                &[]
+            };
+            got.push_str(&streamer.push(&doc.nodes[start..end], links));
+            start = end;
+        }
+        got.push_str(&streamer.push(
+            &doc.nodes[start..],
+            if start == 0 {
+                doc.links.as_slice()
+            } else {
+                &[]
+            },
+        ));
+        got.push_str(&streamer.finish());
+        assert_eq!(
+            got, want,
+            "streamed output diverged (splits={splits:?}, strict={strict})"
+        );
+    }
+
+    #[test]
+    fn streaming_is_byte_identical_to_buffered() {
+        let mut doc = DoclingDocument::new("d");
+        doc.add_heading(1, "Title");
+        doc.add_paragraph("First paragraph.");
+        doc.push(Node::ListItem {
+            ordered: false,
+            number: 1,
+            first_in_list: true,
+            text: "a".into(),
+            level: 0,
+        });
+        doc.push(Node::ListItem {
+            ordered: false,
+            number: 2,
+            first_in_list: false,
+            text: "b".into(),
+            level: 0,
+        });
+        doc.push(Node::Code {
+            language: Some("rust".into()),
+            text: "let x = 1;".into(),
+        });
+        doc.push(Node::Table(Table {
+            rows: vec![vec!["a".into(), "b".into()], vec!["1".into(), "2".into()]],
+        }));
+        doc.push(Node::Picture {
+            caption: Some("Fig 1".into()),
+            image: None,
+        });
+        doc.add_paragraph("Last paragraph.");
+
+        // A run of list items must never straddle a split, so try splits that fall
+        // on safe block boundaries (the streaming PDF assembler guarantees this).
+        for &strict in &[false, true] {
+            for &images in &[ImageMode::Placeholder, ImageMode::Embedded] {
+                for splits in [&[][..], &[1][..], &[2][..], &[4][..], &[1, 4, 6][..]] {
+                    assert_stream_matches(&doc, strict, images, splits);
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn streaming_applies_recovered_links_in_strict_mode() {
+        let mut doc = DoclingDocument::new("d");
+        doc.add_paragraph("See LinkedIn for details.");
+        doc.add_paragraph("And GitHub too.");
+        doc.links = vec![
+            ("LinkedIn".into(), "https://lnkd/".into()),
+            ("GitHub".into(), "https://gh/".into()),
+        ];
+        // The second anchor lives in the second block, so it must be carried across
+        // the page boundary and placed when that block streams out.
+        assert_stream_matches(&doc, true, ImageMode::Placeholder, &[1]);
+    }
+
     #[test]
     fn strict_tightens_punctuation_spacing_legacy_keeps_it() {
         let mut doc = DoclingDocument::new("t");