diff --git a/CLAUDE.md b/CLAUDE.md index 2a7bbb7..e2c00d2 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -10,7 +10,7 @@ Inspired by [phuryn/claude-usage](https://github.com/phuryn/claude-usage) but di ## Status -Working codebase. 68 Python unit tests (`python3 -m unittest discover tests`). Seven UI tabs wired up (Overview, Prompts, Sessions, Projects, Skills, Tips, Settings). Runs on macOS, Windows, and Linux. +Working codebase. 75 Python unit tests (`python3 -m unittest discover tests`). Seven UI tabs wired up (Overview, Prompts, Sessions, Projects, Skills, Tips, Settings). Runs on macOS, Windows, and Linux. ## Architecture @@ -36,7 +36,7 @@ Env vars: `PORT` (default 8080), `HOST` (default 127.0.0.1), `CLAUDE_PROJECTS_DI ## Known limitations -See `docs/KNOWN_LIMITATIONS.md`. Current summary: Skills `tokens_per_call` is populated only for skills installed under the three scanned roots (`~/.claude/skills/`, `~/.claude/scheduled-tasks/`, `~/.claude/plugins/`); project-local skills and subagent-dispatched skills show invocation counts but blank token counts. +See `docs/KNOWN_LIMITATIONS.md`. Current summary: Skills `tokens_per_call` covers the three global roots (`~/.claude/skills/`, `~/.claude/scheduled-tasks/`, `~/.claude/plugins/`) plus project-local `.claude/skills/` directories discovered from cwds in the messages table. Only `Task`-dispatched subagent skills still show blank token counts. ## Verifying changes diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 63323ca..c206e46 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -41,7 +41,6 @@ Component layout: `cli.py` (entry points) → `token_dashboard/scanner.py` (JSON ## Ideas that would genuinely help -- Broadening the Skills catalog scan to cover project-local `.claude/skills/` directories (closes the known limitation). - A CSV or JSON export of any route. - A session-filter UI (currently everything is all-time or implicit-"recent"). - A GitHub Actions workflow that runs the tests on push. diff --git a/cli.py b/cli.py index 87ed122..5f3281c 100644 --- a/cli.py +++ b/cli.py @@ -8,7 +8,7 @@ from pathlib import Path from token_dashboard.db import init_db, default_db_path, overview_totals -from token_dashboard.scanner import scan_dir +from token_dashboard.scanner import rescan_agent_targets, rescan_slash_commands, scan_dir from token_dashboard.tips import all_tips @@ -38,6 +38,26 @@ def cmd_scan(args): print(f"Token Dashboard: scanned {n['files']} files, {n['messages']} messages, {n['tools']} tool calls") +def cmd_rescan_agent_targets(args): + db = _db_path(args) + init_db(db) + n = rescan_agent_targets(db, _projects(args)) + print( + f"Token Dashboard: reset {n['files_reset']} files, " + f"re-parsed {n['messages']} messages, {n['tools']} tool calls" + ) + + +def cmd_rescan_slash_commands(args): + db = _db_path(args) + init_db(db) + n = rescan_slash_commands(db) + print( + f"Token Dashboard: synthesized {n['slash_commands_synthesized']} " + f"Skill rows from historical slash-command messages" + ) + + def cmd_today(args): db = _db_path(args) init_db(db) @@ -94,6 +114,8 @@ def main(): p = argparse.ArgumentParser(prog="token-dashboard", description="Local Claude Code usage dashboard", parents=[common]) sub = p.add_subparsers(dest="cmd", required=True) sub.add_parser("scan", parents=[common]).set_defaults(func=cmd_scan) + sub.add_parser("rescan-agent-targets", parents=[common]).set_defaults(func=cmd_rescan_agent_targets) + sub.add_parser("rescan-slash-commands", parents=[common]).set_defaults(func=cmd_rescan_slash_commands) sub.add_parser("today", parents=[common]).set_defaults(func=cmd_today) sub.add_parser("stats", parents=[common]).set_defaults(func=cmd_stats) sub.add_parser("tips", parents=[common]).set_defaults(func=cmd_tips) diff --git a/docs/KNOWN_LIMITATIONS.md b/docs/KNOWN_LIMITATIONS.md index 4c5939a..2d71fb0 100644 --- a/docs/KNOWN_LIMITATIONS.md +++ b/docs/KNOWN_LIMITATIONS.md @@ -2,11 +2,11 @@ None of these are blockers — the dashboard still gives you useful information. They're the rough edges you'll notice if you look hard. -## Skills token counts are partial +## Skills tokens-per-call is blank when a skill runs only through Task/Agent -The Skills route shows every skill Claude Code invoked, how many times, across how many sessions, and when. The **tokens-per-call** column is populated only for skills whose `SKILL.md` lives under `~/.claude/skills/`, `~/.claude/scheduled-tasks/`, or `~/.claude/plugins/`. Skills registered elsewhere (project-local `.claude/skills/`, or invocations that go through the `Task` tool with a skill-shaped `subagent_type`) show invocation counts but leave the token column blank. +The Skills route shows every skill Claude Code invoked, how many times, across how many sessions, and when. The **tokens-per-call** column is populated for every skill whose `SKILL.md` lives under `~/.claude/skills/`, `~/.claude/scheduled-tasks/`, `~/.claude/plugins/`, or a project-local `.claude/skills/` directory discovered from the cwds in your session history. A skill that runs only through the `Task`/`Agent` tool with a skill-shaped `subagent_type` (never as a direct `Skill` invocation) arrives without a resolvable slug on disk and its tokens-per-call stays blank. -It's still a useful view — you can see which skills dominate your session time — just don't expect a complete per-skill token cost. PRs to broaden the catalog scan welcome. +Cost attribution for orchestrator skills — any skill that dispatches subagents via `Task`/`Agent` — follows the `parent_uuid` chain from every dispatch back to the skill call that emitted it. The `total inc. subagents` column on the Skills tab reflects that. If you upgraded from an older build and the column looks low, run `python3 cli.py rescan-agent-targets` once to re-parse main-session JSONLs whose Agent rows lost their `subagent_type` target. ## Cost for Pro / Max / Max-20x users is shown as API-equivalent, not subscription value diff --git a/tests/test_scanner_parse.py b/tests/test_scanner_parse.py index 8427afd..db54f75 100644 --- a/tests/test_scanner_parse.py +++ b/tests/test_scanner_parse.py @@ -42,6 +42,27 @@ def test_extracts_tool_uses(self): self.assertEqual(parsed[0]["name"], "Read") self.assertEqual(parsed[1]["target"], "npm run lint") + def test_agent_and_task_both_populate_target(self): + """Claude Code renamed Task → Agent; both must resolve subagent_type as target.""" + rec = { + "type": "assistant", "uuid": "u", "sessionId": "s", "timestamp": "t", + "message": { + "model": "claude-opus-4-7", + "usage": {"input_tokens": 1, "output_tokens": 1}, + "content": [ + {"type": "tool_use", "id": "t1", "name": "Agent", + "input": {"subagent_type": "software-architect", "description": "x"}}, + {"type": "tool_use", "id": "t2", "name": "Task", + "input": {"subagent_type": "researcher", "description": "y"}}, + ], + }, + } + _, tools = parse_record(rec, project_slug="p") + self.assertEqual(len(tools), 2) + by_name = {t["tool_name"]: t for t in tools} + self.assertEqual(by_name["Agent"]["target"], "software-architect") + self.assertEqual(by_name["Task"]["target"], "researcher") + class SidechainTests(unittest.TestCase): def test_is_sidechain_flag_propagates(self): @@ -69,5 +90,79 @@ def test_tool_result_estimates_tokens(self): self.assertAlmostEqual(tools[0]["result_tokens"], 1000, delta=10) +class SlashCommandExtractionTests(unittest.TestCase): + """User-typed slash commands (`/foo`) must synthesize a Skill tool_call. + + Claude Code logs them as a user-role record whose content is a string + containing `/`. Two observed orderings + — `` first or `` first — must both match. + """ + + def _user_record(self, content): + return { + "type": "user", + "uuid": "u-cmd", + "sessionId": "s1", + "timestamp": "2026-04-24T07:12:56Z", + "isSidechain": False, + "message": {"role": "user", "content": content}, + } + + def test_slash_command_name_first(self): + rec = self._user_record( + "/demo-cmd\n" + "demo-cmd\n" + "" + ) + _, tools = parse_record(rec, project_slug="p") + self.assertEqual(len(tools), 1) + self.assertEqual(tools[0]["tool_name"], "Skill") + self.assertEqual(tools[0]["target"], "demo-cmd") + self.assertEqual(tools[0]["timestamp"], "2026-04-24T07:12:56Z") + + def test_slash_command_message_first(self): + rec = self._user_record( + "demo-cmd\n" + "/demo-cmd" + ) + _, tools = parse_record(rec, project_slug="p") + self.assertEqual(len(tools), 1) + self.assertEqual(tools[0]["target"], "demo-cmd") + + def test_plugin_namespaced_slug_preserves_colon(self): + rec = self._user_record("/codex:review") + _, tools = parse_record(rec, project_slug="p") + self.assertEqual(tools[0]["target"], "codex:review") + + def test_list_content_with_text_blocks(self): + rec = self._user_record([ + {"type": "text", "text": "/demo-skill"}, + ]) + _, tools = parse_record(rec, project_slug="p") + self.assertEqual(tools[0]["target"], "demo-skill") + + def test_non_user_record_ignored(self): + rec = { + "type": "assistant", "uuid": "a1", "sessionId": "s1", + "timestamp": "t", "isSidechain": False, + "message": {"content": [{"type": "text", + "text": "/foo"}], + "usage": {"input_tokens": 1, "output_tokens": 1}}, + } + _, tools = parse_record(rec, project_slug="p") + # Assistant text doesn't count as a slash invocation. + self.assertEqual([t["tool_name"] for t in tools], []) + + def test_ordinary_user_message_yields_no_skill_row(self): + rec = self._user_record("just a normal question about the code") + _, tools = parse_record(rec, project_slug="p") + self.assertEqual(tools, []) + + def test_malformed_slug_rejected(self): + rec = self._user_record("/not a slug") + _, tools = parse_record(rec, project_slug="p") + self.assertEqual(tools, []) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_scanner_slash_commands.py b/tests/test_scanner_slash_commands.py new file mode 100644 index 0000000..6ce8bbb --- /dev/null +++ b/tests/test_scanner_slash_commands.py @@ -0,0 +1,177 @@ +"""Integration tests for slash-command Skill synthesis. + +Covers both paths: (a) ingest-time synthesis via ``scan_dir`` emitting a +Skill row when a user record carries ``/``, +and (b) one-shot ``rescan_slash_commands`` backfilling existing DBs whose +user messages were ingested before the extractor existed. +""" +import json +import os +import sqlite3 +import tempfile +import time +import unittest + +from token_dashboard.db import connect, init_db +from token_dashboard.scanner import rescan_slash_commands, scan_dir + + +def _write_jsonl(path, records): + with open(path, "w", encoding="utf-8") as f: + for r in records: + f.write(json.dumps(r) + "\n") + + +def _slash_user(uuid, ts, slug, ordering="name-first"): + if ordering == "name-first": + content = ( + f"/{slug}\n" + f"{slug}\n" + f"" + ) + else: + content = ( + f"{slug}\n" + f"/{slug}" + ) + return { + "type": "user", + "uuid": uuid, + "sessionId": "s1", + "timestamp": ts, + "isSidechain": False, + "message": {"role": "user", "content": content}, + } + + +class SlashCommandIngestTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.db = os.path.join(self.tmp, "t.db") + self.proj_root = os.path.join(self.tmp, "projects") + self.proj_dir = os.path.join(self.proj_root, "C--work-sample") + os.makedirs(self.proj_dir) + init_db(self.db) + + def _path(self): + return os.path.join(self.proj_dir, "s1.jsonl") + + def _count_target(self, target): + with sqlite3.connect(self.db) as c: + return c.execute( + "SELECT COUNT(*) FROM tool_calls WHERE tool_name='Skill' AND target=?", + (target,), + ).fetchone()[0] + + def test_scan_emits_skill_row_for_slash_command(self): + _write_jsonl(self._path(), [ + _slash_user("u1", "2026-04-24T07:12:56Z", "demo-cmd"), + ]) + scan_dir(self.proj_root, self.db) + self.assertEqual(self._count_target("demo-cmd"), 1) + + def test_rescan_without_content_change_does_not_duplicate(self): + """Forced rescan (mtime bumped, content identical) must not double-count + the synthetic Skill row — relies on scan_file's per-uuid DELETE.""" + _write_jsonl(self._path(), [ + _slash_user("u1", "2026-04-24T07:12:56Z", "demo-cmd"), + ]) + scan_dir(self.proj_root, self.db) + self.assertEqual(self._count_target("demo-cmd"), 1) + + future = time.time() + 10 + os.utime(self._path(), (future, future)) + scan_dir(self.proj_root, self.db) + self.assertEqual(self._count_target("demo-cmd"), 1) + + def test_plugin_namespaced_slug_round_trips_through_db(self): + _write_jsonl(self._path(), [ + _slash_user("u1", "2026-04-24T07:00:00Z", "codex:review"), + ]) + scan_dir(self.proj_root, self.db) + self.assertEqual(self._count_target("codex:review"), 1) + + +class SlashCommandBackfillTests(unittest.TestCase): + """Verify rescan_slash_commands synthesizes rows from existing messages.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.db = os.path.join(self.tmp, "t.db") + init_db(self.db) + + def _seed_user_message(self, c, *, uuid, session, ts, content): + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp, " + "prompt_text, prompt_chars) " + "VALUES (?, ?, 'p', 'user', ?, ?, ?)", + (uuid, session, ts, content, len(content)), + ) + + def test_backfill_synthesizes_row_from_existing_message(self): + slash = "/demo-cmd" + with connect(self.db) as c: + self._seed_user_message( + c, uuid="u1", session="s1", + ts="2026-04-24T07:12:56Z", content=slash, + ) + c.commit() + + result = rescan_slash_commands(self.db) + self.assertEqual(result["slash_commands_synthesized"], 1) + + with sqlite3.connect(self.db) as c: + c.row_factory = sqlite3.Row + row = c.execute( + "SELECT tool_name, target, session_id, timestamp " + "FROM tool_calls WHERE message_uuid='u1'" + ).fetchone() + self.assertEqual(row["tool_name"], "Skill") + self.assertEqual(row["target"], "demo-cmd") + self.assertEqual(row["session_id"], "s1") + self.assertEqual(row["timestamp"], "2026-04-24T07:12:56Z") + + def test_backfill_is_idempotent(self): + slash = ( + "demo-cmd\n" + "/demo-cmd" + ) + with connect(self.db) as c: + self._seed_user_message( + c, uuid="u1", session="s1", + ts="2026-04-24T07:12:56Z", content=slash, + ) + c.commit() + rescan_slash_commands(self.db) + rescan_slash_commands(self.db) + with sqlite3.connect(self.db) as c: + cnt = c.execute( + "SELECT COUNT(*) FROM tool_calls WHERE message_uuid='u1'" + ).fetchone()[0] + self.assertEqual(cnt, 1, "two backfill calls must leave one row, not two") + + def test_backfill_skips_non_slash_user_messages(self): + with connect(self.db) as c: + self._seed_user_message( + c, uuid="u1", session="s1", + ts="2026-04-24T07:00:00Z", content="normal user prompt", + ) + self._seed_user_message( + c, uuid="u2", session="s1", + ts="2026-04-24T07:01:00Z", + content="/demo-skill", + ) + c.commit() + result = rescan_slash_commands(self.db) + self.assertEqual(result["slash_commands_synthesized"], 1) + with sqlite3.connect(self.db) as c: + targets = [ + r[0] for r in c.execute( + "SELECT target FROM tool_calls WHERE tool_name='Skill'" + ) + ] + self.assertEqual(targets, ["demo-skill"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_server_skills_budget.py b/tests/test_server_skills_budget.py new file mode 100644 index 0000000..492a9c5 --- /dev/null +++ b/tests/test_server_skills_budget.py @@ -0,0 +1,180 @@ +"""Integration tests for /api/skills budget fields.""" +import http.server +import json +import os +import socket +import sqlite3 +import tempfile +import threading +import unittest +import urllib.request +from pathlib import Path + +from token_dashboard.db import init_db +from token_dashboard.server import build_handler +from token_dashboard import skills, skill_budgets + + +def _free_port(): + s = socket.socket() + s.bind(("127.0.0.1", 0)) + port = s.getsockname()[1] + s.close() + return port + + +class ServerSkillBudgetTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.db = os.path.join(self.tmp, "t.db") + init_db(self.db) + + # Seed a project-local SKILL.md so cached_catalog (which walks cwds + # from messages) discovers it. Declared body-text budget = 100. + self.project = Path(self.tmp) / "myrepo" + skill_md = self.project / ".claude" / "skills" / "tight-skill" / "SKILL.md" + skill_md.parent.mkdir(parents=True, exist_ok=True) + skill_md.write_text( + "---\nname: tight-skill\n---\n\n" + "## Token Budget\n< 100 output tokens.\n", + encoding="utf-8", + ) + + with sqlite3.connect(self.db) as c: + # One user message so cwds lookup picks up the project-local skill root. + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp, cwd, output_tokens) " + "VALUES ('u0', 's1', 'p', 'user', '2026-04-10T00:00:00Z', ?, 0)", + (str(self.project / "src"),), + ) + # Invoke the skill, then emit assistant output well over budget × 1.2 (>120 tokens). + c.execute( + "INSERT INTO tool_calls (message_uuid, session_id, project_slug, tool_name, target, timestamp, is_error) " + "VALUES ('a1', 's1', 'p', 'Skill', 'tight-skill', '2026-04-10T00:00:01Z', 0)", + ) + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp, output_tokens) " + "VALUES ('m1', 's1', 'p', 'assistant', '2026-04-10T00:00:02Z', 500)", + ) + c.commit() + + # Reset skill caches so the test doesn't inherit neighbour state. + skills._cache["at"] = 0.0 + skills._cache["data"] = {} + skills._cache["key"] = None + skill_budgets._budget_cache.clear() + + self.port = _free_port() + H = build_handler(self.db, projects_dir="/nonexistent") + self.httpd = http.server.HTTPServer(("127.0.0.1", self.port), H) + threading.Thread(target=self.httpd.serve_forever, daemon=True).start() + + def tearDown(self): + self.httpd.shutdown() + skills._cache["at"] = 0.0 + skills._cache["data"] = {} + skills._cache["key"] = None + skill_budgets._budget_cache.clear() + + def _get(self, path): + return urllib.request.urlopen(f"http://127.0.0.1:{self.port}{path}").read() + + def test_skills_endpoint_includes_budget_fields(self): + rows = json.loads(self._get("/api/skills")) + self.assertIsInstance(rows, list) + self.assertTrue(rows, "expected at least one skill row from seeded tool_calls") + for r in rows: + self.assertIn("budget_output_tokens", r) + self.assertIn("p50_output_tokens", r) + self.assertIn("p95_output_tokens", r) + self.assertIn("over_budget", r) + + def test_over_budget_flag_on_tight_skill(self): + rows = json.loads(self._get("/api/skills")) + by_slug = {r["skill"]: r for r in rows} + self.assertIn("tight-skill", by_slug) + r = by_slug["tight-skill"] + self.assertEqual(r["budget_output_tokens"], 100) + self.assertEqual(r["p50_output_tokens"], 500) + self.assertTrue(r["over_budget"]) + + +class ServerSkillSubagentTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.db = os.path.join(self.tmp, "t.db") + init_db(self.db) + + with sqlite3.connect(self.db) as c: + # One main-chain user message so the project surfaces in catalog walks. + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp) " + "VALUES ('u0', 's1', 'p', 'user', '2026-04-10T00:00:00Z')", + ) + # Skill invocation. + c.execute( + "INSERT INTO tool_calls (message_uuid, session_id, project_slug, tool_name, target, timestamp, is_error) " + "VALUES ('a1', 's1', 'p', 'Skill', 'orchestrator', '2026-04-10T00:00:01Z', 0)", + ) + # Own cost: main-chain assistant emits 100 output tokens on claude-opus-4-5. + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, is_sidechain, timestamp, " + "model, output_tokens) " + "VALUES ('m1', 's1', 'p', 'assistant', 0, '2026-04-10T00:00:02Z', " + "'claude-opus-4-5', 100)", + ) + # Sidechain subagent chain: user injection + assistant response. + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, is_sidechain, " + "timestamp, agent_id) " + "VALUES ('sc-u', 's1', 'p', 'user', 1, '2026-04-10T00:00:03Z', 'agX')", + ) + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, is_sidechain, " + "timestamp, model, output_tokens, agent_id) " + "VALUES ('sc1', 's1', 'p', 'assistant', 1, '2026-04-10T00:00:04Z', " + "'claude-opus-4-5', 400, 'agX')", + ) + c.commit() + + skills._cache["at"] = 0.0 + skills._cache["data"] = {} + skills._cache["key"] = None + skill_budgets._budget_cache.clear() + + self.port = _free_port() + H = build_handler(self.db, projects_dir="/nonexistent") + self.httpd = http.server.HTTPServer(("127.0.0.1", self.port), H) + threading.Thread(target=self.httpd.serve_forever, daemon=True).start() + + def tearDown(self): + self.httpd.shutdown() + skills._cache["at"] = 0.0 + skills._cache["data"] = {} + skills._cache["key"] = None + skill_budgets._budget_cache.clear() + + def _get(self, path): + return urllib.request.urlopen(f"http://127.0.0.1:{self.port}{path}").read() + + def test_skills_endpoint_exposes_subagent_fields(self): + rows = json.loads(self._get("/api/skills")) + by_slug = {r["skill"]: r for r in rows} + self.assertIn("orchestrator", by_slug) + r = by_slug["orchestrator"] + self.assertIn("subagent_cost_usd", r) + self.assertIn("subagent_output_tokens", r) + self.assertIn("total_with_subagents_usd", r) + self.assertEqual(r["subagent_output_tokens"], 400) + # total_with_subagents_usd must equal own + subagent. + self.assertAlmostEqual( + r["total_with_subagents_usd"], + (r["total_cost_usd"] or 0.0) + (r["subagent_cost_usd"] or 0.0), + places=6, + ) + # Subagent cost should be positive (400 output tokens on opus). + self.assertGreater(r["subagent_cost_usd"], 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_skill_budgets.py b/tests/test_skill_budgets.py new file mode 100644 index 0000000..253e988 --- /dev/null +++ b/tests/test_skill_budgets.py @@ -0,0 +1,600 @@ +"""Unit tests for token_dashboard.skill_budgets. + +Parser fixtures use inline strings — never reads ~/.claude/. The actuals +tests seed a tmp SQLite DB and exercise the LEAD window-function boundary. +""" +import os +import tempfile +import unittest + +from token_dashboard.db import connect, init_db +from token_dashboard.skill_budgets import ( + parse_budget_from_text, + skill_actuals, + skill_costs, + skill_subagent_costs, +) + + +class ParseBudgetTests(unittest.TestCase): + def test_parse_inline_budget(self): + body = ( + "---\nname: example-skill\n---\n\n" + "Execute these steps in order. Complete in <800 output tokens. Conversational.\n" + ) + self.assertEqual(parse_budget_from_text(body), 800) + + def test_parse_section_budget(self): + body = ( + "---\nname: skill-foo\n---\n\n" + "Some body.\n\n## Token Budget\n< 100 output tokens. Fire-and-forget.\n" + ) + self.assertEqual(parse_budget_from_text(body), 100) + + def test_parse_budget_with_commas(self): + body = "Complete in <5,500 output tokens. Every claim must trace.\n" + self.assertEqual(parse_budget_from_text(body), 5500) + + def test_parse_no_budget(self): + body = ( + "---\nname: skill-bar\ndescription: Generic.\n---\n\n" + "No declaration in body. Just prose.\n" + ) + self.assertIsNone(parse_budget_from_text(body)) + + def test_parse_inline_wins_over_section(self): + # Both present — inline (top-of-file, more prescriptive) wins. + body = ( + "Execute these steps. Complete in <800 output tokens.\n\n" + "## Token Budget\n< 1,500 output tokens\n" + ) + self.assertEqual(parse_budget_from_text(body), 800) + + +def _seed_messages(c, rows): + """Insert assistant messages. Each row = (uuid, session, ts, output_tokens).""" + for uuid, session, ts, out in rows: + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp, output_tokens) " + "VALUES (?, ?, 'p', 'assistant', ?, ?)", + (uuid, session, ts, out), + ) + + +def _seed_skill_call(c, *, uuid, session, target, ts): + c.execute( + "INSERT INTO tool_calls (message_uuid, session_id, project_slug, tool_name, target, timestamp, is_error) " + "VALUES (?, ?, 'p', 'Skill', ?, ?, 0)", + (uuid, session, target, ts), + ) + + +class SkillActualsTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.db = os.path.join(self.tmp, "s.db") + init_db(self.db) + + def test_skill_actuals_basic(self): + """Single Skill call, 2 subsequent assistant messages → one sample summing both.""" + with connect(self.db) as c: + _seed_skill_call(c, uuid="a1", session="s1", + target="brainstorming", ts="2026-04-10T00:00:00Z") + _seed_messages(c, [ + ("m1", "s1", "2026-04-10T00:00:05Z", 100), + ("m2", "s1", "2026-04-10T00:00:10Z", 200), + ]) + c.commit() + + actuals = skill_actuals(self.db) + self.assertIn("brainstorming", actuals) + stat = actuals["brainstorming"] + self.assertEqual(stat["count"], 1) + self.assertEqual(stat["p50"], 300) + self.assertEqual(stat["p95"], 300) + + def test_skill_actuals_next_skill_terminates_window(self): + """Two Skill calls in one session: first's window ends at second's ts.""" + with connect(self.db) as c: + _seed_skill_call(c, uuid="a1", session="s1", + target="first", ts="2026-04-10T00:00:00Z") + _seed_messages(c, [ + ("m1", "s1", "2026-04-10T00:00:01Z", 100), + ("m2", "s1", "2026-04-10T00:00:02Z", 200), + ("m3", "s1", "2026-04-10T00:00:03Z", 300), + ]) + _seed_skill_call(c, uuid="a2", session="s1", + target="second", ts="2026-04-10T00:00:10Z") + _seed_messages(c, [ + ("m4", "s1", "2026-04-10T00:00:11Z", 50), + ("m5", "s1", "2026-04-10T00:00:12Z", 70), + ]) + c.commit() + + actuals = skill_actuals(self.db) + # first: messages m1+m2+m3 (before the second call) = 600 + # second: messages m4+m5 (after second call, no next call) = 120 + self.assertEqual(actuals["first"]["p50"], 600) + self.assertEqual(actuals["second"]["p50"], 120) + + def test_skill_actuals_end_of_session_window(self): + """Last Skill call in a session with no subsequent call: all remaining output counted.""" + with connect(self.db) as c: + _seed_skill_call(c, uuid="a1", session="s1", + target="tail", ts="2026-04-10T00:00:00Z") + _seed_messages(c, [ + ("m1", "s1", "2026-04-10T00:00:01Z", 500), + ("m2", "s1", "2026-04-10T01:00:00Z", 500), + ]) + c.commit() + + actuals = skill_actuals(self.db) + self.assertEqual(actuals["tail"]["p50"], 1000) + + def test_skill_actuals_cross_session_does_not_leak(self): + """A Skill call in session A must not receive output from session B.""" + with connect(self.db) as c: + _seed_skill_call(c, uuid="a1", session="sA", + target="isolated", ts="2026-04-10T00:00:00Z") + # Same timestamp range, different session — must NOT be counted. + _seed_messages(c, [ + ("m1", "sB", "2026-04-10T00:00:05Z", 9999), + ]) + c.commit() + + actuals = skill_actuals(self.db) + self.assertEqual(actuals["isolated"]["p50"], 0) + self.assertEqual(actuals["isolated"]["count"], 1) + + def test_skill_actuals_excludes_sidechain(self): + """Assistant output with is_sidechain=1 (subagents, auto-compaction) must not count.""" + with connect(self.db) as c: + _seed_skill_call(c, uuid="a1", session="s1", + target="leaky", ts="2026-04-10T00:00:00Z") + # One main-chain message + one huge sidechain message in the window. + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp, output_tokens, is_sidechain) " + "VALUES ('m1', 's1', 'p', 'assistant', '2026-04-10T00:00:01Z', 100, 0)" + ) + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp, output_tokens, is_sidechain) " + "VALUES ('m2', 's1', 'p', 'assistant', '2026-04-10T00:00:02Z', 9999, 1)" + ) + c.commit() + actuals = skill_actuals(self.db) + self.assertEqual(actuals["leaky"]["p50"], 100) + + def test_skill_actuals_real_user_message_terminates_window(self): + """A real-user-typed message (prompt_chars>0, no meta prefix) ends the window.""" + with connect(self.db) as c: + _seed_skill_call(c, uuid="a1", session="s1", + target="chatty", ts="2026-04-10T00:00:00Z") + # Assistant emits 100 tokens, user types something, assistant emits more. + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp, output_tokens) " + "VALUES ('m1', 's1', 'p', 'assistant', '2026-04-10T00:00:01Z', 100)" + ) + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp, prompt_text, prompt_chars) " + "VALUES ('u1', 's1', 'p', 'user', '2026-04-10T00:00:02Z', 'change of plans, do X instead', 30)" + ) + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp, output_tokens) " + "VALUES ('m2', 's1', 'p', 'assistant', '2026-04-10T00:00:03Z', 5000)" + ) + c.commit() + actuals = skill_actuals(self.db) + # Only m1 should count; m2 is past the real-user boundary. + self.assertEqual(actuals["chatty"]["p50"], 100) + + def test_skill_actuals_meta_user_messages_do_not_terminate(self): + """System-injected user messages (SKILL.md body, , etc.) are not boundaries.""" + with connect(self.db) as c: + _seed_skill_call(c, uuid="a1", session="s1", + target="loaded", ts="2026-04-10T00:00:00Z") + # Immediately after the Skill call, Claude Code injects the SKILL.md body + # as a user-role message. This must NOT terminate attribution. + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp, prompt_text, prompt_chars) " + "VALUES ('u-inject', 's1', 'p', 'user', '2026-04-10T00:00:00.500Z', ?, 5000)", + ("Base directory for this skill: /home/x/.claude/skills/loaded\n\n# body...",), + ) + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp, output_tokens) " + "VALUES ('m1', 's1', 'p', 'assistant', '2026-04-10T00:00:01Z', 400)" + ) + c.commit() + actuals = skill_actuals(self.db) + # Injected skill-load user message is filtered out, so m1 is counted. + self.assertEqual(actuals["loaded"]["p50"], 400) + + def test_skill_actuals_respects_since(self): + """Skill calls before `since` are filtered out.""" + with connect(self.db) as c: + _seed_skill_call(c, uuid="a1", session="s1", + target="old", ts="2026-04-10T00:00:00Z") + _seed_skill_call(c, uuid="a2", session="s2", + target="new", ts="2026-04-20T00:00:00Z") + _seed_messages(c, [ + ("m1", "s1", "2026-04-10T00:00:01Z", 111), + ("m2", "s2", "2026-04-20T00:00:01Z", 222), + ]) + c.commit() + + actuals = skill_actuals(self.db, since="2026-04-15T00:00:00Z") + self.assertNotIn("old", actuals) + self.assertEqual(actuals["new"]["p50"], 222) + + +class SkillCostsTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.db = os.path.join(self.tmp, "s.db") + init_db(self.db) + # Minimal pricing table covering one known model + tier fallback. + self.pricing = { + "models": { + "claude-haiku-4-5": { + "input": 1.0, + "output": 5.0, + "cache_read": 0.1, + "cache_create_5m": 1.25, + "cache_create_1h": 2.0, + }, + }, + "tier_fallback": { + "haiku": {"input": 1.0, "output": 5.0, "cache_read": 0.1, + "cache_create_5m": 1.25, "cache_create_1h": 2.0}, + "sonnet": {"input": 3.0, "output": 15.0, "cache_read": 0.3, + "cache_create_5m": 3.75, "cache_create_1h": 6.0}, + "opus": {"input": 15.0, "output": 75.0, "cache_read": 1.5, + "cache_create_5m": 18.75, "cache_create_1h": 30.0}, + }, + } + + def test_skill_costs_basic(self): + """Costs one invocation with known model; verifies the multiplication.""" + with connect(self.db) as c: + _seed_skill_call(c, uuid="a1", session="s1", + target="billable", ts="2026-04-10T00:00:00Z") + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp, " + "model, input_tokens, output_tokens, cache_read_tokens, " + "cache_create_5m_tokens, cache_create_1h_tokens) " + "VALUES ('m1', 's1', 'p', 'assistant', '2026-04-10T00:00:05Z', " + "'claude-haiku-4-5', 1000000, 200000, 0, 0, 0)" + ) + c.commit() + costs = skill_costs(self.db, self.pricing) + # 1M input × $1/M + 200k output × $5/M = $1 + $1 = $2 + self.assertIn("billable", costs) + self.assertAlmostEqual(costs["billable"]["cost_usd"], 2.0, places=4) + self.assertFalse(costs["billable"]["cost_estimated"]) + + def test_skill_costs_unknown_model_falls_back_to_tier(self): + """A model name matching a known tier (opus/sonnet/haiku) uses tier pricing and flags estimated.""" + with connect(self.db) as c: + _seed_skill_call(c, uuid="a1", session="s1", + target="tiered", ts="2026-04-10T00:00:00Z") + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp, " + "model, output_tokens) " + "VALUES ('m1', 's1', 'p', 'assistant', '2026-04-10T00:00:05Z', " + "'claude-opus-4-99-unreleased', 1000000)" + ) + c.commit() + costs = skill_costs(self.db, self.pricing) + # 1M output × $75/M (opus tier fallback) = $75 + self.assertAlmostEqual(costs["tiered"]["cost_usd"], 75.0, places=2) + self.assertTrue(costs["tiered"]["cost_estimated"]) + + def test_skill_costs_aggregates_across_models(self): + """A single skill window hitting two models costs each separately and sums.""" + with connect(self.db) as c: + _seed_skill_call(c, uuid="a1", session="s1", + target="mixed", ts="2026-04-10T00:00:00Z") + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp, " + "model, output_tokens) " + "VALUES ('m1', 's1', 'p', 'assistant', '2026-04-10T00:00:05Z', " + "'claude-haiku-4-5', 1000000)" + ) + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp, " + "model, output_tokens) " + "VALUES ('m2', 's1', 'p', 'assistant', '2026-04-10T00:00:10Z', " + "'claude-opus-4-7', 100000)" + ) + c.commit() + costs = skill_costs(self.db, self.pricing) + # haiku output 1M × $5 = $5, opus output 100k × $75 = $7.5 → total $12.5 + self.assertAlmostEqual(costs["mixed"]["cost_usd"], 12.5, places=2) + self.assertTrue(costs["mixed"]["cost_estimated"]) # opus was tier-fallback + + +def _seed_sidechain(c, *, uuid, session, ts, agent_id, output_tokens=0, + model="claude-opus-4-5", input_tokens=0, msg_type="assistant"): + """Seed a sidechain message. Real subagent messages carry an agentId + (the hash from subagents/agent-.jsonl); attribution joins on it. + """ + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, is_sidechain, " + "timestamp, model, input_tokens, output_tokens, agent_id) " + "VALUES (?, ?, 'p', ?, 1, ?, ?, ?, ?, ?)", + (uuid, session, msg_type, ts, model, input_tokens, output_tokens, agent_id), + ) + + +class SkillSubagentCostsTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.db = os.path.join(self.tmp, "s.db") + init_db(self.db) + self.pricing = { + "models": { + "claude-opus-4-5": { + "input": 15.0, + "output": 75.0, + "cache_read": 1.5, + "cache_create_5m": 18.75, + "cache_create_1h": 30.0, + }, + }, + "tier_fallback": { + "opus": {"input": 15.0, "output": 75.0, "cache_read": 1.5, + "cache_create_5m": 18.75, "cache_create_1h": 30.0}, + "sonnet": {"input": 3.0, "output": 15.0, "cache_read": 0.3, + "cache_create_5m": 3.75, "cache_create_1h": 6.0}, + "haiku": {"input": 1.0, "output": 5.0, "cache_read": 0.1, + "cache_create_5m": 1.25, "cache_create_1h": 2.0}, + }, + } + + def test_single_dispatch_sums_sidechain(self): + """One Skill call → one subagent (agent_id=ag1) → two assistant messages.""" + with connect(self.db) as c: + _seed_skill_call(c, uuid="sk1", session="s1", + target="orch-a", ts="2026-04-10T00:00:00Z") + # Subagent starts inside the window (user-injected prompt). + _seed_sidechain(c, uuid="u1", session="s1", agent_id="ag1", + ts="2026-04-10T00:00:06Z", msg_type="user") + _seed_sidechain(c, uuid="a1", session="s1", agent_id="ag1", + ts="2026-04-10T00:00:10Z", output_tokens=1000) + _seed_sidechain(c, uuid="a2", session="s1", agent_id="ag1", + ts="2026-04-10T00:00:15Z", output_tokens=500) + c.commit() + sub = skill_subagent_costs(self.db, self.pricing) + self.assertIn("orch-a", sub) + self.assertEqual(sub["orch-a"]["output_tokens"], 1500) + # 1500 output × $75/M = $0.1125 + self.assertAlmostEqual(sub["orch-a"]["cost_usd"], 0.1125, places=4) + + def test_sidechain_past_window_end_still_attributed(self): + """Subagent started inside window, finishes AFTER user typed next message.""" + with connect(self.db) as c: + _seed_skill_call(c, uuid="sk1", session="s1", + target="orch-b", ts="2026-04-10T00:00:00Z") + # Subagent started at t+6s (inside window). + _seed_sidechain(c, uuid="u1", session="s1", agent_id="ag1", + ts="2026-04-10T00:00:06Z", msg_type="user") + # User types again, closing the skill's own-cost window. + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp, " + "prompt_text, prompt_chars, is_sidechain) " + "VALUES ('u-real', 's1', 'p', 'user', '2026-04-10T00:01:00Z', " + "'go ahead', 8, 0)" + ) + # Subagent response arrives after the user typed — by lineage still qa's. + _seed_sidechain(c, uuid="a1", session="s1", agent_id="ag1", + ts="2026-04-10T00:02:00Z", output_tokens=5000) + c.commit() + sub = skill_subagent_costs(self.db, self.pricing) + self.assertEqual(sub["orch-b"]["output_tokens"], 5000) + + def test_two_skills_dispatches_disjoint(self): + """Skill A and skill B each dispatch one subagent; attribution doesn't cross.""" + with connect(self.db) as c: + _seed_skill_call(c, uuid="skA", session="s1", + target="A", ts="2026-04-10T00:00:00Z") + _seed_sidechain(c, uuid="uA", session="s1", agent_id="agA", + ts="2026-04-10T00:00:06Z", msg_type="user") + _seed_sidechain(c, uuid="aA", session="s1", agent_id="agA", + ts="2026-04-10T00:00:10Z", output_tokens=100) + # Real user message → closes A's window. + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp, " + "prompt_text, prompt_chars, is_sidechain) " + "VALUES ('u1', 's1', 'p', 'user', '2026-04-10T00:01:00Z', " + "'run B', 6, 0)" + ) + _seed_skill_call(c, uuid="skB", session="s1", + target="B", ts="2026-04-10T00:02:00Z") + _seed_sidechain(c, uuid="uB", session="s1", agent_id="agB", + ts="2026-04-10T00:02:06Z", msg_type="user") + _seed_sidechain(c, uuid="aB", session="s1", agent_id="agB", + ts="2026-04-10T00:02:10Z", output_tokens=900) + c.commit() + sub = skill_subagent_costs(self.db, self.pricing) + self.assertEqual(sub["A"]["output_tokens"], 100) + self.assertEqual(sub["B"]["output_tokens"], 900) + + def test_nested_subagent_attributed_to_root_skill(self): + """Team pattern: outer subagent dispatches an inner subagent during orchestrator's window.""" + with connect(self.db) as c: + _seed_skill_call(c, uuid="sk1", session="s1", + target="orch-c", ts="2026-04-10T00:00:00Z") + # Outer subagent. + _seed_sidechain(c, uuid="u1", session="s1", agent_id="outer", + ts="2026-04-10T00:00:06Z", msg_type="user") + _seed_sidechain(c, uuid="a1", session="s1", agent_id="outer", + ts="2026-04-10T00:00:10Z", output_tokens=300) + # Inner subagent starts at t+12s — still inside team-audit's window. + _seed_sidechain(c, uuid="u2", session="s1", agent_id="inner", + ts="2026-04-10T00:00:12Z", msg_type="user") + _seed_sidechain(c, uuid="a2", session="s1", agent_id="inner", + ts="2026-04-10T00:00:20Z", output_tokens=700) + c.commit() + sub = skill_subagent_costs(self.db, self.pricing) + # outer (300) + inner (700) = 1000; both attributed to the root skill. + self.assertEqual(sub["orch-c"]["output_tokens"], 1000) + + def test_dispatch_outside_skill_window_ignored(self): + """Subagent started before any Skill call is not attributed.""" + with connect(self.db) as c: + # Subagent starts at t=00:00 with no preceding Skill call. + _seed_sidechain(c, uuid="u1", session="s1", agent_id="ag-orphan", + ts="2026-04-10T00:00:00Z", msg_type="user") + _seed_sidechain(c, uuid="a1", session="s1", agent_id="ag-orphan", + ts="2026-04-10T00:00:05Z", output_tokens=9999) + c.commit() + sub = skill_subagent_costs(self.db, self.pricing) + self.assertEqual(sub, {}) + + def test_auto_compaction_sidechain_ignored(self): + """agent_id prefixed acompact is auto-compaction, never counted.""" + with connect(self.db) as c: + _seed_skill_call(c, uuid="sk1", session="s1", + target="brainstorming", ts="2026-04-10T00:00:00Z") + _seed_sidechain(c, uuid="u1", session="s1", agent_id="acompact-abc", + ts="2026-04-10T00:00:05Z", msg_type="user") + _seed_sidechain(c, uuid="ac1", session="s1", agent_id="acompact-abc", + ts="2026-04-10T00:00:10Z", output_tokens=2000) + c.commit() + sub = skill_subagent_costs(self.db, self.pricing) + self.assertNotIn("brainstorming", sub) + + def test_skill_costs_unchanged_for_non_orchestrators(self): + """Regression: skill with no Agent dispatches has skill_costs unchanged and no subagent row.""" + with connect(self.db) as c: + _seed_skill_call(c, uuid="sk1", session="s1", + target="leaf", ts="2026-04-10T00:00:00Z") + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp, " + "model, output_tokens, is_sidechain) " + "VALUES ('m1', 's1', 'p', 'assistant', '2026-04-10T00:00:05Z', " + "'claude-opus-4-5', 200, 0)" + ) + c.commit() + own = skill_costs(self.db, self.pricing) + sub = skill_subagent_costs(self.db, self.pricing) + self.assertIn("leaf", own) + self.assertAlmostEqual(own["leaf"]["cost_usd"], 200 * 75.0 / 1_000_000, places=6) + self.assertNotIn("leaf", sub) + + +class SlashCommandAttributionTests(unittest.TestCase): + """Once a synthetic Skill row is in place for a slash-command invocation, + skill_costs/skill_actuals must attribute the following assistant work to + the slash-command slug, and close the window at the next real-user typed + message (neither `` nor `` counts). + """ + + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.db = os.path.join(self.tmp, "s.db") + init_db(self.db) + self.pricing = { + "models": { + "claude-haiku-4-5": { + "input": 1.0, "output": 5.0, "cache_read": 0.1, + "cache_create_5m": 1.25, "cache_create_1h": 2.0, + }, + }, + "tier_fallback": { + "haiku": {"input": 1.0, "output": 5.0, "cache_read": 0.1, + "cache_create_5m": 1.25, "cache_create_1h": 2.0}, + "sonnet": {"input": 3.0, "output": 15.0, "cache_read": 0.3, + "cache_create_5m": 3.75, "cache_create_1h": 6.0}, + "opus": {"input": 15.0, "output": 75.0, "cache_read": 1.5, + "cache_create_5m": 18.75, "cache_create_1h": 30.0}, + }, + } + + def test_slash_command_row_receives_attribution_window(self): + """User types /demo-cmd at t0; assistant emits 200k output at t1; + user types a real follow-up at t2 → only t1 is attributed.""" + with connect(self.db) as c: + # User message carrying the slash command (stays in messages). + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, " + "timestamp, prompt_text, prompt_chars) " + "VALUES ('u-cmd', 's1', 'p', 'user', '2026-04-24T07:12:56Z', " + "'/demo-cmd', 39)" + ) + # Synthetic Skill row keyed on the same uuid (as ingest would emit). + _seed_skill_call(c, uuid="u-cmd", session="s1", + target="demo-cmd", ts="2026-04-24T07:12:56Z") + # Assistant work inside window. + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, " + "timestamp, model, output_tokens) " + "VALUES ('a1', 's1', 'p', 'assistant', '2026-04-24T07:13:00Z', " + "'claude-haiku-4-5', 200000)" + ) + # Real user follow-up — must terminate the window. + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, " + "timestamp, prompt_text, prompt_chars) " + "VALUES ('u-real', 's1', 'p', 'user', '2026-04-24T07:14:00Z', " + "'thanks, all good', 17)" + ) + # Assistant work AFTER window. + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, " + "timestamp, model, output_tokens) " + "VALUES ('a2', 's1', 'p', 'assistant', '2026-04-24T07:14:05Z', " + "'claude-haiku-4-5', 9999)" + ) + c.commit() + + actuals = skill_actuals(self.db) + self.assertIn("demo-cmd", actuals) + # Only a1 counts (a2 is past the real-user boundary). + self.assertEqual(actuals["demo-cmd"]["p50"], 200000) + + costs = skill_costs(self.db, self.pricing) + # 200k output × $5/M = $1.00 + self.assertAlmostEqual(costs["demo-cmd"]["cost_usd"], 1.0, places=4) + + def test_command_message_prefix_does_not_terminate_window(self): + """Regression: a user record whose prompt_text starts with + `` (one of the observed slash-command orderings) + must be recognised as a meta-message, NOT a real typed message. + Otherwise it would prematurely close a prior skill's window in the + same session.""" + with connect(self.db) as c: + # Earlier skill invocation still open. + _seed_skill_call(c, uuid="sk1", session="s1", + target="demo-skill", ts="2026-04-24T06:00:00Z") + # Assistant does work at t+1s. + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, " + "timestamp, output_tokens) " + "VALUES ('m1', 's1', 'p', 'assistant', '2026-04-24T06:00:01Z', 300)" + ) + # Later in the session, user types /demo-cmd; the record leads + # with , not . + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, " + "timestamp, prompt_text, prompt_chars) " + "VALUES ('u-cmd', 's1', 'p', 'user', '2026-04-24T07:00:00Z', " + "?, 44)", + ("demo-cmd\n" + "/demo-cmd",), + ) + # More assistant work, still inside demo-skill's window if the + # prefix filter is correct. + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, " + "timestamp, output_tokens) " + "VALUES ('m2', 's1', 'p', 'assistant', '2026-04-24T07:00:01Z', 400)" + ) + c.commit() + actuals = skill_actuals(self.db) + # demo-skill's window stays open across the -first + # user record → m1 + m2 both count (700 total). + self.assertEqual(actuals["demo-skill"]["p50"], 700) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_skills.py b/tests/test_skills.py index 1d8c6df..2a08f9d 100644 --- a/tests/test_skills.py +++ b/tests/test_skills.py @@ -3,7 +3,14 @@ import unittest from pathlib import Path -from token_dashboard.skills import scan_catalog, _slugs_for +from token_dashboard.skills import ( + scan_catalog, + _slugs_for, + _project_skill_roots_from_cwds, + cached_catalog, + _cache, +) +from token_dashboard.db import connect, init_db def _write(p: Path, body: str) -> None: @@ -69,6 +76,54 @@ def test_missing_skill_not_in_catalog(self): cat = scan_catalog([self.tmp / "skills"]) self.assertNotIn("never-installed", cat) + def test_project_local_skill_slug(self): + _write(self.tmp / "proj" / ".claude" / "skills" / "my-skill" / "SKILL.md", "p" * 400) + cat = scan_catalog([self.tmp / "proj" / ".claude" / "skills"]) + self.assertIn("my-skill", cat) + self.assertEqual(cat["my-skill"]["tokens"], 100) + + def test_project_skill_roots_innermost_wins(self): + outer = self.tmp / "outer" / ".claude" / "skills" + inner = self.tmp / "outer" / "inner" / ".claude" / "skills" + outer.mkdir(parents=True) + inner.mkdir(parents=True) + roots = _project_skill_roots_from_cwds([str(self.tmp / "outer" / "inner" / "src")]) + self.assertEqual(roots, [inner]) + + def test_project_skill_roots_dedupes_across_cwds(self): + root = self.tmp / "repo" / ".claude" / "skills" + root.mkdir(parents=True) + cwds = [ + str(self.tmp / "repo" / "src"), + str(self.tmp / "repo" / "tests" / "unit"), + ] + roots = _project_skill_roots_from_cwds(cwds) + self.assertEqual(roots, [root]) + + def test_cached_catalog_includes_project_local_from_db(self): + # Reset the module-level cache so this test doesn't inherit neighbour state. + _cache["at"] = 0.0 + _cache["data"] = {} + _cache["key"] = None + + project = self.tmp / "myrepo" + _write(project / ".claude" / "skills" / "repo-skill" / "SKILL.md", "x" * 400) + + db_path = self.tmp / "t.db" + init_db(db_path) + with connect(db_path) as c: + c.execute( + "INSERT INTO messages (uuid, session_id, project_slug, type, timestamp, cwd)" + " VALUES (?, ?, ?, ?, ?, ?)", + ("u1", "s1", "myrepo", "user", "2026-04-23T00:00:00Z", + str(project / "src")), + ) + c.commit() + + cat = cached_catalog(db_path) + self.assertIn("repo-skill", cat) + self.assertEqual(cat["repo-skill"]["tokens"], 100) + if __name__ == "__main__": unittest.main() diff --git a/token_dashboard/db.py b/token_dashboard/db.py index 956b69c..651c6a3 100644 --- a/token_dashboard/db.py +++ b/token_dashboard/db.py @@ -47,6 +47,8 @@ CREATE INDEX IF NOT EXISTS idx_messages_timestamp ON messages(timestamp); CREATE INDEX IF NOT EXISTS idx_messages_model ON messages(model); CREATE INDEX IF NOT EXISTS idx_messages_msgid ON messages(session_id, message_id); +CREATE INDEX IF NOT EXISTS idx_messages_parent ON messages(parent_uuid); +CREATE INDEX IF NOT EXISTS idx_messages_agent ON messages(agent_id); CREATE TABLE IF NOT EXISTS tool_calls ( id INTEGER PRIMARY KEY AUTOINCREMENT, diff --git a/token_dashboard/scanner.py b/token_dashboard/scanner.py index b78a985..f5c93d0 100644 --- a/token_dashboard/scanner.py +++ b/token_dashboard/scanner.py @@ -2,6 +2,7 @@ from __future__ import annotations import json +import re import time from pathlib import Path from typing import List, Optional, Tuple, Union @@ -9,6 +10,16 @@ from .db import connect +# Slash-command user messages (typed as `/foo` in Claude Code) arrive as a +# user-role record whose content looks like `/foo` +# (with optional ``/`` sibling tags, in any +# order). The scanner synthesizes a `tool_name='Skill'` row from these so +# user-invoked skills appear in skill_breakdown / skill_costs / skill_actuals +# alongside assistant-initiated Skill tool_use blocks. Matches plugin-namespaced +# slugs like `codex:review` via the `:` in the character class. +_SLASH_CMD_RE = re.compile(r"/([A-Za-z0-9_:-]+)") + + INSERT_MSG = """ INSERT OR REPLACE INTO messages ( uuid, parent_uuid, session_id, project_slug, cwd, git_branch, cc_version, entrypoint, @@ -39,6 +50,7 @@ "WebFetch": "url", "WebSearch": "query", "Task": "subagent_type", + "Agent": "subagent_type", "Skill": "skill", } @@ -97,6 +109,40 @@ def _extract_tools(rec: dict) -> List[dict]: return out +def _extract_slash_commands(rec: dict) -> List[dict]: + """Return a synthetic `Skill` tool_call row when a user record carries a + `/` tag. At most one per record. + + Claude Code logs user-typed slash commands without emitting an assistant + `tool_use` block, so the base scanner misses them. We key the synthetic + row on the user message's uuid/timestamp so the existing per-message + dedup (``DELETE FROM tool_calls WHERE message_uuid=?``) keeps rescans + idempotent. + """ + if rec.get("type") != "user": + return [] + content = (rec.get("message") or {}).get("content") + if isinstance(content, str): + text = content + elif isinstance(content, list): + text = "".join( + b.get("text", "") for b in content + if isinstance(b, dict) and b.get("type") == "text" + ) + else: + return [] + m = _SLASH_CMD_RE.search(text) + if not m: + return [] + return [{ + "tool_name": "Skill", + "target": m.group(1), + "result_tokens": None, + "is_error": 0, + "timestamp": rec.get("timestamp"), + }] + + def _extract_results(rec: dict) -> List[dict]: out = [] content = (rec.get("message") or {}).get("content") @@ -149,6 +195,7 @@ def parse_record(rec: dict, project_slug: str) -> Tuple[dict, List[dict]]: **_usage(rec), } tools = _extract_tools(rec) + tools.extend(_extract_slash_commands(rec)) tools.extend(_extract_results(rec)) if tools: msg["tool_calls_json"] = json.dumps( @@ -242,6 +289,45 @@ def scan_file(path: Path, project_slug: str, conn, start_byte: int = 0) -> dict: return {"messages": msgs, "tools": tools, "end_offset": end_offset} +def rescan_agent_targets( + db_path: Union[str, Path], + projects_root: Union[str, Path], +) -> dict: + """Re-parse main-session JSONLs that hold ``tool_name='Agent'`` rows with + ``target IS NULL``. + + Older scanner builds recognised only the legacy ``Task`` tool name; + Claude Code renamed it to ``Agent``, leaving historical rows without a + subagent_type to join on. Resetting those files' ``bytes_read`` makes + the next ``scan_dir`` re-parse them end-to-end. Dedup is handled by + ``INSERT OR REPLACE`` on messages + ``DELETE FROM tool_calls`` per + uuid, so repeated runs are safe. + + One-shot utility: wire in at operator time, not on every scan. + """ + with connect(db_path) as conn: + sessions = [ + r["session_id"] for r in conn.execute( + "SELECT DISTINCT session_id FROM tool_calls " + "WHERE tool_name='Agent' AND target IS NULL" + ) + ] + if not sessions: + return {"files_reset": 0, "messages": 0, "tools": 0, "files": 0} + paths: set[str] = set() + for sid in sessions: + for row in conn.execute( + "SELECT path FROM files WHERE path LIKE ?", + (f"%/{sid}.jsonl",), + ): + paths.add(row["path"]) + for p in paths: + conn.execute("UPDATE files SET bytes_read = 0 WHERE path = ?", (p,)) + conn.commit() + result = scan_dir(projects_root, db_path) + return {"files_reset": len(paths), **result} + + def scan_dir(projects_root: Union[str, Path], db_path: Union[str, Path]) -> dict: root = Path(projects_root) totals = {"messages": 0, "tools": 0, "files": 0} @@ -275,3 +361,46 @@ def scan_dir(projects_root: Union[str, Path], db_path: Union[str, Path]) -> dict totals["files"] += 1 conn.commit() return totals + + +def rescan_slash_commands(db_path: Union[str, Path]) -> dict: + """Synthesize ``tool_name='Skill'`` rows for already-ingested slash-command + user messages. No filesystem re-read required — ``prompt_text`` already + holds the ``/`` tag. + + Idempotent: a prior synthetic row on the same ``message_uuid`` is deleted + before re-inserting, so repeated runs are safe. Real assistant-initiated + ``Skill`` tool_use rows have distinct ``message_uuid`` values and aren't + touched. + + One-shot utility for DBs populated before this extractor existed. + """ + synthesized = 0 + with connect(db_path) as conn: + rows = list(conn.execute( + "SELECT uuid, session_id, project_slug, timestamp, prompt_text " + "FROM messages " + "WHERE type='user' AND prompt_text LIKE '%/%'" + )) + for row in rows: + m = _SLASH_CMD_RE.search(row["prompt_text"] or "") + if not m: + continue + conn.execute( + "DELETE FROM tool_calls " + "WHERE message_uuid=? AND tool_name='Skill'", + (row["uuid"],), + ) + conn.execute(INSERT_TOOL, { + "message_uuid": row["uuid"], + "session_id": row["session_id"], + "project_slug": row["project_slug"], + "tool_name": "Skill", + "target": m.group(1), + "result_tokens": None, + "is_error": 0, + "timestamp": row["timestamp"], + }) + synthesized += 1 + conn.commit() + return {"slash_commands_synthesized": synthesized} diff --git a/token_dashboard/server.py b/token_dashboard/server.py index b6156ff..7739c02 100644 --- a/token_dashboard/server.py +++ b/token_dashboard/server.py @@ -122,10 +122,42 @@ def do_GET(self): return _send_json(self, daily_token_breakdown(db_path, since, until)) if path == "/api/skills": rows = skill_breakdown(db_path, since, until) - catalog = cached_catalog() + catalog = cached_catalog(db_path) + # Lazy import so deleting skill_budgets.py keeps the server bootable. + from .skill_budgets import ( + budget_for, + skill_actuals, + skill_costs, + skill_subagent_costs, + ) + actuals = skill_actuals(db_path, since, until) + costs = skill_costs(db_path, pricing, since, until) + sub = skill_subagent_costs(db_path, pricing, since, until) for r in rows: info = catalog.get(r["skill"]) r["tokens_per_call"] = info["tokens"] if info else None + r["budget_output_tokens"] = budget_for(r["skill"], catalog) + a = actuals.get(r["skill"]) + r["p50_output_tokens"] = a["p50"] if a else None + r["p95_output_tokens"] = a["p95"] if a else None + r["over_budget"] = bool( + r["budget_output_tokens"] + and a + and a["p50"] > r["budget_output_tokens"] * 1.2 + ) + c = costs.get(r["skill"]) + r["total_cost_usd"] = c["cost_usd"] if c else None + r["cost_estimated"] = bool(c and c["cost_estimated"]) + s = sub.get(r["skill"]) + r["subagent_cost_usd"] = s["cost_usd"] if s else None + r["subagent_output_tokens"] = s["output_tokens"] if s else 0 + r["total_with_subagents_usd"] = ( + (r["total_cost_usd"] or 0.0) + (r["subagent_cost_usd"] or 0.0) + if (r["total_cost_usd"] is not None or r["subagent_cost_usd"] is not None) + else None + ) + if s and s["cost_estimated"]: + r["cost_estimated"] = True return _send_json(self, rows) if path == "/api/by-model": rows = model_breakdown(db_path, since, until) diff --git a/token_dashboard/skill_budgets.py b/token_dashboard/skill_budgets.py new file mode 100644 index 0000000..1494669 --- /dev/null +++ b/token_dashboard/skill_budgets.py @@ -0,0 +1,387 @@ +"""Skill budget-vs-actual tracking. + +Parses user-declared output-token budgets from SKILL.md body text and +measures each skill's actual output-token footprint per invocation. + +Two declaration formats are supported (no frontmatter field exists across +the catalog today): + 1. Inline: ``Execute these steps in order. Complete in Optional[int]: + """Return declared output-token budget, or None if nothing parsed. + + Inline form wins if both patterns appear in the same file (in the + sampled corpus they are mutually exclusive, but the inline line sits + at the top and is the more prescriptive form). + """ + for rx in (_INLINE, _SECTION): + m = rx.search(text) + if m: + return int(m.group(1).replace(",", "")) + return None + + +_budget_cache: dict[tuple[str, float], Optional[int]] = {} + + +def budget_for(slug: str, catalog=None) -> Optional[int]: + """Look up a skill's declared budget via the catalog, cache by (path, mtime). + + Missing slug, unreadable file, or unparsed body → None. No exceptions. + """ + from .skills import cached_catalog + + cat = catalog if catalog is not None else cached_catalog() + info = cat.get(slug) + if not info: + return None + path = Path(info["path"]) + try: + mtime = path.stat().st_mtime + except OSError: + return None + key = (info["path"], mtime) + if key in _budget_cache: + return _budget_cache[key] + try: + text = path.read_text(encoding="utf-8", errors="replace") + except OSError: + _budget_cache[key] = None + return None + val = parse_budget_from_text(text) + _budget_cache[key] = val + return val + + +def _range_clause(since, until): + where, args = [], [] + if since: + where.append("timestamp >= ?") + args.append(since) + if until: + where.append("timestamp < ?") + args.append(until) + return ((" AND " + " AND ".join(where)) if where else "", args) + + +def _percentile(sorted_xs: list[int], p: int) -> int: + if not sorted_xs: + return 0 + k = (len(sorted_xs) - 1) * (p / 100.0) + lo = int(k) + hi = min(lo + 1, len(sorted_xs) - 1) + if lo == hi: + return sorted_xs[lo] + return int(sorted_xs[lo] * (hi - k) + sorted_xs[hi] * (k - lo)) + + +# User-role messages that are system-injected (not real typing) and must not +# terminate the attribution window. Skill/agent invocations inject the body +# of SKILL.md/AGENT.md as a 20k+ user-role message; other Claude Code +# machinery injects the bracketed tags below. Empirically this covers ~25% +# of non-empty user messages; the other ~75% are the user actually typing. +_META_USER_PREFIXES = ( + "Base directory for this skill:", + "Base directory for this agent:", + "", + "", + "", + "", + "", + "[Request interrupted", +) + + +def skill_costs(db_path, pricing, since=None, until=None) -> dict[str, dict]: + """Return ``{slug: {cost_usd, cost_estimated}}`` — total assistant-side + cost spent within each skill's attribution window. + + Uses the same window-bounds logic as ``skill_actuals`` (exclude sidechain, + cap at next Skill or next real-user-typed message). Groups tokens by + (skill, model) so each bucket is priced with the matching model's rate; + falls back to tier-level pricing for unknown-but-named models. + + Answers "which skill actually costs the most dollars" — a direct + ranking signal that the p50/p95 output view doesn't surface (a skill + with small p50 but many invocations can still dominate monthly spend). + """ + from .pricing import cost_for as _cost_for + + rng, args = _range_clause(since, until) + not_like = " AND ".join( + ["u.prompt_text NOT LIKE ?"] * len(_META_USER_PREFIXES) + ) + like_args = [p + "%" for p in _META_USER_PREFIXES] + sql = f""" + WITH calls AS ( + SELECT session_id, + target AS skill, + timestamp AS start_ts, + LEAD(timestamp) OVER ( + PARTITION BY session_id ORDER BY timestamp + ) AS next_skill_ts + FROM tool_calls + WHERE tool_name = 'Skill' + AND target IS NOT NULL + AND target != '' + {rng} + ), + bounds AS ( + SELECT c.session_id, c.skill, c.start_ts, c.next_skill_ts, + (SELECT MIN(u.timestamp) FROM messages u + WHERE u.session_id = c.session_id + AND u.type = 'user' + AND u.is_sidechain = 0 + AND u.prompt_chars IS NOT NULL + AND u.prompt_chars > 0 + AND u.prompt_text IS NOT NULL + AND u.timestamp > c.start_ts + AND {not_like} + ) AS next_user_ts + FROM calls c + ) + SELECT b.skill, + COALESCE(m.model, 'unknown') AS model, + COALESCE(SUM(m.input_tokens), 0) AS input_tokens, + COALESCE(SUM(m.output_tokens), 0) AS output_tokens, + COALESCE(SUM(m.cache_read_tokens), 0) AS cache_read_tokens, + COALESCE(SUM(m.cache_create_5m_tokens), 0) AS cache_create_5m_tokens, + COALESCE(SUM(m.cache_create_1h_tokens), 0) AS cache_create_1h_tokens + FROM bounds b + LEFT JOIN messages m + ON m.session_id = b.session_id + AND m.type = 'assistant' + AND m.is_sidechain = 0 + AND m.timestamp > b.start_ts + AND (b.next_skill_ts IS NULL OR m.timestamp < b.next_skill_ts) + AND (b.next_user_ts IS NULL OR m.timestamp < b.next_user_ts) + GROUP BY b.skill, m.model + """ + args = [*args, *like_args] + out: dict[str, dict] = {} + with connect(db_path) as conn: + for row in conn.execute(sql, args): + usage = { + "input_tokens": row["input_tokens"], + "output_tokens": row["output_tokens"], + "cache_read_tokens": row["cache_read_tokens"], + "cache_create_5m_tokens": row["cache_create_5m_tokens"], + "cache_create_1h_tokens": row["cache_create_1h_tokens"], + } + c = _cost_for(row["model"] or "", usage, pricing) + entry = out.setdefault(row["skill"], {"cost_usd": 0.0, "cost_estimated": False}) + if c["usd"] is None: + # Model not in pricing and no tier match: flag estimated, skip add. + entry["cost_estimated"] = True + continue + entry["cost_usd"] += c["usd"] + if c["estimated"]: + entry["cost_estimated"] = True + for v in out.values(): + v["cost_usd"] = round(v["cost_usd"], 4) + return out + + +def skill_actuals(db_path, since=None, until=None) -> dict[str, dict]: + """Return ``{slug: {p50, p95, count}}`` of output_tokens per invocation. + + Window boundaries, in priority order: + 1. next Skill call in the same session, + 2. next real-user-typed main-chain message (``prompt_chars > 0`` and + ``prompt_text`` does NOT start with any system-injection prefix), + 3. end of session. + + Sidechain assistant output (subagents, auto-compaction) is excluded — + it is not emitted by the skill itself and would otherwise leak in when + an auto-compact agent fires during the window. + + Note on what ``output_tokens`` counts: the Anthropic API ``output_tokens`` + field includes tool_use JSON blocks and thinking blocks, not just + user-visible text. Skills that declare "Complete in 0 + AND u.prompt_text IS NOT NULL + AND u.timestamp > c.start_ts + AND {not_like} + ) AS next_user_ts + FROM calls c + ) + SELECT b.skill, + COALESCE(SUM(m.output_tokens), 0) AS output_tokens + FROM bounds b + LEFT JOIN messages m + ON m.session_id = b.session_id + AND m.type = 'assistant' + AND m.is_sidechain = 0 + AND m.timestamp > b.start_ts + AND (b.next_skill_ts IS NULL OR m.timestamp < b.next_skill_ts) + AND (b.next_user_ts IS NULL OR m.timestamp < b.next_user_ts) + GROUP BY b.skill, b.session_id, b.start_ts + """ + args = [*args, *like_args] + samples: dict[str, list[int]] = {} + with connect(db_path) as conn: + for row in conn.execute(sql, args): + samples.setdefault(row["skill"], []).append(row["output_tokens"] or 0) + out: dict[str, dict] = {} + for slug, xs in samples.items(): + xs.sort() + out[slug] = { + "p50": _percentile(xs, 50), + "p95": _percentile(xs, 95), + "count": len(xs), + } + return out + + +def skill_subagent_costs(db_path, pricing, since=None, until=None) -> dict[str, dict]: + """Return ``{slug: {cost_usd, cost_estimated, output_tokens}}`` — total + assistant-side cost of all subagent (sidechain) work dispatched by each + skill within its attribution window. + + A subagent is any sidechain conversation with its own ``agent_id`` (the + hash in ``subagents/agent-.jsonl``) except auto-compaction + (``agent_id LIKE 'acompact%'``). Attribution ties each subagent to the + skill window that contains its FIRST sidechain message. + + Window bounds: Skill call → next Skill call in session. This differs + from ``skill_costs``'s window, which ALSO closes at the first real + user-typed message — some interactive orchestrators ask the user a + question mid-execution and THEN dispatch subagents, so closing at + the user's reply would miss the whole point. The next Skill call + remains the correct boundary for "when did this orchestrator hand + off to a new one." + + Nested subagents fall out naturally: the inner subagent's first + sidechain message is itself inside the outer skill's window, so the + inner agent_id attributes to the same orchestrator. + """ + from .pricing import cost_for as _cost_for + + rng, args = _range_clause(since, until) + sql = f""" + WITH calls AS ( + SELECT session_id, + target AS skill, + timestamp AS start_ts, + LEAD(timestamp) OVER ( + PARTITION BY session_id ORDER BY timestamp + ) AS next_skill_ts + FROM tool_calls + WHERE tool_name = 'Skill' + AND target IS NOT NULL + AND target != '' + {rng} + ), + -- Each subagent is identified by (session_id, agent_id). Its "start" + -- is the earliest sidechain message with that agent_id. Auto-compaction + -- agents are excluded by the acompact%-prefix filter. + agent_starts AS ( + SELECT session_id, agent_id, MIN(timestamp) AS start_ts + FROM messages + WHERE is_sidechain = 1 + AND agent_id IS NOT NULL + AND agent_id NOT LIKE 'acompact%' + GROUP BY session_id, agent_id + ), + window_agents AS ( + SELECT c.skill, c.session_id, a.agent_id + FROM calls c + JOIN agent_starts a + ON a.session_id = c.session_id + AND a.start_ts > c.start_ts + AND (c.next_skill_ts IS NULL OR a.start_ts < c.next_skill_ts) + ) + SELECT w.skill, + COALESCE(m.model, 'unknown') AS model, + COALESCE(SUM(m.input_tokens), 0) AS input_tokens, + COALESCE(SUM(m.output_tokens), 0) AS output_tokens, + COALESCE(SUM(m.cache_read_tokens), 0) AS cache_read_tokens, + COALESCE(SUM(m.cache_create_5m_tokens), 0) AS cache_create_5m_tokens, + COALESCE(SUM(m.cache_create_1h_tokens), 0) AS cache_create_1h_tokens + FROM window_agents w + JOIN messages m + ON m.session_id = w.session_id + AND m.agent_id = w.agent_id + AND m.is_sidechain = 1 + AND m.type = 'assistant' + GROUP BY w.skill, m.model + """ + out: dict[str, dict] = {} + with connect(db_path) as conn: + for row in conn.execute(sql, args): + usage = { + "input_tokens": row["input_tokens"], + "output_tokens": row["output_tokens"], + "cache_read_tokens": row["cache_read_tokens"], + "cache_create_5m_tokens": row["cache_create_5m_tokens"], + "cache_create_1h_tokens": row["cache_create_1h_tokens"], + } + c = _cost_for(row["model"] or "", usage, pricing) + entry = out.setdefault(row["skill"], { + "cost_usd": 0.0, + "cost_estimated": False, + "output_tokens": 0, + }) + entry["output_tokens"] += row["output_tokens"] or 0 + if c["usd"] is None: + entry["cost_estimated"] = True + continue + entry["cost_usd"] += c["usd"] + if c["estimated"]: + entry["cost_estimated"] = True + for v in out.values(): + v["cost_usd"] = round(v["cost_usd"], 4) + return out diff --git a/token_dashboard/skills.py b/token_dashboard/skills.py index c1733ab..54990f4 100644 --- a/token_dashboard/skills.py +++ b/token_dashboard/skills.py @@ -14,7 +14,7 @@ import time from pathlib import Path -from typing import Dict, Optional +from typing import Dict, Iterable, Optional _DEFAULT_ROOTS = [ Path.home() / ".claude" / "skills", @@ -91,16 +91,50 @@ def scan_catalog(roots=None) -> Dict[str, dict]: return catalog -_cache: dict = {"at": 0.0, "data": {}} +def _project_skill_roots_from_cwds(cwds: Iterable[str]) -> list[Path]: + """Return the innermost `.claude/skills/` directory for each cwd. + + Matches Claude Code's resolution rule: walk up from cwd and use the first + `.claude/skills/` found, so a nested repo uses its own skills, not a parent's. + """ + roots: set[Path] = set() + for cwd in cwds: + if not cwd: + continue + p = Path(cwd) + for ancestor in (p, *p.parents): + candidate = ancestor / ".claude" / "skills" + if candidate.is_dir(): + roots.add(candidate) + break + return sorted(roots) + + +def _cwds_from_db(db_path) -> list[str]: + from .db import connect + with connect(db_path) as c: + return [r[0] for r in c.execute( + "SELECT DISTINCT cwd FROM messages WHERE cwd IS NOT NULL" + )] + + +_cache: dict = {"at": 0.0, "data": {}, "key": None} _TTL_SECONDS = 60.0 -def cached_catalog() -> Dict[str, dict]: - """scan_catalog() with a simple in-process TTL cache.""" +def cached_catalog(db_path=None) -> Dict[str, dict]: + """scan_catalog() with a simple in-process TTL cache. + + When `db_path` is provided, extra roots are derived from the distinct cwds + in `messages` so project-local `.claude/skills/` directories are included. + """ now = time.time() - if now - _cache["at"] > _TTL_SECONDS: - _cache["data"] = scan_catalog() + key = str(db_path) if db_path else None + if now - _cache["at"] > _TTL_SECONDS or _cache["key"] != key: + extra = _project_skill_roots_from_cwds(_cwds_from_db(db_path)) if db_path else [] + _cache["data"] = scan_catalog(_DEFAULT_ROOTS + extra) _cache["at"] = now + _cache["key"] = key return _cache["data"] diff --git a/web/routes/skills.js b/web/routes/skills.js index 3b6710b..eaad1be 100644 --- a/web/routes/skills.js +++ b/web/routes/skills.js @@ -59,24 +59,34 @@ export default async function (root) {

All skills

-

"Tokens per call" is the size of the skill's SKILL.md file — what Claude Code loads into context each time the skill is invoked.

+

"Tokens per call" is the size of the skill's SKILL.md file — what Claude Code loads into context each time. "Budget" / "p50 out" / "p95 out" track the skill's output footprint: budget is parsed from the SKILL.md body; p50/p95 sum output_tokens from the Skill call until the user types again or another Skill runs, excluding sidechain subagents. Note that output_tokens includes tool_use JSON, so a 2-5× gap over a text-only budget can be tool-call overhead. Red means p50 exceeds budget by more than 20%. "Total $" is the cost the skill itself emitted (input + output + cache) across this range. "Total inc. subagents" adds the cost of any Task/Agent-dispatched subagents whose parent chain traces back into the skill's window — use it to see orchestrator skills (anything that dispatches subagents) at their true weight.

+ + + + + - ${skills.map(s => ` + ${[...skills].sort((a, b) => ((b.total_with_subagents_usd ?? b.total_cost_usd) || 0) - ((a.total_with_subagents_usd ?? a.total_cost_usd) || 0)).map(s => ` + + + + + - `).join('') || ''} + `).join('') || ''}
skill invocations tokens per callbudgetp50 outp95 outtotal $total inc. subagents sessions last used
${fmt.htmlSafe(s.skill)} ${fmt.int(s.invocations)} ${s.tokens_per_call == null ? '—' : fmt.int(s.tokens_per_call)}${s.budget_output_tokens == null ? '—' : fmt.int(s.budget_output_tokens)}${s.p50_output_tokens == null ? '—' : (s.over_budget ? `${fmt.int(s.p50_output_tokens)}` : fmt.int(s.p50_output_tokens))}${s.p95_output_tokens == null ? '—' : fmt.int(s.p95_output_tokens)}${s.total_cost_usd == null ? '—' : fmt.usd(s.total_cost_usd)}${s.cost_estimated ? '*' : ''}${s.total_with_subagents_usd == null ? '—' : (s.subagent_cost_usd ? `${fmt.usd(s.total_with_subagents_usd)}` : fmt.usd(s.total_with_subagents_usd))} ${fmt.int(s.sessions)} ${fmt.ts(s.last_used)}
no skills invoked in this range
no skills invoked in this range