Token Budget¶

Token budget controller — unified cost management and TOS-aware LLM routing.

Cluster: Uncategorised | Type: component | MCP Tools: 11

Overview¶

Drop-in LLMProtocol replacement with persistent budget tracking and TOS-aware routing. Wraps an inner LLMBlock and transparently routes requests to Codex, Claude Code, OpenRouter, or Ollama based on budget degradation level and task complexity.

Supports context compression when approaching token limits and provides a BudgetSnapshot for monitoring usage across all backends. The MCP surface also exposes budget_capabilities for read-only discovery of backend availability, TOS gates, OpenRouter model mapping, compression threshold, billing enforcement, degradation level, and public tool inventory.

When to use:

• Enforcing token budgets across multi-step agent workflows without modifying each step
• Automatically degrading from paid to local backends as budget is consumed
• Monitoring cumulative token usage and cost across Claude Code, Codex, and OpenRouter

Example:

from mvp.token_budget import TokenBudgetBlock

block = TokenBudgetBlock()
result = block.complete(
    prompt="Explain the solver pipeline",
    model="openrouter/anthropic/claude-sonnet-4-20250514",
    max_tokens=500,
)
# result.value -> str (LLM response); budget ledger updated
# inspect block.last_ledger_status and block.last_compression after each call
snapshot = block.get_snapshot()
# snapshot.degradation_level, snapshot.claude_usage_pct, snapshot.openrouter_cost

Caveats and known limitations:

• Inner LLMBlock is typed as Any to avoid circular imports; no compile-time verification of the wrapped block
• Task classification and backend selection logic is internal; no user-facing config for routing rules
• infer() and complete() return Result[str]; they do not carry completion_state, warning_card, evidence, request_id, task_id, or run_id
• Callers that own the block must inspect last_ledger_status after each call; values like ledger_write_failed:RuntimeError should be lifted into the caller-owned qualified-draft warning card and evidence envelope
• Callers must inspect last_compression after each call; when present, lift its before/after token counts and reason into the caller-owned warning_card and evidence
• context_compressor.estimate_tokens is an approximation based on len(text) // 4; non-English or code-heavy prompts can be over- or under-estimated
• Context compression fires automatically when approaching limits; the model receives a truncation notice, but the caller sees it only through last_compression
• Usage recording is best-effort; ledger failures are surfaced through last_ledger_status, not hidden as silent success
• Degradation level routing (PAID_FIRST to LOCAL_FIRST to HARD_LIMIT) is not configurable per request
• The MCP envelope uses only verified, qualified-draft, and blocked-escalated; degraded MCP success is qualified-draft, while blocked MCP failures serialize as blocked-escalated

Works well with: work_loop, learning_layer, payments_x402, llm_router

Public API¶

BudgetAwareLLMBlock(AIBlock[str, str, None])¶

Wraps an inner LLMBlock with persistent budget tracking and TOS-aware routing.

Methods:

infer(data: str | dict) -> Result[str]¶

complete(prompt: str, model: str | None = None, max_tokens: int = 4096, temperature: float = 0.7, system_prompt: str | None = None, **kwargs: Any) -> Result[str]¶

Satisfy LLMProtocol.

budget_status() -> BudgetSnapshot¶

learned_routing_priors(days: int = 7, limit: int = 1000) -> list[dict]¶

BO §13: ADVISORY learned-routing priors from the real usage log.

backend_cost_telemetry(days: int = 7, limit: int = 1000) -> list[dict]¶

BO §1: measured per-backend cost telemetry from the real usage log.

TokenBudgetLedger¶

Persistent SQLite ledger for cross-session token/cost tracking.

Constructor:

Methods:

default() -> TokenBudgetLedger¶

reset_default() -> None¶

close() -> None¶

record_usage(event: UsageEvent) -> None¶

record_outcome(outcome: RouteOutcome) -> None¶

Record a per-route OUTCOME observation (BO §13 outcome learning).

tokens_by_backend(backend: str, window_seconds: float) -> int¶

calls_by_backend(backend: str, window_seconds: float) -> int¶

Count recorded calls for a backend in the trailing window (BO §8 / 8-3).

cost_by_backend(backend: str, window_seconds: float) -> float¶

total_cost(window_seconds: float) -> float¶

tokens_by_component(component: str, window_seconds: float) -> int¶

weekly_usage_fraction(backend: str) -> float¶

session_tokens(session_id: str) -> int¶

save_pause(pause: BudgetPause) -> None¶

get_pending_pauses() -> list[BudgetPause]¶

mark_resumed(pause_id: str) -> None¶

get_config_value(key: str) -> str | None¶

set_config_value(key: str, value: str) -> None¶

apply_config_updates(updates: dict[str, Any]) -> BudgetConfig¶

Validate, persist, and activate runtime budget configuration updates.

snapshot(session_id: str = '') -> BudgetSnapshot¶

usage_history(backend: str = '', days: int = 7, limit: int = 100) -> list[dict]¶

outcome_history(backend: str = '', days: int = 7, limit: int = 1000) -> list[dict]¶

Recorded per-route OUTCOME rows (BO §13). Mirrors :meth:usage_history.

BackendType(str, Enum)¶

BackendEligibility(str, Enum)¶

DegradationLevel(IntEnum)¶

UsageEvent(BaseModel)¶

BudgetSnapshot(BaseModel)¶

BudgetConfig(BaseModel)¶

BudgetPause(BaseModel)¶

BackendSelection(BaseModel)¶

BillingContext(BaseModel)¶

BillingAuthorization(BaseModel)¶

MCP Tools¶