Skip to content

Add GPT-5.6 pricing #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bobergda wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Add GPT-5.6 pricing #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3d014d1
Add GPT-5.6 pricing
bobergda Jul 18, 2026
8c988e6
Price GPT-5.6 long context
bobergda Jul 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 54 additions & 15 deletions src/codex_observatory/cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@
"bar": "38;5;45",
"dim": "38;5;244",
}
PRICING_VERIFIED_AT = "2026-04-29"
PRICING_VERIFIED_AT = "2026-07-18"
API_EQUIVALENT_NOTE = "Estimated API-equivalent spend only; Codex app or subscription billing may differ."
LONG_CONTEXT_INPUT_THRESHOLD = 272_000
LONG_CONTEXT_PRICING_NOTE = (
Expand All @@ -37,6 +37,9 @@
PRICING_SOURCES = [
"https://openai.com/api/pricing/",
"https://developers.openai.com/api/docs/pricing",
"https://developers.openai.com/api/docs/models/gpt-5.6-sol",
"https://developers.openai.com/api/docs/models/gpt-5.6-terra",
"https://developers.openai.com/api/docs/models/gpt-5.6-luna",
"https://developers.openai.com/api/docs/models/gpt-5.5",
"https://developers.openai.com/api/docs/models/gpt-5.5-pro",
"https://developers.openai.com/api/docs/models/gpt-5.4",
Expand Down Expand Up @@ -64,8 +67,9 @@ class ModelPricing:
long_context_cached_input_usd_per_million: float | None = None
long_context_output_usd_per_million: float | None = None
long_context_input_threshold: int | None = None
cache_write_usd_per_million: float | None = None

def rates_for_input_tokens(self, input_tokens: int) -> tuple[float, float, float]:
def rates_for_input_tokens(self, input_tokens: int) -> tuple[float, float, float, float]:
if (
self.long_context_input_threshold is not None
and input_tokens > self.long_context_input_threshold
Expand All @@ -77,8 +81,13 @@ def rates_for_input_tokens(self, input_tokens: int) -> tuple[float, float, float
if self.long_context_cached_input_usd_per_million is not None
else self.cached_input_usd_per_million
)
return self.long_context_input_usd_per_million, cached_rate, self.long_context_output_usd_per_million
return self.input_usd_per_million, self.cached_input_usd_per_million, self.output_usd_per_million
cache_write_rate = (
self.long_context_input_usd_per_million * 1.25
if self.cache_write_usd_per_million is not None
else 0.0
)
return self.long_context_input_usd_per_million, cached_rate, cache_write_rate, self.long_context_output_usd_per_million
return self.input_usd_per_million, self.cached_input_usd_per_million, self.cache_write_usd_per_million or 0.0, self.output_usd_per_million


MODEL_PRICING = {
Expand All @@ -100,6 +109,9 @@ def rates_for_input_tokens(self, input_tokens: int) -> tuple[float, float, float
"gpt-5.4-pro": ModelPricing(30.00, 30.00, 180.00, 60.00, 60.00, 270.00, LONG_CONTEXT_INPUT_THRESHOLD),
"gpt-5.5": ModelPricing(5.00, 0.50, 30.00, 10.00, 1.00, 45.00, LONG_CONTEXT_INPUT_THRESHOLD),
"gpt-5.5-pro": ModelPricing(30.00, 30.00, 180.00, 60.00, 60.00, 270.00, LONG_CONTEXT_INPUT_THRESHOLD),
"gpt-5.6-sol": ModelPricing(5.00, 0.50, 30.00, 10.00, 1.00, 45.00, LONG_CONTEXT_INPUT_THRESHOLD, 6.25),
"gpt-5.6-terra": ModelPricing(2.50, 0.25, 15.00, 5.00, 0.50, 22.50, LONG_CONTEXT_INPUT_THRESHOLD, 3.125),
"gpt-5.6-luna": ModelPricing(1.00, 0.10, 6.00, 2.00, 0.20, 9.00, LONG_CONTEXT_INPUT_THRESHOLD, 1.25),
}
MODEL_ALIASES = {
"gpt-5 mini": "gpt-5-mini",
Expand All @@ -111,6 +123,10 @@ def rates_for_input_tokens(self, input_tokens: int) -> tuple[float, float, float
"gpt-5.4 nano": "gpt-5.4-nano",
"gpt-5.4 pro": "gpt-5.4-pro",
"gpt-5.5 pro": "gpt-5.5-pro",
"gpt-5.6": "gpt-5.6-sol",
"gpt-5.6 sol": "gpt-5.6-sol",
"gpt-5.6 terra": "gpt-5.6-terra",
"gpt-5.6 luna": "gpt-5.6-luna",
}


Expand Down Expand Up @@ -138,6 +154,7 @@ class TokenEvent:
model: str
input_tokens: int
cached_input_tokens: int
cache_write_input_tokens: int
output_tokens: int
reasoning_output_tokens: int
total_tokens: int
Expand All @@ -150,6 +167,7 @@ class PeriodStats:
turns: int = 0
input_tokens: int = 0
cached_input_tokens: int = 0
cache_write_input_tokens: int = 0
output_tokens: int = 0
reasoning_output_tokens: int = 0
tokens: int = 0
Expand Down Expand Up @@ -209,11 +227,13 @@ class ModelStats:
turns: int = 0
input_tokens: int = 0
cached_input_tokens: int = 0
cache_write_input_tokens: int = 0
output_tokens: int = 0
reasoning_output_tokens: int = 0
total_tokens: int = 0
input_cost_usd: float = 0.0
cached_input_cost_usd: float = 0.0
cache_write_input_cost_usd: float = 0.0
output_cost_usd: float = 0.0
total_cost_usd: float = 0.0
cache_savings_usd: float = 0.0
Expand All @@ -224,6 +244,7 @@ class ModelStats:
class CostStats:
input_cost_usd: float = 0.0
cached_input_cost_usd: float = 0.0
cache_write_input_cost_usd: float = 0.0
output_cost_usd: float = 0.0
total_cost_usd: float = 0.0
cache_savings_usd: float = 0.0
Expand Down Expand Up @@ -590,7 +611,7 @@ def load_sessions(codex_home: Path) -> tuple[list[TurnEvent], list[TokenEvent],

for session_file in sorted(sessions_root.rglob("*.jsonl")):
current_model = "unknown"
prev_input = prev_cached = prev_output = prev_reasoning = prev_total = 0
prev_input = prev_cached = prev_cache_write = prev_output = prev_reasoning = prev_total = 0
fallback_time = datetime.fromtimestamp(session_file.stat().st_mtime).replace(microsecond=0)

for line in read_jsonl_lines(session_file):
Expand Down Expand Up @@ -625,12 +646,14 @@ def load_sessions(codex_home: Path) -> tuple[list[TurnEvent], list[TokenEvent],
usage = info.get("total_token_usage") or {}
current_input = int(usage.get("input_tokens") or 0)
current_cached = int(usage.get("cached_input_tokens") or 0)
current_cache_write = int(usage.get("cache_write_input_tokens") or 0)
current_output = int(usage.get("output_tokens") or 0)
current_reasoning = int(usage.get("reasoning_output_tokens") or 0)
current_total = int(usage.get("total_tokens") or 0)

delta_input = current_input - prev_input
delta_cached = current_cached - prev_cached
delta_cache_write = current_cache_write - prev_cache_write
delta_output = current_output - prev_output
delta_reasoning = current_reasoning - prev_reasoning
delta_total = current_total - prev_total
Expand All @@ -639,6 +662,8 @@ def load_sessions(codex_home: Path) -> tuple[list[TurnEvent], list[TokenEvent],
delta_input = current_input
if delta_cached < 0:
delta_cached = current_cached
if delta_cache_write < 0:
delta_cache_write = current_cache_write
if delta_output < 0:
delta_output = current_output
if delta_reasoning < 0:
Expand All @@ -648,11 +673,12 @@ def load_sessions(codex_home: Path) -> tuple[list[TurnEvent], list[TokenEvent],

prev_input = current_input
prev_cached = current_cached
prev_cache_write = current_cache_write
prev_output = current_output
prev_reasoning = current_reasoning
prev_total = current_total

if not any((delta_input, delta_cached, delta_output, delta_reasoning, delta_total)):
if not any((delta_input, delta_cached, delta_cache_write, delta_output, delta_reasoning, delta_total)):
continue

token_events.append(
Expand All @@ -665,6 +691,7 @@ def load_sessions(codex_home: Path) -> tuple[list[TurnEvent], list[TokenEvent],
model=current_model,
input_tokens=delta_input,
cached_input_tokens=delta_cached,
cache_write_input_tokens=delta_cache_write,
output_tokens=delta_output,
reasoning_output_tokens=delta_reasoning,
total_tokens=delta_total,
Expand All @@ -679,6 +706,7 @@ def load_sessions(codex_home: Path) -> tuple[list[TurnEvent], list[TokenEvent],
def add_token_breakdown(stats: PeriodStats, event: TokenEvent) -> None:
stats.input_tokens += event.input_tokens
stats.cached_input_tokens += event.cached_input_tokens
stats.cache_write_input_tokens += event.cache_write_input_tokens
stats.output_tokens += event.output_tokens
stats.reasoning_output_tokens += event.reasoning_output_tokens
stats.tokens += event.total_tokens
Expand Down Expand Up @@ -759,16 +787,18 @@ def resolve_model_pricing(model: str) -> tuple[str | None, ModelPricing | None]:
return None, None


def estimate_token_costs(input_tokens: int, cached_input_tokens: int, output_tokens: int, pricing: ModelPricing) -> tuple[float, float, float, float, float]:
def estimate_token_costs(input_tokens: int, cached_input_tokens: int, cache_write_input_tokens: int, output_tokens: int, pricing: ModelPricing) -> tuple[float, float, float, float, float, float]:
billed_cached_input = max(0, min(cached_input_tokens, input_tokens))
billed_uncached_input = max(0, input_tokens - billed_cached_input)
input_rate, cached_input_rate, output_rate = pricing.rates_for_input_tokens(input_tokens)
billed_cache_write_input = max(0, min(cache_write_input_tokens, input_tokens - billed_cached_input))
billed_uncached_input = max(0, input_tokens - billed_cached_input - billed_cache_write_input)
input_rate, cached_input_rate, cache_write_rate, output_rate = pricing.rates_for_input_tokens(input_tokens)
input_cost = billed_uncached_input / 1_000_000 * input_rate
cached_input_cost = billed_cached_input / 1_000_000 * cached_input_rate
cache_write_input_cost = billed_cache_write_input / 1_000_000 * cache_write_rate
output_cost = output_tokens / 1_000_000 * output_rate
total_cost = input_cost + cached_input_cost + output_cost
total_cost = input_cost + cached_input_cost + cache_write_input_cost + output_cost
cache_savings = billed_cached_input / 1_000_000 * (input_rate - cached_input_rate)
return input_cost, cached_input_cost, output_cost, total_cost, cache_savings
return input_cost, cached_input_cost, cache_write_input_cost, output_cost, total_cost, cache_savings


def apply_model_pricing(model_name: str, stats: ModelStats) -> bool:
Expand All @@ -780,6 +810,7 @@ def apply_model_pricing(model_name: str, stats: ModelStats) -> bool:
(
stats.input_cost_usd,
stats.cached_input_cost_usd,
stats.cache_write_input_cost_usd,
stats.output_cost_usd,
stats.total_cost_usd,
stats.cache_savings_usd,
Expand All @@ -789,16 +820,18 @@ def apply_model_pricing(model_name: str, stats: ModelStats) -> bool:
(
stats.input_cost_usd,
stats.cached_input_cost_usd,
stats.cache_write_input_cost_usd,
stats.output_cost_usd,
stats.total_cost_usd,
stats.cache_savings_usd,
) = estimate_token_costs(stats.input_tokens, stats.cached_input_tokens, stats.output_tokens, pricing)
) = estimate_token_costs(stats.input_tokens, stats.cached_input_tokens, stats.cache_write_input_tokens, stats.output_tokens, pricing)
return True


def add_model_token_event(stats: ModelStats, event: TokenEvent) -> None:
stats.input_tokens += event.input_tokens
stats.cached_input_tokens += event.cached_input_tokens
stats.cache_write_input_tokens += event.cache_write_input_tokens
stats.output_tokens += event.output_tokens
stats.reasoning_output_tokens += event.reasoning_output_tokens
stats.total_tokens += event.total_tokens
Expand All @@ -807,14 +840,16 @@ def add_model_token_event(stats: ModelStats, event: TokenEvent) -> None:
stats.pricing_model = pricing_model
if pricing is None:
return
input_cost, cached_input_cost, output_cost, total_cost, cache_savings = estimate_token_costs(
input_cost, cached_input_cost, cache_write_input_cost, output_cost, total_cost, cache_savings = estimate_token_costs(
event.input_tokens,
event.cached_input_tokens,
event.cache_write_input_tokens,
event.output_tokens,
pricing,
)
stats.input_cost_usd += input_cost
stats.cached_input_cost_usd += cached_input_cost
stats.cache_write_input_cost_usd += cache_write_input_cost
stats.output_cost_usd += output_cost
stats.total_cost_usd += total_cost
stats.cache_savings_usd += cache_savings
Expand All @@ -836,14 +871,16 @@ def summarize_costs(
if pricing is None:
out.unpriced_tokens += event.total_tokens
continue
input_cost, cached_input_cost, output_cost, total_cost, cache_savings = estimate_token_costs(
input_cost, cached_input_cost, cache_write_input_cost, output_cost, total_cost, cache_savings = estimate_token_costs(
event.input_tokens,
event.cached_input_tokens,
event.cache_write_input_tokens,
event.output_tokens,
pricing,
)
out.input_cost_usd += input_cost
out.cached_input_cost_usd += cached_input_cost
out.cache_write_input_cost_usd += cache_write_input_cost
out.output_cost_usd += output_cost
out.total_cost_usd += total_cost
out.cache_savings_usd += cache_savings
Expand All @@ -860,9 +897,10 @@ def estimate_event_cost_total(event: TokenEvent) -> float:
_pricing_model, pricing = resolve_model_pricing(event.model)
if pricing is None:
return 0.0
_input_cost, _cached_input_cost, _output_cost, total_cost, _cache_savings = estimate_token_costs(
_input_cost, _cached_input_cost, _cache_write_input_cost, _output_cost, total_cost, _cache_savings = estimate_token_costs(
event.input_tokens,
event.cached_input_tokens,
event.cache_write_input_tokens,
event.output_tokens,
pricing,
)
Expand Down Expand Up @@ -1445,6 +1483,7 @@ def add_summary(text: str) -> None:
"30d API cost mix: "
f"in {format_usd(report.thirty_days_cost.input_cost_usd)}"
f" | cache {format_usd(report.thirty_days_cost.cached_input_cost_usd)}"
f" | cache write {format_usd(report.thirty_days_cost.cache_write_input_cost_usd)}"
f" | out {format_usd(report.thirty_days_cost.output_cost_usd)}",
"dim",
),
Expand Down
100 changes: 99 additions & 1 deletion tests/test_cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -182,7 +182,7 @@ def test_json_summary_has_expected_totals(self) -> None:
self.assertAlmostEqual(payload["costs"]["all_time"]["total_cost_usd"], 0.01103, places=9)
self.assertAlmostEqual(payload["costs"]["thirty_days"]["cache_savings_usd"], 0.0027, places=9)
self.assertEqual(payload["costs"]["coverage_pct"]["all_time"], 100.0)
self.assertEqual(payload["pricing"]["verified_at"], "2026-04-29")
self.assertEqual(payload["pricing"]["verified_at"], "2026-07-18")
self.assertEqual(payload["pricing"]["long_context_input_threshold"], 272000)
self.assertEqual(
payload["pricing"]["scope_note"],
Expand Down Expand Up @@ -267,6 +267,104 @@ def test_gpt_5_5_pricing_includes_cached_and_long_context_rates(self) -> None:
self.assertAlmostEqual(payload["costs"]["today"]["total_cost_usd"], 3.125, places=9)
self.assertEqual(payload["pricing"]["unpriced_models"], [])

def test_gpt_5_6_family_pricing_includes_cached_input_rates(self) -> None:
temp_dir = Path(tempfile.mkdtemp())
codex_home = temp_dir / ".codex"
sessions_dir = codex_home / "sessions" / "2026" / "07" / "18"
sessions_dir.mkdir(parents=True)
rows = [
{"timestamp": "2026-07-18T00:58:00Z", "type": "turn_context", "payload": {"model": "gpt-5.6-sol"}},
{
"timestamp": "2026-07-18T00:59:00Z",
"type": "event_msg",
"payload": {
"type": "token_count",
"info": {"total_token_usage": {
"input_tokens": 100_000,
"cached_input_tokens": 40_000,
"output_tokens": 10_000,
"total_tokens": 110_000,
}},
},
},
{"timestamp": "2026-07-18T01:00:00Z", "type": "turn_context", "payload": {"model": "gpt-5.6-terra"}},
{
"timestamp": "2026-07-18T01:01:00Z",
"type": "event_msg",
"payload": {
"type": "token_count",
"info": {"total_token_usage": {
"input_tokens": 200_000,
"cached_input_tokens": 80_000,
"output_tokens": 20_000,
"total_tokens": 220_000,
}},
},
},
{"timestamp": "2026-07-18T01:02:00Z", "type": "turn_context", "payload": {"model": "gpt-5.6-luna"}},
{
"timestamp": "2026-07-18T01:03:00Z",
"type": "event_msg",
"payload": {
"type": "token_count",
"info": {"total_token_usage": {
"input_tokens": 300_000,
"cached_input_tokens": 130_000,
"output_tokens": 30_000,
"total_tokens": 330_000,
}},
},
},
]
(sessions_dir / "session-gpt-56.jsonl").write_text(
"\n".join(json.dumps(row) for row in rows) + "\n",
encoding="utf-8",
)

result = self.run_cli("--json", "--codex-home", str(codex_home), "--now", "2026-07-18T10:00:00")
self.assertEqual(result.returncode, 0, result.stderr)
payload = json.loads(result.stdout)
models = {row["model"]: row for row in payload["models_30d"]}

self.assertAlmostEqual(models["gpt-5.6-sol"]["total_cost_usd"], 0.62, places=9)
self.assertAlmostEqual(models["gpt-5.6-terra"]["total_cost_usd"], 0.31, places=9)
self.assertAlmostEqual(models["gpt-5.6-luna"]["total_cost_usd"], 0.115, places=9)
self.assertEqual(payload["pricing"]["unpriced_models"], [])

def test_gpt_5_6_long_context_and_cache_write_pricing(self) -> None:
temp_dir = Path(tempfile.mkdtemp())
codex_home = temp_dir / ".codex"
sessions_dir = codex_home / "sessions" / "2026" / "07" / "18"
sessions_dir.mkdir(parents=True)
rows = [
{"timestamp": "2026-07-18T01:00:00Z", "type": "turn_context", "payload": {"model": "gpt-5.6-sol"}},
{
"timestamp": "2026-07-18T01:01:00Z",
"type": "event_msg",
"payload": {"type": "token_count", "info": {"total_token_usage": {
"input_tokens": 300_000,
"cached_input_tokens": 100_000,
"cache_write_input_tokens": 50_000,
"output_tokens": 100_000,
"total_tokens": 400_000,
}}},
},
]
(sessions_dir / "session-gpt-56-long.jsonl").write_text(
"\n".join(json.dumps(row) for row in rows) + "\n",
encoding="utf-8",
)

result = self.run_cli("--json", "--codex-home", str(codex_home), "--now", "2026-07-18T10:00:00")
self.assertEqual(result.returncode, 0, result.stderr)
model = json.loads(result.stdout)["models_30d"][0]

self.assertAlmostEqual(model["input_cost_usd"], 1.5, places=9)
self.assertAlmostEqual(model["cached_input_cost_usd"], 0.1, places=9)
self.assertAlmostEqual(model["cache_write_input_cost_usd"], 0.625, places=9)
self.assertAlmostEqual(model["output_cost_usd"], 4.5, places=9)
self.assertAlmostEqual(model["total_cost_usd"], 6.725, places=9)

def test_json_month_selector_expands_daily_rows_for_that_month(self) -> None:
codex_home = self.make_fixture()
result = self.run_cli("--json", "--codex-home", str(codex_home), "--now", "2026-03-19T10:00:00", "--month", "2026-03")
Expand Down
Loading