sqlite_graphrag/output/
stream.rs

1//! NDJSON streaming: one self-contained JSON record per line.
2
3use super::sink;
4use crate::errors::AppError;
5use serde::Serialize;
6
7/// Writes compact JSON to stdout, silently ignoring serialization and I/O errors.
8/// Designed for NDJSON streaming where partial output is acceptable.
9///
10/// GAP-SG-142: NDJSON deliberately bypasses [`crate::agent_surface`]. The
11/// stream contract is one record per line and the shaping surface is defined
12/// over a complete envelope; filtering or capping a stream line by line would
13/// change what "one record" means for every consumer already parsing it.
14///
15/// The exclusion is a decision, not an oversight — do not "fix" it wholesale.
16/// The flags split into two groups:
17///
18/// * **Set-wide, never applicable here:** `--max-items`, `--sort` and
19///   `--dedupe-by` need the complete result set before they can decide what to
20///   emit. A stream has no complete set by construction.
21/// * **Per-line, safe to add later:** `--select` and `--truncate-content` are
22///   stateless per record.
23///
24/// GAP-SG-215 took the contract decision that second bullet asked for, and
25/// [`emit_stream_record`] / [`emit_stream_trailer`] below are it. `--select` and
26/// `--truncate-content` act per record; everything else is refused before the
27/// first byte; the summary line carries the one `agent_surface` record for the
28/// whole stream. `--max-output-bytes` as a RUNNING budget across the stream is
29/// still open, and still needs its own decision about what a consumer sees when
30/// the budget is exhausted mid-stream — truncated line, terminator record, or
31/// silent stop. Today it is refused rather than half-applied.
32///
33/// This function keeps its unshaped, infallible shape for the emitters that
34/// genuinely want neither: `enrich` progress events and `--print-schema`
35/// listings, which are diagnostics rather than a record stream with a trailer.
36#[inline]
37pub fn emit_json_line<T: Serialize>(value: &T) {
38    if let Ok(json) = serde_json::to_string(value) {
39        sink::write_line_lossy(json.as_bytes());
40    }
41}
42
43/// Writes one record of a stream, with the per-record knobs applied.
44///
45/// GAP-SG-215. Deliberately does NOT go through `super::envelope::render`: that
46/// path exists to shape and annotate ONE complete envelope, and reaching it once
47/// per line is the defect this replaces. A record line leaves here carrying the
48/// record and nothing else.
49///
50/// The command must have called [`crate::agent_surface::stream::open`] first, so
51/// that an unusable knob has already been refused. Emitting without opening
52/// shapes nothing rather than misbehaving, and `tests/stream_contract_gate.rs`
53/// is what keeps that from being a quiet way to lose the surface.
54///
55/// # Errors
56/// Returns `Err` when serialization fails or on a non-`BrokenPipe` I/O error.
57#[inline]
58pub fn emit_stream_record<T: Serialize>(value: &T) -> Result<(), AppError> {
59    let surface = crate::agent_surface::get();
60    if surface.select.is_empty() && surface.truncate_content == 0 {
61        // Nothing to apply, so the record never becomes a `Value` at all and the
62        // bytes are exactly what the command's own `Serialize` impl produces.
63        // This is the path an unflagged `export` takes for every one of its
64        // lines, which is why it is worth keeping free of the round-trip.
65        return sink::write_line(serde_json::to_string(value)?.as_bytes());
66    }
67    let state = crate::agent_surface::stream::get();
68    let shaped = crate::agent_surface::stream::shape_record_with(
69        state,
70        surface,
71        serde_json::to_value(value)?,
72    );
73    sink::write_line(serde_json::to_string(&shaped)?.as_bytes())
74}
75
76/// Writes the line that ends a stream, carrying its one `agent_surface` record.
77///
78/// GAP-SG-215. Never shaped: the trailer describes the stream, so a `--select`
79/// aimed at the records must neither fail on it nor rewrite it. Both were
80/// measured — `--select name export` exited 2 on this line, and
81/// `--select namespace export` silently deleted `summary: true` from it.
82///
83/// # Errors
84/// Returns `Err` when serialization fails or on a non-`BrokenPipe` I/O error.
85#[inline]
86pub fn emit_stream_trailer<T: Serialize>(value: &T) -> Result<(), AppError> {
87    let surface = crate::agent_surface::get();
88    let ceiling = crate::agent_surface::universe::get();
89    let shaped = crate::agent_surface::stream::trailer_with(
90        crate::agent_surface::stream::get(),
91        surface,
92        crate::agent_surface::target::record(surface, ceiling),
93        serde_json::to_value(value)?,
94    );
95    sink::write_line(serde_json::to_string(&shaped)?.as_bytes())
96}
sqlite_graphrag/output/stream.rs

sqlite_graphrag/output/
stream.rs
