Woke Up to a $10,000 AWS Bill? How to Stop Runaway Amazon Bedrock Agent Loops Before They Drain Your Wallet

import logging
import sys
from typing import Dict, Any

# Setup diagnostics logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - [%(levelname)s] - %(message)s')
logger = logging.getLogger("FinOps_Guardrail_Engine")

class RunawayAgentException(Exception):
"""Custom exception raised when an agent violates financial or iteration boundaries."""
pass

class GuardedAgentRuntime:
"""
Wraps an autonomous agent execution layer with explicit iteration counters
and fiscal safety stops.
"""
def __init__(self, max_allowed_turns: int, max_session_budget_usd: float):
self.max_turns = max_allowed_turns
self.budget_limit = max_session_budget_usd
self.accumulated_cost = 0.0
self.current_turn = 0

def _calculate_step_cost(self, prompt_tokens: int, completion_tokens: int, model_id: str) -> float:
"""
Calculates exact API costs using current model token pricing.
NOTE: The hardcoded rates below are purely illustrative examples. Runtimes must
be updated dynamically based on real-time Amazon Bedrock pricing structures.
"""
pricing_table = {
"anthropic.claude-3-5-sonnet": {"input": 3.00, "output": 15.00},
"amazon.titan-text-premier": {"input": 2.50, "output": 10.00}
}

rates = pricing_table.get(model_id, {"input": 5.00, "output": 15.00})
input_cost = (prompt_tokens / 1_000_000) * rates["input"]
output_cost = (completion_tokens / 1_000_000) * rates["output"]

return input_cost + output_cost

def invoke_guarded_agent(self, task_input: str, model_id: str) -> Dict[str, Any]:
"""
Executes the autonomous reasoning loop while evaluating budget metrics.
"""
logger.info(f"Starting task execution. Maximum safety budget set to: ${self.budget_limit:.4f}")

# Base context footprint before multi-turn history accumulation
base_prompt_tokens = 3000

while self.current_turn < self.max_turns:
self.current_turn += 1
logger.info(f"Executing Loop Turn: {self.current_turn}/{self.max_turns}")

# Simulate context window expansion across progressive iterations
simulated_input_tokens = base_prompt_tokens + (self.current_turn * 750)
simulated_output_tokens = 400

# Evaluate step expenditure
step_fee = self._calculate_step_cost(simulated_input_tokens, simulated_output_tokens, model_id)
self.accumulated_cost += step_fee

logger.info(f"Turn Cost: ${step_fee:.4f} | Accumulated Session Total: ${self.accumulated_cost:.4f}")

# Safety Stop 1: Budget Cap Exceeded
if self.accumulated_cost >= self.budget_limit:
logger.error(f"CRITICAL: Budget threshold breached! Halted at ${self.accumulated_cost:.4f}")
raise RunawayAgentException(f"Execution killed. Total cost crossed safe limit of ${self.budget_limit}")

# Simulate a persistent error state causing a retry loop
task_resolved = False
if task_resolved:
return {"status": "SUCCESS", "final_bill": self.accumulated_cost}

# Safety Stop 2: Max Iterations Exceeded
logger.error(" CRITICAL: Agent hit maximum turn capacity without resolving the task.")
raise RunawayAgentException("Execution killed to prevent an infinite loop scenario.")

# ==========================================
# APPLICATION ENTRYPOINT & RUNTIME TESTING
# ==========================================
if __name__ == "__main__":
# Enforce strict constraints for development environments
safety_layer = GuardedAgentRuntime(max_allowed_turns=4, max_session_budget_usd=0.0500)

try:
safety_layer.invoke_guarded_agent(
task_input="Sync external ERP data pipelines and run anomaly checks",
model_id="anthropic.claude-3-5-sonnet"
)
except RunawayAgentException as guard_alert:
print(f"\n[Circuit Breaker Active]: {guard_alert}")
sys.exit(0)