The $200 crash did not happen. It almost happened. A support automation agent got stuck in a retry loop during a test run. The projection was clear: same bad loop, same document search, same model calls, left running overnight. The estimate landed near $200 for one avoidable failure.
The trace was a mess. The agent called the right tool with the wrong input, retried against stale context, summarized old results, and kept paying for each turn. The final answer looked polished enough to pass a sleepy review. The execution behind it was not polished at all.
That is when the agent stopped being a chat feature and became a system that needs a black box. Not a dashboard. Not a full observability stack. Not another hosted service. Just one local file that can answer: What did the agent try? Which tool did it call? What input did the tool receive? Did the tool fail? How long did it take? Did the run cross a cost or turn limit? Can I query the run after everything is over?
This is the 71-line implementation that captures tool calls, stops runaway loops, and enables SQL queries over crash data without external services.
The Observability Gap in Agent Workflows
Most agent frameworks give you logs. Logs are fine for happy paths. They fall apart when you need to reconstruct a multi-turn failure three days later.
The questions you need to answer are relational:
- Which tool calls happened before the timeout?
- Did the agent retry the same input twice?
- What was the cumulative cost before the circuit breaker fired?
- Which turn reused stale context?
Logs are append-only text. You need structured data and a query layer.
The standard answer is OpenTelemetry spans or a vendor observability SDK. Both work. Both add dependencies, configuration, and external services. For a single-agent workflow running on a developer laptop or a small batch server, that is overhead you do not need yet.
The alternative is a local black box: a Python decorator that intercepts tool calls, writes structured records to a DuckDB file, and lets you query failures with SQL.
Architecture: Interception, Recording, and Query
The black box sits between the agent framework and the tool functions. It does not modify framework code. It wraps tool functions with a decorator that records invocations.
Flow:
- Agent decides to call a tool.
- Decorator intercepts the call.
- Decorator records: tool name, input, timestamp, run ID.
- Tool executes.
- Decorator records: output, error, duration, token count (if available).
- Decorator checks circuit breaker: turn limit, cost limit, timeout.
- If breaker trips, raise exception and stop the run.
- Write all records to DuckDB file.
Storage schema:
CREATE TABLE tool_calls (
run_id TEXT,
turn_number INTEGER,
tool_name TEXT,
input_json TEXT,
output_json TEXT,
error_message TEXT,
duration_ms REAL,
timestamp TIMESTAMP,
token_count INTEGER,
cost_estimate REAL
);DuckDB is embedded. No server. No network. The file lives next to your agent script. You query it with the DuckDB CLI or Python client.
Implementation: The 71-Line Decorator
The decorator wraps any tool function. It uses Python’s functools.wraps to preserve the original signature. It uses a global context object to track the current run ID and turn number.
Core logic:
import functools
import time
import json
import duckdb
from datetime import datetime
class AgentBlackBox:
def __init__(self, db_path="agent_trace.duckdb", max_turns=50, max_cost=10.0):
self.db_path = db_path
self.max_turns = max_turns
self.max_cost = max_cost
self.current_run_id = None
self.turn_counter = 0
self.total_cost = 0.0
self._init_db()
def _init_db(self):
conn = duckdb.connect(self.db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS tool_calls (
run_id TEXT,
turn_number INTEGER,
tool_name TEXT,
input_json TEXT,
output_json TEXT,
error_message TEXT,
duration_ms REAL,
timestamp TIMESTAMP,
token_count INTEGER,
cost_estimate REAL
)
""")
conn.close()
def start_run(self, run_id):
self.current_run_id = run_id
self.turn_counter = 0
self.total_cost = 0.0
def record_tool(self, func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
if self.turn_counter >= self.max_turns:
raise RuntimeError(f"Circuit breaker: exceeded {self.max_turns} turns")
self.turn_counter += 1
start = time.perf_counter()
error_msg = None
output = None
token_count = 0
cost = 0.0
try:
output = func(*args, **kwargs)
# Extract token count if tool returns metadata
if isinstance(output, dict) and "tokens" in output:
token_count = output["tokens"]
cost = token_count * 0.00002 # Example pricing
self.total_cost += cost
except Exception as e:
error_msg = str(e)
raise
finally:
duration = (time.perf_counter() - start) * 1000
self._write_record(
tool_name=func.__name__,
input_data={"args": args, "kwargs": kwargs},
output_data=output,
error_message=error_msg,
duration_ms=duration,
token_count=token_count,
cost_estimate=cost
)
if self.total_cost > self.max_cost:
raise RuntimeError(f"Circuit breaker: exceeded ${self.max_cost} cost")
return output
return wrapper
def _write_record(self, tool_name, input_data, output_data, error_message, duration_ms, token_count, cost_estimate):
conn = duckdb.connect(self.db_path)
conn.execute("""
INSERT INTO tool_calls VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", [
self.current_run_id,
self.turn_counter,
tool_name,
json.dumps(input_data),
json.dumps(output_data) if output_data else None,
error_message,
duration_ms,
datetime.now(),
token_count,
cost_estimate
])
conn.close()
# Usage
black_box = AgentBlackBox(max_turns=20, max_cost=5.0)
@black_box.record_tool
def search_docs(query):
# Simulate document search
time.sleep(0.15)
return {"results": ["doc1", "doc2"], "tokens": 500}
@black_box.record_tool
def summarize(text):
# Simulate LLM call
time.sleep(0.3)
return {"summary": "...", "tokens": 1200}The decorator is 71 lines with whitespace and comments removed. It handles:
- Turn counting
- Cost tracking
- Duration measurement
- Error capture
- Structured logging to DuckDB
Sanitization: Stripping Secrets Before Persistence
Agent tool calls often include API keys, user emails, or document content. You cannot write raw inputs to a local file without sanitization.
Sanitization strategies:
| Data Type | Sanitization Method | Trade-off |
|---|---|---|
| API keys | Redact entirely | Lose ability to debug auth failures |
| User emails | Hash with salt | Preserve uniqueness, lose readability |
| Document content | Store first 100 chars + hash | Preserve context, lose full text |
| Tool outputs | Redact PII fields | Requires schema knowledge |
| Error messages | Regex filter for secrets | Fragile, misses novel patterns |
The black box adds a sanitization hook before writing:
def _sanitize(self, data):
if isinstance(data, dict):
sanitized = {}
for key, value in data.items():
if key in ["api_key", "password", "token"]:
sanitized[key] = "[REDACTED]"
elif key == "email":
sanitized[key] = hashlib.sha256(value.encode()).hexdigest()[:16]
elif key == "content" and isinstance(value, str):
sanitized[key] = value[:100] + "..." if len(value) > 100 else value
else:
sanitized[key] = self._sanitize(value)
return sanitized
elif isinstance(data, list):
return [self._sanitize(item) for item in data]
return dataThis runs before json.dumps() in _write_record(). It is not perfect. It misses secrets in nested structures or non-standard field names. For production use, combine with a secret scanning library like detect-secrets or truffleHog.
Querying the Crash: SQL Over Agent Traces
Once the black box writes to DuckDB, you query it like any relational database.
Find all failed tool calls:
SELECT run_id, turn_number, tool_name, error_message, duration_ms
FROM tool_calls
WHERE error_message IS NOT NULL
ORDER BY timestamp DESC;Find runs that exceeded 10 turns:
SELECT run_id, COUNT(*) as turn_count, SUM(cost_estimate) as total_cost
FROM tool_calls
GROUP BY run_id
HAVING COUNT(*) > 10
ORDER BY total_cost DESC;Find duplicate tool calls (retry loops):
SELECT run_id, tool_name, input_json, COUNT(*) as call_count
FROM tool_calls
GROUP BY run_id, tool_name, input_json
HAVING COUNT(*) > 1;Find the slowest tool calls:
SELECT tool_name, AVG(duration_ms) as avg_duration, MAX(duration_ms) as max_duration
FROM tool_calls
GROUP BY tool_name
ORDER BY avg_duration DESC;DuckDB supports window functions, CTEs, and JSON extraction. You can query nested tool inputs without flattening the schema:
SELECT run_id, turn_number,
json_extract_string(input_json, '$.kwargs.query') as query
FROM tool_calls
WHERE tool_name = 'search_docs';Circuit Breaker: Stopping Runaway Runs
The black box enforces two limits: max turns and max cost. Both are configurable at initialization.
Turn limit: Prevents infinite loops. If the agent calls tools more than max_turns times in one run, the decorator raises RuntimeError and stops execution.
Cost limit: Prevents budget overruns. If cumulative cost_estimate exceeds max_cost, the decorator raises RuntimeError.
Both limits are checked after the tool executes but before returning to the agent. This ensures the tool call is recorded even if the breaker trips.
Why not check before execution? Because you want the trace to show the exact call that triggered the limit. If you check before, the trace stops one turn early and you lose context.
Comparison: Black Box vs. OpenTelemetry vs. Vendor SDKs
| Approach | Setup Complexity | Query Flexibility | Cost | Dependency Count |
|---|---|---|---|---|
| 71-line black box | 1 file, 1 decorator | Full SQL via DuckDB | Free | 1 (DuckDB) |
| OpenTelemetry | Collector + exporter config | Limited (trace UI) | Free (self-hosted) | 3+ (SDK, exporter, backend) |
| Langfuse / Helicone | SDK + API key | Vendor UI + API | Paid after free tier | 1 (vendor SDK) |
The black box wins on simplicity and local control. It loses on distributed tracing, real-time dashboards, and team collaboration. If you are running one agent on one machine, the black box is enough. If you are running 50 agents across 10 servers, you need OpenTelemetry or a vendor platform.
Failure Modes and Mitigations
Failure mode: DuckDB file grows unbounded.
Mitigation: Add a retention policy. Delete records older than 30 days or runs older than 100 entries.
def _prune_old_runs(self):
conn = duckdb.connect(self.db_path)
conn.execute("""
DELETE FROM tool_calls
WHERE timestamp < NOW() - INTERVAL 30 DAYS
""")
conn.close()Failure mode: Sanitization misses a secret.
Mitigation: Run a post-write scan with detect-secrets or a regex filter. Log warnings but do not block writes.
Failure mode: Circuit breaker trips too early.
Mitigation: Make limits configurable per run. Pass max_turns and max_cost to start_run() instead of hardcoding at initialization.
Failure mode: DuckDB write fails (disk full, permission error).
Mitigation: Wrap _write_record() in a try-except. Log the error but do not crash the agent. The tool call still executes.
Technical Verdict
Use this pattern when:
- You are running a single agent or small batch of agents on one machine.
- You need to debug failures after the fact, not in real time.
- You want zero external dependencies beyond DuckDB.
- You need SQL-level query flexibility over agent traces.
- You are prototyping observability before committing to a vendor platform.
Avoid this pattern when:
- You are running distributed agents across multiple servers.
- You need real-time alerting or dashboards.
- You need team-wide trace sharing and collaboration.
- You are already using OpenTelemetry for other services.
- You need compliance-grade audit logs (the black box is not tamper-proof).
The 71-line black box is not a replacement for production observability. It is a stepping stone. It gives you structured traces, cost control, and query power without the overhead of a full platform. When you outgrow it, the migration path is clear: export the DuckDB schema to OpenTelemetry spans or a vendor SDK. The data model is already relational. The hard part is already done.