AI agent parallel tool call budget control: why concurrent calls break sequential budget checks and how to fix it

import asyncio
import threading
from dataclasses import dataclass, field
from typing import Any, Callable, Coroutine
import anthropic
from runguard import BudgetExceededError

@dataclass
class ConcurrentBudgetTracker:
    """
    Thread-safe budget tracker with reservation support for parallel tool calls.
    Uses a lock to make reservation+check atomic.
    """
    cap_usd: float
    _lock: threading.Lock = field(default_factory=threading.Lock, repr=False)
    _spent_usd: float = 0.0
    _reserved_usd: float = 0.0

    @property
    def available_usd(self) -> float:
        return self.cap_usd - self._spent_usd - self._reserved_usd

    def reserve(self, amount_usd: float) -> None:
        """Atomically reserve budget for an upcoming call. Raises if over cap."""
        with self._lock:
            if self._spent_usd + self._reserved_usd + amount_usd > self.cap_usd:
                raise BudgetExceededError(
                    f"Budget cap ${self.cap_usd:.2f} would be exceeded: "
                    f"spent ${self._spent_usd:.4f} + reserved ${self._reserved_usd:.4f} "
                    f"+ new ${amount_usd:.4f}"
                )
            self._reserved_usd += amount_usd

    def release_and_record(self, reserved_usd: float, actual_usd: float) -> None:
        """Release a reservation and record the actual cost."""
        with self._lock:
            self._reserved_usd = max(0.0, self._reserved_usd - reserved_usd)
            self._spent_usd += actual_usd

    def release_reservation(self, reserved_usd: float) -> None:
        """Release a reservation without recording cost (call cancelled)."""
        with self._lock:
            self._reserved_usd = max(0.0, self._reserved_usd - reserved_usd)

    def summary(self) -> str:
        return (f"spent=${self._spent_usd:.4f} "
                f"reserved=${self._reserved_usd:.4f} "
                f"available=${self.available_usd:.4f} "
                f"cap=${self.cap_usd:.2f}")


HAIKU_IN  = 0.25 / 1_000_000
HAIKU_OUT = 1.25 / 1_000_000

async def execute_tool_with_budget(
    tool_name: str,
    tool_fn: Callable[..., Coroutine],
    tool_args: dict,
    tracker: ConcurrentBudgetTracker,
    estimated_cost_usd: float,
) -> dict[str, Any]:
    """
    Execute a single tool call with budget reservation and reconciliation.
    Designed to run concurrently via asyncio.gather().
    """
    # Reserve budget before executing (raises BudgetExceededError if over cap)
    tracker.reserve(estimated_cost_usd)
    try:
        result = await tool_fn(**tool_args)
        # Assume tool calls have a known or estimable cost
        # For LLM tool calls, parse actual usage from response
        actual_cost = estimated_cost_usd  # Replace with real usage in production
        tracker.release_and_record(estimated_cost_usd, actual_cost)
        return {"tool": tool_name, "result": result, "cost": actual_cost}
    except Exception as e:
        # Call failed — release reservation without recording cost
        tracker.release_reservation(estimated_cost_usd)
        raise


async def run_parallel_tools(
    tool_calls: list[dict],
    tracker: ConcurrentBudgetTracker,
) -> list[dict]:
    """
    Execute tool calls in parallel with per-call budget reservation.
    Reserves ALL estimated costs upfront before launching any calls.
    On budget exhaustion, cancels the entire batch rather than partial execution.
    """
    estimated_costs = [tc.get("estimated_cost_usd", 0.02) for tc in tool_calls]
    total_estimated = sum(estimated_costs)

    # Try to reserve the entire batch atomically
    try:
        tracker.reserve(total_estimated)
    except BudgetExceededError as e:
        print(f"[budget] Parallel batch rejected: {e}")
        raise

    # All costs are now reserved — launch in parallel
    try:
        tasks = [
            execute_tool_with_budget(
                tool_name=tc["name"],
                tool_fn=tc["fn"],
                tool_args=tc.get("args", {}),
                tracker=tracker,
                estimated_cost_usd=cost,
            )
            for tc, cost in zip(tool_calls, estimated_costs)
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        # Release the batch reservation (individual calls already recorded)
        tracker.release_reservation(total_estimated)

        # Separate successes from failures
        successes = [r for r in results if isinstance(r, dict)]
        failures  = [r for r in results if isinstance(r, Exception)]
        if failures:
            print(f"[parallel] {len(failures)} of {len(tool_calls)} calls failed")
        return successes
    except Exception:
        # Unexpected failure — release entire batch reservation
        tracker.release_reservation(total_estimated)
        raise


# Example: agent making three parallel searches with budget control
async def agent_turn_with_parallel_tools(
    query: str,
    budget_cap_usd: float = 0.50,
) -> None:
    client = anthropic.Anthropic()
    tracker = ConcurrentBudgetTracker(cap_usd=budget_cap_usd)

    async def mock_search(q: str) -> str:
        await asyncio.sleep(0.1)  # simulate API latency
        return f"Search results for: {q}"

    # Simulate the agent deciding to make 3 parallel searches
    parallel_batch = [
        {"name": "search_1", "fn": mock_search, "args": {"q": f"{query} overview"}, "estimated_cost_usd": 0.01},
        {"name": "search_2", "fn": mock_search, "args": {"q": f"{query} examples"}, "estimated_cost_usd": 0.01},
        {"name": "search_3", "fn": mock_search, "args": {"q": f"{query} best practices"}, "estimated_cost_usd": 0.01},
    ]

    try:
        results = await run_parallel_tools(parallel_batch, tracker)
        print(f"Completed {len(results)} parallel tool calls. {tracker.summary()}")
    except BudgetExceededError as e:
        print(f"[BUDGET] Parallel batch blocked: {e}")