fix(backend/executor): validate stored workflow input depth (#15911)

### Why / What / How

**Why:** Fixes #15648. Excessively nested workflow inputs surface
storage or serialization errors; the issue also documents how a saved
JSON fallback string can break execution listings. Runs and schedules
need the same validation, including values expanded from JSON strings.

**What:** Reject unstorable input before creating a run or saving a
schedule, while preserving previously working 101–126-level graph
inputs.

**How:** Check raw inputs and masks in
`validate_and_construct_node_execution_input` before fetching the
workflow, then check starting-node values after defaults, dynamic pins
and type coercion. The verified limit is 127 containers per stored JSON
value: 126 levels inside a graph input, 125 inside a mask field, and 127
for a separately stored starting-node value. PostgreSQL 15.18 and the
generated Prisma 0.15.0 client confirmed the boundary through the real
execution writer and reads; the next level failed without adding a row.

Existing corrupt rows are not repaired by this change.

### Changes 🏗️

- Preserve @troyan-dy's original commit and iterative depth walk,
integrating `dev` with a normal merge.
- Apply storage-aware checks to the shared run/schedule validator and
its constructed inputs.
- Replace mocked-validator regressions with real-validator tests
covering raw input, masks, coercion, defaults, dynamic pins and schedule
create/update/ticks.
- Add PostgreSQL roundtrip tests and API tests that create workflows,
submit runs, reject excessive depth and read both execution lists.
- No configuration changes.

Validation: 34 shared-validator/schedule cases, 5 PostgreSQL/API cases
and 541 existing affected tests passed, with no skips. Eight schedule
cases were rerun successfully after fixing their event-loop fixture.
Whole-backend lint and Python 3.12 types passed (zero type errors). Auth
context is supplied by test fixtures; there was no browser or
executor-worker end-to-end run, and the full backend matrix remains a
hosted CI gate.

Formatter limitation: the pinned Ruff 0.17.2 default-rule migration also
fails on the integrated base. A task-owned copy reproduced the `poetry
run format` failure (9,024 findings, 6,217 auto-fixed, 2,807 remaining).
The remaining whole-subproject format stages completed, and all
resulting formatting fixes are included. A whole-tree baseline
comparison found 7,842 Ruff diagnostics on both base and candidate, with
zero new diagnostics; the strict pre-commit Ruff rule set passes. The
unrelated partial Ruff migration is excluded as described by the
repository's format-scope assessment.

All applicable repository pre-commit hooks passed after task-local
launcher repairs, except the staged-merge Gitleaks scan: it reports two
matches in `frontend/src/lib/secret-detection.test.ts`, which is
byte-for-byte unchanged from the integrated `dev` base. A Gitleaks scan
of the effective PR diff found no leaks. The regenerated OpenAPI schema
also matches the integrated base byte-for-byte.

Independent Codex review is required before merge.

### Agents and large language models used

Codex with GPT-6 Astra for implementation and validation. The original
author disclosed Claude Code with Claude Fable 5.1. Claude Code with
Claude Opus 5.5 supplied the prior assessment.

### Checklist 📋

#### For code changes:
- [x] I have clearly listed my changes in the PR description
- [x] I have made a test plan
- [x] I have tested my changes according to the test plan:
  - [x] Shared-validator and schedule regressions
- [x] Real PostgreSQL/Prisma roundtrip and API execution-list
regressions
- [x] Existing JSON, executor, scheduler, input-size/count, graph-route
and architecture suites
- [x] Whole relevant format stages, lint, Python 3.12 types and Ruff
baseline comparison
- [x] Repository pre-commit hooks and effective-diff secret scan
(inherited Gitleaks fixture matches documented above)

<details>
  <summary>Example test plan</summary>
  
  - [ ] Create from scratch and execute an agent with at least 3 blocks
- [ ] Import an agent from file upload, and confirm it executes
correctly
  - [ ] Upload agent to marketplace
- [ ] Import an agent from marketplace and confirm it executes correctly
  - [ ] Edit an agent from monitor, and confirm it executes correctly
</details>

#### For frontend changes (design system, see
`docs/engineering/design-system.md`):

Not applicable: this PR has no frontend changes.

- [ ] `pnpm lint` passes, including the `design-system/*` rules; I did
not raise any count in
`scripts/eslint-rules/design-system-baseline.json` except for a moved,
renamed or split file (explained under **Changes**)
- [ ] No `dark:`/`light:` classes, raw palette classes (`bg-zinc-200`)
or hex/`rgb()` colours (`bg-[#fff]`); colours use role tokens
(`bg-surface`, `text-foreground-secondary`)
- [ ] No imports from `src/components/__legacy__`, `legacy/` paths or
`@/components/ui`, and no `LegacyButton` / `legacy-*` Button variants
- [ ] Copy uses the `Text` atom with a `variant`/`tone`; primitives get
a `variant`/`size`. `className` holds layout, plus on `Text` only
truncation, line clamping, wrapping, breaking,
`text-balance`/`text-pretty` and alignment
- [ ] Before/after screenshots at 1440 and 390 px are attached for
changes to `src/components/**`, `globals.css` or `tailwind*.ts`, or I
wrote "no pixel change" and how I checked
- [ ] User-facing copy says workflow and Otto, never agent or graph
(root `AGENTS.md`, "User-facing copy")

#### For configuration changes:

- [x] `.env.default` is updated or already compatible with my changes
- [x] `docker-compose.yml` is updated or already compatible with my
changes
- [x] I have included a list of my configuration changes in the PR
description (under **Changes**)

<details>
  <summary>Examples of configuration changes</summary>

  - Changing ports
  - Adding new services that need to communicate with each other
  - Secrets or environment variable changes
  - New or infrastructure changes such as databases
</details>


### Fixture cleanup follow-up

Setup failures now clean up the generated test user and any connection
opened by the fixture; shared connections remain open. Nine
failure/ownership probes and whole backend formatting, CI lint, and
configured types passed. The earlier 580-test evidence remains for the
unchanged production implementation; database tests were not rerun for
this fixture-only follow-up.

---------

Co-authored-by: Nicholas Tindle <[email protected]>
Co-authored-by: GPT-6 Astra (Codex) <[email protected]>

autogpt_platform/backend/backend/api/features/graphs/routes_test.py

@@ -2,6 +2,7 @@
 
 import json
 from datetime import datetime
+from typing import Any
 from unittest.mock import AsyncMock, Mock
 
 import fastapi
@@ -1179,3 +1180,39 @@ def test_execute_graph_records_a_failure_and_re_raises(
         client.post("/graphs/graph-1/execute/1", json={})
 
     assert record.call_args.kwargs["status"] == "error"
+
+
+def test_execute_graph_refuses_an_input_nested_too_deep(
+    mocker: pytest_mock.MockFixture,
+) -> None:
+    """#15648: an input nested ~1000 levels deep answered 422 but still saved
+    an INCOMPLETE run whose ``inputs`` column held a string, which broke every
+    later listing of the user's runs. The real ``add_graph_execution`` must
+    refuse it as a validation error before anything is written."""
+    credit_model = Mock()
+    credit_model.get_credits = AsyncMock(return_value=100)
+    mocker.patch(
+        "backend.api.features.graphs.routes.get_credit_model",
+        AsyncMock(return_value=credit_model),
+    )
+    mocker.patch(
+        "backend.executor.utils.is_user_paywalled", AsyncMock(return_value=False)
+    )
+    mocker.patch("backend.executor.utils.is_user_banned", AsyncMock(return_value=False))
+    created = mocker.patch(
+        "backend.executor.utils.execution_db.create_graph_execution", AsyncMock()
+    )
+    mocker.patch("backend.api.features.graphs.routes.record_graph_operation", Mock())
+    value: Any = "abyss"
+    for _ in range(300):
+        value = [value]
+
+    response = client.post(
+        "/graphs/graph-1/execute/1", json={"inputs": {"shift_tags": value}}
+    )
+
+    assert response.status_code == 400
+    detail = response.json()["detail"]
+    assert detail["type"] == "validation_error"
+    assert "shift_tags" in detail["message"]
+    created.assert_not_awaited()

autogpt_platform/backend/backend/executor/input_depth.py

@@ -0,0 +1,49 @@
+from collections.abc import Iterable, Mapping
+
+from backend.data.block import BlockInput
+from backend.data.execution import NodesInputMasks
+from backend.data.model import GraphInput
+from backend.util.exceptions import GraphValidationError
+from backend.util.json import nesting_depth
+
+# Prisma's JSON parser accepts 127 containers, including each column's wrappers.
+MAX_STORED_JSON_DEPTH = 127
+
+
+def validate_graph_input_depth(
+    graph_inputs: GraphInput, nodes_input_masks: NodesInputMasks
+) -> None:
+    for name, value in graph_inputs.items():
+        if nesting_depth(value) + 1 > MAX_STORED_JSON_DEPTH:
+            raise GraphValidationError(
+                f"Workflow input '{name}' is nested too deeply to be saved "
+                f"(maximum {MAX_STORED_JSON_DEPTH - 1} levels)"
+            )
+    _validate_node_input_depth(nodes_input_masks.items(), MAX_STORED_JSON_DEPTH - 2)
+
+
+def validate_starting_node_input_depth(
+    starting_nodes_input: list[tuple[str, BlockInput]],
+) -> None:
+    _validate_node_input_depth(starting_nodes_input, MAX_STORED_JSON_DEPTH)
+
+
+def _validate_node_input_depth(
+    node_inputs: Iterable[tuple[str, Mapping[str, object]]], max_depth: int
+) -> None:
+    node_errors = {
+        node_id: {
+            name: f"Input is nested too deeply to be saved (maximum {max_depth} levels)"
+            for name, value in values.items()
+            if nesting_depth(value) > max_depth
+        }
+        for node_id, values in node_inputs
+    }
+    if node_errors := {
+        node_id: errors for node_id, errors in node_errors.items() if errors
+    }:
+        n_errors = sum(len(errors) for errors in node_errors.values())
+        raise GraphValidationError(
+            f"Workflow validation failed: {n_errors} node inputs are nested too deeply to be saved",
+            node_errors=node_errors,
+        )

autogpt_platform/backend/backend/executor/input_depth_db_test.py

@@ -0,0 +1,187 @@
+import json
+from uuid import uuid4
+
+import fastapi
+import httpx
+import pytest
+import pytest_asyncio
+from autogpt_libs.auth import get_request_context, get_user_id, requires_user
+from autogpt_libs.auth.models import RequestContext
+from prisma.errors import MissingRequiredValueError
+
+from backend.api.features.graph_executions.routes import router as executions_router
+from backend.api.features.graphs.routes import router as graphs_router
+from backend.api.utils.exceptions import add_exception_handlers
+from backend.blocks.data_manipulation import CreateListBlock
+from backend.blocks.io import AgentInputBlock
+from backend.data.db import prisma
+from backend.data.execution import create_graph_execution, get_graph_execution
+
+
+@pytest_asyncio.fixture(loop_scope="session")
+async def owner():
+    connected_here = not prisma.is_connected()
+    user_id = str(uuid4())
+    try:
+        if connected_here:
+            await prisma.connect()
+        await prisma.user.create(
+            data={"id": user_id, "email": f"depth-{user_id}@example.com"}
+        )
+        for block in (AgentInputBlock(), CreateListBlock()):
+            await prisma.agentblock.upsert(
+                where={"id": block.id},
+                data={
+                    "create": {
+                        "id": block.id,
+                        "name": block.name,
+                        "inputSchema": json.dumps(block.input_schema.jsonschema()),
+                        "outputSchema": json.dumps(block.output_schema.jsonschema()),
+                    },
+                    "update": {},
+                },
+            )
+        yield user_id
+    finally:
+        try:
+            if prisma.is_connected():
+                await prisma.user.delete_many(where={"id": user_id})
+        finally:
+            if connected_here and prisma.is_connected():
+                await prisma.disconnect()
+
+
+@pytest_asyncio.fixture(loop_scope="session")
+async def api(owner):
+    app = fastapi.FastAPI()
+    app.include_router(graphs_router)
+    app.include_router(executions_router)
+    add_exception_handlers(app)
+
+    async def user_id():
+        return owner
+
+    async def context():
+        return RequestContext(
+            user_id=owner,
+            org_id=f"depth-org-{owner}",
+            team_id=None,
+            is_org_owner=True,
+            is_org_admin=True,
+            is_org_billing_manager=False,
+            is_team_admin=True,
+            is_team_billing_manager=False,
+            seat_status="ACTIVE",
+        )
+
+    app.dependency_overrides[get_user_id] = user_id
+    app.dependency_overrides[requires_user] = user_id
+    app.dependency_overrides[get_request_context] = context
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app), base_url="http://test"
+    ) as client:
+        yield client
+
+
+def nested(depth):
+    value = "abyss"
+    for _ in range(depth):
+        value = [value]
+    return value
+
+
+@pytest.mark.asyncio(loop_scope="session")
+@pytest.mark.parametrize(
+    "column,depth", [("inputs", 126), ("mask", 125), ("node", 127)]
+)
+async def test_prisma_roundtrip_at_stored_json_limit(owner, column, depth):
+    graph = await prisma.agentgraph.create(
+        data={
+            "userId": owner,
+            "Nodes": {"create": [{"agentBlockId": CreateListBlock().id}]},
+        },
+        include={"Nodes": True},
+    )
+    node_id = graph.Nodes[0].id
+
+    async def write(value):
+        return await create_graph_execution(
+            graph_id=graph.id,
+            graph_version=graph.version,
+            user_id=owner,
+            inputs={"topic": value} if column == "inputs" else {},
+            nodes_input_masks={node_id: {"values": value}} if column == "mask" else {},
+            starting_nodes_input=(
+                [(node_id, {"values": value})] if column == "node" else []
+            ),
+        )
+
+    value = nested(depth)
+    run = await write(value)
+    loaded = await get_graph_execution(owner, run.id, include_node_executions=True)
+    assert loaded is not None
+    if column == "inputs":
+        assert loaded.inputs == {"topic": value}
+    elif column == "mask":
+        assert loaded.nodes_input_masks == {node_id: {"values": value}}
+    else:
+        assert loaded.node_executions[0].input_data["values"] == value
+
+    with pytest.raises(MissingRequiredValueError, match="recursion limit exceeded"):
+        await write(nested(depth + 1))
+    assert await prisma.agentgraphexecution.count(where={"userId": owner}) == 1
+
+
+@pytest.mark.asyncio(loop_scope="session")
+@pytest.mark.parametrize("source", ["raw_input", "encoded_default"])
+async def test_rejected_run_keeps_both_execution_lists_readable(api, owner, source):
+    async def create_workflow(depth):
+        block = AgentInputBlock() if source == "raw_input" else CreateListBlock()
+        defaults = (
+            {"name": "topic"}
+            if source == "raw_input"
+            else {"values": json.dumps(nested(depth))}
+        )
+        response = await api.post(
+            "/graphs",
+            json={
+                "graph": {
+                    "name": "Input depth regression",
+                    "description": "Execution storage boundary regression",
+                    "nodes": [{"block_id": block.id, "input_default": defaults}],
+                    "links": [],
+                }
+            },
+        )
+        assert response.status_code == 200, response.text
+        return response.json()["id"]
+
+    depth = 126 if source == "raw_input" else 127
+    graph_id = await create_workflow(depth)
+    inputs = {"topic": nested(depth)} if source == "raw_input" else {}
+    accepted = await api.post(f"/graphs/{graph_id}/execute/1", json={"inputs": inputs})
+    assert accepted.status_code == 200, accepted.text
+    run_id = accepted.json()["id"]
+    assert accepted.json()["inputs"] == inputs
+    assert accepted.json()["status"] == "QUEUED"
+
+    rejected_graph = graph_id
+    if source == "encoded_default":
+        rejected_graph = await create_workflow(depth + 1)
+    rejected = await api.post(
+        f"/graphs/{rejected_graph}/execute/1",
+        json={"inputs": {"topic": nested(depth + 1)} if source == "raw_input" else {}},
+    )
+    assert rejected.status_code == 400, rejected.text
+    assert rejected.json()["detail"]["type"] == "validation_error"
+    assert await prisma.agentgraphexecution.count(where={"userId": owner}) == 1
+
+    all_runs = await api.get("/executions")
+    graph_runs = await api.get(f"/graphs/{graph_id}/executions")
+    assert all_runs.status_code == graph_runs.status_code == 200
+    assert [run["id"] for run in all_runs.json()] == [run_id]
+    assert [run["id"] for run in graph_runs.json()["executions"]] == [run_id]
+    read_back = await get_graph_execution(owner, run_id, include_node_executions=True)
+    assert read_back is not None
+    field = "value" if source == "raw_input" else "values"
+    assert read_back.node_executions[0].input_data[field] == nested(depth)

autogpt_platform/backend/backend/executor/input_depth_test.py

@@ -0,0 +1,309 @@
+import asyncio
+import json
+from datetime import UTC, datetime
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from backend.blocks.data_manipulation import CreateListBlock
+from backend.blocks.io import AgentInputBlock
+from backend.data.graph import GraphModel, NodeModel
+from backend.executor.scheduler import (
+    GraphExecutionJobArgs,
+    Scheduler,
+    _build_trigger,
+    _execute_graph,
+)
+from backend.executor.utils import (
+    add_graph_execution,
+    validate_and_construct_node_execution_input,
+)
+from backend.util.exceptions import GraphValidationError
+from backend.util.json import nesting_depth
+
+
+@pytest.fixture
+def graph_source(mocker):
+    graph = GraphModel(
+        id="graph-1",
+        version=1,
+        is_active=True,
+        name="Depth boundary",
+        description="",
+        user_id="user-1",
+        created_at=datetime.now(UTC),
+        nodes=[
+            NodeModel(
+                id="input-node",
+                block_id=AgentInputBlock().id,
+                graph_id="graph-1",
+                graph_version=1,
+                input_default={"name": "topic"},
+            ),
+            NodeModel(
+                id="list-node",
+                block_id=CreateListBlock().id,
+                graph_id="graph-1",
+                graph_version=1,
+                input_default={"values": []},
+            ),
+        ],
+    )
+    source = AsyncMock()
+    source.get_graph.return_value = graph
+    source.resolve_default_tenancy.return_value = (None, None)
+    mocker.patch("backend.executor.utils.graph_db", source)
+    mocker.patch("backend.executor.utils.execution_db", source)
+    mocker.patch(
+        "backend.executor.utils.get_database_manager_async_client", return_value=source
+    )
+    return source
+
+
+def nested(depth, kind="list"):
+    value = "abyss"
+    for _ in range(depth):
+        value = [value] if kind == "list" else {"value": value}
+    return value
+
+
+@pytest.mark.parametrize("depth", [0, 100, 101, 126])
+@pytest.mark.parametrize("kind", ["list", "dict"])
+async def test_storable_graph_inputs_are_preserved(graph_source, depth, kind):
+    value = nested(depth, kind)
+    _, starting, _, _ = await validate_and_construct_node_execution_input(
+        graph_id="graph-1", user_id="user-1", graph_inputs={"topic": value}
+    )
+    assert dict(starting)["input-node"]["value"] == value
+    graph_source.validate_graph_execution_permissions.assert_awaited_once_with(
+        user_id="user-1", graph_id="graph-1", graph_version=1, is_sub_graph=False
+    )
+
+
+@pytest.mark.parametrize("depth", [127, 300, 5000])
+async def test_raw_input_is_rejected_before_graph_fetch(graph_source, depth):
+    with pytest.raises(GraphValidationError, match="topic"):
+        await validate_and_construct_node_execution_input(
+            graph_id="graph-1",
+            user_id="user-1",
+            graph_inputs={"topic": nested(depth)},
+        )
+    graph_source.get_graph.assert_not_awaited()
+
+
+async def test_literal_masks_account_for_both_storage_wrappers(graph_source):
+    _, starting, masks, _ = await validate_and_construct_node_execution_input(
+        graph_id="graph-1",
+        user_id="user-1",
+        graph_inputs={},
+        nodes_input_masks={"list-node": {"values": nested(125)}},
+    )
+    assert nesting_depth(masks) == 127
+    assert nesting_depth(dict(starting)["list-node"]["values"]) == 125
+
+    with pytest.raises(GraphValidationError) as exc:
+        await validate_and_construct_node_execution_input(
+            graph_id="graph-1",
+            user_id="user-1",
+            graph_inputs={},
+            nodes_input_masks={"list-node": {"values": nested(126), "max_size": 1}},
+        )
+    assert set(exc.value.node_errors) == {"list-node"}
+    assert set(exc.value.node_errors["list-node"]) == {"values"}
+
+
+async def test_all_deep_mask_fields_are_reported_before_graph_fetch(graph_source):
+    with pytest.raises(GraphValidationError) as exc:
+        await validate_and_construct_node_execution_input(
+            graph_id="graph-1",
+            user_id="user-1",
+            graph_inputs={},
+            nodes_input_masks={
+                "list-node": {"values": nested(5000), "max_size": 1},
+                "unused-node": {"payload": nested(126)},
+            },
+        )
+    assert {key: set(errors) for key, errors in exc.value.node_errors.items()} == {
+        "list-node": {"values"},
+        "unused-node": {"payload"},
+    }
+    graph_source.get_graph.assert_not_awaited()
+
+
+@pytest.mark.parametrize(
+    "source,depth",
+    [(source, depth) for source in ("mask", "default") for depth in (127, 128, 300)]
+    + [("dynamic_pin", 127), ("dynamic_pin", 128)],
+)
+async def test_constructed_values_are_checked_after_coercion(
+    graph_source, source, depth
+):
+    graph = graph_source.get_graph.return_value
+    mask = {}
+    if source == "dynamic_pin":
+        graph.nodes[1].input_default = {"values_$_0": nested(depth - 1)}
+    elif source == "default":
+        graph.nodes[1].input_default = {"values": json.dumps(nested(depth))}
+    else:
+        mask = {"list-node": {"values": json.dumps(nested(depth))}}
+
+    async def construct():
+        return await validate_and_construct_node_execution_input(
+            graph_id="graph-1",
+            user_id="user-1",
+            graph_inputs={},
+            nodes_input_masks=mask,
+        )
+
+    if depth == 127:
+        _, starting, _, _ = await construct()
+        assert nesting_depth(dict(starting)["list-node"]["values"]) == depth
+    else:
+        with pytest.raises(GraphValidationError) as exc:
+            await construct()
+        assert set(exc.value.node_errors["list-node"]) == {"values"}
+
+
+@pytest.mark.parametrize("source", ["raw", "mask", "encoded"])
+async def test_rejected_run_never_reaches_persistence(graph_source, source):
+    inputs = {"topic": nested(127)} if source == "raw" else {}
+    masks = (
+        {"list-node": {"values": nested(126)}}
+        if source == "mask"
+        else (
+            {"list-node": {"values": json.dumps(nested(128))}}
+            if source == "encoded"
+            else {}
+        )
+    )
+    with pytest.raises(GraphValidationError):
+        await add_graph_execution(
+            graph_id="graph-1",
+            user_id="user-1",
+            inputs=inputs,
+            nodes_input_masks=masks,
+        )
+    graph_source.create_graph_execution.assert_not_awaited()
+
+
+@pytest.fixture
+def schedule(mocker):
+    scheduler = Scheduler(register_system_tasks=False, throwaway_jobstores=True)
+    scheduler.scheduler = MagicMock()
+    job = MagicMock(id="schedule-1")
+    job.name = "Daily"
+    job.trigger = _build_trigger(cron="0 9 * * *", run_at=None, user_timezone="UTC")
+    job.next_run_time = None
+    job.kwargs = GraphExecutionJobArgs(
+        schedule_id="schedule-1",
+        user_id="user-1",
+        graph_id="graph-1",
+        graph_version=1,
+        cron="0 9 * * *",
+        input_data={"topic": "previous"},
+    ).model_dump(mode="json")
+    scheduler.scheduler.get_job.return_value = job
+    scheduler.scheduler.modify_job.return_value = job
+    mocker.patch.object(scheduler, "_persist_schedule", return_value=job)
+    mocker.patch.object(scheduler, "_invalidate_jobs_cache")
+    loop = asyncio.new_event_loop()
+    mocker.patch(
+        "backend.executor.scheduler.run_async", side_effect=loop.run_until_complete
+    )
+    mocker.patch("backend.executor.schedule_events.record_schedule_created")
+    try:
+        yield scheduler
+    finally:
+        loop.close()
+        if scheduler._throwaway_dir is not None:
+            scheduler._throwaway_dir.cleanup()
+
+
+@pytest.mark.parametrize("operation", ["create", "update"])
+@pytest.mark.parametrize("depth", [101, 126, 127])
+def test_schedule_validates_before_persisting(graph_source, schedule, operation, depth):
+    def save():
+        if operation == "create":
+            return schedule.add_graph_execution_schedule(
+                graph_id="graph-1",
+                graph_version=1,
+                user_id="user-1",
+                cron="0 9 * * *",
+                input_data={"topic": nested(depth)},
+                input_credentials={},
+                user_timezone="UTC",
+            )
+        return schedule.update_graph_execution_schedule(
+            schedule_id="schedule-1",
+            user_id="user-1",
+            input_data={"topic": nested(depth)},
+        )
+
+    if depth == 127:
+        with pytest.raises(GraphValidationError, match="topic"):
+            save()
+        schedule._persist_schedule.assert_not_called()
+        schedule.scheduler.modify_job.assert_not_called()
+        assert schedule.scheduler.get_job.return_value.kwargs["input_data"] == {
+            "topic": "previous"
+        }
+    else:
+        save()
+        if operation == "create":
+            schedule._persist_schedule.assert_called_once()
+        else:
+            schedule.scheduler.modify_job.assert_called_once()
+
+
+@pytest.mark.parametrize("operation", ["create", "update"])
+def test_schedule_refuses_defaults_that_expand_past_storage(
+    graph_source, schedule, operation
+):
+    graph_source.get_graph.return_value.nodes[1].input_default = {
+        "values": json.dumps(nested(128))
+    }
+    with pytest.raises(GraphValidationError) as exc:
+        if operation == "create":
+            schedule.add_graph_execution_schedule(
+                graph_id="graph-1",
+                graph_version=1,
+                user_id="user-1",
+                cron="0 9 * * *",
+                input_data={},
+                input_credentials={},
+                user_timezone="UTC",
+            )
+        else:
+            schedule.update_graph_execution_schedule(
+                schedule_id="schedule-1", user_id="user-1", input_data={}
+            )
+    assert set(exc.value.node_errors["list-node"]) == {"values"}
+    schedule._persist_schedule.assert_not_called()
+    schedule.scheduler.modify_job.assert_not_called()
+
+
+@pytest.mark.parametrize("depth", [101, 126])
+async def test_existing_storable_schedule_reaches_run_creation(
+    graph_source, mocker, depth
+):
+    graph_source.create_graph_execution.side_effect = RuntimeError(
+        "persistence boundary"
+    )
+    mocker.patch(
+        "backend.executor.scheduler.get_database_manager_async_client",
+        return_value=graph_source,
+    )
+    scheduler_client = AsyncMock()
+    mocker.patch(
+        "backend.executor.scheduler.get_scheduler_client", return_value=scheduler_client
+    )
+    await _execute_graph(
+        schedule_id="schedule-1",
+        user_id="user-1",
+        graph_id="graph-1",
+        graph_version=1,
+        cron="0 9 * * *",
+        input_data={"topic": nested(depth)},
+    )
+    graph_source.create_graph_execution.assert_awaited_once()
+    scheduler_client.delete_schedule.assert_not_called()

autogpt_platform/backend/backend/executor/utils.py

@@ -64,6 +64,10 @@ from backend.data.model import (
 )
 from backend.data.rabbitmq import Exchange, ExchangeType, Queue, RabbitMQConfig
 from backend.data.user import is_user_banned
+from backend.executor.input_depth import (
+    validate_graph_input_depth,
+    validate_starting_node_input_depth,
+)
 from backend.integrations.credentials_store import (
     CredentialsSnapshot,
     is_system_credential,
@@ -991,6 +995,8 @@ async def validate_and_construct_node_execution_input(
         GraphValidationError: If the graph has validation issues.
         ValueError: If there are other validation errors.
     """
+    validate_graph_input_depth(graph_inputs, nodes_input_masks or {})
+
     if prisma.is_connected():
         gdb = graph_db
     else:
@@ -1039,6 +1045,7 @@ async def validate_and_construct_node_execution_input(
         nodes_input_masks=nodes_input_masks,
         dry_run=dry_run,
     )
+    validate_starting_node_input_depth(starting_nodes_input)
     # Also enforced by create_graph_execution; here a schedule or a request is
     # refused before anything is written.
     check_node_input_counts(starting_nodes_input)

autogpt_platform/backend/backend/util/json.py

@@ -289,6 +289,29 @@ def sanitize_string(value: str) -> str:
     return POSTGRES_CONTROL_CHARS.sub("", value)
 
 
+def nesting_depth(data: Any) -> int:
+    """How many levels of lists and dicts ``data`` contains: 0 for a scalar,
+    1 for a flat list or dict, 2 for a list of dicts, and so on.
+
+    Walks iteratively, so it also measures data too deep for ``to_dict`` or
+    ``orjson`` to serialize.
+    """
+    deepest = 0
+    pending: list[tuple[Any, int]] = [(data, 0)]
+    while pending:
+        current, depth = pending.pop()
+        if isinstance(current, dict):
+            children = current.values()
+        elif isinstance(current, (list, tuple)):
+            children = current
+        else:
+            continue
+        depth += 1
+        deepest = max(deepest, depth)
+        pending.extend((child, depth) for child in children)
+    return deepest
+
+
 def sanitize_json(data: Any) -> Any:
     try:
         # Use two-pass approach for consistent string sanitization:

autogpt_platform/backend/backend/util/test_json.py

@@ -1223,3 +1223,29 @@ class TestValidateWithJsonschema:
         except jsonschema.ValidationError as error:
             return str(error)
         return None
+
+
+class TestNestingDepth:
+    @pytest.mark.parametrize(
+        "data,depth",
+        [
+            ("abyss", 0),
+            (1, 0),
+            (None, 0),
+            ([], 1),
+            ({}, 1),
+            ([1, 2], 1),
+            ({"a": [1]}, 2),
+            ([[["x"]]], 3),
+            ({"a": {"b": {"c": 1}}, "d": [1]}, 3),
+            (("t", ["u"]), 2),
+        ],
+    )
+    def test_counts_nested_lists_and_dicts(self, data: Any, depth: int):
+        assert json_util.nesting_depth(data) == depth
+
+    def test_walks_data_deeper_than_the_recursion_limit(self):
+        value: Any = "abyss"
+        for _ in range(5000):
+            value = [value]
+        assert json_util.nesting_depth({"shift_tags": value}) == 5001