Building Agent-Native Apps Without Burning Your Budget: Practical Patterns for Sandboxed AI Agent Architectures

                    ┌─────────────────────────┐
                    │   Task Complexity       │
                    │   Classifier            │
                    └────────┬────────────────┘
                             │
              ┌──────────────┼──────────────┐
              ▼              ▼              ▼
     ┌────────────┐ ┌────────────┐ ┌────────────┐
     │  Micro     │ │  Standard  │ │  Extended  │
     │  (1 model  │ │  (1 model  │ │  (multi-   │
     │   call,    │ │   + tools, │ │   model,   │
     │   ≤2K tok) │ │   ≤15K tok)│ │   ≤50K tok)│
     └────────────┘ └────────────┘ └────────────┘

from dataclasses import dataclass
from enum import Enum
import time


class ComputeTier(Enum):
    MICRO = "micro"        # Single LLM call, no tools
    STANDARD = "standard"  # LLM + tools, bounded iterations
    EXTENDED = "extended"  # Multi-model, multi-step reasoning


@dataclass
class TierConfig:
    tier: ComputeTier
    max_iterations: int
    max_tokens_total: int
    max_tool_calls: int
    allowed_models: list[str]
    timeout_seconds: int
    max_concurrent_subagents: int


TIER_REGISTRY: dict[ComputeTier, TierConfig] = {
    ComputeTier.MICRO: TierConfig(
        tier=ComputeTier.MICRO,
        max_iterations=1,
        max_tokens_total=2_000,
        max_tool_calls=0,
        allowed_models=["gpt-4o-mini", "claude-haiku"],
        timeout_seconds=30,
        max_concurrent_subagents=0,
    ),
    ComputeTier.STANDARD: TierConfig(
        tier=ComputeTier.STANDARD,
        max_iterations=8,
        max_tokens_total=15_000,
        max_tool_calls=12,
        allowed_models=["gpt-4o", "claude-sonnet"],
        timeout_seconds=120,
        max_concurrent_subagents=2,
    ),
    ComputeTier.EXTENDED: TierConfig(
        tier=ComputeTier.EXTENDED,
        max_iterations=25,
        max_tokens_total=50_000,
        max_tool_calls=40,
        allowed_models=["gpt-4o", "claude-sonnet", "claude-opus"],
        timeout_seconds=300,
        max_concurrent_subagents=5,
    ),
}


class TierClassifier:
    """Classifies incoming agent tasks into compute tiers."""

    def __init__(self, model_client):
        self.model_client = model_client

    def classify(self, user_request: str, context: dict) -> ComputeTier:
        classification_prompt = f"""Classify this task into one of three tiers:
- MICRO: Simple single-step tasks (summarize, translate, extract)
- STANDARD: Multi-step tasks requiring tool use (search, research, code review)
- EXTENDED: Complex multi-phase tasks (system design, deep research, multi-file refactoring)

Request: "{user_request[:500]}"
Context summary: {context.get('summary', 'none')}

Return ONLY the tier name."""

        response = self.model_client.complete(
            model="gpt-4o-mini",
            prompt=classification_prompt,
            max_tokens=20,
        )

        tier_str = response.strip().upper()
        if "EXTENDED" in tier_str:
            return ComputeTier.EXTENDED
        elif "STANDARD" in tier_str:
            return ComputeTier.STANDARD
        return ComputeTier.MICRO

import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable


class CircuitState(Enum):
    CLOSED = "closed"
    HALF_OPEN = "half_open"
    OPEN = "open"


@dataclass
class TokenBudget:
    """Token budget with circuit breaker enforcement."""
    total_budget: int
    consumed: int = 0
    state: CircuitState = CircuitState.CLOSED
    warning_threshold: float = 0.70  # 70% triggers HALF_OPEN
    hard_threshold: float = 1.00     # 100% triggers OPEN
    velocity_window_seconds: int = 60
    max_velocity_tokens_per_second: int = 500
    _token_log: list[tuple[float, int]] = field(default_factory=list)

    def consume(self, amount: int) -> bool:
        """Attempt to consume tokens. Returns False if budget exceeded."""
        if self.state == CircuitState.OPEN:
            return False

        self.consumed += amount
        now = time.time()
        self._token_log.append((now, amount))

        # Clean old log entries
        cutoff = now - self.velocity_window_seconds
        self._token_log = [(t, a) for t, a in self._token_log if t >= cutoff]

        ratio = self.consumed / self.total_budget

        if ratio >= self.hard_threshold:
            self.state = CircuitState.OPEN
            return False
        elif ratio >= self.warning_threshold:
            self.state = CircuitState.HALF_OPEN

        return True

    def get_velocity(self) -> float:
        """Calculate current token consumption rate."""
        if len(self._token_log) < 2:
            return 0.0
        elapsed = self._token_log[-1][0] - self._token_log[0][0]
        if elapsed == 0:
            return 0.0
        total_in_window = sum(a for _, a in self._token_log)
        return total_in_window / elapsed

    def is_velocity_exceeded(self) -> bool:
        return self.get_velocity() > self.max_velocity_tokens_per_second


class BudgetEnforcedAgent:
    """Agent loop with token budget circuit breaker."""

    def __init__(self, model_client, budget: TokenBudget, tools: list):
        self.model_client = model_client
        self.budget = budget
        self.tools = tools
        self.iteration = 0

    def run(self, task: str, max_iterations: int = 10) -> dict:
        messages = [{"role": "user", "content": task}]
        result = {"status": "completed", "output": None, "iterations": 0}

        for self.iteration in range(1, max_iterations + 1):
            # Check circuit breaker state
            if self.budget.state == CircuitState.OPEN:
                result["status"] = "budget_exhausted"
                result["partial_output"] = messages[-1].get("content", "")
                break

            # In HALF_OPEN state, reduce context by trimming older messages
            effective_messages = messages
            if self.budget.state == CircuitState.HALF_OPEN:
                effective_messages = self._trim_context(messages, max_messages=4)

            # Check velocity
            if self.budget.is_velocity_exceeded():
                result["status"] = "velocity_throttled"
                result["partial_output"] = messages[-1].get("content", "")
                break

            # Make LLM call
            response = self.model_client.chat(
                messages=effective_messages,
                tools=self.tools if self.budget.state != CircuitState.HALF_OPEN else self.tools[:2],
            )

            # Track token consumption
            tokens_used = response.get("usage", {}).get("total_tokens", 0)
            if not self.budget.consume(tokens_used):
                result["status"] = "budget_exhausted"
                result["partial_output"] = response.get("content", "")
                break

            messages.append({"role": "assistant", "content": response.get("content")})

            # Check if agent is done
            if not response.get("tool_calls"):
                result["output"] = response.get("content")
                result["iterations"] = self.iteration
                break

        return result

    def _trim_context(self, messages: list, max_messages: int) -> list:
        """Reduce context in HALF_OPEN state to slow consumption."""
        if len(messages) <= max_messages:
            return messages
        # Keep system message + last N messages
        if messages[0].get("role") == "system":
            return [messages[0]] + messages[-(max_messages - 1):]
        return messages[-max_messages:]

import asyncio
import hashlib
import json
from dataclasses import dataclass
from typing import Any, Callable


@dataclass
class ToolBudget:
    max_calls: int
    calls_made: int = 0
    max_output_tokens: int = 2_000
    max_execution_seconds: float = 10.0
    max_retries: int = 2


class SandboxedTool:
    """Wraps a tool function with resource constraints."""

    def __init__(self, name: str, func: Callable, budget: ToolBudget):
        self.name = name
        self.func = func
        self.budget = budget
        self._call_log: list[dict] = []

    async def execute(self, arguments: dict) -> dict:
        # Enforce call limit
        if self.budget.calls_made >= self.budget.max_calls:
            return {
                "error": "tool_call_limit_exceeded",
                "tool": self.name,
                "max_calls": self.budget.max_calls,
            }

        # Enforce execution time limit with asyncio timeout
        for attempt in range(self.budget.max_retries + 1):
            try:
                self.budget.calls_made += 1
                start = asyncio.get_event_loop().time()

                result = await asyncio.wait_for(
                    self.func(**arguments),
                    timeout=self.budget.max_execution_seconds,
                )

                elapsed = asyncio.get_event_loop().time() - start

                # Enforce output size limit
                result_str = json.dumps(result) if not isinstance(result, str) else result
                if len(result_str) > self.budget.max_output_tokens * 4:  # ~4 chars per token
                    result = self._truncate_output(result, self.budget.max_output_tokens)

                self._call_log.append({
                    "attempt": attempt,
                    "elapsed": elapsed,
                    "output_size": len(result_str),
                    "success": True,
                })

                return {"result": result, "metadata": {"elapsed": elapsed}}

            except asyncio.TimeoutError:
                self._call_log.append({"attempt": attempt, "error": "timeout"})
                if attempt == self.budget.max_retries:
                    return {"error": "tool_timeout", "tool": self.name}

            except Exception as e:
                self._call_log.append({"attempt": attempt, "error": str(e)})
                if attempt == self.budget.max_retries:
                    return {"error": "tool_failed", "tool": self.name, "detail": str(e)}

        return {"error": "unexpected_exit", "tool": self.name}

    def _truncate_output(self, result: Any, max_tokens: int) -> Any:
        """Truncate tool output to prevent context bloat."""
        max_chars = max_tokens * 4
        if isinstance(result, str):
            return result[:max_chars] + "\n[TRUNCATED]"
        result_str = json.dumps(result)
        if len(result_str) <= max_chars:
            return result
        # For structured data, keep the first N items
        if isinstance(result, list):
            truncated = result[:10]
            return {"items": truncated, "truncated": True, "total_count": len(result)}
        return json.loads(result_str[:max_chars])

    @property
    def stats(self) -> dict:
        return {
            "tool": self.name,
            "calls_made": self.budget.calls_made,
            "max_calls": self.budget.max_calls,
            "avg_elapsed": sum(c["elapsed"] for c in self._call_log if "elapsed" in c) / max(len([c for c in self._call_log if "elapsed" in c]), 1),
        }


class ToolRegistry:
    """Central registry of sandboxed tools with per-tool budgets."""

    def __init__(self):
        self._tools: dict[str, SandboxedTool] = {}

    def register(self, name: str, func: Callable, budget: ToolBudget) -> None:
        self._tools[name] = SandboxedTool(name, func, budget)

    async def execute(self, name: str, arguments: dict) -> dict:
        if name not in self._tools:
            return {"error": "tool_not_found", "tool": name}
        return await self._tools[name].execute(arguments)

    def get_all_stats(self) -> dict:
        return {name: tool.stats for name, tool in self._tools.items()}

class SideEffectGate:
    """Requires explicit confirmation before executing side-effect tools."""

    def __init__(self, allowed_actions: set[str]):
        self.allowed_actions = allowed_actions

    async def execute(self, action: str, payload: dict, confirm: dict) -> dict:
        if action not in self.allowed_actions:
            return {"error": "action_not_allowed", "action": action}

        # Verify the agent's confirmation matches the actual payload
        if not self._confirmations_match(payload, confirm):
            return {
                "error": "confirmation_mismatch",
                "expected": confirm,
                "actual": payload,
            }

        # Now actually execute
        return await self._do_execute(action, payload)

    def _confirmations_match(self, payload: dict, confirm: dict) -> bool:
        """Check that agent's stated intent matches actual payload."""
        for key, expected in confirm.items():
            actual = payload.get(key)
            if actual != expected:
                return False
        return True

import hashlib
import time
from collections import OrderedDict
from dataclasses import dataclass
from typing import Optional

import numpy as np


@dataclass
class CacheEntry:
    embedding: np.ndarray
    response: str
    model: str
    timestamp: float
    hit_count: int = 0
    tokens_saved: int = 0


class SemanticCache:
    """Embedding-based cache for LLM responses."""

    def __init__(
        self,
        embedding_model: str = "text-embedding-3-small",
        similarity_threshold: float = 0.92,
        max_entries: int = 1000,
        ttl_seconds: int = 3600,
    ):
        self.embedding_model = embedding_model
        self.similarity_threshold = similarity_threshold
        self.max_entries = max_entries
        self.ttl_seconds = ttl_seconds
        self._cache: OrderedDict[str, CacheEntry] = OrderedDict()
        self._hits = 0
        self._misses = 0

    async def get_or_compute(
        self,
        prompt: str,
        model: str,
        compute_fn: Callable[[], str],
        token_count: int = 0,
    ) -> str:
        """Check cache first, compute if miss."""
        embedding = await self._embed(prompt)

        # Find best match
        best_match: Optional[CacheEntry] = None
        best_score = 0.0

        for entry in self._cache.values():
            if time.time() - entry.timestamp > self.ttl_seconds:
                continue
            score = np.dot(embedding, entry.embedding) / (
                np.linalg.norm(embedding) * np.linalg.norm(entry.embedding)
            )
            if score > best_score:
                best_score = score
                best_match = entry

        if best_match and best_score >= self.similarity_threshold:
            best_match.hit_count += 1
            best_match.tokens_saved += token_count
            self._hits += 1
            self._cache.move_to_end(best_match.key)
            return best_match.response

        # Cache miss — compute and store
        self._misses += 1
        response = await compute_fn()

        cache_key = hashlib.md5(prompt.encode()).hexdigest()
        self._cache[cache_key] = CacheEntry(
            embedding=embedding,
            response=response,
            model=model,
            timestamp=time.time(),
        )

        # Evict oldest if over capacity
        while len(self._cache) > self.max_entries:
            self._cache.popitem(last=False)

        return response

    async def _embed(self, text: str) -> np.ndarray:
        """Generate embedding for text (delegated to embedding API)."""
        # In production, this calls your embedding provider
        # Placeholder for illustration
        return np.random.randn(1536)  # Replace with actual embedding call

    @property
    def hit_rate(self) -> float:
        total = self._hits + self._misses
        return self._hits / total if total > 0 else 0.0

    @property
    def stats(self) -> dict:
        return {
            "hits": self._hits,
            "misses": self._misses,
            "hit_rate": self.hit_rate,
            "entries": len(self._cache),
            "total_tokens_saved": sum(e.tokens_saved for e in self._cache.values()),
        }

class SharedContextCache:
    """Cross-agent cache for shared knowledge."""

    def __init__(self, semantic_cache: SemanticCache):
        self.semantic_cache = semantic_cache
        self._agent_registry: dict[str, dict] = {}

    async def agent_query(
        self, agent_id: str, question: str, compute_fn: Callable[[], str]
    ) -> str:
        """Query with cross-agent deduplication."""
        result = await self.semantic_cache.get_or_compute(
            prompt=f"[agent:{agent_id}] {question}",
            model="shared",
            compute_fn=compute_fn,
        )

        # Track what each agent has queried
        if agent_id not in self._agent_registry:
            self._agent_registry[agent_id] = {"queries": [], "cache_hits": 0}
        self._agent_registry[agent_id]["queries"].append(question)

        return result

    def get_agent_stats(self, agent_id: str) -> dict:
        return self._agent_registry.get(agent_id, {})

class EscalationLevel(Enum):
    DIRECT = 0       # One LLM call
    AUGMENTED = 1    # RAG + one LLM call
    REASONED = 2     # Multi-step with tools
    ENSEMBLE = 3     # Multi-model synthesis


class ProgressiveEscalationAgent:
    """Escalates through complexity levels until result quality is sufficient."""

    def __init__(self, model_client, retriever, quality_evaluator, budget: TokenBudget):
        self.model_client = model_client
        self.retriever = retriever
        self.quality_evaluator = quality_evaluator
        self.budget = budget

    async def execute(self, task: str) -> dict:
        last_result = None

        for level in EscalationLevel:
            if not self.budget.consume(0):  # Check budget still available
                return {"status": "budget_exhausted", "output": last_result}

            # Execute at current escalation level
            result = await self._execute_at_level(task, level)
            last_result = result.get("output")

            # Evaluate quality
            quality = await self._evaluate_quality(task, result)

            if quality >= self._threshold_for_level(level):
                return {
                    "status": "completed",
                    "output": result.get("output"),
                    "escalation_level": level.name,
                    "quality_score": quality,
                }

        # Exhausted all levels — return best result
        return {
            "status": "max_escalation_reached",
            "output": last_result,
            "escalation_level": EscalationLevel.ENSEMBLE.name,
        }

    async def _execute_at_level(self, task: str, level: EscalationLevel) -> dict:
        if level == EscalationLevel.DIRECT:
            return await self._direct_answer(task)
        elif level == EscalationLevel.AUGMENTED:
            return await self._augmented_answer(task)
        elif level == EscalationLevel.REASONED:
            return await self._reasoned_answer(task)
        elif level == EscalationLevel.ENSEMBLE:
            return await self._ensemble_answer(task)

    async def _direct_answer(self, task: str) -> dict:
        response = await self.model_client.chat(
            messages=[{"role": "user", "content": task}],
            model="gpt-4o-mini",
            max_tokens=500,
        )
        return {"output": response["content"], "model": "gpt-4o-mini"}

    async def _augmented_answer(self, task: str) -> dict:
        context = await self.retriever.search(task, top_k=5)
        prompt = f"Use this context to answer:\n\nContext:\n{context}\n\nQuestion: {task}"
        response = await self.model_client.chat(
            messages=[{"role": "user", "content": prompt}],
            model="gpt-4o",
            max_tokens=1000,
        )
        return {"output": response["content"], "model": "gpt-4o", "context_used": True}

    async def _reasoned_answer(self, task: str) -> dict:
        # Multi-step with tool use
        agent = BudgetEnforcedAgent(
            self.model_client,
            TokenBudget(total_budget=self.budget.total_budget // 3),
            tools=self.model_client.get_tools(),
        )
        result = await agent.run(task, max_iterations=6)
        return {"output": result.get("output"), "iterations": result.get("iterations")}

    async def _ensemble_answer(self, task: str) -> dict:
        # Multiple models, synthesized
        responses = await asyncio.gather(
            self._direct_answer(task),
            self._augmented_answer(task),
        )

        synthesis_prompt = f"""Two responses to the same question:

Response 1: {responses[0].get('output')}
Response 2: {responses[1].get('output')}

Question: {task}

Synthesize the best answer, noting where responses agree and disagree."""

        response = await self.model_client.chat(
            messages=[{"role": "user", "content": synthesis_prompt}],
            model="claude-sonnet",
            max_tokens=1500,
        )
        return {"output": response["content"], "model": "claude-sonnet", "ensemble": True}

    def _threshold_for_level(self, level: EscalationLevel) -> float:
        thresholds = {
            EscalationLevel.DIRECT: 0.85,
            EscalationLevel.AUGMENTED: 0.80,
            EscalationLevel.REASONED: 0.75,
            EscalationLevel.ENSEMBLE: 0.70,
        }
        return thresholds[level]

    async def _evaluate_quality(self, task: str, result: dict) -> float:
        """Evaluate result quality using a lightweight scoring model."""
        eval_prompt = f"""Score this response from 0.0 to 1.0 for quality, accuracy, and completeness.

Question: {task}
Response: {result.get('output', '')}

Return ONLY a number between 0.0 and 1.0."""

        response = await self.model_client.chat(
            messages=[{"role": "user", "content": eval_prompt}],
            model="gpt-4o-mini",
            max_tokens=10,
        )
        try:
            return float(response["content"].strip())
        except ValueError:
            return 0.0

┌──────────────────────────────────────────────────────────────┐
│                      Client / User                            │
└─────────────────────────┬────────────────────────────────────┘
                          │
┌─────────────────────────▼────────────────────────────────────┐
│              Request Gateway + Rate Limiter                   │
│  (per-user/per-org budgets, request dedup, auth)             │
└─────────────────────────┬────────────────────────────────────┘
                          │
┌─────────────────────────▼────────────────────────────────────┐
│              Task Classifier (Micro/Standard/Extended)        │
│  (lightweight model call, tier assignment)                    │
└─────────────────────────┬────────────────────────────────────┘
                          │
┌─────────────────────────▼────────────────────────────────────┐
│           Progressive Escalation Engine                       │
│  (starts cheap, escalates on quality failure)                 │
└─────────────────────────┬────────────────────────────────────┘
                          │
┌─────────────────────────▼────────────────────────────────────┐
│           Token Budget + Circuit Breaker                      │
│  (global budget, velocity limits, state machine)              │
└─────────────────────────┬────────────────────────────────────┘
                          │
          ┌───────────────┼───────────────┐
          ▼               ▼               ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ Semantic     │ │ Sandboxed    │ │ Sub-Agent    │
│ Cache        │ │ Tool Registry│ │ Spawner      │
│ (dedup)      │ │ (limits)     │ │ (concurrency)│
└──────┬───────┘ └──────┬───────┘ └──────┬───────┘
       │                │                │
       ▼                ▼                ▼
┌────────────────────────────────────────────────────────────┐
│              LLM Provider(s) + Tool Backends                │
└────────────────────────────────────────────────────────────┘

import asyncio
import time
import uuid
from dataclasses import dataclass, field
from typing import Any, Callable, Optional


@dataclass
class AgentRequest:
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    user_id: str = ""
    task: str = ""
    context: dict = field(default_factory=dict)
    priority: str = "normal"  # low, normal, high, critical
    created_at: float = field(default_factory=time.time)


@dataclass
class AgentResult:
    request_id: str
    status: str  # completed, budget_exhausted, timeout, error
    output: Optional[str] = None
    tier: str = ""
    escalation_level: str = ""
    tokens_used: int = 0
    tools_called: int = 0
    elapsed_seconds: float = 0.0
    cost_estimate_usd: float = 0.0
    metadata: dict = field(default_factory=dict)


class SandboxOrchestrator:
    """Main orchestrator combining all budget-control patterns."""

    def __init__(
        self,
        model_client,
        retriever,
        tool_registry: ToolRegistry,
        semantic_cache: SemanticCache,
        org_budget_per_month: float = 1000.0,
    ):
        self.model_client = model_client
        self.retriever = retriever
        self.tool_registry = tool_registry
        self.semantic_cache = semantic_cache
        self.org_budget_per_month = org_budget_per_month
        self.tier_classifier = TierClassifier(model_client)
        self._monthly_spent = 0.0
        self._request_count = 0
        self._cost_log: list[dict] = []

    async def process(self, request: AgentRequest) -> AgentResult:
        start = time.time()

        # Step 1: Check organizational budget
        if self._monthly_spent >= self.org_budget_per_month:
            return AgentResult(
                request_id=request.id,
                status="org_budget_exhausted",
                metadata={"monthly_spent": self._monthly_spent, "limit": self.org_budget_per_month},
            )

        # Step 2: Check semantic cache
        cached = await self.semantic_cache.get_or_compute(
            prompt=request.task,
            model="cache-check",
            compute_fn=lambda: "CACHE_MISS",
        )
        if cached != "CACHE_MISS":
            return AgentResult(
                request_id=request.id,
                status="completed",
                output=cached,
                metadata={"cache_hit": True, "elapsed": time.time() - start},
            )

        # Step 3: Classify tier
        tier = self.tier_classifier.classify(request.task, request.context)
        tier_config = TIER_REGISTRY[tier]

        # Step 4: Create per-request token budget
        # Scale budget by priority
        priority_multiplier = {"low": 0.5, "normal": 1.0, "high": 1.5, "critical": 2.0}
        request_budget = TokenBudget(
            total_budget=int(tier_config.max_tokens_total * priority_multiplier.get(request.priority, 1.0)),
            max_velocity_tokens_per_second=200 if tier == ComputeTier.MICRO else 500,
        )

        # Step 5: Execute with progressive escalation
        escalation_agent = ProgressiveEscalationAgent(
            model_client=self.model_client,
            retriever=self.retriever,
            quality_evaluator=self.model_client,
            budget=request_budget,
        )

        try:
            result = await asyncio.wait_for(
                escalation_agent.execute(request.task),
                timeout=tier_config.timeout_seconds,
            )
        except asyncio.TimeoutError:
            result = {"status": "timeout", "output": None}

        # Step 6: Calculate and track cost
        elapsed = time.time() - start
        cost = self._estimate_cost(request_budget.consumed, tier)
        self._monthly_spent += cost
        self._request_count += 1
        self._cost_log.append({
            "request_id": request.id,
            "user_id": request.user_id,
            "cost": cost,
            "tokens": request_budget.consumed,
            "tier": tier.value,
            "elapsed": elapsed,
        })

        return AgentResult(
            request_id=request.id,
            status=result.get("status", "error"),
            output=result.get("output"),
            tier=tier.value,
            escalation_level=result.get("escalation_level", ""),
            tokens_used=request_budget.consumed,
            tools_called=self.tool_registry.get_all_stats(),
            elapsed_seconds=elapsed,
            cost_estimate_usd=cost,
            metadata=result.get("metadata", {}),
        )

    def _estimate_cost(self, tokens: int, tier: ComputeTier) -> float:
        """Rough cost estimation based on tier's typical model."""
        # Approximate pricing (update for your provider)
        cost_per_1k_tokens = {
            ComputeTier.MICRO: 0.0006,   # gpt-4o-mini
            ComputeTier.STANDARD: 0.005,  # gpt-4o
            ComputeTier.EXTENDED: 0.015,  # claude-sonnet
        }
        return (tokens / 1000) * cost_per_1k_tokens.get(tier, 0.005)

    def get_budget_report(self) -> dict:
        """Generate budget utilization report."""
        avg_cost = sum(c["cost"] for c in self._cost_log) / max(len(self._cost_log), 1)
        return {
            "monthly_spent": self._monthly_spent,
            "monthly_limit": self.org_budget_per_month,
            "utilization_pct": (self._monthly_spent / self.org_budget_per_month) * 100,
            "total_requests": self._request_count,
            "avg_cost_per_request": avg_cost,
            "cache_hit_rate": self.semantic_cache.hit_rate,
            "projected_monthly_cost": avg_cost * 30 * 24 * 60,  # naive projection
        }

# Prometheus alerting rules for agent budget monitoring
 groups:
 - name: agent-budget-alerts
   rules:
   - alert: HighBudgetUtilization
     expr: |
       agent_monthly_spend / agent_monthly_budget > 0.80
     for: 5m
     labels:
       severity: warning
     annotations:
       summary: "Agent monthly budget at {{ $value | humanizePercentage }}"

   - alert: RunawayAgentLoop
     expr: |
       rate(agent_tokens_consumed_total[5m]) > 2000
     for: 2m
     labels:
       severity: critical
     annotations:
       summary: "Token velocity exceeded: {{ $value }} tokens/sec"

   - alert: BudgetExhaustionRate
     expr: |
       rate(agent_budget_exhausted_total[1h]) / rate(agent_requests_total[1h]) > 0.05
     for: 10m
     labels:
       severity: warning
     annotations:
       summary: "5%+ of requests hitting budget limits"