fix: bound lifecycle JSON capture

docs/json-contracts.md

@@ -21,9 +21,14 @@ Click error rendering and command stdout behavior, is unchanged.
 Capture is activated only when JSON mode is selected, including an environment
 variable or Click `default_map`. Human and NDJSON invocations write directly to
 the caller's stdout, preserving progress visibility and flush behavior. JSON
-capture uses a 1 MiB in-memory spool and transparently rolls larger output to a
-temporary file; the complete captured text remains available in the v1 envelope
-and the temporary file is removed when the invocation ends.
+capture allows at most 8 MiB of UTF-8 command stdout. It keeps the first 1 MiB
+in memory and rolls the remainder to a temporary file, so both temporary-disk
+use and finalization memory remain bounded. The temporary file is removed when
+the invocation ends.
+
+If a command exceeds the limit, base-cli emits one `base-cli.error` envelope
+with `code: "capture_limit"` and exit code `1`; it never silently truncates
+the captured text. Use the NDJSON contract for larger record sets.
 
 ## Output and errors
 
@@ -43,7 +48,7 @@ Both envelopes use `schema_version: 1` and stable fields:
 
 Failures use `schema: "base-cli.error"`, `type: "error"`, and a deterministic
 `code` derived from the lifecycle outcome (`usage_error`, `click_error`,
-`aborted`, `interrupted`, `unexpected_error`, and so on). `details` always
+`capture_limit`, `aborted`, `interrupted`, `unexpected_error`, and so on). `details` always
 contains the numeric `exit_code` and captured command stdout. A command's
 human output is represented as a JSON string, so it cannot introduce prose or
 ANSI escapes as a second stdout record.

lib/python/base_cli/_run.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import inspect
+import io
 import os
 import sys
 import tempfile
@@ -30,7 +31,66 @@
 from .lifecycle_options import LifecycleOption, LifecycleOptions
 from .redaction import option_aliases_from_decls
 
-_MAX_JSON_CAPTURE_BYTES = 1_048_576
+_MAX_JSON_CAPTURE_BYTES = 8 * 1_048_576
+
+
+class JsonCaptureLimitError(RuntimeError):
+    """Raised when a JSON invocation exceeds its bounded stdout contract."""
+
+
+class _BoundedJsonCapture(io.TextIOBase):
+    """Text stream that bounds UTF-8 output before it reaches the spool."""
+
+    encoding = "utf-8"
+    errors = "strict"
+
+    def __init__(self, limit_bytes: int) -> None:
+        super().__init__()
+        self._limit_bytes = limit_bytes
+        self._bytes_written = 0
+        self._stream = cast(
+            TextIO,
+            tempfile.SpooledTemporaryFile(
+                max_size=min(limit_bytes, 1_048_576),
+                mode="w+",
+                encoding="utf-8",
+                newline="",
+            ),
+        )
+
+    def write(self, value: str) -> int:
+        encoded_size = len(value.encode("utf-8"))
+        if self._bytes_written + encoded_size > self._limit_bytes:
+            raise JsonCaptureLimitError(
+                f"JSON stdout exceeded the {_format_bytes(self._limit_bytes)} limit; use NDJSON for large record sets."
+            )
+        written = self._stream.write(value)
+        self._bytes_written += encoded_size
+        return written
+
+    def flush(self) -> None:
+        self._stream.flush()
+
+    def read(self, size: int | None = -1) -> str:
+        return self._stream.read(-1 if size is None else size)
+
+    def seek(self, offset: int, whence: int = io.SEEK_SET) -> int:
+        return self._stream.seek(offset, whence)
+
+    def tell(self) -> int:
+        return self._stream.tell()
+
+    def close(self) -> None:
+        try:
+            super().close()
+        finally:
+            self._stream.close()
+
+
+def _format_bytes(value: int) -> str:
+    if value % 1_048_576 == 0:
+        return f"{value // 1_048_576} MiB"
+    return f"{value} bytes"
 
 
 def run_app(
@@ -142,6 +202,12 @@ def run_app(
             if exc.code is not None and not isinstance(exc.code, int):
                 print(str(exc.code), file=sys.stderr)
             return system_exit_code(exc)
+        except JsonCaptureLimitError as exc:
+            if state.json_output:
+                outcome = InvocationOutcome("capture_limit", "error", ExitCode.FAILURE)
+                _emit_json_error(state, outcome, str(exc), output_capture)
+                return outcome.exit_code
+            raise
         except Exception as exc:
             if reraise_unexpected:
                 raise
@@ -234,15 +300,7 @@ def _command_default_map(command: Any) -> Mapping[str, Any] | None:
 
 
 def _new_json_capture() -> TextIO:
-    return cast(
-        TextIO,
-        tempfile.SpooledTemporaryFile(
-            max_size=_MAX_JSON_CAPTURE_BYTES,
-            mode="w+",
-            encoding="utf-8",
-            newline="",
-        ),
-    )
+    return cast(TextIO, _BoundedJsonCapture(_MAX_JSON_CAPTURE_BYTES))
 
 
 def _explicit_lifecycle_value(

tests/test_json_contracts.py

@@ -10,7 +10,7 @@
 from unittest import mock
 
 import base_cli
-from base_cli._run import _json_requested
+from base_cli._run import JsonCaptureLimitError, _BoundedJsonCapture, _json_requested
 from base_cli.json_contracts import MAX_JSON_LOG_MESSAGE_LENGTH
 
 
@@ -339,6 +339,40 @@ def main(ctx: base_cli.Context) -> None:
         self.assertEqual(result.stdout, "")
         self.assertIn("No such option", result.stderr or result.output)
 
+    def test_json_capture_rejects_overflow_without_partial_writes(self) -> None:
+        capture = _BoundedJsonCapture(4)
+        try:
+            self.assertEqual(capture.write("éé"), 2)
+            with self.assertRaises(JsonCaptureLimitError):
+                capture.write("x")
+            capture.seek(0)
+            self.assertEqual(capture.read(), "éé")
+        finally:
+            capture.close()
+
+    def test_json_capture_overflow_is_one_machine_readable_error(self) -> None:
+        app = base_cli.App(
+            name="json-capture-limit",
+            log_to_file=False,
+            lifecycle_options=self._lifecycle_options(),
+        )
+
+        @app.command()
+        def main(ctx: base_cli.Context) -> None:
+            del ctx
+            print("x" * 17, end="")
+
+        with tempfile.TemporaryDirectory() as home:
+            with mock.patch("base_cli._run._MAX_JSON_CAPTURE_BYTES", 16):
+                result = base_cli.testing.invoke(app, ["--json"], home=Path(home))
+
+        self.assertEqual(result.exit_code, 1)
+        envelope = json.loads(result.stdout)
+        self.assertEqual(envelope["schema"], "base-cli.error")
+        self.assertEqual(envelope["code"], "capture_limit")
+        self.assertEqual(envelope["details"]["stdout"], "")
+        self.assertIn("16 bytes", envelope["message"])
+
     def test_json_mode_captures_default_map_values(self) -> None:
         app = base_cli.App(
             name="json-default-map",