@@ -59,6 +59,10 @@ still requires that owner's qualified source provider. See
 - Provider delivery and readback receipts must match the approved revision and
   digest; the lifecycle helper performs no external write.
 - Layout acceptance never implies content approval or publishing authority.
+- JSON read from stdin (`-`) is refused before parsing above 16 MiB
+  (16,777,216 bytes); `--max-input-bytes N` sets a different positive limit on
+  the subcommand. The limit counts input bytes, not process memory, and does
+  not apply to JSON files.
 - Built-in cover pages require `0.82–0.94` meaningful vertical density; the
   template field is `density.role_overrides.cover`.
 - Built-in `page_sequence` defaults require a cover first, make it the density
 
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import argparse
+import codecs
 import json
 import sys
 from collections.abc import Callable
@@ -43,6 +44,49 @@
     render_content_ops_walkthrough_artifact_markdown,
 )
 
+MAX_INPUT_BYTES = 16 * 1024 * 1024
+
+
+def _input_byte_limit(text: str) -> int:
+    # An explicit, positive byte count; there is no unbounded setting.
+    if not text.isdecimal() or int(text) < 1:
+        raise argparse.ArgumentTypeError(f"must be a positive whole number of bytes, got {text!r}")
+    return int(text)
+
+
+def _add_input_limit(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        "--max-input-bytes",
+        type=_input_byte_limit,
+        default=MAX_INPUT_BYTES,
+        help=(
+            "Refuse JSON read from stdin ('-') larger than this many bytes before "
+            f"parsing (default {MAX_INPUT_BYTES}). Bounds input size, not process memory."
+        ),
+    )
+
+
+def _read_stdin_text(limit: int, label: str) -> str:
+    # Read through the text stream so stdin keeps its decoding, but charge each
+    # chunk by its encoded size: one character can take up to four bytes. One
+    # incremental encoder spans all chunks, so a BOM is charged only once.
+    encoding = getattr(sys.stdin, "encoding", None) or "utf-8"
+    errors = getattr(sys.stdin, "errors", None) or "surrogatepass"
+    encoder = codecs.getincrementalencoder(encoding)(errors)
+    chunks: list[str] = []
+    consumed = 0
+    while consumed <= limit:
+        chunk = sys.stdin.read(max(1, (limit + 1 - consumed) // 4))
+        if not chunk:
+            consumed += len(encoder.encode("", final=True))
+            break
+        chunks.append(chunk)
+        consumed += len(encoder.encode(chunk))
+    if consumed > limit:
+        raise ValueError(f"{label} exceeds the {limit}-byte limit")
+    return "".join(chunks)
+
+
 PrintPayload = Callable[
     [dict[str, object], str, Callable[[dict[str, object]], str]],
     None,
@@ -51,9 +95,9 @@
 AddFormat = Callable[[argparse.ArgumentParser], None]
 
 
-def _load_json_object(path_text: str) -> dict[str, Any]:
+def _load_json_object(path_text: str, limit: int) -> dict[str, Any]:
     if path_text == "-":
-        payload = json.loads(sys.stdin.read())
+        payload = json.loads(_read_stdin_text(limit, "content-ops stdin JSON input"))
     else:
         payload = json.loads(Path(path_text).expanduser().read_text(encoding="utf-8"))
     if not isinstance(payload, dict):
@@ -109,6 +153,7 @@ def register_content_ops_commands(
     reference_parser.add_argument("--observed-at")
     reference_parser.add_argument("--output-json", help="Create a new mode-0600 private result artifact; never overwrite an existing catalog.")
     add_subcommand_format(reference_parser)
+    _add_input_limit(reference_parser)
     preview_parser = content_ops_sub.add_parser(
         "preview",
         help="Preview metadata-only connector trials and content-ops projection.",
@@ -234,6 +279,7 @@ def register_content_ops_commands(
         ),
     )
     add_subcommand_format(aggregate_parser)
+    _add_input_limit(aggregate_parser)
     aggregate_parser.add_argument(
         "--public-packet-json",
         action="append",
@@ -414,6 +460,7 @@ def register_content_ops_commands(
         ),
     )
     add_subcommand_format(item_transition_parser)
+    _add_input_limit(item_transition_parser)
     item_transition_parser.add_argument(
         "--item-json",
         required=True,
@@ -432,6 +479,7 @@ def register_content_ops_commands(
         ),
     )
     add_subcommand_format(queue_parser)
+    _add_input_limit(queue_parser)
     queue_parser.add_argument(
         "--item-json",
         action="append",
@@ -492,6 +540,7 @@ def register_content_ops_commands(
         help="Check rendered-page measurements against a typed layout plan.",
     )
     add_subcommand_format(layout_check_parser)
+    _add_input_limit(layout_check_parser)
     layout_check_parser.add_argument(
         "--plan-json",
         required=True,
@@ -520,10 +569,10 @@ def handle_content_ops_command(
                     raise ValueError("inventory requires a catalog file, --goal-id, --store-id and --observed-at")
                 payload = inspect_reference_materials(library_path=Path(args.library_json).expanduser(), goal_id=args.goal_id, store_id=args.store_id, observed_at=args.observed_at)
             else:
-                request = _load_json_object(args.input_json) if args.input_json else {}
+                request = _load_json_object(args.input_json, args.max_input_bytes) if args.input_json else {}
                 if args.operation in {"capture", "draft"} and not args.input_json:
                     raise ValueError("capture/draft require --input-json")
-                request["library"] = _load_json_object(args.library_json)
+                request["library"] = _load_json_object(args.library_json, args.max_input_bytes)
                 if args.operation == "search":
                     request.update(query=args.query, structure=args.structure)
                 payload = content_reference_operation(args.operation, request)
@@ -565,10 +614,10 @@ def handle_content_ops_command(
         elif args.content_ops_command == "aggregate-packets":
             payload = build_content_ops_packet_aggregation_packet(
                 public_handle_packets=[
-                    _load_json_object(path) for path in args.public_packet_json
+                    _load_json_object(path, args.max_input_bytes) for path in args.public_packet_json
                 ],
                 private_connector_gate_packets=[
-                    _load_json_object(path) for path in args.private_gate_packet_json
+                    _load_json_object(path, args.max_input_bytes) for path in args.private_gate_packet_json
                 ],
                 surface_id=args.surface_id,
                 generated_at=args.generated_at,
@@ -615,14 +664,14 @@ def handle_content_ops_command(
             renderer = render_content_ops_item_packet_markdown
         elif args.content_ops_command == "item-transition":
             payload = apply_content_ops_item_event(
-                _load_json_object(args.item_json),
-                _load_json_object(args.event_json),
+                _load_json_object(args.item_json, args.max_input_bytes),
+                _load_json_object(args.event_json, args.max_input_bytes),
             )
             renderer = render_content_ops_item_packet_markdown
         elif args.content_ops_command == "queue-status":
             payload = build_content_ops_queue_status_packet(
                 items=[
-                    _load_json_object(path) for path in args.item_json
+                    _load_json_object(path, args.max_input_bytes) for path in args.item_json
                 ],
                 queue_id=args.queue_id,
                 generated_at=args.generated_at,
@@ -646,8 +695,8 @@ def handle_content_ops_command(
             renderer = render_layout_packet_markdown
         elif args.content_ops_command == "layout-check":
             payload = check_layout_packet(
-                _load_json_object(args.plan_json),
-                _load_json_object(args.measurement_json),
+                _load_json_object(args.plan_json, args.max_input_bytes),
+                _load_json_object(args.measurement_json, args.max_input_bytes),
             )
             renderer = render_layout_packet_markdown
         else:
 
@@ -203,6 +203,12 @@ The CLI streams the append-only log and applies the 4,096-row capacity limit
 only after goal, segment-window, and relevant-kind filtering. Malformed durable
 rows and an oversized relevant window fail closed.
 
+Every `--*-json` input, from a file or `-` for stdin, is refused before parsing
+when it exceeds 16 MiB (16,777,216 bytes). Pass `--max-input-bytes N` on the
+subcommand to set a different positive limit for a legitimately larger or
+deliberately smaller request; there is no unlimited setting. The limit counts
+input bytes, not total process memory. Request files stay strict UTF-8.
+
 ### Automatic post-writeback intent
 
 Automatic milestone evaluation is default-off. A project may opt the built-in
 
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import argparse
+import codecs
 import json
 import os
 import sys
@@ -40,6 +41,49 @@
 )
 from ...paths import resolve_runtime_root
 
+MAX_INPUT_BYTES = 16 * 1024 * 1024
+
+
+def _input_byte_limit(text: str) -> int:
+    # An explicit, positive byte count; there is no unbounded setting.
+    if not text.isdecimal() or int(text) < 1:
+        raise argparse.ArgumentTypeError(f"must be a positive whole number of bytes, got {text!r}")
+    return int(text)
+
+
+def _add_input_limit(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        "--max-input-bytes",
+        type=_input_byte_limit,
+        default=MAX_INPUT_BYTES,
+        help=(
+            "Refuse a JSON input file or stdin larger than this many bytes before "
+            f"parsing (default {MAX_INPUT_BYTES}). Bounds input size, not process memory."
+        ),
+    )
+
+
+def _read_stdin_text(limit: int, label: str) -> str:
+    # Read through the text stream so stdin keeps its decoding, but charge each
+    # chunk by its encoded size: one character can take up to four bytes. One
+    # incremental encoder spans all chunks, so a BOM is charged only once.
+    encoding = getattr(sys.stdin, "encoding", None) or "utf-8"
+    errors = getattr(sys.stdin, "errors", None) or "surrogatepass"
+    encoder = codecs.getincrementalencoder(encoding)(errors)
+    chunks: list[str] = []
+    consumed = 0
+    while consumed <= limit:
+        chunk = sys.stdin.read(max(1, (limit + 1 - consumed) // 4))
+        if not chunk:
+            consumed += len(encoder.encode("", final=True))
+            break
+        chunks.append(chunk)
+        consumed += len(encoder.encode(chunk))
+    if consumed > limit:
+        raise ValueError(f"{label} exceeds the {limit}-byte limit")
+    return "".join(chunks)
+
+
 PrintPayload = Callable[
     [dict[str, object], str, Callable[[dict[str, object]], str]],
     None,
@@ -59,11 +103,17 @@ def __call__(
 ProviderCommandHandler = Callable[..., int | None]
 
 
-def _load_json_object(path_text: str) -> dict[str, Any]:
+def _load_json_object(path_text: str, limit: int) -> dict[str, Any]:
+    label = "periodic-report JSON input"
     if path_text == "-":
-        payload = json.loads(sys.stdin.read())
+        text = _read_stdin_text(limit, label)
     else:
-        payload = json.loads(Path(path_text).expanduser().read_text(encoding="utf-8"))
+        with Path(path_text).expanduser().open("rb") as source:
+            raw = source.read(limit + 1)
+        if len(raw) > limit:
+            raise ValueError(f"{label} exceeds the {limit}-byte limit")
+        text = raw.decode("utf-8")
+    payload = json.loads(text)
     if not isinstance(payload, dict):
         raise ValueError(f"{path_text} must contain a JSON object")
     return payload
@@ -85,6 +135,7 @@ def register_periodic_report_commands(
         help="Normalize one periodic_report_v0 attempt without provider effects.",
     )
     add_subcommand_format(compose)
+    _add_input_limit(compose)
     compose.add_argument(
         "--request-json",
         required=True,
@@ -95,6 +146,7 @@ def register_periodic_report_commands(
         help="Evaluate cadence and material progress triggers without effects.",
     )
     add_subcommand_format(evaluate)
+    _add_input_limit(evaluate)
     evaluate.add_argument(
         "--request-json",
         required=True,
@@ -105,6 +157,7 @@ def register_periodic_report_commands(
         help="Promote durable rollout events into a periodic-report trigger.",
     )
     add_subcommand_format(evaluate_runtime)
+    _add_input_limit(evaluate_runtime)
     evaluate_runtime.add_argument(
         "--request-json",
         required=True,
@@ -153,6 +206,7 @@ def register_periodic_report_commands(
         help="Preview or apply the runtime-root machine periodic-report policy.",
     )
     add_subcommand_format(configure_machine_defaults)
+    _add_input_limit(configure_machine_defaults)
     configure_machine_defaults.add_argument(
         "--config-json",
         required=True,
@@ -187,6 +241,7 @@ def register_periodic_report_commands(
         ),
     )
     add_subcommand_format(plan_goal_delivery)
+    _add_input_limit(plan_goal_delivery)
     plan_goal_delivery.add_argument(
         "--request-json",
         required=True,
@@ -202,6 +257,7 @@ def register_periodic_report_commands(
         help="Inspect a built-in preset or project profile without provider effects.",
     )
     add_subcommand_format(inspect_profile)
+    _add_input_limit(inspect_profile)
     profile_source = inspect_profile.add_mutually_exclusive_group(required=True)
     profile_source.add_argument(
         "--profile-json",
@@ -220,6 +276,7 @@ def register_periodic_report_commands(
         ),
     )
     add_subcommand_format(archive)
+    _add_input_limit(archive)
     archive.add_argument(
         "--request-json",
         required=True,
@@ -465,22 +522,22 @@ def handle_periodic_report_command(
     try:
         if args.periodic_report_command == "archive-openviking":
             payload = _archive_openviking(
-                _load_json_object(args.request_json),
+                _load_json_object(args.request_json, args.max_input_bytes),
                 args,
             )
         elif args.periodic_report_command == "inspect-profile":
             payload = (
                 build_periodic_report_preset_activation(args.preset)
                 if args.preset
                 else build_periodic_report_activation(
-                    _load_json_object(args.profile_json)
+                    _load_json_object(args.profile_json, args.max_input_bytes)
                 )
             )
         elif args.periodic_report_command == "evaluate-trigger":
-            request = _load_json_object(args.request_json)
+            request = _load_json_object(args.request_json, args.max_input_bytes)
             payload = build_periodic_report_trigger_decision(request)
         elif args.periodic_report_command == "evaluate-runtime-trigger":
-            request = _load_json_object(args.request_json)
+            request = _load_json_object(args.request_json, args.max_input_bytes)
             payload = build_periodic_report_runtime_trigger_decision(
                 request,
                 rollout_events=iter_rollout_events(
@@ -550,7 +607,7 @@ def handle_periodic_report_command(
             if args.periodic_report_command == "configure-machine-defaults":
                 payload = configure_periodic_report_machine_defaults(
                     runtime_root=runtime_root,
-                    machine_defaults=_load_json_object(args.config_json),
+                    machine_defaults=_load_json_object(args.config_json, args.max_input_bytes),
                     execute=bool(args.execute),
                     expected_plan_revision=args.expected_plan_revision,
                 )
@@ -569,11 +626,11 @@ def handle_periodic_report_command(
                 registry, runtime_root_arg, registry_path=registry_path
             )
             payload = build_goal_periodic_report_delivery_plan(
-                _load_json_object(args.request_json),
+                _load_json_object(args.request_json, args.max_input_bytes),
                 machine_defaults=read_periodic_report_machine_defaults(runtime_root),
             )
         else:
-            request = _load_json_object(args.request_json)
+            request = _load_json_object(args.request_json, args.max_input_bytes)
             payload = build_periodic_report_run(request)
     except PeriodicReportSubscriptionConfigurationError as exc:
         payload = {