feat: add versioned NDJSON streaming output

CHANGELOG.md

@@ -19,6 +19,8 @@ and versions are tracked in the repo-root `VERSION` file.
   actionable installation hint when YAML configuration or output is selected.
 - Add an explicit `App.async_command()` adapter and `run_async()` helper for
   deterministic async callbacks without changing the synchronous core.
+- Add versioned NDJSON output and typed writer protocols for bounded,
+  flush-per-record machine output.
 
 ### Added
 

docs/json-contracts.md

@@ -45,6 +45,11 @@ ANSI escapes as a second stdout record.
 context, otherwise it is `null`. Unexpected failures intentionally expose only
 the generic message `Unexpected internal error.`; diagnostics stay in logs.
 
+For large or long-running record sets, use the `ndjson` output contract in
+[`output-contracts.md`](output-contracts.md). NDJSON is intentionally a stream
+of versioned records rather than a single success/error envelope; command
+errors and diagnostics still use the normal stderr and exit-code boundary.
+
 The lower-level `success_envelope()`, `error_envelope()`, `dumps_envelope()`,
 and `redact_json_value()` helpers are public for commands that need to publish
 their own structured `details` records. Secret-looking keys (`token`,

docs/output-contracts.md

@@ -1,8 +1,8 @@
 # Output contracts
 
-`base_cli.output.render_records()` supports `text`, `csv`, `tsv`, `yaml`, and
-`json` formats. Install `base-cli[yaml]` before selecting `yaml`; the other
-formats are available from the core package. The requested `text` format is presentation-aware: it renders
+`base_cli.output.render_records()` supports `text`, `csv`, `tsv`, `yaml`, `json`,
+and `ndjson` formats. Install `base-cli[yaml]` before selecting `yaml`; the
+other formats are available from the core package. The requested `text` format is presentation-aware: it renders
 a table on a TTY and tab-delimited rows when stdout is redirected or piped.
 
 Delimited output is intentionally automation-friendly:
@@ -14,6 +14,19 @@ Delimited output is intentionally automation-friendly:
 - values use the standard `csv` quoting rules, while ANSI escape sequences and
   other control characters are replaced with spaces.
 
+`ndjson` is the bounded machine-output format for large or long-running
+results. It consumes the input iterable once and writes one flushed JSON object
+per record without first building a list. Each line has this stable shape:
+
+```json
+{"schema_version":1,"schema":"base-cli.record","record":{"name":"base"}}
+```
+
+Use `base_cli.NdjsonWriter` when a consumer produces records incrementally.
+The `StructuredRecord` and `StructuredResultWriter` types describe the public
+producer boundary. Diagnostics remain on stderr; a consumer should not mix log
+lines into the NDJSON stream.
+
 Terminal tables use Unicode display-cell width rather than Python string length.
 Long cells are bounded by `max_cell_width` (80 by default), and the complete
 table is fitted to the detected terminal width (120 columns as a safe fallback)

lib/python/base_cli/__init__.py

@@ -124,8 +124,13 @@ def _resolve_version() -> str:
 )
 from .logging import configure_logger, log_critical, log_debug, log_error, log_info, log_warning
 from .output import (
+    NDJSON_SCHEMA,
+    NDJSON_SCHEMA_VERSION,
     PUBLIC_OUTPUT_FORMATS,
+    NdjsonWriter,
     OutputFormatError,
+    StructuredRecord,
+    StructuredResultWriter,
     is_terminal,
     output_format_choices,
     render_document,
@@ -205,6 +210,9 @@ def _resolve_version() -> str:
     "JSON_OUTPUT_SCHEMA",
     "JsonLogFormatter",
     "MAX_JSON_LOG_MESSAGE_LENGTH",
+    "NDJSON_SCHEMA",
+    "NDJSON_SCHEMA_VERSION",
+    "NdjsonWriter",
     "dumps_envelope",
     "dumps_record",
     "dumps_records",
@@ -237,6 +245,8 @@ def _resolve_version() -> str:
     "normalize_command_filters",
     "OutputFormatError",
     "PUBLIC_OUTPUT_FORMATS",
+    "StructuredRecord",
+    "StructuredResultWriter",
     "ProjectInfo",
     "ProjectDiscovery",
     "RECORD_SCHEMAS",

lib/python/base_cli/output.py

@@ -10,12 +10,15 @@
 import sys
 import unicodedata
 from collections.abc import Iterable, Mapping, Sequence
-from typing import Any, TextIO
+from dataclasses import dataclass
+from typing import Any, Protocol, TextIO, TypeAlias
 
 from ._dependencies import require_yaml
 from .integrations import try_render_rich_table
 
-PUBLIC_OUTPUT_FORMATS = ("text", "csv", "tsv", "yaml", "json")
+PUBLIC_OUTPUT_FORMATS = ("text", "csv", "tsv", "yaml", "json", "ndjson")
+NDJSON_SCHEMA = "base-cli.record"
+NDJSON_SCHEMA_VERSION = 1
 _ANSI_ESCAPE_RE = re.compile(r"\x1b(?:\[[0-?]*[ -/]*[@-~]|\][^\x07]*(?:\x07|\x1b\\))")
 _DEFAULT_TERMINAL_WIDTH = 120
 _DEFAULT_MAX_CELL_WIDTH = 80
@@ -25,6 +28,45 @@ class OutputFormatError(ValueError):
     """Raised when a public output format is not supported."""
 
 
+StructuredRecord: TypeAlias = Mapping[str, Any]
+
+
+class StructuredResultWriter(Protocol):
+    """Typed sink for one structured result at a time."""
+
+    def write(self, record: StructuredRecord) -> None:
+        """Write one record without retaining the stream in memory."""
+
+
+@dataclass
+class NdjsonWriter:
+    """Write versioned structured records as newline-delimited JSON."""
+
+    stream: TextIO
+    schema: str = NDJSON_SCHEMA
+    schema_version: int = NDJSON_SCHEMA_VERSION
+
+    def __post_init__(self) -> None:
+        if not self.schema.strip():
+            raise ValueError("schema must be a non-empty string")
+        if self.schema_version < 1:
+            raise ValueError("schema_version must be greater than 0")
+
+    def write(self, record: StructuredRecord) -> None:
+        """Write one record and flush it for pipeline consumers."""
+
+        if not isinstance(record, Mapping):
+            raise TypeError(f"structured records must be mappings, got {type(record).__name__}")
+        payload = {
+            "schema_version": self.schema_version,
+            "schema": self.schema,
+            "record": dict(record),
+        }
+        self.stream.write(json.dumps(payload, separators=(",", ":")))
+        self.stream.write("\n")
+        self.stream.flush()
+
+
 def output_format_choices() -> str:
     """Return the public choices in help/error-message order."""
 
@@ -80,9 +122,9 @@ def render_records(
 
     The returned format name is also written to *stream* when supplied (or
     stdout when omitted). JSON and YAML retain the mapping shape supplied by
-    the caller; delimited formats stream one row at a time, use the explicit
-    ``columns`` order, sanitize terminal control sequences, and never emit a
-    header or footer. ``minimum_widths`` applies only to terminal table
+    the caller; NDJSON and delimited formats stream one row at a time, use the
+    explicit ``columns`` order where applicable, sanitize terminal control
+    sequences, and never emit a header or footer. ``minimum_widths`` applies only to terminal table
     columns. Terminal cells use Unicode display-cell widths and are bounded by
     ``terminal_width`` and ``max_cell_width`` with deterministic ellipsis
     truncation. ``rich=True`` opts terminal text into the optional Rich
@@ -99,6 +141,12 @@ def render_records(
             writer.writerow([_delimited_value(record.get(key)) for _header, key in columns])
         return resolved
 
+    if resolved == "ndjson":
+        ndjson_writer = NdjsonWriter(target)
+        for record in records:
+            ndjson_writer.write(record)
+        return resolved
+
     record_list = [dict(record) for record in records]
     if resolved == "json":
         target.write(json.dumps(record_list, separators=(",", ":")))

tests/test_api_stability.py

@@ -63,6 +63,9 @@
         "JSON_OUTPUT_SCHEMA",
         "JsonLogFormatter",
         "MAX_JSON_LOG_MESSAGE_LENGTH",
+        "NDJSON_SCHEMA",
+        "NDJSON_SCHEMA_VERSION",
+        "NdjsonWriter",
         "dumps_envelope",
         "dumps_record",
         "dumps_records",
@@ -94,6 +97,8 @@
         "normalize_command_filters",
         "OutputFormatError",
         "PUBLIC_OUTPUT_FORMATS",
+        "StructuredRecord",
+        "StructuredResultWriter",
         "ProjectInfo",
         "ProjectDiscovery",
         "RECORD_SCHEMAS",

tests/test_output.py

@@ -4,7 +4,15 @@
 import json
 import unittest
 
-from base_cli.output import OutputFormatError, render_document, render_records, resolve_output_format
+from base_cli.output import (
+    NDJSON_SCHEMA,
+    NDJSON_SCHEMA_VERSION,
+    NdjsonWriter,
+    OutputFormatError,
+    render_document,
+    render_records,
+    resolve_output_format,
+)
 
 
 class _Stream(io.StringIO):
@@ -32,6 +40,16 @@ def write(self, value: str) -> int:
         self.rows += value.count("\n")
         return len(value)
 
+
+class _FlushTrackingSink(io.StringIO):
+    def __init__(self) -> None:
+        super().__init__()
+        self.flushes = 0
+
+    def flush(self) -> None:
+        self.flushes += 1
+        super().flush()
+
     def isatty(self) -> bool:
         return False
 
@@ -219,6 +237,31 @@ def test_json_preserves_record_shape(self) -> None:
 
         self.assertEqual(json.loads(stream.getvalue()), list(RECORDS))
 
+    def test_ndjson_streams_versioned_records_and_flushes_each_row(self) -> None:
+        stream = _FlushTrackingSink()
+        render_records(
+            (record for record in RECORDS),
+            requested_format="ndjson",
+            columns=(),
+            stream=stream,
+        )
+
+        lines = [json.loads(line) for line in stream.getvalue().splitlines()]
+        self.assertEqual(len(lines), 2)
+        self.assertEqual(
+            lines[0],
+            {"schema_version": NDJSON_SCHEMA_VERSION, "schema": NDJSON_SCHEMA, "record": dict(RECORDS[0])},
+        )
+        self.assertEqual(stream.flushes, 2)
+
+    def test_ndjson_writer_rejects_invalid_schema_and_records(self) -> None:
+        stream = io.StringIO()
+        with self.assertRaisesRegex(ValueError, "schema_version"):
+            NdjsonWriter(stream, schema_version=0)
+        writer = NdjsonWriter(stream)
+        with self.assertRaisesRegex(TypeError, "mappings"):
+            writer.write([("name", "invalid")])  # type: ignore[arg-type]
+
     def test_yaml_preserves_record_shape(self) -> None:
         import yaml
 
@@ -229,7 +272,7 @@ def test_yaml_preserves_record_shape(self) -> None:
         self.assertEqual(yaml.safe_load(stream.getvalue()), list(RECORDS))
 
     def test_resolve_rejects_unknown_format(self) -> None:
-        with self.assertRaisesRegex(OutputFormatError, "Expected one of: text, csv, tsv, yaml, json"):
+        with self.assertRaisesRegex(OutputFormatError, "Expected one of: text, csv, tsv, yaml, json, ndjson"):
             resolve_output_format("xml")
 
     def test_empty_tty_result_keeps_footer(self) -> None: