tokentab local token cost ledger

15@dataclass
16class TokenCounts:
17    input: int = 0
18    output: int = 0
19    cache_read: int = 0   # tokens read back from the prompt cache (cheap)
20    cache_write: int = 0  # tokens written into the cache (a bit pricier than input)
21
22    def total(self) -> int:
23        return self.input + self.output + self.cache_read + self.cache_write
24
25
26@dataclass
27class UsageRecord:
28    provider: str          # "claude", "codex", "gemini", "cursor"
29    model: str             # model name as the tool reported it
30    project: str           # project / working directory (best effort)
31    timestamp: datetime    # when the call happened
32    activity: str          # rough guess at what the turn was doing
33    tokens: TokenCounts
34    id: str                # stable-ish id so we can drop duplicates
35
36
37@dataclass
38class PricedRecord(UsageRecord):
39    cost: float = 0.0
40
41
42@dataclass
43class ModelPrice:
44    """A model's price, in US dollars per 1 million tokens."""
45    input: float
46    output: float
47    cache_read: float
48    cache_write: float

14# each entry: (name, label, collect_callable)
15ALL_PROVIDERS = [
16    (claude.NAME, claude.LABEL, claude.collect),
17    (codex.NAME, codex.LABEL, codex.collect),
18    (gemini.NAME, gemini.LABEL, gemini.collect),
19    (cursor.NAME, cursor.LABEL, cursor.collect),
20]
21
22LABELS = {name: label for name, label, _ in ALL_PROVIDERS}
23
24
25def label_for(name: str) -> str:
26    return LABELS.get(name, name)
27
28
29def collect_all(only: str | None = None) -> list[PricedRecord]:
30    """Run every provider (or a chosen one), flatten to a single list, and attach
31    a dollar cost to each. Providers that aren't installed just return [], so this
32    quietly returns whatever you actually use.
33    """
34    providers = ALL_PROVIDERS
35    if only:
36        providers = [p for p in ALL_PROVIDERS if p[0] == only.lower()]
37
38    priced: list[PricedRecord] = []
39    for name, label, collect in providers:
40        try:
41            records = collect()
42        except Exception as err:  # one broken tool shouldn't sink the whole report
43            print(f"warning: couldn't read {label}: {err}", file=sys.stderr)
44            continue
45        for r in records:
46            priced.append(
47                PricedRecord(
48                    provider=r.provider,
49                    model=r.model,
50                    project=r.project,
51                    timestamp=r.timestamp,
52                    activity=r.activity,
53                    tokens=r.tokens,
54                    id=r.id,
55                    cost=cost_of(r.tokens, r.model),
56                )
57            )
58
59    return priced

45def collect() -> list[UsageRecord]:
46    root = _config_dir() / "projects"
47    if not root.is_dir():
48        return []
49
50    records: list[UsageRecord] = []
51    seen: set[str] = set()
52
53    for file in walk_files(root, lambda n: n.endswith(".jsonl")):
54        project = pretty_project(_derive_project(file))
55
56        try:
57            text = file.read_text(encoding="utf-8")
58        except OSError:
59            continue
60
61        lines = text.split("\n")
62
63        # first pass: first user message (for activity) + tools used this session
64        first_user_text = ""
65        tool_names: list[str] = []
66        for line in lines:
67            if not line.strip():
68                continue
69            obj = try_json(line)
70            msg = obj.get("message") if isinstance(obj, dict) else None
71            if not msg:
72                continue
73            if msg.get("role") == "user" and not first_user_text:
74                first_user_text = _extract_text(msg.get("content"))
75            for block in msg.get("content") or []:
76                if block.get("type") == "tool_use" and block.get("name"):
77                    tool_names.append(block["name"])
78        activity = classify(tool_names, first_user_text)
79
80        # second pass: token usage from assistant turns
81        for line in lines:
82            if not line.strip():
83                continue
84            obj = try_json(line)
85            msg = obj.get("message") if isinstance(obj, dict) else None
86            usage = msg.get("usage") if msg else None
87            if not usage:
88                continue
89
90            rec_id = msg.get("id") or f"{file}:{len(records)}"
91            if rec_id in seen:
92                continue
93            seen.add(rec_id)
94
95            records.append(
96                UsageRecord(
97                    provider=NAME,
98                    model=msg.get("model") or "unknown",
99                    project=project,
100                    timestamp=to_date(obj.get("timestamp")) or EPOCH,
101                    activity=activity,
102                    id=rec_id,
103                    tokens=TokenCounts(
104                        input=usage.get("input_tokens", 0),
105                        output=usage.get("output_tokens", 0),
106                        cache_read=usage.get("cache_read_input_tokens", 0),
107                        cache_write=usage.get("cache_creation_input_tokens", 0),
108                    ),
109                )
110            )

38def collect() -> list[UsageRecord]:
39    roots = [_config_dir() / "sessions", _config_dir() / "archived_sessions"]
40    records: list[UsageRecord] = []
41
42    for root in roots:
43        if not root.is_dir():
44            continue
45
46        for file in walk_files(root, lambda n: n.startswith("rollout-") and n.endswith(".jsonl")):
47            try:
48                text = file.read_text(encoding="utf-8")
49            except OSError:
50                continue
51
52            lines = text.split("\n")
53            project = "unknown"
54            model = "unknown"
55            first_user_text = ""
56            tool_names: list[str] = []
57
58            # context pass
59            for line in lines:
60                if not line.strip():
61                    continue
62                ev = try_json(line)
63                if not isinstance(ev, dict):
64                    continue
65                if ev.get("cwd"):
66                    project = pretty_project(ev["cwd"])
67                if ev.get("model"):
68                    model = ev["model"]
69                payload = ev.get("payload") or {}
70                if ev.get("type") == "function_call" or payload.get("type") == "function_call":
71                    name = payload.get("name") or (ev.get("info") or {}).get("name")
72                    if name:
73                        tool_names.append(str(name))
74                if (ev.get("type") == "message" or payload.get("role") == "user") and not first_user_text:
75                    content = payload.get("content") or (ev.get("info") or {}).get("content")
76                    if isinstance(content, str):
77                        first_user_text = content[:500]
78
79            activity = classify(tool_names, first_user_text)
80
81            # usage pass — cumulative totals diffed into per-call deltas
82            prev_in = prev_out = prev_cache = 0
83            i = 0
84            for line in lines:
85                if not line.strip():
86                    continue
87                ev = try_json(line)
88                if not isinstance(ev, dict):
89                    continue
90                tc = _find_token_count(ev)
91                if not tc:
92                    continue
93
94                tot_in = tc.get("input_tokens", tc.get("input", 0)) or 0
95                tot_out = tc.get("output_tokens", tc.get("output", 0)) or 0
96                tot_cache = tc.get("cached_input_tokens", tc.get("cache_read", 0)) or 0
97
98                d_in = max(0, tot_in - prev_in)
99                d_out = max(0, tot_out - prev_out)
100                d_cache = max(0, tot_cache - prev_cache)
101                prev_in, prev_out, prev_cache = tot_in, tot_out, tot_cache
102
103                if d_in + d_out + d_cache == 0:
104                    continue
105
106                records.append(
107                    UsageRecord(
108                        provider=NAME,
109                        model=model,
110                        project=project,
111                        timestamp=to_date(ev.get("timestamp")) or EPOCH,
112                        activity=activity,
113                        id=f"{file}:{i}",
114                        tokens=TokenCounts(input=d_in, output=d_out, cache_read=d_cache, cache_write=0),
115                    )
116                )
117                i += 1

30def collect() -> list[UsageRecord]:
31    root = home(".gemini", "tmp")
32    if not root.is_dir():
33        return []
34
35    records: list[UsageRecord] = []
36
37    for file in walk_files(root, lambda n: n.startswith("session-") and n.endswith(".json")):
38        try:
39            raw = file.read_text(encoding="utf-8")
40        except OSError:
41            continue
42        doc = try_json(raw)
43        messages = doc.get("messages") if isinstance(doc, dict) else None
44        if not messages:
45            continue
46
47        project = pretty_project(_derive_project(file))
48
49        first_user_text = ""
50        tool_names: list[str] = []
51        for m in messages:
52            if m.get("role") == "user" and not first_user_text:
53                first_user_text = (m.get("text") or "")[:500]
54            for p in m.get("parts") or []:
55                fc = p.get("functionCall")
56                if fc and fc.get("name"):
57                    tool_names.append(fc["name"])
58        activity = classify(tool_names, first_user_text)
59
60        for i, m in enumerate(messages):
61            t = m.get("tokens")
62            if not t:
63                continue
64            cached = t.get("cached", 0) or 0
65            raw_input = t.get("input", 0) or 0
66            fresh_input = max(0, raw_input - cached)  # input is inclusive of cache
67            output = t.get("output", 0) or 0
68
69            if fresh_input + output + cached == 0:
70                continue
71
72            records.append(
73                UsageRecord(
74                    provider=NAME,
75                    model=m.get("model") or "gemini-2.5-pro",
76                    project=project,
77                    timestamp=to_date(m.get("timestamp")) or EPOCH,
78                    activity=activity,
79                    id=f"{file}:{i}",
80                    tokens=TokenCounts(input=fresh_input, output=output, cache_read=cached, cache_write=0),
81                )
82            )
83

16TABLE: dict[str, ModelPrice] = {
17    # --- Anthropic (Claude) ---
18    "claude-opus-4-6":   ModelPrice(15, 75, 1.5, 18.75),
19    "claude-opus-4-5":   ModelPrice(15, 75, 1.5, 18.75),
20    "claude-opus-4-1":   ModelPrice(15, 75, 1.5, 18.75),
21    "claude-opus-4":     ModelPrice(15, 75, 1.5, 18.75),
22    "claude-sonnet-4-6": ModelPrice(3, 15, 0.3, 3.75),
23    "claude-sonnet-4-5": ModelPrice(3, 15, 0.3, 3.75),
24    "claude-sonnet-4":   ModelPrice(3, 15, 0.3, 3.75),
25    "claude-haiku-4-5":  ModelPrice(1, 5, 0.1, 1.25),
26    "claude-3-5-haiku":  ModelPrice(0.8, 4, 0.08, 1),
27
28    # --- OpenAI (Codex / GPT) ---
29    "gpt-5.6":      ModelPrice(1.25, 10, 0.125, 0),
30    "gpt-5.1":      ModelPrice(1.25, 10, 0.125, 0),
31    "gpt-5":        ModelPrice(1.25, 10, 0.125, 0),
32    "gpt-5-mini":   ModelPrice(0.25, 2, 0.025, 0),
33    "gpt-5-codex":  ModelPrice(1.25, 10, 0.125, 0),
34    "gpt-4o":       ModelPrice(2.5, 10, 1.25, 0),
35    "gpt-4o-mini":  ModelPrice(0.15, 0.6, 0.075, 0),
36    "o3":           ModelPrice(2, 8, 0.5, 0),
37    "o4-mini":      ModelPrice(1.1, 4.4, 0.275, 0),
38
39    # --- Google (Gemini) ---
40    "gemini-3.5-pro":   ModelPrice(2.5, 15, 0.25, 0),
41    "gemini-3.5-flash": ModelPrice(0.3, 2.5, 0.03, 0),
42    "gemini-2.5-pro":   ModelPrice(1.25, 10, 0.125, 0),
43    "gemini-2.5-flash": ModelPrice(0.3, 2.5, 0.03, 0),
44}
45
46# When a log says "claude-opus-4-6-20260514" or "anthropic/claude-sonnet-4-6" we
47# won't find an exact key, so we try progressively looser matches. Longest keys
48# first, so "claude-opus-4-6" wins over "claude-opus-4".
49_KEYS_BY_LENGTH = sorted(TABLE.keys(), key=len, reverse=True)
50
51_unknown_models: set[str] = set()
52
53
54def price_for(model: str) -> ModelPrice | None:
55    if not model:
56        return None
57    m = model.lower()
58
59    if m in TABLE:
60        return TABLE[m]
61
62    # strip a leading vendor prefix like "anthropic/" or "openai/"
63    bare = m.split("/")[-1] if "/" in m else m
64    if bare in TABLE:
65        return TABLE[bare]
66
67    # substring match — handles date-stamped and suffixed names
68    for key in _KEYS_BY_LENGTH:
69        if key in bare:
70            return TABLE[key]
71
72    _unknown_models.add(model)
73    return None
74
75
76def unpriced_models() -> list[str]:
77    """Models we saw but couldn't price. The CLI warns so $0 rows aren't a silent lie."""
78    return sorted(_unknown_models)
79
80
81def cost_of(tokens: TokenCounts, model: str) -> float:
82    """Cost in dollars for one set of token counts on a given model."""
83    p = price_for(model)
84    if p is None:
85        return 0.0
86    return (
87        tokens.input * p.input
88        + tokens.output * p.output
89        + tokens.cache_read * p.cache_read
90        + tokens.cache_write * p.cache_write

102def build_report(
103    records: list[PricedRecord],
104    rng: DateRange,
105    project: str | None = None,
106    exclude: list[str] | None = None,
107) -> Report:
108    exclude = exclude or []
109
110    def keep(r: PricedRecord) -> bool:
111        ts = r.timestamp
112        if ts.tzinfo is None:
113            ts = ts.replace(tzinfo=timezone.utc)
114        if ts < rng.frm or ts > rng.to:
115            return False
116        if project and project.lower() not in r.project.lower():
117            return False
118        if any(x.lower() in r.project.lower() for x in exclude):
119            return False
120        return True
121
122    in_range = [r for r in records if keep(r)]
123
124    totals = Totals(calls=len(in_range))
125    tools: dict[str, Row] = {}
126    models: dict[str, Row] = {}
127    projects: dict[str, Row] = {}
128    activities: dict[str, Row] = {}
129    days: dict[str, DayRow] = {}
130
131    def bump(m: dict[str, Row], key: str, r: PricedRecord) -> None:
132        row = m.get(key)
133        if row is None:
134            row = Row(label=key)
135            m[key] = row
136        row.cost += r.cost
137        row.tokens += r.tokens.total()
138        row.calls += 1
139
140    for r in in_range:
141        totals.cost += r.cost
142        totals.tokens.input += r.tokens.input
143        totals.tokens.output += r.tokens.output
144        totals.tokens.cache_read += r.tokens.cache_read
145        totals.tokens.cache_write += r.tokens.cache_write
146
147        bump(tools, r.provider, r)
148        bump(models, r.model, r)
149        bump(projects, r.project, r)
150        bump(activities, r.activity, r)
151
152        ts = r.timestamp if r.timestamp.tzinfo else r.timestamp.replace(tzinfo=timezone.utc)
153        key = _day_key(ts)
154        d = days.get(key)
155        if d is None:
156            d = DayRow(date=key)
157            days[key] = d
158        d.cost += r.cost
159        d.tokens += r.tokens.total()
160        d.calls += 1
161
162    input_side = totals.tokens.input + totals.tokens.cache_read
163    totals.cache_hit_rate = (totals.tokens.cache_read / input_side) if input_side > 0 else 0.0
164
165    by_cost = lambda rows: sorted(rows, key=lambda x: x.cost, reverse=True)
166
167    return Report(
168        range=rng,
169        totals=totals,
170        by_tool=by_cost(tools.values()),
171        by_model=by_cost(models.values()),
172        by_project=by_cost(projects.values()),
173        by_activity=by_cost(activities.values()),
174        by_day=sorted(days.values(), key=lambda d: d.date),
175    )

30def render_report(report: Report, console: Console) -> None:
31    totals = report.totals
32
33    if report.range.frm.timestamp() == 0:
34        range_label = "all time"
35    else:
36        range_label = f"{_fmt_day(report.range.frm)} → {_fmt_day(report.range.to)}"
37
38    console.print()
39    console.print(Text("  tokentab  ", style="bold white") + Text(range_label, style="grey58"))
40    console.print()
41
42    # ---- headline totals ----
43    console.print(Text("Totals", style="bold"))
44    parts = Text("  ")
45    parts.append("cost ", style="grey58")
46    parts.append(money(totals.cost) + "   ·   ", style="green")
47    parts.append("tokens ", style="grey58")
48    parts.append(compact_tokens(totals.tokens.total()) + "   ·   ")
49    parts.append("calls ", style="grey58")
50    parts.append(f"{totals.calls:,}" + "   ·   ")
51    parts.append("cache hit ", style="grey58")
52    parts.append(pct(totals.cache_hit_rate))
53    console.print(parts)
54    console.print()
55
56    if totals.calls == 0:
57        console.print(Text("  No usage found for this window.", style="yellow"))
58        console.print(
59            Text(
60                "  Either nothing ran in this period, or none of the supported tools have logs on disk yet.",
61                style="grey58",
62            )
63        )
64        console.print()
65        return
66
67    # ---- by tool ----
68    console.print(Text("By tool", style="bold"))
69    t = Table(box=SIMPLE_HEAVY, show_edge=True, pad_edge=False)
70    t.add_column("Tool")
71    t.add_column("Cost", justify="right", style="green")
72    t.add_column("Tokens", justify="right")
73    t.add_column("Share", justify="right")
74    for r in report.by_tool:
75        t.add_row(label_for(r.label), money(r.cost), compact_tokens(r.tokens), share(r.cost, totals.cost))
76    console.print(t)
77
78    # ---- top models ----
79    console.print(Text("Top models", style="bold"))
80    t = Table(box=SIMPLE_HEAVY, pad_edge=False)
81    t.add_column("Model")
82    t.add_column("Cost", justify="right", style="green")
83    t.add_column("Tokens", justify="right")
84    t.add_column("Calls", justify="right")
85    for r in report.by_model[:8]:
86        t.add_row(r.label, money(r.cost), compact_tokens(r.tokens), str(r.calls))
87    console.print(t)
88
89    # ---- top projects ----
90    if len(report.by_project) > 1:
91        console.print(Text("Top projects", style="bold"))
92        t = Table(box=SIMPLE_HEAVY, pad_edge=False)
93        t.add_column("Project")
94        t.add_column("Cost", justify="right", style="green")
95        t.add_column("Tokens", justify="right")
96        for r in report.by_project[:8]:
97            t.add_row(r.label, money(r.cost), compact_tokens(r.tokens))
98        console.print(t)
99
100    # ---- by activity ----
101    console.print(Text("By activity", style="bold"))
102    t = Table(box=SIMPLE_HEAVY, pad_edge=False)
103    t.add_column("Activity")
104    t.add_column("Cost", justify="right", style="green")
105    t.add_column("Share", justify="right")
106    for r in report.by_activity:
107        t.add_row(r.label, money(r.cost), share(r.cost, totals.cost))
108    console.print(t)
109
110    # ---- per day ----
111    if len(report.by_day) > 1:
112        console.print(Text("Per day", style="bold"))
113        max_cost = max((d.cost for d in report.by_day), default=0.0)
114        for d in report.by_day[-14:]:
115            line = Text("  ")
116            line.append(d.date + "  ", style="grey58")
117            line.append(_bar(d.cost, max_cost, 24))
118            line.append("  ")
119            line.append(money(d.cost), style="green")
120            console.print(line)
121        console.print()

24def _make_handler(default_period: str):
25    class Handler(BaseHTTPRequestHandler):
26        def log_message(self, *_args):
27            pass  # quiet — we don't want a request log spewing in the terminal
28
29        def do_GET(self):
30            parsed = urlparse(self.path)
31
32            try:
33                if parsed.path == "/api/report":
34                    qs = parse_qs(parsed.query)
35                    period = (qs.get("period") or [default_period])[0]
36                    records = collect_all()
37                    report = build_report(records, parse_range(period))
38                    body = json.dumps(report_to_dict(report)).encode("utf-8")
39                    self._send(200, "application/json", body)
40                    return
41
42                if parsed.path == "/":
43                    self._send(200, "text/html; charset=utf-8", render_page().encode("utf-8"))
44                    return
45
46                self._send(404, "text/plain", b"not found")
47            except Exception as err:  # never let one bad request kill the server
48                self._send(500, "text/plain", f"error: {err}".encode("utf-8"))
49
50        def _send(self, code: int, ctype: str, body: bytes):
51            self.send_response(code)
52            self.send_header("Content-Type", ctype)
53            self.send_header("Content-Length", str(len(body)))
54            self.end_headers()
55            self.wfile.write(body)
56
57    return Handler
58
59
60def start_web(port: int, open_browser: bool, period: str) -> None:
61    handler = _make_handler(period)
62
63    # port 0 (or a taken port) -> let the OS pick a free one
64    try:
65        server = ThreadingHTTPServer(("127.0.0.1", port), handler)
66    except OSError:
67        server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
68
69    actual_port = server.server_address[1]
70    link = f"http://localhost:{actual_port}"
71    print(f"tokentab is running at {link}")
72    print("press ctrl-c to stop")
73
74    if open_browser:
75        threading.Timer(0.4, lambda: _safe_open(link)).start()
76
77    try:
78        server.serve_forever()
79    except KeyboardInterrupt:
80        print("\nstopped")
81        server.shutdown()
82
83
84def _safe_open(link: str) -> None:
85    try:
86        webbrowser.open(link)
87    except Exception:
88        pass  # opening a browser is a nicety, not a requirement

1from __future__ import annotations
2
3import setup
4import subprocess
5import random 
6import string
7import argparse 
8import sys 
9from pathlib import Path
10
11
12setup.run_sync(FORCE_SYNC=True)
13
14def _banner() -> Panel:
15    return Panel.fit(
16        f"[bold cyan]text-humanizer[/bold cyan] [dim]v{__version__}[/dim]\n"
17        "[dim]Deepseek — type /help for commands, /exit to quit[/dim]",
18        border_style="cyan",
19    )
20
21
22HELP_TEXT = """\
23Available commands:
24  /help           Show this help
25  /reset          Clear the conversation history
26  /history        Show number of messages in the current session
27  /tools          List tools available to the agent
28  /exit, /quit    Exit the CLI
29Anything else is sent to the agent.
30"""
31
32
33def _repl(agent: Agent) -> None:
34    console.print(_banner())
35    while True:
36        try:
37            user_in = Prompt.ask("[bold green]you[/bold green]")
38        except (EOFError, KeyboardInterrupt):
39            console.print("\n[dim]bye[/dim]")
40            return
41
42        if not user_in.strip():
43            continue
44
45        if user_in.startswith("/"):
46            cmd = user_in.strip().lower()
47            if cmd in ("/exit", "/quit"):
48                console.print("[dim]bye[/dim]")
49                return
50            if cmd == "/help":
51                console.print(HELP_TEXT)
52                continue
53            if cmd == "/reset":
54                agent.reset()
55                console.print("[dim]history cleared[/dim]")
56                continue
57            if cmd == "/history":
58                console.print(f"[dim]{len(agent.history)} messages[/dim]")
59                continue
60            if cmd == "/tools":
61                for t in agent.tools:
62                    console.print(f"  [cyan]{t.name}[/cyan] — {t.description}")
63                continue
64            console.print(f"[yellow]unknown command: {cmd}[/yellow]")
65            continue
66
67        try:
68            reply = agent.send(user_in)
69        except Exception as exc:  # noqa: BLE001
70            console.print(f"[red]error:[/red] {exc}")
71            continue
72
73        console.print(Panel(Markdown(reply or "_(no text)_"), border_style="magenta", title="claude"))
74
75def main(argv: list[str] | None = None) -> int:
76    parser = argparse.ArgumentParser(prog="claude-engineer", description="Interactive Claude Opus 4.7 coding agent.")
77    parser.add_argument("--model", help="Override the model (default: claude-opus-4-7)")
78    parser.add_argument("--env", default=".env", help="Path to .env file (default: .env)")
79    parser.add_argument("--verbose", action="store_true", help="Print tool calls as they happen")
80    parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}")
81    args = parser.parse_args(argv)
82
83    try:
84        cfg = Config.load(env_file=args.env)
85    except RuntimeError as exc:
86        print(f"error: {exc}", file=sys.stderr)
87        return 2
88
89    if args.model:
90        cfg.model = args.model
91    if args.verbose:
92        cfg.verbose = True
93
94    agent = Agent(config=cfg)
95    _repl(agent)
96    return 0
97
98
99if __name__ == "__main__":
100    raise SystemExit(main())