feat(setup): refuse an output_format_spec at or over 4096 characters

docs/changelog/1.0.2.md

@@ -55,6 +55,12 @@ the model self-corrects from (never a raised traceback).
   numeric/string/array/object constraints. Plus `reject_control_chars` (C0 control chars in strings) and
   `reject_unsafe_keys` (recursively rejects content keys carrying non-BMP/control characters that storage would silently
   drop).
+- **`SetupContentValidator.reject_oversized_output_format_spec`** — refuses a content `output_format_spec` of 4096
+  characters or more (the write would succeed and leave the setup unusable, with nothing pointing back at the field).
+  Recursive through nested objects and arrays; non-string values are left to the schema check. Unlike `validate`, it
+  needs no module context or schema fetch, so it is called unconditionally from `create_setup` **and** `update_setup`
+  on both `DefaultSetup` and `GrpcSetup` — the only choke point every write passes through — and raises `ValueError`
+  before any RPC.
 
 ### Supporting Agno toolkits & tool loading
 

src/digitalkin/services/setup/default_setup.py

@@ -17,6 +17,7 @@
     SetupVersionData,
     SetupVersionPage,
 )
+from digitalkin.utils.setup_content_validator import SetupContentValidator
 
 
 class DefaultSetup(SetupStrategy):
@@ -87,6 +88,7 @@ async def create_setup(self, setup_dict: dict[str, Any]) -> SetupData:
         Raises:
             ValueError: If name or content is invalid.
         """
+        SetupContentValidator.reject_oversized_output_format_spec(setup_dict.get("content") or {})
         setup_id = self._new_id()
         try:
             setup = SetupData(
@@ -137,6 +139,7 @@ async def update_setup(self, setup_dict: dict[str, Any]) -> SetupData:
         if not name or not isinstance(content, dict):
             msg = "setup_id, name and content (object) are required"
             raise ValueError(msg)
+        SetupContentValidator.reject_oversized_output_format_spec(content)
         setup.name = name
         # A new revision rather than an in-place edit, matching UpdateSetup on the wire.
         history = self.versions.setdefault(setup.id, [setup.current_setup_version])

src/digitalkin/services/setup/grpc_setup.py

@@ -19,6 +19,7 @@
 from digitalkin.services.setup.exceptions import SetupServiceError
 from digitalkin.services.setup.setup_strategy import SetupData, SetupStrategy, SetupVersionData, SetupVersionPage
 from digitalkin.utils.proto_utils import ProtoUtils
+from digitalkin.utils.setup_content_validator import SetupContentValidator
 
 
 class GrpcSetup(SetupStrategy, GrpcClientWrapper):
@@ -172,13 +173,16 @@ async def create_setup(self, setup_dict: dict[str, Any]) -> SetupData:
             The created setup with its initial version.
 
         Raises:
-            ValueError: If name or content is missing.
+            ValueError: If name or content is missing, or output_format_spec is oversized.
             ServerError: If gRPC operation fails.
             SetupServiceError: If the server reports failure or an unexpected error occurs.
         """
         if not setup_dict.get("name") or not isinstance(setup_dict.get("content"), dict):
             msg = "name and content (object) are required"
             raise ValueError(msg)
+        # Outside handle_grpc_errors: an input guard must stay a ValueError, not become a
+        # SetupServiceError via the catch-all.
+        SetupContentValidator.reject_oversized_output_format_spec(setup_dict["content"])
         async with self.handle_grpc_errors("Setup Creation"):
             content_struct = Struct()
             content_struct.update(setup_dict["content"])
@@ -201,7 +205,7 @@ async def update_setup(self, setup_dict: dict[str, Any]) -> SetupData:
             The updated setup with its current version.
 
         Raises:
-            ValueError: If setup_id, name or content is missing.
+            ValueError: If setup_id, name or content is missing, or output_format_spec is oversized.
             ServerError: If gRPC operation fails.
             SetupServiceError: If the server reports failure or an unexpected error occurs.
         """
@@ -212,6 +216,7 @@ async def update_setup(self, setup_dict: dict[str, Any]) -> SetupData:
         ):
             msg = "setup_id, name and content (object) are required"
             raise ValueError(msg)
+        SetupContentValidator.reject_oversized_output_format_spec(setup_dict["content"])
         async with self.handle_grpc_errors("Setup Update"):
             content_struct = Struct()
             content_struct.update(setup_dict["content"])

src/digitalkin/utils/setup_content_validator.py

@@ -39,6 +39,8 @@ class SetupContentValidator:
     _STRICT_SCALARS: ClassVar[tuple[type, ...]] = (int, float, bool)
     _FIRST_PRINTABLE: ClassVar[int] = 0x20  # code points below this are C0 control characters.
     _LAST_BMP: ClassVar[int] = 0xFFFF  # code points above this are astral (non-BMP, e.g. emoji).
+    _OUTPUT_FORMAT_SPEC_KEY: ClassVar[str] = "output_format_spec"
+    _OUTPUT_FORMAT_SPEC_MAX: ClassVar[int] = 4096  # at or past this the written setup is unusable.
     _NUMERIC_CONSTRAINTS: ClassVar[dict[str, str]] = {
         "minimum": "ge",
         "maximum": "le",
@@ -113,6 +115,45 @@ def reject_unsafe_keys(cls, content: dict[str, Any]) -> dict[str, Any]:
                 stack.extend((path, item) for item in node)
         return content
 
+    @classmethod
+    def reject_oversized_output_format_spec(cls, content: dict[str, Any]) -> dict[str, Any]:
+        """Refuse an ``output_format_spec`` that reaches the length the setup cannot carry.
+
+        The write itself succeeds, so the breakage surfaces later as an unusable setup with no
+        diagnostic pointing back at the field. Reject up front, naming the path and the actual
+        size. Recurses through nested objects and arrays; a non-string value is left to the
+        schema check, which owns typing.
+
+        Args:
+            content: The setup ``content`` about to be written.
+
+        Returns:
+            The content unchanged when every ``output_format_spec`` fits.
+
+        Raises:
+            ValueError: An ``output_format_spec`` string is at or over the limit.
+        """
+        stack: list[tuple[str, Any]] = [("", content)]
+        while stack:
+            path, node = stack.pop()
+            if isinstance(node, dict):
+                for key, value in node.items():
+                    where = f"{path}.{key}" if path else key
+                    if (
+                        key == cls._OUTPUT_FORMAT_SPEC_KEY
+                        and isinstance(value, str)
+                        and len(value) >= cls._OUTPUT_FORMAT_SPEC_MAX
+                    ):
+                        msg = (
+                            f"content field {where!r} is {len(value)} characters; it must stay under "
+                            f"{cls._OUTPUT_FORMAT_SPEC_MAX} or the setup breaks"
+                        )
+                        raise ValueError(msg)
+                    stack.append((where, value))
+            elif isinstance(node, list):
+                stack.extend((path, item) for item in node)
+        return content
+
     @classmethod
     def validate(cls, content: dict[str, Any], schema: dict[str, Any]) -> None:
         """Validate ``content`` against ``schema``.

tests/services/setup/test_default_setup.py

@@ -15,6 +15,28 @@ async def test_creates_service_setup(self) -> None:
         assert setup.current_setup_version.content == {"branding": True}
 
 
+class TestOutputFormatSpecGuard:
+    """An oversized output_format_spec is refused on both write paths, not just update."""
+
+    async def test_create_refuses_an_oversized_spec(self) -> None:
+        strategy = DefaultSetup()
+        with pytest.raises(ValueError, match="must stay under 4096"):
+            await strategy.create_setup({"name": "n", "content": {"output_format_spec": "x" * 4096}})
+        assert strategy.setups == {}, "nothing may be stored when the guard trips"
+
+    async def test_update_refuses_an_oversized_spec(self) -> None:
+        strategy = DefaultSetup()
+        setup = await strategy.create_setup({"name": "n", "content": {"output_format_spec": "ok"}})
+
+        with pytest.raises(ValueError, match="must stay under 4096"):
+            await strategy.update_setup(
+                {"setup_id": setup.id, "name": "n", "content": {"output_format_spec": "x" * 4096}}
+            )
+        # The guard runs before the revision is cut, so no half-written version survives.
+        assert setup.current_setup_version.content == {"output_format_spec": "ok"}
+        assert (await strategy.list_setup_versions({"setup_id": setup.id})).total_count == 1
+
+
 class TestVersionHistory:
     """The local strategy keeps a version history so the two version RPCs are meaningful."""
 

tests/services/setup/test_grpc_setup.py

@@ -140,6 +140,13 @@ async def test_create_setup_missing_fields_no_rpc(self, client: GrpcSetup) -> No
         with pytest.raises(ValueError, match="name and content"):
             await client.create_setup({"name": "x", "content": "not-a-dict"})
 
+    @pytest.mark.grpc
+    @pytest.mark.validation
+    async def test_create_setup_oversized_output_format_spec_no_rpc(self, client: GrpcSetup) -> None:
+        """The guard trips before the channel is touched, and stays a ValueError."""
+        with pytest.raises(ValueError, match="must stay under 4096"):
+            await client.create_setup({"name": "x", "content": {"output_format_spec": "x" * 4096}})
+
     @pytest.mark.grpc
     @pytest.mark.edge_case
     async def test_create_setup_permission_denied(self, client: GrpcSetup) -> None:
@@ -275,6 +282,14 @@ async def test_update_setup_missing_fields_no_rpc(self, client: GrpcSetup) -> No
         with pytest.raises(ValueError, match="setup_id, name and content"):
             await client.update_setup({"setup_id": "s1", "name": "", "content": {}})
 
+    @pytest.mark.grpc
+    @pytest.mark.validation
+    async def test_update_setup_oversized_output_format_spec_no_rpc(self, client: GrpcSetup) -> None:
+        with pytest.raises(ValueError, match="must stay under 4096"):
+            await client.update_setup(
+                {"setup_id": "s1", "name": "x", "content": {"output_format_spec": "x" * 4096}}
+            )
+
 
 class TestDeleteSetup:
     """delete_setup returns the server's success flag."""

tests/utils/test_setup_content_validator.py

@@ -0,0 +1,46 @@
+"""Content rules enforced on a setup's ``content`` before it is written."""
+
+import pytest
+
+from digitalkin.utils.setup_content_validator import SetupContentValidator
+
+_LIMIT = 4096
+
+
+class TestRejectOversizedOutputFormatSpec:
+    """``output_format_spec`` must stay under the length the written setup can carry."""
+
+    def test_accepts_a_spec_under_the_limit(self) -> None:
+        content = {"output_format_spec": "x" * (_LIMIT - 1)}
+        assert SetupContentValidator.reject_oversized_output_format_spec(content) is content
+
+    def test_rejects_a_spec_exactly_at_the_limit(self) -> None:
+        """The boundary is exclusive: 4096 is already too long."""
+        with pytest.raises(ValueError, match="must stay under 4096"):
+            SetupContentValidator.reject_oversized_output_format_spec({"output_format_spec": "x" * _LIMIT})
+
+    def test_rejects_a_spec_over_the_limit_and_reports_its_size(self) -> None:
+        with pytest.raises(ValueError, match="is 5000 characters"):
+            SetupContentValidator.reject_oversized_output_format_spec({"output_format_spec": "x" * 5000})
+
+    def test_names_the_offending_path_when_nested(self) -> None:
+        content = {"agent": {"output_format_spec": "x" * _LIMIT}}
+        with pytest.raises(ValueError, match=r"'agent\.output_format_spec'"):
+            SetupContentValidator.reject_oversized_output_format_spec(content)
+
+    def test_reaches_into_lists(self) -> None:
+        content = {"agents": [{"name": "ok"}, {"output_format_spec": "x" * _LIMIT}]}
+        with pytest.raises(ValueError, match="output_format_spec"):
+            SetupContentValidator.reject_oversized_output_format_spec(content)
+
+    def test_ignores_a_non_string_value(self) -> None:
+        """Typing is the schema check's job; this rule only measures strings."""
+        content = {"output_format_spec": {"nested": "x" * 5000}}
+        assert SetupContentValidator.reject_oversized_output_format_spec(content) is content
+
+    def test_leaves_other_long_fields_alone(self) -> None:
+        content = {"prompt": "x" * 100_000}
+        assert SetupContentValidator.reject_oversized_output_format_spec(content) is content
+
+    def test_empty_content_is_a_no_op(self) -> None:
+        assert SetupContentValidator.reject_oversized_output_format_spec({}) == {}