Mini-Project: A Cost Report for Reservo Runs

Description

Seven lessons built, separately, every piece: tokens as the unit (02), estimate_tokens with its limits confirmed (03), claude-sonnet-5's fixed pricing (04), estimate_cost_cents and cost_for_run with a per-tool-call breakdown (05), aggregating a batch and scaling to thousands of runs (06), and cost as an operational signal with its own boundary (07). This mini-project brings them all together into a complete observability/cost_calculator.py, and runs it over the same batch of four Reservo runs that accompanied the last two lessons — Ana, Sofía, Diego, Carla — each correlated by the deterministic trace_id traced_run (Module 2) already built.

The result is a comprehensive cost report: per run, with its tool-call breakdown; per batch, correctly aggregated; scaled to the magnitude Reservo actually operates at; and with lesson 07's signal criterion applied over the whole batch. By the end of this lesson, you're going to have this guide's second complete artifact — the first was Module 2's observability/run_logger.py — ready for Module 4 to build the third one alongside it.

Connection to the module

This is the synthesis of the eight lessons. There's no new mechanism piece — cost_for_run, aggregate_reports, flag_expensive_runs are exactly lessons 05, 06, and 07's; this mini-project's job is assembling them into a single file and running them together, end to end, over real data.

observability/cost_calculator.py, complete

This is the complete file, with each lesson's pieces in the order they were built:

# observability/cost_calculator.py
"""Costo por run del agente de Reservo (Módulo 3): tokens estimados con
len(texto) // 4 (L03), pricing fijo de claude-sonnet-5 (L04), costo en
centavos con desglose por tool call (L05), agregación de lote y escalado
(L06), y el costo como señal operacional (L07). Reusa run_logger.py
(Módulo 2) para el trace_id -- no construye ninguna correlación nueva."""
import itertools
import json
import statistics
from dataclasses import dataclass, field

INPUT_PRICE_CENTS_PER_MILLION_TOKENS = 300    # $3.00 / 1M tokens -- claude-sonnet-5, precio de lista
OUTPUT_PRICE_CENTS_PER_MILLION_TOKENS = 1500  # $15.00 / 1M tokens -- claude-sonnet-5, precio de lista


def estimate_tokens(text):
    """L03: estimación de ORDEN DE MAGNITUD. Nunca un conteo exacto de un
    tokenizer real."""
    return len(text) // 4


def estimate_cost_cents(input_tokens, output_tokens):
    """L04+L05: la fórmula central del módulo, en centavos int."""
    return (
        input_tokens * INPUT_PRICE_CENTS_PER_MILLION_TOKENS
        + output_tokens * OUTPUT_PRICE_CENTS_PER_MILLION_TOKENS
    ) // 1_000_000


@dataclass
class StepCost:
    """L05: el costo estimado de UN tool call dentro de un run."""
    step: int
    tool: str
    input_tokens: int
    output_tokens: int
    cost_cents: int


@dataclass
class CostReport:
    """L05: el costo estimado de un run completo, identificado por el
    mismo trace_id de run_logger.py (Módulo 2)."""
    trace_id: str
    question: str
    steps: list = field(default_factory=list)
    input_tokens: int = 0
    output_tokens: int = 0
    cost_cents: int = 0


def cost_for_run(trace_id, question, history):
    """L05: recorre history (de run_reservo_agent, SIN tocarlo) y calcula
    el costo del run completo, con desglose por tool call."""
    steps = []
    total_input_text = question
    total_output_text = ""
    pending = {}
    step_counter = itertools.count(1)

    for turn in history:
        content = turn["content"]
        if isinstance(content, str):
            continue
        for block in content:
            if block["type"] == "tool_use":
                step = next(step_counter)
                args_text = json.dumps(block["input"])
                total_output_text += args_text
                pending[block["id"]] = {"step": step, "tool": block["name"], "args_text": args_text}
            elif block["type"] == "tool_result":
                entry = pending[block["tool_use_id"]]
                result_text = block["content"]
                total_input_text += result_text
                step_in = estimate_tokens(result_text)
                step_out = estimate_tokens(entry["args_text"])
                steps.append(StepCost(
                    step=entry["step"], tool=entry["tool"],
                    input_tokens=step_in, output_tokens=step_out,
                    cost_cents=estimate_cost_cents(step_in, step_out),
                ))
            elif block["type"] == "text":
                total_output_text += block["text"]

    input_tokens = estimate_tokens(total_input_text)
    output_tokens = estimate_tokens(total_output_text)
    return CostReport(
        trace_id=trace_id, question=question, steps=steps,
        input_tokens=input_tokens, output_tokens=output_tokens,
        cost_cents=estimate_cost_cents(input_tokens, output_tokens),
    )


def aggregate_reports(reports):
    """L06: suma TOKENS de todos los reports primero, aplica
    estimate_cost_cents UNA SOLA VEZ -- nunca suma cost_cents ya
    redondeados."""
    total_input = sum(r.input_tokens for r in reports)
    total_output = sum(r.output_tokens for r in reports)
    return total_input, total_output, estimate_cost_cents(total_input, total_output)


def project_cost_cents(input_tokens_per_run, output_tokens_per_run, n_runs):
    """L06: proyecta el costo de n_runs runs con un perfil de tokens dado
    -- multiplica tokens primero, redondea al final."""
    return estimate_cost_cents(input_tokens_per_run * n_runs, output_tokens_per_run * n_runs)


def flag_expensive_runs(reports, threshold_ratio=1.5):
    """L07: marca los reports cuyo total de tokens supera threshold_ratio
    veces el promedio del lote -- un criterio de FORMA, determinista."""
    totals = [r.input_tokens + r.output_tokens for r in reports]
    avg = statistics.mean(totals)
    flagged = []
    for r, total in zip(reports, totals):
        ratio = total / avg
        if ratio >= threshold_ratio:
            flagged.append((r, ratio))
    return flagged

Ten lines of discipline from this guide summed up in each piece: every number here is int, and every estimate is labeled, in its own comment, for what it is — an order-of-magnitude approximation, never an exact count.

Worked example: the complete batch, end to end

Run the same four tasks from lessons 06 and 07, this time with the complete report: per run, batch aggregate, cost signal, and scale projection.

import logging
import reservo_agent as ra
import run_logger as rl
import cost_calculator as cc

rl.logger.setLevel(logging.CRITICAL)  # silenciamos traced_run para este reporte final

# Los mismos cuatro guiones de las lecciones 06 y 07: Ana (con un error corregido),
# Sofía (limpio), Diego (reserva y cancela), Carla (compara seis combinaciones).
tasks = [
    ("Reserva Focus pro 3h para Ana", script_a),
    ("Reserva Boardroom pro 1h para Sofía", script_b),
    ("Reserva y cancela Studio basic 1h para Diego", script_d),
    ("Compara todas las salas antes de reservar la mejor opción para Carla", script_compare),
]

reports = []
for i, (question, script) in enumerate(tasks, start=1):
    with rl.traced_run(question, i) as trace_id:
        final, history = ra.run_reservo_agent(question, script)
    reports.append(cc.cost_for_run(trace_id, question, history))

print("=== reporte de costo por run ===")
for r in reports:
    print(f"{r.trace_id}  in={r.input_tokens:>4} out={r.output_tokens:>4} cost_cents={r.cost_cents}  {r.question!r}")

print()
print("=== agregado del lote ===")
total_in, total_out, total_cost = cc.aggregate_reports(reports)
print("input_tokens totales :", total_in)
print("output_tokens totales:", total_out)
print("costo total del lote :", total_cost, "centavos")

print()
print("=== señales: runs anómalamente caros (>= 1.5x el promedio) ===")
flagged = cc.flag_expensive_runs(reports, threshold_ratio=1.5)
for r, ratio in flagged:
    print(f"  {r.trace_id}: {ratio:.2f}x -- {r.question!r}")

print()
print("=== proyección a escala, usando el perfil promedio del lote ===")
avg_in, avg_out = total_in / len(reports), total_out / len(reports)
for n in (1_000, 10_000, 100_000):
    cost_n = cc.project_cost_cents(int(avg_in), int(avg_out), n)
    print(f"  {n:>7} runs -> {cost_n:>6} centavos = ${cost_n / 100:.2f}")

What to expect:

=== reporte de costo por run ===
run-8487582448eb  in=  64 out=  56 cost_cents=0  'Reserva Focus pro 3h para Ana'
run-ae6ff85cf0b0  in=  53 out=  49 cost_cents=0  'Reserva Boardroom pro 1h para Sofía'
run-2c27934d8a39  in=  27 out=  31 cost_cents=0  'Reserva y cancela Studio basic 1h para Diego'
run-cecde864aa84  in=  88 out= 118 cost_cents=0  'Compara todas las salas antes de reservar la mejor opción para Carla'

=== agregado del lote ===
input_tokens totales : 232
output_tokens totales: 254
costo total del lote : 0 centavos

=== señales: runs anómalamente caros (>= 1.5x el promedio) ===
  run-cecde864aa84: 1.70x -- 'Compara todas las salas antes de reservar la mejor opción para Carla'

=== proyección a escala, usando el perfil promedio del lote ===
     1000 runs ->    112 centavos = $1.12
    10000 runs ->   1126 centavos = $11.26
   100000 runs ->  11265 centavos = $112.65

This is the complete report lesson 01 promised when it opened the module: every CostReport correlated by its trace_id — the same one that would show up in RUN_LOG.jsonl if traced_run were logging at INFO instead of CRITICAL — the batch aggregated with the correct arithmetic (0 cents, honest for four runs this size), the cost signal identifying exactly Carla's run as the outlier, and the projection showing that same batch profile, at 100,000 runs, costs a real $112.65. Four "per run" lines, three analysis blocks — all from a single file, without touching a single line of reservo_agent.py or run_logger.py.

Common mistakes

1. Thinking this mini-project "already optimizes" the agent's cost. No — as lesson 07 insisted, everything this file does is measure and classify. Not a single line of cost_calculator.py caches a tool_result, switches models, or batches questions — that is, precisely, the boundary toward cost-optimization-caching-guide.

2. Calculating the cost report without wrapping the run in traced_run. It's possible to call cost_for_run directly over a history obtained without traced_run — passing any string as trace_id — but you lose the real correlation with Module 2's RUN_LOG.jsonl. The correct pattern, used throughout this mini-project, is getting the trace_id from traced_run and passing it to cost_for_run unmodified.

3. Reordering aggregate_reports and flag_expensive_runs expecting the same result. The order they're called in doesn't matter — each operates over reports, an already-calculated list — but it does matter that both receive the batch's complete list: calling flag_expensive_runs over a subset would change the average each run gets compared against, and therefore which runs get flagged.

4. Using project_cost_cents with a single run's profile when the batch has runs very different from each other. The worked example uses the batch's average (avg_in, avg_out) to project, not Ana's run in particular — with a batch where the most expensive run is 1.70x the average, projecting with the wrong run can over- or underestimate the real cost at scale.

5. Forgetting cost_calculator.py depends on the complete history, not on RUN_LOG.jsonl. This lesson's Exercise 3 confirms, with real numbers, why trying to reconstruct a run's cost only from Module 2's persisted lines underestimates the result — the agent's final response text was never logged as its own event.

Exercises

Exercise 1: Add a fifth run and recalculate the complete report (Easy)

Add a fifth task to the batch: Luis cancels booking 999, which doesn't exist (is_error: True, with no real booking involved). Run cost_for_run over that run, add it to reports, and recalculate the batch aggregate with aggregate_reports.

script_luis = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "cancel_booking", "input": {"id": 999}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "No encontré esa reserva."}]},
]
with rl.traced_run("Cancela la reserva 999 para Luis", 5) as trace_id_luis:
    final_luis, history_luis = ra.run_reservo_agent("Cancela la reserva 999 para Luis", script_luis)

report_luis = cc.cost_for_run(trace_id_luis, "Cancela la reserva 999 para Luis", history_luis)
reports.append(report_luis)

print(f"{report_luis.trace_id}  in={report_luis.input_tokens:>4} out={report_luis.output_tokens:>4} "
      f"cost_cents={report_luis.cost_cents}  {report_luis.question!r}")

total_in, total_out, total_cost = cc.aggregate_reports(reports)
print("input_tokens totales (5 runs) :", total_in)
print("output_tokens totales (5 runs):", total_out)
print("costo total del lote (5 runs) :", total_cost, "centavos")

run-65388909596a  in=  16 out=   8 cost_cents=0  'Cancela la reserva 999 para Luis'

input_tokens totales (5 runs) : 248
output_tokens totales (5 runs): 262
costo total del lote (5 runs) : 0 centavos

Exercise 2: Find the run that contributes most to the batch's cost (Medium)

With Exercise 1's five CostReports, find the run with the highest total tokens (input_tokens + output_tokens), without assuming in advance which one it is.

busiest = max(reports, key=lambda r: r.input_tokens + r.output_tokens)
total_busiest = busiest.input_tokens + busiest.output_tokens
print(f"run que más contribuye: {busiest.trace_id} ({total_busiest} tokens) -- {busiest.question!r}")

run que más contribuye: run-cecde864aa84 (206 tokens) -- 'Compara todas las salas antes de reservar la mejor opción para Carla'

Exercise 3: Confirm why RUN_LOG.jsonl alone, without history, underestimates cost (Hard)

RUN_LOG.jsonl (Module 2) records every tool_use and tool_result, but never records the agent's final response text — the text block from the turn with stop_reason: "end_turn" — as its own run_logger.py event. Reconstruct Ana's run's output_tokens two ways: (a) summing only each tool_use's arguments (what could actually be reconstructed from RUN_LOG.jsonl), and (b) summing those same arguments plus the response's final text (what cost_for_run does calculate, because it has access to the complete history). Compare both results.

import json

args_del_run_de_ana = [
    json.dumps({}),                                                          # list_rooms
    json.dumps({"room": "Focus", "tier": "premium", "hours": 3}),             # get_quote (rechazado)
    json.dumps({"room": "Focus", "tier": "pro", "hours": 3}),                 # get_quote (corregido)
    json.dumps({"room": "Focus", "tier": "pro", "hours": 3, "member": "Ana"}), # book_room
]
texto_final = "Reservé Focus pro por 3 horas para Ana. Total $60.00. Confirmación #1."

solo_args = "".join(args_del_run_de_ana)
args_mas_texto_final = solo_args + texto_final

print("output_tokens SOLO con args de tool_use (lo que RUN_LOG.jsonl podría dar):",
      cc.estimate_tokens(solo_args))
print("output_tokens CON el texto final (lo que cost_for_run SÍ calcula)      :",
      cc.estimate_tokens(args_mas_texto_final))
print("tokens que se pierden si solo se usa RUN_LOG.jsonl:",
      cc.estimate_tokens(args_mas_texto_final) - cc.estimate_tokens(solo_args))

output_tokens SOLO con args de tool_use (lo que RUN_LOG.jsonl podría dar): 38
output_tokens CON el texto final (lo que cost_for_run SÍ calcula)      : 56
tokens que se pierden si solo se usa RUN_LOG.jsonl: 18

Summary and next step

• We assembled a complete observability/cost_calculator.py: estimate_tokens (L03), the fixed pricing (L04), estimate_cost_cents + cost_for_run with a breakdown (L05), aggregate_reports + project_cost_cents (L06), and flag_expensive_runs (L07) — eight lessons, one file.
• We ran it over the complete batch of four Reservo runs, correlated by trace_id: a per-run report, a batch aggregate (0 cents, honest), a cost signal (Carla's run, 1.70x the average), and a scale projection ($112.65 at 100,000 runs).
• We confirmed, with real numbers, a genuine limit in this guide: a run's cost is calculated over its complete history, available only while the process that generated it stays alive — reconstructing it solely from RUN_LOG.jsonl would underestimate the result, because the response's final text was never recorded as its own event.

With this, Module 3 closes. You have a complete observability/cost_calculator.py, and the run-tested evidence that it precisely answers the question Module 1 left open: "how much did this run cost, and why?"

Next module: Module 4 — Measuring Latency Honestly. With cost now solved, this module completes the second half of Module 1's "measure" layer: how long a run took, per tool and in total, with the same honesty about what's measured and what's modeled that you already saw with cost — never with a real stopwatch inside a "What to expect" block.

Additional resources

1. Anthropic — Pricing — The source for claude-sonnet-5's list price, the constant this module fixed in lesson 04 and reused, unchanged, in every following lesson.
2. Anthropic — Token counting — The real token count, the reference this module's entire estimate is honestly measured against.
3. Python — dataclasses — StepCost and CostReport, the structures that organize every result in this module.
4. Python — statistics — statistics.mean, the foundation of flag_expensive_runs; statistics.median and statistics.quantiles are Module 4's central content.
5. Python 3.14 — What's New — The version every line of code in this module ran on, including this mini-project's final report.