Budget Controls & Cost Optimization

[router]
strategy = "cost_optimized"
fallback_chain = ["anthropic", "openai", "google"]
max_cost_per_session = 1.00     # Hard limit: $1 per session
max_tokens_per_session = 500000  # Hard limit: 500K tokens
simple_task_model = "claude-haiku-4-5-20251001"  # Use cheap model for simple tasks

from harness.providers.budget import (
    BudgetSnapshot,
    BudgetExhaustedError,
    TokenBudgetTracker,
)

# BudgetSnapshot — immutable point-in-time view
@dataclass
class BudgetSnapshot:
    input_tokens_used: int = 0
    output_tokens_used: int = 0
    total_tokens_used: int = 0
    cost_used: float = 0.0
    tokens_remaining: int = 0   # 0 when no limit is configured
    cost_remaining: float = 0.0  # 0.0 when no limit is configured

# TokenBudgetTracker — stateful accumulator
class TokenBudgetTracker:
    def __init__(self, *, max_tokens: int = 0, max_cost: float = 0.0): ...
    @property
    def total_tokens(self) -> int: ...       # Tokens used so far
    @property
    def total_cost(self) -> float: ...       # Cost accrued so far
    def record_usage(
        self, input_tokens=0, output_tokens=0, cost=0.0
    ) -> BudgetSnapshot: ...                 # Returns snapshot after recording
    def snapshot(self) -> BudgetSnapshot: ... # Current state without modifying
    def is_exhausted(self) -> bool: ...      # True if any limit exceeded
    def check_budget(self) -> None: ...      # Raises BudgetExhaustedError if exhausted
    def reset(self) -> None: ...             # Reset counters (new session)

import asyncio
import harness
from harness.providers.budget import TokenBudgetTracker, BudgetExhaustedError

async def main():
    # Create a budget: max $0.50 or 100K tokens (whichever hits first)
    budget = TokenBudgetTracker(max_tokens=100_000, max_cost=0.50)

    try:
        async for msg in harness.run(
            "Analyze and document this codebase",
            provider="anthropic",
            model="claude-sonnet-4-20250514",
        ):
            if isinstance(msg, harness.Result):
                # Record usage from each Result message
                budget.record_usage(
                    # Note: total_tokens doesn't split input/output.
                    # This 50/50 approximation is for budget tracking only.
                    input_tokens=msg.total_tokens // 2,
                    output_tokens=msg.total_tokens // 2,
                    cost=msg.total_cost,
                )

                snap = budget.snapshot()
                print(f"Used: ${snap.cost_used:.4f} / ${snap.cost_remaining:.4f} remaining")
                print(f"Tokens: {snap.total_tokens_used:,} / {snap.tokens_remaining:,} remaining")

                # Manually check before issuing the next turn
                budget.check_budget()

    except BudgetExhaustedError as e:
        print(f"Budget exceeded! {e}")
        print(f"Final: ${e.snapshot.cost_used:.4f} spent, "
              f"{e.snapshot.total_tokens_used:,} tokens used")

asyncio.run(main())

from harness.providers.budget import BudgetExhaustedError

try:
    budget.check_budget()
except BudgetExhaustedError as e:
    snap = e.snapshot

    if snap.cost_remaining <= 0:
        print(f"Cost limit hit! Spent ${snap.cost_used:.4f}")
        print("Switching to free local model via Ollama...")
        # Re-run with Ollama provider (free)
        async for msg in harness.run(task, provider="ollama", model="llama3.2"):
            ...

    elif snap.tokens_remaining <= 0:
        print(f"Token limit hit! Used {snap.total_tokens_used:,} tokens")
        print("Compacting context and retrying with summary...")
        # Summarize results so far and continue with compact context
        summary = summarize_progress(results_so_far)
        budget.reset()
        async for msg in harness.run(
            f"Continue from: {summary}",
            provider="anthropic",
            model="claude-haiku-4-5-20251001",  # Cheaper model for remainder
        ):
            ...

from enum import Enum

class RoutingStrategy(Enum):
    MANUAL         = "manual"          # You choose the model explicitly
    COST_OPTIMIZED = "cost_optimized"  # Routes simple tasks to cheaper models
    QUALITY_FIRST  = "quality_first"   # Always uses best model, budget permitting
    LATENCY_FIRST  = "latency_first"   # Uses fastest model for each task

class ModelRouter:
    def __init__(
        self,
        primary,                         # Primary ProviderAdapter
        *,
        strategy=RoutingStrategy.MANUAL,
        simple_task_provider=None,       # Cheap provider for simple tasks
        budget=None,                     # TokenBudgetTracker instance
    ): ...

from harness.providers.fallback import FallbackProvider

# Try Anthropic first, then OpenAI, then Google
fallback = FallbackProvider([
    anthropic_provider,
    openai_provider,
    google_provider,
])

# active_provider reflects whichever is currently serving requests
print(f"Active: {fallback.active_provider}")

# If Anthropic is down, automatically tries OpenAI
# If OpenAI is down, automatically tries Google
async for msg in harness.run(
    "Review this PR",
    # Advanced: pass a pre-configured provider instance
    # (internal API — prefer TOML fallback_chain config instead)
    _provider=fallback,
):
    print(msg)

[router]
fallback_chain = ["anthropic", "openai", "google"]
# Harness will try providers in this order on failure