From fd48a67819414167469f5d2b697a8301954f835f Mon Sep 17 00:00:00 2001 From: lda Date: Sat, 5 Sep 2026 23:15:57 +0700 Subject: [PATCH] docs: complete run step budget slice --- docs/README.md | 2 +- docs/current_roadmap.md | 22 +- .../plans/2026-09-05-run-step-budget.md | 50 +- docs/wf_cli.md | 34 ++ docs/wf_core_architecture.md | 6 + .../references/workflow-lifecycle.md | 17 +- src/wf_core/__init__.py | 3 +- src/wf_core/run_codec.py | 35 +- src/wf_core/run_limits.py | 28 ++ src/wf_core/run_state.py | 27 +- src/wf_core/runtime/engine.py | 2 +- src/wf_core/runtime/limits.py | 16 +- src/wf_core/runtime/ops/runs.py | 2 +- tests/core/test_run_limits.py | 168 +++++++ tests/core/test_run_step_budget.py | 442 ++---------------- tests/core/test_run_step_budget_async.py | 65 +-- tests/core/test_run_step_budget_codec.py | 351 ++++++++++++++ tests/wf_api/test_resume_concurrency.py | 3 +- tests/wf_api/test_source_admin_api.py | 3 +- 19 files changed, 735 insertions(+), 541 deletions(-) rename docs/{ => historical}/superpowers/plans/2026-09-05-run-step-budget.md (89%) create mode 100644 src/wf_core/run_limits.py create mode 100644 tests/core/test_run_limits.py create mode 100644 tests/core/test_run_step_budget_codec.py diff --git a/docs/README.md b/docs/README.md index 36a6dd39..6e848dd7 100644 --- a/docs/README.md +++ b/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): diff --git a/docs/current_roadmap.md b/docs/current_roadmap.md index 22f29c95..bdd76861 100644 --- a/docs/current_roadmap.md +++ b/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, -scheduler, lineage, and barrier foundations in this order. +The next two slices build on the foreach control-region, structured-context, +scheduler, lineage, barrier, and persisted run step budget foundations in +this order. -### 1. Add a persisted run step budget - -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 +### 1. 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 diff --git a/docs/superpowers/plans/2026-09-05-run-step-budget.md b/docs/historical/superpowers/plans/2026-09-05-run-step-budget.md similarity index 89% rename from docs/superpowers/plans/2026-09-05-run-step-budget.md rename to docs/historical/superpowers/plans/2026-09-05-run-step-budget.md index f4181889..eb4f1c7f 100644 --- a/docs/superpowers/plans/2026-09-05-run-step-budget.md +++ b/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` diff --git a/docs/wf_cli.md b/docs/wf_cli.md index a4f152e9..244eac38 100644 --- a/docs/wf_cli.md +++ b/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 diff --git a/docs/wf_core_architecture.md b/docs/wf_core_architecture.md index 156efd01..c1a9077f 100644 --- a/docs/wf_core_architecture.md +++ b/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. diff --git a/skills/wf-workflow/references/workflow-lifecycle.md b/skills/wf-workflow/references/workflow-lifecycle.md index 029296ba..8d57eb78 100644 --- a/skills/wf-workflow/references/workflow-lifecycle.md +++ b/skills/wf-workflow/references/workflow-lifecycle.md @@ -54,7 +54,13 @@ validated, runnable deployment. 7. Save and validate a deployment. - `wf deploy save --artifact --version 1 --binding =` (or `wf deploy create` alias) - `wf deploy validate ` -8. Run the deployment. +8. Run the deployment with an optional step budget. + - CLI: `wf run start --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 ` for compact stored result. - `wf run trace --from 0 --limit 25` for explicit debug slices. -- `wf run resume ` only for interrupted runs. +- `wf run resume ` only for interrupted runs. Resume reuses the + persisted budget and accepts no replacement value. Do not ask for unbounded traces. diff --git a/src/wf_core/__init__.py b/src/wf_core/__init__.py index 6f76cc7a..c4885973 100644 --- a/src/wf_core/__init__.py +++ b/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, diff --git a/src/wf_core/run_codec.py b/src/wf_core/run_codec.py index fc0b45cb..665c6cc4 100644 --- a/src/wf_core/run_codec.py +++ b/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 - 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. + future frame numbers and silently inflate or distort the budget. The + limits object holds exactly ``max_steps``: unknown fields are corrupt + rather than silently dropped, since no stored-data contract emits them. + 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") diff --git a/src/wf_core/run_limits.py b/src/wf_core/run_limits.py new file mode 100644 index 00000000..4a3bd2cb --- /dev/null +++ b/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"] diff --git a/src/wf_core/run_state.py b/src/wf_core/run_state.py index e5c8cdd0..35fa559b 100644 --- a/src/wf_core/run_state.py +++ b/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 - -if TYPE_CHECKING: - from wf_core.runtime.limits import RunLimits +from wf_core.run_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 diff --git a/src/wf_core/runtime/engine.py b/src/wf_core/runtime/engine.py index 17869080..bced54f8 100644 --- a/src/wf_core/runtime/engine.py +++ b/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 diff --git a/src/wf_core/runtime/limits.py b/src/wf_core/runtime/limits.py index 2237ec99..9cbcb93f 100644 --- a/src/wf_core/runtime/limits.py +++ b/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 diff --git a/src/wf_core/runtime/ops/runs.py b/src/wf_core/runtime/ops/runs.py index 359c93e3..0638a857 100644 --- a/src/wf_core/runtime/ops/runs.py +++ b/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 diff --git a/tests/core/test_run_limits.py b/tests/core/test_run_limits.py new file mode 100644 index 00000000..103f87ae --- /dev/null +++ b/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 diff --git a/tests/core/test_run_step_budget.py b/tests/core/test_run_step_budget.py index ca170505..0d6ced47 100644 --- a/tests/core/test_run_step_budget.py +++ b/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.runtime.limits import ( - RunLimits, - admit_step_attempt, - remaining_step_attempts, -) +from wf_core.run_state import RunState 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( - workflow, - {"items": ["a", "b"]}, - { - "record": lambda payload, _ctx: { - "outcome": "ok", - "output": {"seen": payload["value"]}, - } - }, - ) + def record(payload: dict[str, Any], _context: object) -> dict[str, Any]: + return {"outcome": "ok", "output": {"seen": payload["value"]}} + + run = execute_workflow(workflow, {"items": ["a", "b"]}, {"record": record}) 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 handler(_payload: dict, _context: object) -> dict: + def make( + 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( diff --git a/tests/core/test_run_step_budget_async.py b/tests/core/test_run_step_budget_async.py index f836b302..f89446ba 100644 --- a/tests/core/test_run_step_budget_async.py +++ b/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( - workflow: Workflow, items: list[Any], *, max_steps: int -): - run = create_run_state(workflow, {"items": items}, limits=RunLimits(max_steps=max_steps)) +def _prepare_limited_run(workflow: Workflow, items: list[Any], *, max_steps: int): + 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]: - return {"outcome": "ok", "output": payload} - - await step_workflow_async(workflow, run, {"record": _noop}, index=index) + await step_workflow_async(workflow, run, {"record": _noop_record}, 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]: - return {"outcome": "ok", "output": payload} - - await step_workflow_async(workflow, run, {"record": _noop}, index=index) + await step_workflow_async(workflow, run, {"record": _noop_record}, 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]: - return {"outcome": "ok", "output": payload} - - await step_workflow_async(workflow, run, {"record": _noop}, index=index) + await step_workflow_async(workflow, run, {"record": _noop_record}, 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]: - return {"outcome": "ok", "output": payload} - - await step_workflow_async(workflow, run, {"record": _noop}, index=index) + await step_workflow_async(workflow, run, {"record": _noop_record}, 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]: - return {"outcome": "ok", "output": payload} - - await step_workflow_async(workflow, run, {"record": _noop}, index=index) + await step_workflow_async(workflow, run, {"record": _noop_record}, 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]: diff --git a/tests/core/test_run_step_budget_codec.py b/tests/core/test_run_step_budget_codec.py new file mode 100644 index 00000000..4837b26b --- /dev/null +++ b/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) diff --git a/tests/wf_api/test_resume_concurrency.py b/tests/wf_api/test_resume_concurrency.py index ec111ffd..ed341503 100644 --- a/tests/wf_api/test_resume_concurrency.py +++ b/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") diff --git a/tests/wf_api/test_source_admin_api.py b/tests/wf_api/test_source_admin_api.py index 16e11ba0..6a7427be 100644 --- a/tests/wf_api/test_source_admin_api.py +++ b/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")