feat: add cost budget enforcement with audit/enforce modes

AGENTS.md

@@ -176,6 +176,22 @@ Tests mirror source structure in `tests/`:
 
 Use `pytest.mark.performance` for performance tests (exclude with `-m "not performance"`).
 
+### Test Fixture Patterns
+
+When writing integration tests that construct `WorkflowConfig` programmatically, follow these conventions (see `tests/test_engine/test_limits.py` for canonical examples):
+
+- `AgentDef` uses `prompt=` (not `instructions=`), `output={"key": OutputField(type="string")}` (dict, not list), and `routes=[RouteDef(...)]` (not raw dicts).
+- `WorkflowDef` requires `entry_point=` and places `limits=` inside `workflow=`. `agents=` and `output=` are top-level on `WorkflowConfig`.
+- The engine entry point is `await engine.run({})` (not `execute`).
+- To test with controlled token/cost data, patch `provider.execute` to return a custom `AgentOutput` with explicit `input_tokens`, `output_tokens`, and `model` fields.
+
+### Resume / Checkpoint Parity
+
+When adding new fields to `LimitEnforcer`:
+
+- **Transient fields** (reset each run): add to `from_dict()` as parameters sourced from the current workflow config, like `timeout_seconds`, `budget_usd`, `budget_mode`. Update the call site in `cli/run.py` → `resume_workflow_async()`.
+- **Persistent fields** (survive across resume): add to both `to_dict()` and `from_dict()` deserialization, like `max_iterations`, `current_iteration`, `execution_history`.
+
 ## Code Style
 
 - Python 3.12+

CHANGELOG.md

@@ -318,6 +318,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `docs/cli-reference.md` `--web-bg` section now documents the `human_gate`
   incompatibility and the new pre-fork validation behavior.
 
+### Added
+- Workflow `limits.budget_usd` and `limits.budget_mode` (`audit` | `enforce`)
+  cap cumulative LLM cost across a run. `audit` (default) emits a
+  `budget_exceeded` event and continues so users can profile costs before
+  enforcing; `enforce` saves a checkpoint and stops with
+  `BudgetExceededError`. Resuming with `conductor resume` starts a fresh
+  budget window (cumulative spend resets to $0), so the remaining work runs
+  under a full budget — raising `budget_usd` first is optional. Sub-workflow
+  spend is merged into the parent so a parent budget accounts for delegated
+  cost. Schema, engine enforcement at all five existing limit-check points,
+  resume parity for restored budget state, and the new `BudgetExceededError`
+  type are wired end-to-end. See
+  [docs/workflow-syntax.md](docs/workflow-syntax.md#cost-budget) and
+  [docs/configuration.md](docs/configuration.md) for the graduation path.
+
 ## [0.1.17](https://github.com/microsoft/conductor/compare/v0.1.16...v0.1.17) - 2026-05-21
 
 ### Added

docs/configuration.md

@@ -449,8 +449,10 @@ Safety limits prevent runaway execution:
 ```yaml
 workflow:
   limits:
-    max_iterations: 10  # Default: 10, max: 100
-    timeout_seconds: 600  # Default: 600, max: 3600
+    max_iterations: 10  # Default: 10, max: 500
+    timeout_seconds: 600  # Default: None (unlimited)
+    budget_usd: 5.00      # Default: None (no budget tracking)
+    budget_mode: audit    # Default: audit. Options: audit, enforce
 ```
 
 **max_iterations**:
@@ -461,6 +463,24 @@ workflow:
 - Total workflow timeout
 - Includes all agent executions
 
+**budget_usd** and **budget_mode**:
+- Tracks cumulative cost and acts when the budget is exceeded
+- `audit` mode (default): emits a `budget_exceeded` event and logs a warning,
+  but the workflow continues — use this to discover cost profiles
+- `enforce` mode: emits a `budget_exceeded` event, saves a checkpoint,
+  and stops the workflow with a `BudgetExceededError`. Resuming with
+  `conductor resume` starts a fresh budget window (cumulative spend resets
+  to $0), so raising the budget first is optional
+- Sub-workflow spend is merged into the parent budget, so a parent-level
+  budget accounts for delegated `type: workflow` cost
+- When `budget_usd` is not set, no budget tracking occurs
+
+**Recommended graduation path**:
+
+1. Run workflows without a budget to see costs in the summary
+2. Add `budget_usd` in `audit` mode to track overshoots without breaking workflows
+3. Switch to `enforce` mode once you know your cost profile
+
 ## Complete Examples
 
 ### Claude Configuration
@@ -577,7 +597,8 @@ export CONDUCTOR_LOG_LEVEL=DEBUG  # INFO, DEBUG, WARNING, ERROR
 
 1. **Set conservative limits** initially (`max_iterations: 10`)
 2. **Use timeout** to prevent long-running workflows
-3. **Test with dry-run** before production
+3. **Set a cost budget** — start with `budget_usd` in `audit` mode to learn your cost profile, then switch to `enforce`
+4. **Test with dry-run** before production
 
 ## Troubleshooting
 

docs/workflow-syntax.md

@@ -36,6 +36,8 @@ workflow:
   limits:
     max_iterations: 10              # Default: 10, max: 500
     timeout_seconds: 600            # Optional: Maximum wall-clock time (seconds)
+    budget_usd: 5.00                # Optional: Cost cap in USD (no tracking when unset)
+    budget_mode: audit              # audit (default) | enforce
 
   hooks:
     on_start: "{{ template }}"      # Optional: Expression evaluated on start
@@ -1016,6 +1018,8 @@ workflow:
   limits:
     max_iterations: 50              # Maximum agent executions (1-500, default: 10)
     timeout_seconds: 1800           # Maximum wall-clock time in seconds (optional)
+    budget_usd: 5.00                # Cumulative cost cap in USD (optional)
+    budget_mode: audit              # audit | enforce (default: audit)
 ```
 
 ### Iteration Counting
@@ -1031,6 +1035,28 @@ workflow:
 - Includes all agent execution time and overhead
 - `None` (default) means no timeout
 
+### Cost Budget
+
+- `budget_usd` caps cumulative LLM cost across the run. When unset (default), no
+  budget tracking occurs.
+- `budget_mode: audit` (default) emits a `budget_exceeded` event and logs a
+  warning on first overshoot, but the workflow continues — use this to discover
+  cost profiles before enforcing.
+- `budget_mode: enforce` emits a `budget_exceeded` event, saves a checkpoint,
+  and stops the workflow with `BudgetExceededError`. Resuming with
+  `conductor resume <workflow.yaml>` starts a fresh budget window (cumulative
+  spend resets to $0); raising `budget_usd` first is optional.
+- Sub-workflow (`type: workflow`) spend is merged into the parent's budget, so
+  a parent-level budget accounts for cost incurred by delegated workflows.
+- Recommended graduation path:
+  1. Run without `budget_usd` to observe costs in the summary
+  2. Add `budget_usd` in `audit` mode to track overshoots non-disruptively
+  3. Switch to `enforce` once the cost profile is understood
+
+See [configuration.md](configuration.md#limits) for the budget
+configuration reference and notes on how budget tracking integrates with the
+provider usage callbacks.
+
 ### Periodic Checkpoints
 
 By default Conductor writes a checkpoint **only when a workflow fails** with an

src/conductor/cli/run.py

@@ -521,6 +521,46 @@ def verbose_log_agent_timeout(
         _file_console.print(text)
 
 
+def verbose_log_budget_exceeded(
+    budget_usd: float,
+    spent_usd: float,
+    budget_mode: str,
+    current_agent: str | None = None,
+) -> None:
+    """Log a cost-budget overshoot.
+
+    Renders the ``budget_exceeded`` event so audit-mode overshoots are
+    visible on the console/log instead of only reaching the logging
+    lastResort stderr handler.
+
+    Args:
+        budget_usd: Configured budget limit in USD.
+        spent_usd: Cumulative spend that crossed the budget.
+        budget_mode: Active mode (``audit`` or ``enforce``).
+        current_agent: Agent executing when the budget was exceeded.
+    """
+    from rich.text import Text
+
+    from conductor.cli.app import is_verbose
+
+    should_console = is_verbose()
+    should_file = _file_console is not None
+    if not should_console and not should_file:
+        return
+
+    style = "red bold" if budget_mode == "enforce" else "yellow bold"
+    text = Text()
+    text.append("  💸 budget exceeded ", style=style)
+    text.append(f"(${spent_usd:.2f} of ${budget_usd:.2f}, {budget_mode} mode)", style="dim")
+    if current_agent:
+        text.append(f" at agent '{current_agent}'", style="dim")
+
+    if should_console:
+        _verbose_console.print(text)
+    if _file_console is not None:
+        _file_console.print(text)
+
+
 def verbose_log_parallel_summary(
     group_name: str,
     success_count: int,
@@ -946,6 +986,14 @@ def on_event(self, event: WorkflowEvent) -> None:
                 d.get("elapsed", 0.0),
             )
 
+        elif t == "budget_exceeded":
+            verbose_log_budget_exceeded(
+                d.get("budget_usd", 0.0),
+                d.get("spent_usd", 0.0),
+                d.get("budget_mode", "audit"),
+                d.get("current_agent"),
+            )
+
         elif t == "wait_completed":
             interrupted = d.get("interrupted", False)
             waited = d.get("waited_seconds", d.get("elapsed", 0.0))
@@ -2029,6 +2077,8 @@ async def resume_workflow_async(
         restored_limits = LimitEnforcer.from_dict(
             cp.limits,
             timeout_seconds=config.workflow.limits.timeout_seconds,
+            budget_usd=config.workflow.limits.budget_usd,
+            budget_mode=config.workflow.limits.budget_mode,
         )
 
         # Construct the web dashboard early (subscribes to the emitter on

src/conductor/config/schema.py

@@ -22,6 +22,13 @@
 from conductor.providers.context_tier import ContextTier
 from conductor.providers.reasoning import ReasoningEffort
 
+BudgetMode = Literal["audit", "enforce"]
+"""How the engine responds when a workflow cost budget is exceeded.
+
+Shared between :class:`LimitsConfig` and :class:`conductor.engine.limits.LimitEnforcer`
+so the literal type is defined in exactly one place.
+"""
+
 # Maximum allowed wait-step duration (24 hours). Anything longer almost
 # certainly wants ``limits.timeout_seconds`` reconsidered first.
 MAX_WAIT_DURATION_SECONDS = 24 * 60 * 60
@@ -323,6 +330,27 @@ class LimitsConfig(BaseModel):
     a hard time limit.
     """
 
+    budget_usd: float | None = Field(default=None, gt=0.0)
+    """Maximum cost budget for the workflow in USD.
+
+    When set, the engine tracks cumulative cost and acts according to
+    ``budget_mode`` when the budget is exceeded. Must be strictly positive
+    (a zero budget would trip after the first priced token, which is never
+    a useful limit). Default is None (no budget tracking).
+    """
+
+    budget_mode: BudgetMode = "audit"
+    """How the engine responds when ``budget_usd`` is exceeded.
+
+    - ``audit``: emit a ``budget_exceeded`` event and log a warning,
+      but allow the workflow to continue. Use this to discover cost
+      profiles before applying hard limits.
+    - ``enforce``: emit a ``budget_exceeded`` event, save a checkpoint,
+      and stop the workflow with a ``BudgetExceededError``.
+
+    Only takes effect when ``budget_usd`` is set. Default is ``audit``.
+    """
+
 
 class PricingOverride(BaseModel):
     """Custom pricing for a specific model.

src/conductor/engine/limits.py

@@ -11,8 +11,9 @@
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from dataclasses import dataclass, field
-from typing import Any
+from typing import Any, NamedTuple
 
+from conductor.config.schema import BudgetMode
 from conductor.exceptions import (
     MaxIterationsError,
 )
@@ -21,6 +22,25 @@
 )
 
 
+class BudgetCheckResult(NamedTuple):
+    """Result of a :meth:`LimitEnforcer.check_budget` call.
+
+    Attributes:
+        exceeded: True when ``spent_usd`` is over ``budget_usd``.
+        should_emit: True when the caller should surface a ``budget_exceeded``
+            event/warning now. True on the first overshoot and again each time
+            spend crosses another full ``budget_usd`` increment (so audit-mode
+            users get periodic updates instead of a single stale figure).
+        budget_usd: The configured budget (None when no budget is set).
+        spent_usd: The cumulative spend that was checked.
+    """
+
+    exceeded: bool
+    should_emit: bool
+    budget_usd: float | None
+    spent_usd: float
+
+
 @dataclass
 class LimitEnforcer:
     """Enforces iteration and timeout limits on workflow execution.
@@ -52,6 +72,12 @@ class LimitEnforcer:
     timeout_seconds: int | None = None
     """Maximum wall-clock time for entire workflow. None means unlimited."""
 
+    budget_usd: float | None = None
+    """Maximum cost budget in USD. None means no budget tracking."""
+
+    budget_mode: BudgetMode = "audit"
+    """Budget enforcement mode: 'audit' (warn only) or 'enforce' (stop)."""
+
     current_iteration: int = 0
     """Current iteration count."""
 
@@ -64,6 +90,13 @@ class LimitEnforcer:
     current_agent: str | None = None
     """Currently executing agent name."""
 
+    _budget_last_emitted_usd: float | None = field(default=None, repr=False)
+    """Spend at the last ``budget_exceeded`` emission, or None if never emitted.
+
+    Used to re-arm audit-mode emission once spend climbs by another full
+    ``budget_usd`` increment, instead of latching after a single event.
+    """
+
     def to_dict(self) -> dict[str, Any]:
         """Serialize limit state to a JSON-compatible dict.
 
@@ -84,18 +117,29 @@ def to_dict(self) -> dict[str, Any]:
     def from_dict(
         cls,
         data: dict[str, Any],
-        timeout_seconds: int | None = None,
+        *,
+        timeout_seconds: int | None,
+        budget_usd: float | None,
+        budget_mode: BudgetMode,
     ) -> LimitEnforcer:
         """Reconstruct a LimitEnforcer from a serialized dict.
 
         Uses ``max_iterations`` from the checkpoint (it may have been
         increased by the user) and ``timeout_seconds`` from the current
         workflow config so that the resumed run gets a fresh timeout
-        window.
+        window. ``budget_usd`` and ``budget_mode`` come from the current
+        config so that the resumed run gets a fresh budget window.
+
+        The transient fields are keyword-only and required: callers must
+        source them from the current workflow config. Defaulting them here
+        previously let non-CLI callers silently disable the budget on
+        resume.
 
         Args:
             data: Dict previously produced by ``to_dict()``.
             timeout_seconds: Timeout from the workflow config (fresh window).
+            budget_usd: Budget from the workflow config (fresh window).
+            budget_mode: Budget mode from the workflow config.
 
         Returns:
             A new LimitEnforcer with restored iteration state and a fresh
@@ -104,6 +148,8 @@ def from_dict(
         enforcer = cls(
             max_iterations=data.get("max_iterations", 10),
             timeout_seconds=timeout_seconds,
+            budget_usd=budget_usd,
+            budget_mode=budget_mode,
         )
         enforcer.current_iteration = data.get("current_iteration", 0)
         enforcer.execution_history = list(data.get("execution_history", []))
@@ -239,6 +285,41 @@ def check_timeout(self) -> None:
                 current_agent=self.current_agent,
             )
 
+    def check_budget(self, spent_usd: float) -> BudgetCheckResult:
+        """Check if the workflow cost budget has been exceeded.
+
+        In ``enforce`` mode the caller should raise ``BudgetExceededError``
+        after emitting the event. In ``audit`` mode the caller should
+        log a warning and continue.
+
+        Emission re-arms per budget increment: ``should_emit`` is True on
+        the first overshoot and again once cumulative spend climbs by
+        another full ``budget_usd`` beyond the last emission. This keeps
+        audit-mode output current as spend grows, rather than latching on
+        a single stale figure.
+
+        Args:
+            spent_usd: Current cumulative cost from UsageTracker.
+
+        Returns:
+            A ``BudgetCheckResult`` carrying the overshoot flag, the
+            should-emit flag, and the budget/spend figures so the caller
+            does not need to re-derive them.
+        """
+        if self.budget_usd is None:
+            return BudgetCheckResult(False, False, None, spent_usd)
+
+        if spent_usd <= self.budget_usd:
+            return BudgetCheckResult(False, False, self.budget_usd, spent_usd)
+
+        should_emit = (
+            self._budget_last_emitted_usd is None
+            or spent_usd >= self._budget_last_emitted_usd + self.budget_usd
+        )
+        if should_emit:
+            self._budget_last_emitted_usd = spent_usd
+        return BudgetCheckResult(True, should_emit, self.budget_usd, spent_usd)
+
     def get_elapsed_time(self) -> float:
         """Get the elapsed time since workflow start.