docs: complete run step budget slice

docs/README.md

@@ -120,7 +120,7 @@ docs as the active references:
   implemented same-scope nested foreach context, path, schema, and
   authoring-ref semantics.
 - [`superpowers/specs/2026-09-04-run-step-budget-design.md`](superpowers/specs/2026-09-04-run-step-budget-design.md):
-  proposed persisted run-wide protection against unbounded graph execution.
+  implemented persisted run-wide step budget against unbounded graph execution.
 - [`superpowers/specs/2026-05-24-native-subgraphs-design.md`](superpowers/specs/2026-05-24-native-subgraphs-design.md):
   native subgraph design.
 - [`superpowers/specs/2026-05-26-durable-workflow-runs-and-resume-design.md`](superpowers/specs/2026-05-26-durable-workflow-runs-and-resume-design.md):

docs/current_roadmap.md

@@ -40,27 +40,18 @@ author -> validate -> save artifact -> deploy -> run -> inspect or resume
 
 ## Active runtime sequence
 
-The next three slices build on the foreach control-region, structured-context,
+The next two slices build on the foreach control-region, structured-context,
-scheduler, lineage, and barrier foundations in this order.
+scheduler, lineage, barrier, and persisted run step budget foundations in
+this order.
 
-### 1. Add a persisted run step budget
+### 1. Consolidate runtime identity resolution
-
-Implement the proposed run-wide limit:
-
-- [`run step budget design`](superpowers/specs/2026-09-04-run-step-budget-design.md)
-
-The budget must cover every frame and subgraph scope in one run, survive
-checkpoint and resume, and stop valid but non-terminating graph cycles with a
-clear runtime failure.
-
-### 2. Consolidate runtime identity resolution
 
 Introduce one internal resolver for a frame, lineage, runtime scope, and
 foreach activation environment. The resolver should validate the canonical
 identity chain once so fork/gather code does not pass related identifiers
 independently or repeat ownership walks.
 
-### 3. Implement explicit fork and gather
+### 2. Implement explicit fork and gather
 
 Reuse the scheduler, activation, lineage, and reducer-aware barrier machinery:
 
@@ -141,6 +132,9 @@ The active sequence can assume these foundations:
 - Removal of the pass-through `JoinNode`; future `GatherNode` starts with its
   actual synchronization contract and no placeholder compatibility
 - Durable stopped-run inspection and resume
+- Persisted run-wide step budget (`RunLimits`, `steps_executed`,
+  computed `steps_remaining`) covering every frame and subgraph scope,
+  surviving checkpoint and resume, with exhaustion as a failed run
 - Python client reconstruction of capabilities, artifacts, deployments, and
   runs through the API
 

docs/historical/superpowers/plans/2026-09-05-run-step-budget.md

@@ -55,7 +55,7 @@ async workflow runtimes, FastAPI JSON-RPC, the Python workflow client, pytest.
 - Produces: `remaining_step_attempts(run) -> int`.
 - Produces: `load_run_state_with_upgrade(payload) -> tuple[RunState, bool]`.
 
-- [ ] **Step 1: Write failing model, admission, and codec tests**
+- [x] **Step 1: Write failing model, admission, and codec tests**
 
 Cover positive validation, the default, a budget of one, denied admission not
 incrementing, error details, v2 round-trip, v1 default injection, and v2
@@ -73,7 +73,7 @@ with pytest.raises(WorkflowStepLimitExceeded):
     admit_step_attempt(run, run.current_frame(), workflow.start)
 ```
 
-- [ ] **Step 2: Run the new tests and verify they fail for missing symbols**
+- [x] **Step 2: Run the new tests and verify they fail for missing symbols**
 
 Run:
 
@@ -81,7 +81,7 @@ Run:
 uv run pytest tests/core/test_run_step_budget.py tests/core/test_run_codec.py -q
 ```
 
-- [ ] **Step 3: Implement the minimal core model and admission module**
+- [x] **Step 3: Implement the minimal core model and admission module**
 
 ```python
 @dataclass(frozen=True, slots=True)
@@ -109,14 +109,14 @@ Add `limits`, `steps_executed`, and computed `steps_remaining` to `RunState`;
 add `step_number: int | None` to `ExecutionFrame`; pass optional limits through
 `create_run_state()`.
 
-- [ ] **Step 4: Implement strict v2 output and explicit v1 loading**
+- [x] **Step 4: Implement strict v2 output and explicit v1 loading**
 
 `dump_run_state()` writes envelope version 2. Version 1 may omit the three new
 fields and receives defaults. Version 2 validates that `limits`,
 `steps_executed`, and each serialized frame's `step_number` field are present
 before using the dataclass adapter. Return `upgraded=True` only for v1.
 
-- [ ] **Step 5: Run the focused tests and commit**
+- [x] **Step 5: Run the focused tests and commit**
 
 Run:
 
@@ -143,18 +143,18 @@ Commit: `feat: add persisted run step budget state`
 - Produces: `TraceEntry.step_number: int`.
 - Produces: `InterruptRequest.step_number: int`.
 
-- [ ] **Step 1: Add failing sync behavior tests**
+- [x] **Step 1: Add failing sync behavior tests**
 
 Test NodeUse, condition, foreach controller/body, subgraph entry/return,
 interrupt/resume, explicit End, legacy `END`, handler failure, handled `error`
 outcome, a closed cycle, an exiting loop, and denial without handler invocation.
 Assert trace numbers rather than inferring counts from trace length.
 
-- [ ] **Step 2: Verify the focused tests fail before dispatch is counted**
+- [x] **Step 2: Verify the focused tests fail before dispatch is counted**
 
 Run: `uv run pytest tests/core/test_run_step_budget.py -q`
 
-- [ ] **Step 3: Admit after resolving the selected step and before dispatch**
+- [x] **Step 3: Admit after resolving the selected step and before dispatch**
 
 Call `admit_step_attempt()` exactly once in the non-batched paths of
 `step_workflow()` and `step_workflow_async()`. Make `append_trace()` fail closed
@@ -163,7 +163,7 @@ trace produced during the dispatch. Store the interrupt activation's number on
 `InterruptRequest`; its resume-completion trace reuses that value and does not
 admit another attempt.
 
-- [ ] **Step 4: Verify sync semantics and commit**
+- [x] **Step 4: Verify sync semantics and commit**
 
 Run:
 
@@ -190,7 +190,7 @@ Commit: `feat: enforce step budget during sync dispatch`
 - Consumes: `admit_step_attempt(...) -> int`.
 - Produces: bounded `_claim_matching_async_item_frames(..., limit: int)`.
 
-- [ ] **Step 1: Write failing async reservation tests**
+- [x] **Step 1: Write failing async reservation tests**
 
 Use handlers gated by `asyncio.Event` to prove that a three-unit remainder
 starts only the first three eligible frames, assigns numbers in queue order,
@@ -198,18 +198,18 @@ keeps reservations after a handler failure, settles siblings before raising,
 and discards later sibling state/trace commits after the first unhandled result
 in reservation order.
 
-- [ ] **Step 2: Verify the tests fail because batching claims every sibling**
+- [x] **Step 2: Verify the tests fail because batching claims every sibling**
 
 Run: `uv run pytest tests/core/test_run_step_budget_async.py -q`
 
-- [ ] **Step 3: Bound claims and reserve before creating handler tasks**
+- [x] **Step 3: Bound claims and reserve before creating handler tasks**
 
 Before `_step_async_foreach_item_batch()` creates any coroutine, require one
 unit for `first_frame`, claim at most `remaining - 1` matching frames, then call
 `admit_step_attempt()` for the resulting ordered frame list. Do not launch any
 task until every selected frame has its number.
 
-- [ ] **Step 4: Verify async and parity suites and commit**
+- [x] **Step 4: Verify async and parity suites and commit**
 
 Run:
 
@@ -242,14 +242,14 @@ Commit: `feat: reserve async workflow step attempts`
 - Produces: optional `max_steps` on run creation only.
 - Produces: `max_steps`, `steps_executed`, and `steps_remaining` in run results.
 
-- [ ] **Step 1: Write failing API and migration tests**
+- [x] **Step 1: Write failing API and migration tests**
 
 Pin requested/effective limit inspection, interrupted resume preserving the
 counter, resume accepting no replacement, and a v1 interrupted checkpoint being
 rewritten as v2 before runtime dispatch. Make the fake runtime assert it has not
 been called until the upgraded checkpoint exists.
 
-- [ ] **Step 2: Verify the API tests fail on the missing fields**
+- [x] **Step 2: Verify the API tests fail on the missing fields**
 
 Run:
 
@@ -257,7 +257,7 @@ Run:
 uv run pytest tests/wf_api/test_runs.py tests/wf_api/test_run_lifecycle.py -q
 ```
 
-- [ ] **Step 3: Thread limits through creation and project inspection fields**
+- [x] **Step 3: Thread limits through creation and project inspection fields**
 
 `WorkflowRunApi.run_deployment(..., max_steps: int | None = None)` constructs
 `RunLimits(max_steps=max_steps)` when supplied and otherwise uses the default.
@@ -270,14 +270,14 @@ passes it to the core async executor. `_run_payload()` always includes:
 "steps_remaining": run.steps_remaining,
 ```
 
-- [ ] **Step 4: Persist a v1 upgrade before resume dispatch**
+- [x] **Step 4: Persist a v1 upgrade before resume dispatch**
 
 In `restore_interrupted_run()`, load the raw latest checkpoint with the upgrade
 flag. If true, call `persist_stopped_run()` with the same run id and pinned
 environment, producing a v2 interrupted checkpoint before returning the run to
 the caller. Ordinary inspection may decode v1 prospectively without mutation.
 
-- [ ] **Step 5: Verify API behavior and commit**
+- [x] **Step 5: Verify API behavior and commit**
 
 Run:
 
@@ -313,13 +313,13 @@ Commit: `feat: persist and inspect run step budgets`
   `.steps_remaining`.
 - Produces: CLI `wf run start --max-steps INTEGER`.
 
-- [ ] **Step 1: Write failing round-trip and client reconstruction tests**
+- [x] **Step 1: Write failing round-trip and client reconstruction tests**
 
 Assert the request includes `max_steps` only when supplied; the response
 decoder requires all three inspection fields; refresh/resume preserve them;
 and CLI rejects zero before making an API request.
 
-- [ ] **Step 2: Verify transport/client tests fail**
+- [x] **Step 2: Verify transport/client tests fail**
 
 Run:
 
@@ -329,12 +329,12 @@ uv run pytest tests/wf_transport_rpc_http/test_client.py
 uv run pytest tests/wf_transport_rpc_http/test_app.py tests/wf_cli/test_app.py -q
 ```
 
-- [ ] **Step 3: Thread the optional creation value and reconstruct results**
+- [x] **Step 3: Thread the optional creation value and reconstruct results**
 
 Keep `max_steps` off resume signatures. Validate the CLI option with Typer
 `min=1`; server-side `RunLimits` remains authoritative for non-CLI callers.
 
-- [ ] **Step 4: Regenerate contracts, verify, and commit**
+- [x] **Step 4: Regenerate contracts, verify, and commit**
 
 Run:
 
@@ -361,14 +361,14 @@ Commit: `feat: expose run step budgets to clients`
 
 **Interfaces:** None.
 
-- [ ] **Step 1: Document creation, inspection, exhaustion, and resume**
+- [x] **Step 1: Document creation, inspection, exhaustion, and resume**
 
 Show `Deployment.run(..., max_steps=50_000)`, `wf run start --max-steps`, the
 three inspection fields, and that resume cannot reset the budget. Remove the
 step-budget item from the active roadmap and leave runtime identity resolution
 as the next fork/gather prerequisite.
 
-- [ ] **Step 2: Run focused and full verification**
+- [x] **Step 2: Run focused and full verification**
 
 Run:
 
@@ -385,7 +385,7 @@ pnpm --dir web test
 pnpm --dir web typecheck
 ```
 
-- [ ] **Step 3: Retire the completed plan and commit**
+- [x] **Step 3: Retire the completed plan and commit**
 
 Commit: `docs: complete run step budget slice`
 

docs/wf_cli.md

@@ -702,6 +702,29 @@ wf run start concat_ws.default \
   --input '{"items":["red","blue"],"separator":" + "}'
 ```
 
+Start with an explicit run-wide step budget (default `10_000` when omitted):
+
+```bash
+wf run start concat_ws.default \
+  --input '{"items":["red","blue"]}' \
+  --max-steps 50000
+```
+
+`--max-steps` must be at least 1; the server validates it again. The budget
+covers every frame and subgraph scope in the run and stops runaway cycles
+with a failed run instead of a routable workflow outcome.
+
+The Python client accepts the same creation-only value:
+
+```python
+run = await deployment.run({"items": ["red", "blue"]}, max_steps=50_000)
+assert (run.max_steps, run.steps_executed, run.steps_remaining) == (
+    50_000,
+    run.steps_executed,
+    50_000 - run.steps_executed,
+)
+```
+
 List durable stopped runs:
 
 ```bash
@@ -721,6 +744,17 @@ Inspect a run without trace detail:
 wf run inspect run_123
 ```
 
+Inspection reports the effective step budget alongside status and output:
+
+- `max_steps`: effective limit stored with the run
+- `steps_executed`: admitted step attempts so far
+- `steps_remaining`: unspent budget, floored at zero
+
+Resume reuses the persisted budget and accepts no replacement value.
+`wf run resume` takes only a payload and outcome; it never resets the
+counter. Budget exhaustion fails the run with a step-budget error and never
+invokes the denied handler.
+
 Poll a run until it stops:
 
 ```bash

docs/wf_core_architecture.md

@@ -13,7 +13,9 @@ and user-facing control belong in `wf_mcp`.
 | --- | --- |
 | `wf_core.models` | Pydantic workflow schema package: schemas, condition expressions, executable steps, workflow graph, and node results. |
 | `wf_core.run_state` | Serializable execution state: run status, frames, trace entries, interrupt requests, and runtime context. |
+| `wf_core.run_limits` | Immutable run-wide step budget (`RunLimits`). |
 | `wf_core.runtime` | Public execution interface: execute, resume, and step in sync or async mode. |
+| `wf_core.runtime.limits` | Admits one counted attempt per dispatched step. |
 | `wf_core.runtime.scheduler` | Internal frame scheduler: ready queue, selected cursor, frame creation, block/wake helpers, and typed foreach frame metadata. |
 | `wf_core.runtime.ops` | Executor-only operations used behind `wf_core.runtime`: node execution, state writes, frame movement, foreach, interrupts, indexes, and schema checks. |
 | `wf_core.validation` | Structural workflow validation split by validation concern. |
@@ -35,6 +37,10 @@ flat modules.
    `RunState.ready_frame_ids`, marks that frame `RUNNING`, and updates
    compatibility cursor fields such as `current_frame_id`.
 5. `step_workflow` resolves the selected frame and dispatches by step type.
+   Admission runs first: `runtime.limits.admit_step_attempt` consumes one unit
+   of the run-wide budget (`RunLimits`, default `10_000`) and stamps the frame
+   with its step number before any handler runs. Exhaustion fails the run and
+   never invokes the denied handler.
 6. A normal non-terminal step marks the same frame `PENDING` and puts it back at
    the end of the ready queue.
 7. Terminal, blocked, interrupted, and failed frames are not re-enqueued.

skills/wf-workflow/references/workflow-lifecycle.md

@@ -54,7 +54,13 @@ validated, runnable deployment.
 7. Save and validate a deployment.
    - `wf deploy save <deployment_id> --artifact <artifact_id> --version 1 --binding <logical_source>=<concrete_source>` (or `wf deploy create` alias)
    - `wf deploy validate <deployment_id>`
-8. Run the deployment.
+8. Run the deployment with an optional step budget.
+   - CLI: `wf run start <deployment_id> --input-file input.json`
+     with `--max-steps 50000` (default `10_000` when omitted; at least 1)
+   - Python: `await deployment.run(input, max_steps=50_000)`
+   - The budget covers every frame and subgraph scope in the run. Exhaustion
+     fails the run; it is never a routable workflow outcome and the denied
+     handler never runs.
 9. Inspect the run summary first; read bounded traces only when needed.
 
 ## Raw Plan Escape Hatch
@@ -104,16 +110,19 @@ sources and choose the concrete source explicitly.
 - **Workflow capability**: graph-ready `NodeSpec` or saved wrapper artifact.
 - **Artifact**: immutable saved workflow or wrapper.
 - **Deployment**: mutable binding from artifact version to concrete sources.
-- **Run**: durable stopped execution record with status, output, and trace count.
+- **Run**: durable stopped execution record with status, output, trace count,
+  and step budget (`max_steps`, `steps_executed`, `steps_remaining`).
 
 ## Result Handling
 
 `wf run start` returns compact status by default. Capture `run_id` even for
-completed or failed runs. Use:
+completed or failed runs. Inspection exposes the effective budget stored
+with the run (`max_steps`, `steps_executed`, `steps_remaining`). Use:
 
 - `wf run inspect <run_id>` for compact stored result.
 - `wf run trace <run_id> --from 0 --limit 25` for explicit debug slices.
-- `wf run resume <run_id>` only for interrupted runs.
+- `wf run resume <run_id>` only for interrupted runs. Resume reuses the
+  persisted budget and accepts no replacement value.
 
 Do not ask for unbounded traces.
 

src/wf_core/__init__.py

@@ -38,6 +38,7 @@ from .run_codec import (
     load_run_state,
     load_run_state_with_upgrade,
 )
+from .run_limits import RunLimits
 from .run_state import (
     ExecutionFrame,
     ForeachContext,
@@ -65,7 +66,7 @@ from .runtime import (
     step_workflow,
     step_workflow_async,
 )
-from .runtime.limits import RunLimits, admit_step_attempt, remaining_step_attempts
+from .runtime.limits import admit_step_attempt, remaining_step_attempts
 from .tokens import END, START
 from .validation import (
     ValidationIssue,

src/wf_core/run_codec.py

@@ -5,8 +5,8 @@ from typing import Any, Literal, cast
 
 from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
 
+from .run_limits import RunLimits
 from .run_state import ROOT_SCOPE_ID, RunState
-from .runtime.limits import RunLimits
 
 
 class PersistedRunState(BaseModel):
@@ -57,7 +57,13 @@ def _check_step_number(value: object, *, steps_executed: int, what: str) -> None
     """
     if value is None:
         return
-    if not _is_strict_int(value) or not 1 <= value <= steps_executed:  # type: ignore[operator]
+    if not _is_strict_int(value):
+        raise ValueError(
+            "invalid persisted workflow run state: "
+            f"{what} has incoherent step number {value!r}"
+        )
+    number = cast(int, value)
+    if not 1 <= number <= steps_executed:
         raise ValueError(
             "invalid persisted workflow run state: "
             f"{what} has incoherent step number {value!r}"
@@ -101,27 +107,32 @@ def _require_v2_budget_fields(state: dict[str, Any]) -> None:
     corruption, not another request for defaults. Values are validated
     strictly on the raw envelope (exact ints, ranges, coherence) because lax
     coercion would otherwise accept bools, numeric strings, negatives, or
-    future frame numbers and silently inflate or distort the budget. Trace
+    future frame numbers and silently inflate or distort the budget. The
-    and interrupt entries always carry the key (``None`` only for unadmitted
+    limits object holds exactly ``max_steps``: unknown fields are corrupt
-    or upgraded pre-budget history), so a missing key is likewise corrupt
+    rather than silently dropped, since no stored-data contract emits them.
-    even though the dataclass default would otherwise mask it.
+    Trace and interrupt entries always carry the key (``None`` only for
+    unadmitted or upgraded pre-budget history), so a missing key is likewise
+    corrupt even though the dataclass default would otherwise mask it.
     """
     limits = state.get("limits")
-    if not isinstance(limits, dict) or "max_steps" not in limits:
+    if not isinstance(limits, dict) or set(limits.keys()) != {"max_steps"}:
         raise ValueError("invalid persisted workflow run state: missing step budget")
     max_steps = limits["max_steps"]
-    if not _is_strict_int(max_steps) or max_steps < 1:  # type: ignore[operator]
+    if not _is_strict_int(max_steps):
+        raise ValueError(
+            "invalid persisted workflow run state: corrupt step budget limit"
+        )
+    max_steps_value = cast(int, max_steps)
+    if max_steps_value < 1:
         raise ValueError(
             "invalid persisted workflow run state: corrupt step budget limit"
         )
     steps_executed = state.get("steps_executed")
     if not _is_strict_int(steps_executed):
         raise ValueError("invalid persisted workflow run state: missing step budget")
-    if not 0 <= steps_executed <= max_steps:  # type: ignore[operator]
-        raise ValueError(
-            "invalid persisted workflow run state: corrupt step counter"
-        )
     exec_count = cast(int, steps_executed)
+    if not 0 <= exec_count <= max_steps_value:
+        raise ValueError("invalid persisted workflow run state: corrupt step counter")
     frames = state.get("frames")
     if not isinstance(frames, dict):
         raise ValueError("invalid persisted workflow run state: missing frames")

src/wf_core/run_limits.py

@@ -0,0 +1,28 @@
+"""Neutral run-wide step budget value model.
+
+`RunLimits` is immutable policy captured when a run is created. It lives
+here (next to `run_state`, not under `runtime`) so `run_state` can import
+it at module top without executing the `wf_core.runtime` package whose
+engine imports `run_state` back. Admission policy stays in
+`wf_core.runtime.limits`.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, slots=True)
+class RunLimits:
+    """Immutable step budget captured when a run is created."""
+
+    max_steps: int = 10_000
+
+    def __post_init__(self) -> None:
+        if isinstance(self.max_steps, bool) or not isinstance(self.max_steps, int):
+            raise TypeError("max_steps must be an integer")
+        if self.max_steps < 1:
+            raise ValueError("max_steps must be positive")
+
+
+__all__ = ["RunLimits"]

src/wf_core/run_state.py

@@ -2,14 +2,12 @@ from __future__ import annotations
 
 from dataclasses import asdict, dataclass, field
 from enum import StrEnum
-from typing import TYPE_CHECKING, Any
+from typing import Any
 
 from wf_core.models.reducers import ReducerRef
 from wf_core.models.workflow_refs import WorkflowRef
 from wf_core.paths import StatePath
-
+from wf_core.run_limits import RunLimits
-if TYPE_CHECKING:
-    from wf_core.runtime.limits import RunLimits
 
 ROOT_SCOPE_ID = "root"
 ROOT_LINEAGE_ID = "root"
@@ -193,23 +191,12 @@ class InterruptRequest:
     step_number: int | None = None
 
 
-def _default_run_limits() -> RunLimits:
-    """Build the default budget without a top-level runtime import.
-
-    Importing ``wf_core.runtime.limits`` at module top would execute the
-    ``wf_core.runtime`` package, whose engine imports this module back.
-    """
-    from wf_core.runtime.limits import RunLimits
-
-    return RunLimits()
-
-
 @dataclass(slots=True)
 class RunState:
     """Mutable execution state for one workflow run.
 
     ``limits``/``steps_executed`` form the persisted run-wide step budget (see
-    ``wf_core.runtime.limits``); ``steps_remaining`` is computed from them.
+    ``wf_core.run_limits``); ``steps_remaining`` is computed from them.
     """
 
     workflow_name: str
@@ -229,7 +216,7 @@ class RunState:
     activated_incoming_edge: str | None = None
     error: str | None = None
     interrupt: InterruptRequest | None = None
-    limits: RunLimits = field(default_factory=_default_run_limits)
+    limits: RunLimits = field(default_factory=RunLimits)
     steps_executed: int = 0
 
     @property
@@ -257,9 +244,3 @@ class RunState:
 
     def to_dict(self) -> dict[str, Any]:
         return asdict(self)
-
-
-# Deferred runtime import: binding ``RunLimits`` here (after every class is
-# defined) lets ``wf_core.runtime`` engine modules import this module back
-# without a cycle, and gives pydantic a resolvable annotation for the codec.
-from wf_core.runtime.limits import RunLimits  # noqa: E402

src/wf_core/runtime/engine.py

@@ -5,8 +5,8 @@ from typing import Any
 
 from wf_core.errors import WorkflowExecutionError
 from wf_core.models.workflow import Workflow
+from wf_core.run_limits import RunLimits
 from wf_core.run_state import ROOT_SCOPE_ID, RunState, RunStatus
-from wf_core.runtime.limits import RunLimits
 from wf_core.runtime.ops.flow import finalize_run
 from wf_core.runtime.ops.merges import ReducerDefinition
 from wf_core.runtime.ops.nodes import AsyncNodeHandler, NodeHandler

src/wf_core/runtime/limits.py

@@ -10,7 +10,6 @@ checkpoints; the counter is persisted inside the existing stopped-run envelope.
 
 from __future__ import annotations
 
-from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
 from wf_core.errors import WorkflowStepLimitExceeded
@@ -19,19 +18,6 @@ if TYPE_CHECKING:
     from wf_core.run_state import ExecutionFrame, RunState
 
 
-@dataclass(frozen=True, slots=True)
-class RunLimits:
-    """Immutable step budget captured when a run is created."""
-
-    max_steps: int = 10_000
-
-    def __post_init__(self) -> None:
-        if isinstance(self.max_steps, bool) or not isinstance(self.max_steps, int):
-            raise TypeError("max_steps must be an integer")
-        if self.max_steps < 1:
-            raise ValueError("max_steps must be positive")
-
-
 def admit_step_attempt(run: RunState, frame: ExecutionFrame, node_id: str) -> int:
     """Admit one step attempt for ``frame`` about to dispatch ``node_id``.
 
@@ -50,4 +36,4 @@ def admit_step_attempt(run: RunState, frame: ExecutionFrame, node_id: str) -> in
 
 def remaining_step_attempts(run: RunState) -> int:
     """Return the unspent budget, floored at zero (never negative)."""
-    return max(run.limits.max_steps - run.steps_executed, 0)
+    return run.steps_remaining

src/wf_core/runtime/ops/runs.py

@@ -4,6 +4,7 @@ from copy import deepcopy
 
 from wf_core.models.workflow import Workflow
 from wf_core.paths import set_nested_value
+from wf_core.run_limits import RunLimits
 from wf_core.run_state import (
     ROOT_FRAME_ID,
     ROOT_LINEAGE_ID,
@@ -15,7 +16,6 @@ from wf_core.run_state import (
     RunStatus,
     RuntimeScope,
 )
-from wf_core.runtime.limits import RunLimits
 from wf_core.runtime.scheduler import add_frame
 
 

tests/core/test_run_limits.py

@@ -0,0 +1,168 @@
+"""Run limit model and admission tests."""
+
+from __future__ import annotations
+
+from typing import Any, cast
+
+import pytest
+
+from wf_core import (
+    END,
+    Edge,
+    NodeDef,
+    NodeUse,
+    RunLimits,
+    SchemaRef,
+    StateSchema,
+    Workflow,
+    WorkflowExecutionError,
+)
+from wf_core.errors import WorkflowStepLimitExceeded
+from wf_core.runtime.limits import admit_step_attempt, remaining_step_attempts
+from wf_core.runtime.ops.runs import create_run_state
+
+
+def _minimal_workflow(name: str = "budget") -> Workflow:
+    return Workflow(
+        name=name,
+        input_schema=SchemaRef(type="object", properties={}),
+        state_schema=StateSchema.from_field_map({}),
+        output_schema=SchemaRef(type="object", properties={}),
+        node_defs=[
+            NodeDef(
+                name="finish",
+                input_schema=SchemaRef(type="object", properties={}),
+                output_schema=SchemaRef(type="object", properties={}),
+                outcomes=["ok"],
+            )
+        ],
+        start="finish",
+        nodes=[
+            NodeUse.model_validate({"id": "finish", "type": "node", "node": "finish"})
+        ],
+        edges=[Edge.model_validate({"from": "finish", "outcome": "ok", "to": END})],
+    )
+
+
+def test_run_limits_default() -> None:
+    limits = RunLimits()
+
+    assert limits.max_steps == 10_000
+
+
+def test_run_limits_rejects_non_positive() -> None:
+    with pytest.raises(ValueError):
+        RunLimits(max_steps=0)
+    with pytest.raises(ValueError):
+        RunLimits(max_steps=-3)
+
+
+def test_run_limits_rejects_bool_and_non_int() -> None:
+    with pytest.raises(TypeError):
+        RunLimits(max_steps=cast(Any, True))
+    with pytest.raises(TypeError):
+        RunLimits(max_steps=cast(Any, False))
+    with pytest.raises(TypeError):
+        RunLimits(max_steps=cast(Any, "10"))
+    with pytest.raises(TypeError):
+        RunLimits(max_steps=cast(Any, 10.0))
+
+
+def test_create_run_state_defaults_to_budget() -> None:
+    workflow = _minimal_workflow()
+
+    run = create_run_state(workflow, {})
+
+    assert run.limits.max_steps == 10_000
+    assert run.steps_executed == 0
+    assert run.steps_remaining == 10_000
+    assert run.current_frame().step_number is None
+    assert remaining_step_attempts(run) == 10_000
+
+
+def test_create_run_state_captures_limits() -> None:
+    workflow = _minimal_workflow()
+    limits = RunLimits(max_steps=5)
+
+    run = create_run_state(workflow, {}, limits=limits)
+
+    assert run.limits.max_steps == 5
+    assert run.steps_remaining == 5
+
+
+def test_budget_of_one() -> None:
+    workflow = _minimal_workflow()
+    limits = RunLimits(max_steps=1)
+    run = create_run_state(workflow, {}, limits=limits)
+    number = admit_step_attempt(run, run.current_frame(), workflow.start)
+
+    assert number == 1
+    assert run.steps_executed == 1
+    assert run.steps_remaining == 0
+    with pytest.raises(WorkflowStepLimitExceeded):
+        admit_step_attempt(run, run.current_frame(), workflow.start)
+
+
+def test_denied_admission_does_not_increment() -> None:
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=1))
+    admit_step_attempt(run, run.current_frame(), workflow.start)
+
+    with pytest.raises(WorkflowStepLimitExceeded):
+        admit_step_attempt(run, run.current_frame(), workflow.start)
+
+    assert run.steps_executed == 1
+    assert run.current_frame().step_number == 1
+    assert run.steps_remaining == 0
+    assert remaining_step_attempts(run) == 0
+
+
+def test_admission_assigns_step_numbers() -> None:
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=3))
+
+    first = admit_step_attempt(run, run.current_frame(), workflow.start)
+    second = admit_step_attempt(run, run.current_frame(), workflow.start)
+
+    assert first == 1
+    assert second == 2
+    assert run.steps_executed == 2
+    assert run.current_frame().step_number == 2
+    assert run.steps_remaining == 1
+    assert remaining_step_attempts(run) == 1
+
+
+def test_step_limit_error_details() -> None:
+    workflow = _minimal_workflow(name="budget_details")
+    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=1))
+    frame = run.current_frame()
+    admit_step_attempt(run, frame, workflow.start)
+
+    with pytest.raises(WorkflowStepLimitExceeded) as exc_info:
+        admit_step_attempt(run, frame, workflow.start)
+
+    assert isinstance(exc_info.value, WorkflowExecutionError)
+    message = str(exc_info.value)
+    assert "budget_details" in message
+    assert "1" in message
+    assert frame.id in message
+    assert frame.scope_id in message
+    assert workflow.start in message
+
+
+def test_remaining_never_negative() -> None:
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=1))
+    run.steps_executed = 5
+
+    assert run.steps_remaining == 0
+    assert remaining_step_attempts(run) == 0
+
+
+def test_remaining_delegates_to_run_property() -> None:
+    """`remaining_step_attempts` is the computed `steps_remaining` convenience."""
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=4))
+    admit_step_attempt(run, run.current_frame(), workflow.start)
+
+    assert remaining_step_attempts(run) == run.steps_remaining == 3

tests/core/test_run_step_budget.py

@@ -1,6 +1,16 @@
+"""Sync step-budget dispatch tests.
+
+Covers run-wide counting for every current step kind, trace numbering,
+and the engine-level limits seam. Model/admission unit tests live in
+`test_run_limits.py`; codec and migration tests live in
+`test_run_step_budget_codec.py`.
+"""
+
 from __future__ import annotations
 
 import inspect
+from collections.abc import Callable
+from typing import Any
 
 import pytest
 
@@ -15,30 +25,23 @@ from wf_core import (
     NodeUse,
     PreparedSubgraph,
     ReducerRef,
+    RunLimits,
     RunStatus,
     SchemaRef,
     StateField,
     StateSchema,
     SubgraphNode,
     Workflow,
-    WorkflowExecutionError,
-    dump_run_state,
     execute_workflow,
     execute_workflow_async,
     execute_workflow_result_async,
-    load_run_state,
     resume_workflow,
     resume_workflow_async,
     resume_workflow_result_async,
     step_workflow,
 )
 from wf_core.errors import WorkflowStepLimitExceeded
-from wf_core.run_codec import load_run_state_with_upgrade
+from wf_core.run_state import RunState
-from wf_core.runtime.limits import (
-    RunLimits,
-    admit_step_attempt,
-    remaining_step_attempts,
-)
 from wf_core.runtime.ops.runs import create_run_state
 from wf_core.runtime.preparation import prepare_resume
 
@@ -65,343 +68,15 @@ def _minimal_workflow(name: str = "budget") -> Workflow:
     )
 
 
-def test_run_limits_default() -> None:
-    limits = RunLimits()
-
-    assert limits.max_steps == 10_000
-
-
-def test_run_limits_rejects_non_positive() -> None:
-    with pytest.raises(ValueError):
-        RunLimits(max_steps=0)
-    with pytest.raises(ValueError):
-        RunLimits(max_steps=-3)
-
-
-def test_run_limits_rejects_bool_and_non_int() -> None:
-    with pytest.raises(TypeError):
-        RunLimits(max_steps=True)  # type: ignore[arg-type]
-    with pytest.raises(TypeError):
-        RunLimits(max_steps=False)  # type: ignore[arg-type]
-    with pytest.raises(TypeError):
-        RunLimits(max_steps="10")  # type: ignore[arg-type]
-    with pytest.raises(TypeError):
-        RunLimits(max_steps=10.0)  # type: ignore[arg-type]
-
-
-def test_create_run_state_defaults_to_budget() -> None:
-    workflow = _minimal_workflow()
-
-    run = create_run_state(workflow, {})
-
-    assert run.limits.max_steps == 10_000
-    assert run.steps_executed == 0
-    assert run.steps_remaining == 10_000
-    assert run.current_frame().step_number is None
-    assert remaining_step_attempts(run) == 10_000
-
-
-def test_create_run_state_captures_limits() -> None:
-    workflow = _minimal_workflow()
-    limits = RunLimits(max_steps=5)
-
-    run = create_run_state(workflow, {}, limits=limits)
-
-    assert run.limits.max_steps == 5
-    assert run.steps_remaining == 5
-
-
-def test_budget_of_one() -> None:
-    workflow = _minimal_workflow()
-    limits = RunLimits(max_steps=1)
-    run = create_run_state(workflow, {}, limits=limits)
-    number = admit_step_attempt(run, run.current_frame(), workflow.start)
-
-    assert number == 1
-    assert run.steps_executed == 1
-    assert run.steps_remaining == 0
-    with pytest.raises(WorkflowStepLimitExceeded):
-        admit_step_attempt(run, run.current_frame(), workflow.start)
-
-
-def test_denied_admission_does_not_increment() -> None:
-    workflow = _minimal_workflow()
-    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=1))
-    admit_step_attempt(run, run.current_frame(), workflow.start)
-
-    with pytest.raises(WorkflowStepLimitExceeded):
-        admit_step_attempt(run, run.current_frame(), workflow.start)
-
-    assert run.steps_executed == 1
-    assert run.current_frame().step_number == 1
-    assert run.steps_remaining == 0
-    assert remaining_step_attempts(run) == 0
-
-
-def test_admission_assigns_step_numbers() -> None:
-    workflow = _minimal_workflow()
-    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=3))
-
-    first = admit_step_attempt(run, run.current_frame(), workflow.start)
-    second = admit_step_attempt(run, run.current_frame(), workflow.start)
-
-    assert first == 1
-    assert second == 2
-    assert run.steps_executed == 2
-    assert run.current_frame().step_number == 2
-    assert run.steps_remaining == 1
-    assert remaining_step_attempts(run) == 1
-
-
-def test_step_limit_error_details() -> None:
-    workflow = _minimal_workflow(name="budget_details")
-    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=1))
-    frame = run.current_frame()
-    admit_step_attempt(run, frame, workflow.start)
-
-    with pytest.raises(WorkflowStepLimitExceeded) as exc_info:
-        admit_step_attempt(run, frame, workflow.start)
-
-    assert isinstance(exc_info.value, WorkflowExecutionError)
-    message = str(exc_info.value)
-    assert "budget_details" in message
-    assert "1" in message
-    assert frame.id in message
-    assert frame.scope_id in message
-    assert workflow.start in message
-
-
-def test_remaining_never_negative() -> None:
-    workflow = _minimal_workflow()
-    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=1))
-    run.steps_executed = 5
-
-    assert run.steps_remaining == 0
-    assert remaining_step_attempts(run) == 0
-
-
-def test_dump_writes_v2_envelope() -> None:
-    workflow = _minimal_workflow()
-    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=7))
-    admit_step_attempt(run, run.current_frame(), workflow.start)
-
-    stored = dump_run_state(run)
-
-    assert stored["version"] == 2
-
-
-def test_v2_round_trip() -> None:
-    workflow = _minimal_workflow()
-    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=7))
-    admit_step_attempt(run, run.current_frame(), workflow.start)
-
-    stored = dump_run_state(run)
-    restored, upgraded = load_run_state_with_upgrade(stored)
-
-    assert upgraded is False
-    assert restored.limits.max_steps == 7
-    assert restored.steps_executed == 1
-    assert restored.steps_remaining == 6
-    assert restored.frames["root"].step_number == 1
-
-    via_legacy = load_run_state(stored)
-    assert via_legacy.limits.max_steps == 7
-    assert via_legacy.steps_executed == 1
-    assert via_legacy.frames["root"].step_number == 1
-
-
-def _strip_to_v1(stored: dict) -> dict:
-    state = dict(stored["state"])
-    state.pop("limits", None)
-    state.pop("steps_executed", None)
-    frames = {
-        frame_id: {key: value for key, value in frame.items() if key != "step_number"}
-        for frame_id, frame in dict(state["frames"]).items()
-    }
-    state["frames"] = frames
-    return {"version": 1, "state": state}
-
-
-def test_v1_payload_receives_defaults_once() -> None:
-    workflow = _minimal_workflow()
-    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=7))
-    stored = _strip_to_v1(dump_run_state(run))
-
-    restored, upgraded = load_run_state_with_upgrade(stored)
-
-    assert upgraded is True
-    assert restored.limits.max_steps == 10_000
-    assert restored.steps_executed == 0
-    assert restored.steps_remaining == 10_000
-    assert restored.frames["root"].step_number is None
-
-    via_legacy = load_run_state(stored)
-    assert via_legacy.limits.max_steps == 10_000
-    assert via_legacy.steps_executed == 0
-
-
-def test_v2_missing_limits_is_corrupt() -> None:
-    workflow = _minimal_workflow()
-    run = create_run_state(workflow, {})
-    stored = dump_run_state(run)
-    stored["state"].pop("limits")
-
-    with pytest.raises(ValueError):
-        load_run_state_with_upgrade(stored)
-
-
-def test_v2_missing_steps_executed_is_corrupt() -> None:
-    workflow = _minimal_workflow()
-    run = create_run_state(workflow, {})
-    stored = dump_run_state(run)
-    stored["state"].pop("steps_executed")
-
-    with pytest.raises(ValueError):
-        load_run_state_with_upgrade(stored)
-
-
-def test_v2_missing_frame_step_number_is_corrupt() -> None:
-    workflow = _minimal_workflow()
-    run = create_run_state(workflow, {})
-    stored = dump_run_state(run)
-    del stored["state"]["frames"]["root"]["step_number"]
-
-    with pytest.raises(ValueError):
-        load_run_state_with_upgrade(stored)
-
-
-def test_v1_smuggled_budget_fields_receive_defaults() -> None:
-    """V1 envelopes predate budgets; smuggled fields must not survive."""
-    from typing import Any, cast
-
-    workflow = _minimal_workflow()
-    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=7))
-    stored = _strip_to_v1(dump_run_state(run))
-    state = cast(dict[str, Any], stored["state"])
-    state["limits"] = {"max_steps": 999_999}
-    state["steps_executed"] = 999
-
-    restored, upgraded = load_run_state_with_upgrade(stored)
-
-    assert upgraded is True
-    assert restored.limits.max_steps == 10_000
-    assert restored.steps_executed == 0
-
-
-def test_v2_bool_max_steps_is_corrupt() -> None:
-    from typing import Any, cast
-
-    workflow = _minimal_workflow()
-    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=2))
-    stored = dump_run_state(run)
-    cast(dict[str, Any], stored["state"])["limits"] = {"max_steps": True}
-
-    with pytest.raises(ValueError):
-        load_run_state_with_upgrade(stored)
-
-
-def test_v2_str_max_steps_is_corrupt() -> None:
-    from typing import Any, cast
-
-    workflow = _minimal_workflow()
-    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=2))
-    stored = dump_run_state(run)
-    cast(dict[str, Any], stored["state"])["limits"] = {"max_steps": "10"}
-
-    with pytest.raises(ValueError):
-        load_run_state_with_upgrade(stored)
-
-
-def test_v2_negative_steps_executed_is_corrupt() -> None:
-    from typing import Any, cast
-
-    workflow = _minimal_workflow()
-    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=2))
-    stored = dump_run_state(run)
-    cast(dict[str, Any], stored["state"])["steps_executed"] = -100
-
-    with pytest.raises(ValueError):
-        load_run_state_with_upgrade(stored)
-
-
-def test_v2_exceeding_steps_executed_is_corrupt() -> None:
-    from typing import Any, cast
-
-    workflow = _minimal_workflow()
-    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=2))
-    stored = dump_run_state(run)
-    cast(dict[str, Any], stored["state"])["steps_executed"] = 5
-
-    with pytest.raises(ValueError):
-        load_run_state_with_upgrade(stored)
-
-
-def test_v2_bool_steps_executed_is_corrupt() -> None:
-    from typing import Any, cast
-
-    workflow = _minimal_workflow()
-    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=2))
-    stored = dump_run_state(run)
-    cast(dict[str, Any], stored["state"])["steps_executed"] = True
-
-    with pytest.raises(ValueError):
-        load_run_state_with_upgrade(stored)
-
-
-def test_v2_incoherent_frame_step_number_is_corrupt() -> None:
-    from typing import Any, cast
-
-    workflow = _minimal_workflow()
-    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=2))
-    admit_step_attempt(run, run.current_frame(), workflow.start)
-    stored = dump_run_state(run)
-    state = cast(dict[str, Any], stored["state"])
-    frames = cast(dict[str, Any], state["frames"])
-    cast(dict[str, Any], frames["root"])["step_number"] = 999
-
-    with pytest.raises(ValueError):
-        load_run_state_with_upgrade(stored)
-
-
-def test_v2_bool_frame_step_number_is_corrupt() -> None:
-    from typing import Any, cast
-
-    workflow = _minimal_workflow()
-    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=2))
-    admit_step_attempt(run, run.current_frame(), workflow.start)
-    stored = dump_run_state(run)
-    state = cast(dict[str, Any], stored["state"])
-    frames = cast(dict[str, Any], state["frames"])
-    cast(dict[str, Any], frames["root"])["step_number"] = True
-
-    with pytest.raises(ValueError):
-        load_run_state_with_upgrade(stored)
-
-
-def test_v2_pending_frame_none_number_round_trips() -> None:
-    """Fresh V2 runs legitimately persist unadmitted (None) frame numbers."""
-    workflow = _minimal_workflow()
-    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=2))
-
-    restored, upgraded = load_run_state_with_upgrade(dump_run_state(run))
-
-    assert upgraded is False
-    assert restored.steps_executed == 0
-    assert restored.frames["root"].step_number is None
-
-
-# --- Task 2: sync dispatch and trace numbering ---
-
-
 def _empty_schema() -> SchemaRef:
     return SchemaRef(type="object", properties={})
 
 
-def _ok_handler(payload: dict, _context: object) -> dict:
+def _ok_handler(payload: dict[str, Any], _context: object) -> dict[str, Any]:
     return {"outcome": "ok", "output": {}}
 
 
-def _trace_numbers(run) -> list:
+def _trace_numbers(run: RunState) -> list[int | None]:
     return [entry.step_number for entry in run.trace]
 
 
@@ -557,16 +232,10 @@ def _serial_foreach_workflow() -> Workflow:
 def test_sync_foreach_controller_and_body_share_counter() -> None:
     workflow = _serial_foreach_workflow()
 
-    run = execute_workflow(
+    def record(payload: dict[str, Any], _context: object) -> dict[str, Any]:
-        workflow,
+        return {"outcome": "ok", "output": {"seen": payload["value"]}}
-        {"items": ["a", "b"]},
+
-        {
+    run = execute_workflow(workflow, {"items": ["a", "b"]}, {"record": record})
-            "record": lambda payload, _ctx: {
-                "outcome": "ok",
-                "output": {"seen": payload["value"]},
-            }
-        },
-    )
 
     assert run.status == RunStatus.COMPLETED
     assert run.steps_executed == 5
@@ -741,7 +410,7 @@ def test_sync_explicit_end_counts() -> None:
         edges=[Edge.model_validate({"from": "finish", "outcome": "done", "to": "end"})],
     )
 
-    def finish(_payload: dict, _context: object) -> dict:
+    def finish(_payload: dict[str, Any], _context: object) -> dict[str, Any]:
         return {"outcome": "done", "output": {}}
 
     run = execute_workflow(workflow, {}, {"finish": finish})
@@ -766,7 +435,7 @@ def test_sync_legacy_end_creates_no_extra_attempt() -> None:
 def test_sync_handler_failure_consumes_attempt() -> None:
     workflow = _minimal_workflow()
 
-    def explode(_payload: dict, _context: object) -> dict:
+    def explode(_payload: dict[str, Any], _context: object) -> dict[str, Any]:
         raise ValueError("boom")
 
     run = create_run_state(workflow, {})
@@ -801,7 +470,7 @@ def test_sync_handled_error_outcome_counts_once() -> None:
         ],
     )
 
-    def fail_soft(_payload: dict, _context: object) -> dict:
+    def fail_soft(_payload: dict[str, Any], _context: object) -> dict[str, Any]:
         return {"outcome": "error", "output": {}}
 
     run = execute_workflow(workflow, {}, {"risky": fail_soft})
@@ -850,8 +519,10 @@ def test_sync_closed_cycle_fails_at_limit() -> None:
     workflow = _cyclic_workflow()
     calls: list[str] = []
 
-    def make(name: str):  # type: ignore[no-untyped-def]
+    def make(
-        def handler(_payload: dict, _context: object) -> dict:
+        name: str,
+    ) -> Callable[[dict[str, Any], object], dict[str, Any]]:
+        def handler(_payload: dict[str, Any], _context: object) -> dict[str, Any]:
             calls.append(name)
             return {"outcome": "ok", "output": {}}
 
@@ -925,7 +596,7 @@ def _counting_loop_workflow() -> Workflow:
 def test_sync_exiting_loop_completes_within_budget() -> None:
     workflow = _counting_loop_workflow()
 
-    def bump(payload: dict, _context: object) -> dict:
+    def bump(payload: dict[str, Any], _context: object) -> dict[str, Any]:
         count = payload.get("count", 0)
         assert isinstance(count, int)
         return {"outcome": "ok", "output": {"count": count + 1}}
@@ -939,9 +610,9 @@ def test_sync_exiting_loop_completes_within_budget() -> None:
 
 def test_sync_denial_never_invokes_handler() -> None:
     workflow = _chain_workflow()
-    b_calls: list[dict] = []
+    b_calls: list[dict[str, Any]] = []
 
-    def b_handler(payload: dict, _context: object) -> dict:
+    def b_handler(payload: dict[str, Any], _context: object) -> dict[str, Any]:
         b_calls.append(payload)
         return {"outcome": "ok", "output": {}}
 
@@ -955,57 +626,6 @@ def test_sync_denial_never_invokes_handler() -> None:
     assert b_calls == []
 
 
-def _strip_budget_fields(stored: dict) -> dict:
-    state = dict(_strip_to_v1(stored)["state"])
-    state["trace"] = [
-        {key: value for key, value in entry.items() if key != "step_number"}
-        for entry in state.get("trace", [])
-    ]
-    if state.get("interrupt") is not None:
-        state["interrupt"] = {
-            key: value
-            for key, value in state["interrupt"].items()
-            if key != "step_number"
-        }
-    return {"version": 1, "state": state}
-
-
-def test_v1_traces_and_interrupt_receive_none_step_numbers() -> None:
-    workflow = _interrupt_workflow()
-    run = execute_workflow(workflow, {}, {"work": _ok_handler})
-    stored = _strip_budget_fields(dump_run_state(run))
-
-    restored, upgraded = load_run_state_with_upgrade(stored)
-
-    assert upgraded is True
-    assert restored.trace[0].step_number is None
-    assert restored.interrupt is not None
-    assert restored.interrupt.step_number is None
-
-
-def test_v2_missing_trace_step_number_is_corrupt() -> None:
-    workflow = _interrupt_workflow()
-    run = execute_workflow(workflow, {}, {"work": _ok_handler})
-    stored = dump_run_state(run)
-    del stored["state"]["trace"][0]["step_number"]
-
-    with pytest.raises(ValueError):
-        load_run_state_with_upgrade(stored)
-
-
-def test_v2_missing_interrupt_step_number_is_corrupt() -> None:
-    workflow = _interrupt_workflow()
-    run = execute_workflow(workflow, {}, {"work": _ok_handler})
-    stored = dump_run_state(run)
-    del stored["state"]["interrupt"]["step_number"]
-
-    with pytest.raises(ValueError):
-        load_run_state_with_upgrade(stored)
-
-
-# --- Task 4: engine-level limits seam ---
-
-
 def test_execute_workflow_accepts_explicit_limits() -> None:
     workflow = _chain_workflow()
 
@@ -1034,7 +654,7 @@ def test_execute_workflow_defaults_to_ten_thousand() -> None:
 def test_execute_workflow_enforces_limits() -> None:
     workflow = _cyclic_workflow()
 
-    def ok_handler(_payload: dict, _context: object) -> dict:
+    def ok_handler(_payload: dict[str, Any], _context: object) -> dict[str, Any]:
         return {"outcome": "ok", "output": {}}
 
     with pytest.raises(WorkflowStepLimitExceeded):
@@ -1049,7 +669,7 @@ def test_execute_workflow_enforces_limits() -> None:
 async def test_execute_workflow_async_accepts_explicit_limits() -> None:
     workflow = _chain_workflow()
 
-    async def ok_async(_payload: dict, _context: object) -> dict:
+    async def ok_async(_payload: dict[str, Any], _context: object) -> dict[str, Any]:
         return {"outcome": "ok", "output": {}}
 
     run = await execute_workflow_async(
@@ -1068,7 +688,7 @@ async def test_execute_workflow_async_accepts_explicit_limits() -> None:
 async def test_execute_workflow_result_async_reports_exhaustion() -> None:
     workflow = _cyclic_workflow()
 
-    async def ok_async(_payload: dict, _context: object) -> dict:
+    async def ok_async(_payload: dict[str, Any], _context: object) -> dict[str, Any]:
         return {"outcome": "ok", "output": {}}
 
     run = await execute_workflow_result_async(

tests/core/test_run_step_budget_async.py

@@ -9,6 +9,7 @@ later sibling commits after the first unhandled result in reservation order.
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Callable
 from typing import Any
 
 import pytest
@@ -21,6 +22,7 @@ from wf_core import (
     NodeDef,
     NodeUse,
     ReducerRef,
+    RunLimits,
     RunStatus,
     SchemaRef,
     StateField,
@@ -31,11 +33,19 @@ from wf_core import (
     step_workflow_async,
 )
 from wf_core.errors import WorkflowStepLimitExceeded
-from wf_core.runtime.limits import RunLimits, remaining_step_attempts
+from wf_core.runtime.limits import remaining_step_attempts
 from wf_core.runtime.ops.runs import create_run_state
 from wf_core.runtime.preparation import prepare_new_run, prepare_resume
 
 
+async def _noop_record(payload: dict[str, Any], _context: object) -> dict[str, Any]:
+    return {"outcome": "ok", "output": payload}
+
+
+def _sync_noop_record(payload: dict[str, Any], _context: object) -> dict[str, Any]:
+    return {"outcome": "ok", "output": payload}
+
+
 def _concurrent_workflow(*, max_active: int, name: str = "async_budget") -> Workflow:
     return Workflow(
         name=name,
@@ -102,17 +112,19 @@ def _concurrent_workflow(*, max_active: int, name: str = "async_budget") -> Work
     )
 
 
-def _prepare_limited_run(
+def _prepare_limited_run(workflow: Workflow, items: list[Any], *, max_steps: int):
-    workflow: Workflow, items: list[Any], *, max_steps: int
+    run = create_run_state(
-):
+        workflow, {"items": items}, limits=RunLimits(max_steps=max_steps)
-    run = create_run_state(workflow, {"items": items}, limits=RunLimits(max_steps=max_steps))
+    )
     prepare_new_run(workflow, {"items": items}, run)
-    index = prepare_resume(workflow, run, resume_payload=None, resume_outcome="submitted")
+    index = prepare_resume(
+        workflow, run, resume_payload=None, resume_outcome="submitted"
+    )
     assert index is not None
     return run, index
 
 
-async def _wait_for(predicate, *, timeout: float = 2.0) -> None:  # type: ignore[no-untyped-def]
+async def _wait_for(predicate: Callable[[], bool], *, timeout: float = 2.0) -> None:
     async def _poll() -> None:
         while not predicate():
             await asyncio.sleep(0.005)
@@ -126,10 +138,7 @@ async def test_async_batch_bounded_to_remaining_budget() -> None:
     items = ["a", "b", "c", "d", "e"]
     run, index = _prepare_limited_run(workflow, items, max_steps=4)
 
-    async def _noop(payload: dict[str, Any], _ctx: object) -> dict[str, Any]:
+    await step_workflow_async(workflow, run, {"record": _noop_record}, index=index)
-        return {"outcome": "ok", "output": payload}
-
-    await step_workflow_async(workflow, run, {"record": _noop}, index=index)
     assert run.steps_executed == 1
     assert remaining_step_attempts(run) == 3
     assert run.ready_frame_ids == [f"root:each#0:{i}" for i in range(5)]
@@ -189,10 +198,7 @@ async def test_async_batch_denies_first_when_remaining_zero() -> None:
     items = ["a", "b", "c"]
     run, index = _prepare_limited_run(workflow, items, max_steps=1)
 
-    async def _noop(payload: dict[str, Any], _ctx: object) -> dict[str, Any]:
+    await step_workflow_async(workflow, run, {"record": _noop_record}, index=index)
-        return {"outcome": "ok", "output": payload}
-
-    await step_workflow_async(workflow, run, {"record": _noop}, index=index)
     assert run.steps_executed == 1
     assert remaining_step_attempts(run) == 0
 
@@ -224,10 +230,7 @@ async def test_async_batch_numbers_follow_queue_order_not_completion() -> None:
     items = ["a", "b", "c"]
     run, index = _prepare_limited_run(workflow, items, max_steps=20)
 
-    async def _noop(payload: dict[str, Any], _ctx: object) -> dict[str, Any]:
+    await step_workflow_async(workflow, run, {"record": _noop_record}, index=index)
-        return {"outcome": "ok", "output": payload}
-
-    await step_workflow_async(workflow, run, {"record": _noop}, index=index)
     assert run.steps_executed == 1
 
     allow_a = asyncio.Event()
@@ -282,10 +285,7 @@ async def test_async_batch_reservations_kept_after_failure() -> None:
     items = ["a", "b", "c"]
     run, index = _prepare_limited_run(workflow, items, max_steps=20)
 
-    async def _noop(payload: dict[str, Any], _ctx: object) -> dict[str, Any]:
+    await step_workflow_async(workflow, run, {"record": _noop_record}, index=index)
-        return {"outcome": "ok", "output": payload}
-
-    await step_workflow_async(workflow, run, {"record": _noop}, index=index)
     base_steps = run.steps_executed
     assert base_steps == 1
 
@@ -321,10 +321,7 @@ async def test_async_batch_settles_siblings_and_discards_later_commits() -> None
     items = ["a", "b", "c"]
     run, index = _prepare_limited_run(workflow, items, max_steps=20)
 
-    async def _noop(payload: dict[str, Any], _ctx: object) -> dict[str, Any]:
+    await step_workflow_async(workflow, run, {"record": _noop_record}, index=index)
-        return {"outcome": "ok", "output": payload}
-
-    await step_workflow_async(workflow, run, {"record": _noop}, index=index)
 
     release = asyncio.Event()
     started: list[str] = []
@@ -373,7 +370,9 @@ async def test_sync_async_parity_for_serial_execution() -> None:
     def _serial_workflow(name: str) -> Workflow:
         return Workflow(
             name=name,
-            input_schema=SchemaRef(type="object", properties={"items": {"type": "array"}}),
+            input_schema=SchemaRef(
+                type="object", properties={"items": {"type": "array"}}
+            ),
             state_schema=StateSchema.from_field_map(
                 {
                     "items": StateField(type="array"),
@@ -383,7 +382,9 @@ async def test_sync_async_parity_for_serial_execution() -> None:
                     ),
                 }
             ),
-            output_schema=SchemaRef(type="object", properties={"seen": {"type": "array"}}),
+            output_schema=SchemaRef(
+                type="object", properties={"seen": {"type": "array"}}
+            ),
             node_defs=[
                 NodeDef(
                     name="record",
@@ -426,7 +427,9 @@ async def test_sync_async_parity_for_serial_execution() -> None:
                 ),
             ],
             edges=[
-                Edge.model_validate({"from": "each", "outcome": "loop", "to": "record"}),
+                Edge.model_validate(
+                    {"from": "each", "outcome": "loop", "to": "record"}
+                ),
                 Edge.model_validate({"from": "record", "outcome": "ok", "to": "each"}),
                 Edge.model_validate({"from": "each", "outcome": "done", "to": END}),
             ],
@@ -438,7 +441,7 @@ async def test_sync_async_parity_for_serial_execution() -> None:
     sync_run = execute_workflow(
         sync_workflow,
         {"items": ["a", "b"]},
-        {"record": lambda payload, _ctx: {"outcome": "ok", "output": payload}},
+        {"record": _sync_noop_record},
     )
 
     async def record(payload: dict[str, Any], _ctx: object) -> dict[str, Any]:

tests/core/test_run_step_budget_codec.py

@@ -0,0 +1,351 @@
+"""Step budget codec and migration tests."""
+
+from __future__ import annotations
+
+from typing import Any, cast
+
+import pytest
+
+from wf_core import (
+    END,
+    Edge,
+    InterruptNode,
+    NodeDef,
+    NodeUse,
+    RunLimits,
+    SchemaRef,
+    StateSchema,
+    Workflow,
+    dump_run_state,
+    execute_workflow,
+    load_run_state,
+)
+from wf_core.run_codec import load_run_state_with_upgrade
+from wf_core.run_state import RunState
+from wf_core.runtime.limits import admit_step_attempt
+from wf_core.runtime.ops.runs import create_run_state
+
+
+def _minimal_workflow(name: str = "budget") -> Workflow:
+    return Workflow(
+        name=name,
+        input_schema=SchemaRef(type="object", properties={}),
+        state_schema=StateSchema.from_field_map({}),
+        output_schema=SchemaRef(type="object", properties={}),
+        node_defs=[
+            NodeDef(
+                name="finish",
+                input_schema=SchemaRef(type="object", properties={}),
+                output_schema=SchemaRef(type="object", properties={}),
+                outcomes=["ok"],
+            )
+        ],
+        start="finish",
+        nodes=[
+            NodeUse.model_validate({"id": "finish", "type": "node", "node": "finish"})
+        ],
+        edges=[Edge.model_validate({"from": "finish", "outcome": "ok", "to": END})],
+    )
+
+
+def _empty_schema() -> SchemaRef:
+    return SchemaRef(type="object", properties={})
+
+
+def _ok_handler(payload: dict[str, Any], _context: object) -> dict[str, Any]:
+    return {"outcome": "ok", "output": {}}
+
+
+def _interrupt_workflow() -> Workflow:
+    return Workflow(
+        name="interrupt_counts",
+        input_schema=_empty_schema(),
+        state_schema=StateSchema.from_field_map({}),
+        output_schema=_empty_schema(),
+        outcomes=["ok"],
+        node_defs=[
+            NodeDef(
+                name="work",
+                input_schema=_empty_schema(),
+                output_schema=_empty_schema(),
+                outcomes=["ok"],
+            )
+        ],
+        start="ask",
+        nodes=[
+            InterruptNode(id="ask", type="interrupt", kind="approval"),
+            NodeUse(id="work", type="node", node="work"),
+        ],
+        edges=[
+            Edge.model_validate({"from": "ask", "outcome": "submitted", "to": "work"}),
+            Edge.model_validate({"from": "work", "outcome": "ok", "to": END}),
+        ],
+    )
+
+
+def _state_dict(stored: dict[str, object]) -> dict[str, Any]:
+    return cast(dict[str, Any], stored["state"])
+
+
+def _strip_to_v1(stored: dict[str, object]) -> dict[str, Any]:
+    state = dict(_state_dict(stored))
+    state.pop("limits", None)
+    state.pop("steps_executed", None)
+    frames = {
+        frame_id: {key: value for key, value in frame.items() if key != "step_number"}
+        for frame_id, frame in dict(cast(dict[str, Any], state["frames"])).items()
+    }
+    state["frames"] = frames
+    return {"version": 1, "state": state}
+
+
+def _strip_budget_fields(stored: dict[str, object]) -> dict[str, Any]:
+    state = dict(_strip_to_v1(stored)["state"])
+    state["trace"] = [
+        {key: value for key, value in entry.items() if key != "step_number"}
+        for entry in cast(list[dict[str, Any]], state.get("trace", []))
+    ]
+    if state.get("interrupt") is not None:
+        state["interrupt"] = {
+            key: value
+            for key, value in cast(dict[str, Any], state["interrupt"]).items()
+            if key != "step_number"
+        }
+    return {"version": 1, "state": state}
+
+
+def test_dump_writes_v2_envelope() -> None:
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=7))
+    admit_step_attempt(run, run.current_frame(), workflow.start)
+
+    stored = dump_run_state(run)
+
+    assert stored["version"] == 2
+
+
+def test_v2_round_trip() -> None:
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=7))
+    admit_step_attempt(run, run.current_frame(), workflow.start)
+
+    stored = dump_run_state(run)
+    restored, upgraded = load_run_state_with_upgrade(stored)
+
+    assert upgraded is False
+    assert restored.limits.max_steps == 7
+    assert restored.steps_executed == 1
+    assert restored.steps_remaining == 6
+    assert restored.frames["root"].step_number == 1
+
+    via_legacy = load_run_state(stored)
+    assert via_legacy.limits.max_steps == 7
+    assert via_legacy.steps_executed == 1
+    assert via_legacy.frames["root"].step_number == 1
+
+
+def test_v1_payload_receives_defaults_once() -> None:
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=7))
+    stored = _strip_to_v1(dump_run_state(run))
+
+    restored, upgraded = load_run_state_with_upgrade(stored)
+
+    assert upgraded is True
+    assert restored.limits.max_steps == 10_000
+    assert restored.steps_executed == 0
+    assert restored.steps_remaining == 10_000
+    assert restored.frames["root"].step_number is None
+
+    via_legacy = load_run_state(stored)
+    assert via_legacy.limits.max_steps == 10_000
+    assert via_legacy.steps_executed == 0
+
+
+def test_v2_missing_limits_is_corrupt() -> None:
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {})
+    stored = dump_run_state(run)
+    _state_dict(stored).pop("limits")
+
+    with pytest.raises(ValueError):
+        load_run_state_with_upgrade(stored)
+
+
+def test_v2_missing_steps_executed_is_corrupt() -> None:
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {})
+    stored = dump_run_state(run)
+    _state_dict(stored).pop("steps_executed")
+
+    with pytest.raises(ValueError):
+        load_run_state_with_upgrade(stored)
+
+
+def test_v2_missing_frame_step_number_is_corrupt() -> None:
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {})
+    stored = dump_run_state(run)
+    del cast(dict[str, Any], _state_dict(stored)["frames"])["root"]["step_number"]
+
+    with pytest.raises(ValueError):
+        load_run_state_with_upgrade(stored)
+
+
+def test_v1_smuggled_budget_fields_receive_defaults() -> None:
+    """V1 envelopes predate budgets; smuggled fields must not survive."""
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=7))
+    stored = _strip_to_v1(dump_run_state(run))
+    state = cast(dict[str, Any], stored["state"])
+    state["limits"] = {"max_steps": 999_999}
+    state["steps_executed"] = 999
+
+    restored, upgraded = load_run_state_with_upgrade(stored)
+
+    assert upgraded is True
+    assert restored.limits.max_steps == 10_000
+    assert restored.steps_executed == 0
+
+
+def test_v2_bool_max_steps_is_corrupt() -> None:
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=2))
+    stored = dump_run_state(run)
+    _state_dict(stored)["limits"] = {"max_steps": True}
+
+    with pytest.raises(ValueError):
+        load_run_state_with_upgrade(stored)
+
+
+def test_v2_str_max_steps_is_corrupt() -> None:
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=2))
+    stored = dump_run_state(run)
+    _state_dict(stored)["limits"] = {"max_steps": "10"}
+
+    with pytest.raises(ValueError):
+        load_run_state_with_upgrade(stored)
+
+
+def test_v2_unknown_limits_field_is_corrupt() -> None:
+    """Strict v2: unknown fields inside limits are corrupt, not preserved."""
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=2))
+    stored = dump_run_state(run)
+    _state_dict(stored)["limits"] = {"max_steps": 2, "future_quota": 99}
+
+    with pytest.raises(ValueError):
+        load_run_state_with_upgrade(stored)
+
+
+def test_v2_negative_steps_executed_is_corrupt() -> None:
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=2))
+    stored = dump_run_state(run)
+    _state_dict(stored)["steps_executed"] = -100
+
+    with pytest.raises(ValueError):
+        load_run_state_with_upgrade(stored)
+
+
+def test_v2_exceeding_steps_executed_is_corrupt() -> None:
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=2))
+    stored = dump_run_state(run)
+    _state_dict(stored)["steps_executed"] = 5
+
+    with pytest.raises(ValueError):
+        load_run_state_with_upgrade(stored)
+
+
+def test_v2_bool_steps_executed_is_corrupt() -> None:
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=2))
+    stored = dump_run_state(run)
+    _state_dict(stored)["steps_executed"] = True
+
+    with pytest.raises(ValueError):
+        load_run_state_with_upgrade(stored)
+
+
+def test_v2_incoherent_frame_step_number_is_corrupt() -> None:
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=2))
+    admit_step_attempt(run, run.current_frame(), workflow.start)
+    stored = dump_run_state(run)
+    state = _state_dict(stored)
+    frames = cast(dict[str, Any], state["frames"])
+    cast(dict[str, Any], frames["root"])["step_number"] = 999
+
+    with pytest.raises(ValueError):
+        load_run_state_with_upgrade(stored)
+
+
+def test_v2_bool_frame_step_number_is_corrupt() -> None:
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=2))
+    admit_step_attempt(run, run.current_frame(), workflow.start)
+    stored = dump_run_state(run)
+    state = _state_dict(stored)
+    frames = cast(dict[str, Any], state["frames"])
+    cast(dict[str, Any], frames["root"])["step_number"] = True
+
+    with pytest.raises(ValueError):
+        load_run_state_with_upgrade(stored)
+
+
+def test_v2_pending_frame_none_number_round_trips() -> None:
+    """Fresh V2 runs legitimately persist unadmitted (None) frame numbers."""
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {}, limits=RunLimits(max_steps=2))
+
+    restored, upgraded = load_run_state_with_upgrade(dump_run_state(run))
+
+    assert upgraded is False
+    assert restored.steps_executed == 0
+    assert restored.frames["root"].step_number is None
+
+
+def test_v1_traces_and_interrupt_receive_none_step_numbers() -> None:
+    workflow = _interrupt_workflow()
+    run = execute_workflow(workflow, {}, {"work": _ok_handler})
+    stored = _strip_budget_fields(dump_run_state(run))
+
+    restored, upgraded = load_run_state_with_upgrade(stored)
+
+    assert upgraded is True
+    assert restored.trace[0].step_number is None
+    assert restored.interrupt is not None
+    assert restored.interrupt.step_number is None
+
+
+def test_v2_missing_trace_step_number_is_corrupt() -> None:
+    workflow = _interrupt_workflow()
+    run = execute_workflow(workflow, {}, {"work": _ok_handler})
+    stored = dump_run_state(run)
+    del cast(dict[str, Any], cast(list[Any], _state_dict(stored)["trace"])[0])[
+        "step_number"
+    ]
+
+    with pytest.raises(ValueError):
+        load_run_state_with_upgrade(stored)
+
+
+def test_v2_missing_interrupt_step_number_is_corrupt() -> None:
+    workflow = _interrupt_workflow()
+    run = execute_workflow(workflow, {}, {"work": _ok_handler})
+    stored = dump_run_state(run)
+    del cast(dict[str, Any], _state_dict(stored)["interrupt"])["step_number"]
+
+    with pytest.raises(ValueError):
+        load_run_state_with_upgrade(stored)
+
+
+def test_restored_run_state_type() -> None:
+    workflow = _minimal_workflow()
+    run = create_run_state(workflow, {})
+
+    restored = load_run_state(dump_run_state(run))
+
+    assert isinstance(restored, RunState)

tests/wf_api/test_resume_concurrency.py

@@ -17,7 +17,7 @@ from wf_artifacts import (
     WorkflowDeployment,
 )
 from wf_authoring import NodeSpec
-from wf_core import InterruptRequest, RunState, RunStatus
+from wf_core import InterruptRequest, RunLimits, RunState, RunStatus
 from wf_platform import CapabilitySource
 
 
@@ -57,6 +57,7 @@ class BlockingResumeRuntime:
         deployment: WorkflowDeployment | None = None,
         artifact: WorkflowArtifact | None = None,
         saved_subgraph_tree: SavedSubgraphTree | None = None,
+        limits: RunLimits | None = None,
     ) -> RunState:
         raise AssertionError("test should not start new workflow runs")
 

tests/wf_api/test_source_admin_api.py

@@ -12,7 +12,7 @@ from wf_api.saved_subgraphs import SavedSubgraphTree
 from wf_api.source_admin import WorkflowSourceDiagnosticsProvider
 from wf_artifacts import WorkflowArtifact, WorkflowDeployment
 from wf_authoring import NodeSpec
-from wf_core import RunState
+from wf_core import RunLimits, RunState
 from wf_platform import (
     CapabilityBuckets,
     CapabilitySource,
@@ -43,6 +43,7 @@ class DummyRuntime:
         deployment: WorkflowDeployment | None = None,
         artifact: WorkflowArtifact | None = None,
         saved_subgraph_tree: SavedSubgraphTree | None = None,
+        limits: RunLimits | None = None,
     ) -> RunState:
         raise AssertionError("source admin tests must not run workflows")