feat(grok): read opencode session usage for cost capture

Confirmed by inspecting a local opencode run: opencode persists per-session
usage in SQLite at ~/.local/share/opencode/opencode.db — the `session` table
carries cost + tokens_input/output/reasoning/cache_read/cache_write. xAI's
response usage object (prompt_tokens, completion_tokens,
prompt_tokens_details.cached_tokens, completion_tokens_details.reasoning_tokens)
maps directly onto those columns.

Add opencode_usage.read_session_usage / cost_for_session: read the opencode DB
and price the tokens via roboco.billing.pricing (our cost stays authoritative;
opencode's own `cost` column is kept for reference). Tested against a fixture DB
mirroring the real schema (single session, summed sessions, missing/empty DB).

Remaining wiring (for the live spawn): mount the opencode data dir on grok
spawn + call cost_for_session at reap to record the usage rollup.
This commit is contained in:
Renn F
2026-06-18 09:04:49 +02:00
parent 07f55117de
commit af6cad98d8
2 changed files with 227 additions and 0 deletions
+116
View File
@@ -0,0 +1,116 @@
"""Read token usage from an opencode SQLite store — Grok agent cost capture.
opencode (v1.x) persists per-session usage in a SQLite DB at
``~/.local/share/opencode/opencode.db`` (confirmed by inspecting a local run:
the ``session`` table carries ``cost`` and ``tokens_input`` / ``tokens_output``
/ ``tokens_reasoning`` / ``tokens_cache_read`` / ``tokens_cache_write``).
A Grok agent runs opencode, so its usage lands there rather than in a Claude
Code transcript. The orchestrator reads this at agent finalize and feeds the
token counts to :func:`roboco.billing.pricing.calculate_cost` — keeping our
pricing authoritative — while opencode's own ``cost`` column is kept for
reference. A per-agent container has a single opencode store, so summing all
session rows is correct without needing to map an opencode session id.
"""
from __future__ import annotations
import sqlite3
from dataclasses import dataclass
from pathlib import Path
from roboco.billing.pricing import calculate_cost
# Default location inside the agent container (HOME=/home/agent).
DEFAULT_DB_PATH = "/home/agent/.local/share/opencode/opencode.db"
# Column order here MUST match the unpacking in read_session_usage below.
_SELECT_ALL = (
"SELECT tokens_input, tokens_output, tokens_cache_read, "
"tokens_cache_write, tokens_reasoning, cost FROM session"
)
_SELECT_ONE = _SELECT_ALL + " WHERE id = ?"
@dataclass(frozen=True)
class OpencodeUsage:
"""Aggregated token usage read from an opencode store."""
tokens_input: int
tokens_output: int
tokens_cache_read: int
tokens_cache_write: int
tokens_reasoning: int
opencode_cost: float # opencode's own computed cost (reference only)
def read_session_usage(
db_path: str | Path = DEFAULT_DB_PATH,
session_id: str | None = None,
) -> OpencodeUsage | None:
"""Read aggregated usage from an opencode SQLite store.
Reads the ``session`` table — a specific row when ``session_id`` is given,
otherwise the sum across all sessions (one store per agent container).
Returns ``None`` if the DB is missing, the table absent, or there are no
rows — never raises, so callers don't need to guard finalize on it.
"""
path = Path(db_path)
if not path.exists():
return None
try:
con = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
try:
if session_id is not None:
cur = con.execute(_SELECT_ONE, (session_id,))
else:
cur = con.execute(_SELECT_ALL)
rows = cur.fetchall()
finally:
con.close()
except sqlite3.Error:
return None
if not rows:
return None
totals = [0, 0, 0, 0, 0]
cost = 0.0
for row in rows:
for i in range(5):
totals[i] += int(row[i] or 0)
cost += float(row[5] or 0.0)
return OpencodeUsage(
tokens_input=totals[0],
tokens_output=totals[1],
tokens_cache_read=totals[2],
tokens_cache_write=totals[3],
tokens_reasoning=totals[4],
opencode_cost=cost,
)
def cost_for_session(
model: str,
db_path: str | Path = DEFAULT_DB_PATH,
session_id: str | None = None,
) -> tuple[OpencodeUsage | None, float]:
"""Return (usage, roboco_cost_usd) for a Grok agent's opencode session.
``roboco_cost_usd`` is computed from our own pricing table so cost is
consistent with the Claude path. Returns ``(None, 0.0)`` when no usage is
recorded yet.
"""
usage = read_session_usage(db_path, session_id)
if usage is None:
return None, 0.0
cost = calculate_cost(
model,
tokens_input=usage.tokens_input,
tokens_output=usage.tokens_output,
tokens_cache_read=usage.tokens_cache_read,
tokens_cache_write=usage.tokens_cache_write,
)
return usage, cost
+111
View File
@@ -0,0 +1,111 @@
"""Tests for opencode usage capture (reading the opencode SQLite session table).
The fixture DB mirrors the real opencode v1.x ``session`` table columns observed
from a local run (cost + tokens_input/output/reasoning/cache_read/cache_write).
"""
from __future__ import annotations
import sqlite3
from typing import TYPE_CHECKING
from roboco.llm.providers.opencode_usage import (
cost_for_session,
read_session_usage,
)
if TYPE_CHECKING:
from pathlib import Path
_M = 1_000_000
_TOL = 1e-4
_ZERO_COST = 0.0
# Single-session fixture: input, output, reasoning, cache_read, cache_write.
_IN, _OUT, _REASON, _CREAD, _CWRITE = 100, 50, 10, 20, 5
# Second session for the summation test.
_S2_IN, _S2_OUT, _S2_CREAD = 200, 70, 10
# grok-build-0.1: 1M input ($1.00) + 1M output ($2.00) = $3.00.
_GROK_COST_1M_1M = 3.00
def _make_db(
path: Path, rows: list[tuple[str, int, int, int, int, int, float]]
) -> None:
con = sqlite3.connect(path)
con.execute(
"""
CREATE TABLE session (
id text PRIMARY KEY,
tokens_input integer DEFAULT 0 NOT NULL,
tokens_output integer DEFAULT 0 NOT NULL,
tokens_reasoning integer DEFAULT 0 NOT NULL,
tokens_cache_read integer DEFAULT 0 NOT NULL,
tokens_cache_write integer DEFAULT 0 NOT NULL,
cost real DEFAULT 0 NOT NULL
)
"""
)
con.executemany(
"INSERT INTO session "
"(id, tokens_input, tokens_output, tokens_reasoning, "
"tokens_cache_read, tokens_cache_write, cost) "
"VALUES (?, ?, ?, ?, ?, ?, ?)",
rows,
)
con.commit()
con.close()
def test_read_missing_db_returns_none(tmp_path: Path) -> None:
assert read_session_usage(tmp_path / "nope.db") is None
def test_read_single_session(tmp_path: Path) -> None:
db = tmp_path / "opencode.db"
# (id, input, output, reasoning, cache_read, cache_write, cost)
_make_db(db, [("s1", _IN, _OUT, _REASON, _CREAD, _CWRITE, 0.0007)])
usage = read_session_usage(db, session_id="s1")
assert usage is not None
assert usage.tokens_input == _IN
assert usage.tokens_output == _OUT
assert usage.tokens_cache_read == _CREAD
assert usage.tokens_cache_write == _CWRITE
assert usage.tokens_reasoning == _REASON
def test_read_sums_all_sessions_when_no_id(tmp_path: Path) -> None:
db = tmp_path / "opencode.db"
_make_db(
db,
[
("s1", _IN, _OUT, 0, 0, 0, 0.0),
("s2", _S2_IN, _S2_OUT, 0, _S2_CREAD, 0, 0.0),
],
)
usage = read_session_usage(db)
assert usage is not None
assert usage.tokens_input == _IN + _S2_IN
assert usage.tokens_output == _OUT + _S2_OUT
assert usage.tokens_cache_read == _S2_CREAD
def test_read_empty_table_returns_none(tmp_path: Path) -> None:
db = tmp_path / "opencode.db"
_make_db(db, [])
assert read_session_usage(db) is None
def test_cost_for_session_uses_roboco_pricing(tmp_path: Path) -> None:
db = tmp_path / "opencode.db"
# 1M input + 1M output for grok-build-0.1 → our $3.00, not opencode's 99.0.
_make_db(db, [("s1", _M, _M, 0, 0, 0, 99.0)])
usage, cost = cost_for_session("grok-build-0.1", db, session_id="s1")
assert usage is not None
assert abs(cost - _GROK_COST_1M_1M) < _TOL
def test_cost_for_session_missing_db(tmp_path: Path) -> None:
usage, cost = cost_for_session("grok-build-0.1", tmp_path / "nope.db")
assert usage is None
assert cost == _ZERO_COST