From 560df2e99a299cc1bdfc8c849397cce617857f55 Mon Sep 17 00:00:00 2001 From: Alexander Holbreich Date: Mon, 7 Sep 2026 20:56:02 +0200 Subject: [PATCH 1/2] Add a DeepSeek usage collector for the agents panel --- bin/omarchy-agent-usage-deepseek | 563 ++++++++++++++++++ shell/plugins/agents/README.md | 8 +- shell/plugins/agents/assets/deepseek.svg | 4 + .../agent-usage-deepseek-scanner-test.sh | 201 +++++++ 4 files changed, 775 insertions(+), 1 deletion(-) create mode 100755 bin/omarchy-agent-usage-deepseek create mode 100644 shell/plugins/agents/assets/deepseek.svg create mode 100755 test/shell.d/agent-usage-deepseek-scanner-test.sh diff --git a/bin/omarchy-agent-usage-deepseek b/bin/omarchy-agent-usage-deepseek new file mode 100755 index 00000000000..2d2fac5e7f5 --- /dev/null +++ b/bin/omarchy-agent-usage-deepseek @@ -0,0 +1,563 @@ +#!/usr/bin/python3 +# omarchy:summary=Print the DeepSeek usage record as JSON +# omarchy:args=[--force] [--limits-only] +# omarchy:hidden=true +"""Collect DeepSeek usage into one display-ready JSON record. + +Local stats come from pi/omp session files whose assistant messages ran on the +deepseek provider, and from opencode sessions that ran on a deepseek provider. +DeepSeek billing is prepaid, so the record carries the live credit balance from +GET https://api.deepseek.com/user/balance when a key is available, plus the +optional funded/spent detail derived from ~/.config/omarchy/agents/deepseek.json +(the same shape the fireworks collector uses). The agents panel only ever reads +the JSON this prints. + +Token accounting deliberately matches pi's own totals: input + output + +cacheRead + cacheWrite, with the separate "reasoning" field left out, exactly +like the codex collector treats pi sessions. +""" + +from __future__ import annotations + +import argparse +import json +import os +import shutil +import subprocess +import sys +import urllib.error +import urllib.request +from datetime import datetime, timedelta, timezone +from decimal import Decimal, InvalidOperation +from pathlib import Path +from typing import Any + +AGENT_ID = "deepseek" +AGENT_NAME = "DeepSeek" +AUTH_HELP = ( + "Add a DeepSeek key in pi, set DEEPSEEK_API_KEY, " + "or sign in to DeepSeek in opencode." +) +API_BASE_URL = "https://api.deepseek.com" + + +class DeepSeekError(Exception): + pass + + +def number(value: Any) -> int: + try: + return int(value or 0) + except Exception: + return 0 + + +def model_name(raw: Any) -> str: + value = str(raw or AGENT_ID) + return value if value else AGENT_ID + + +def local_day(value: Any) -> str: + if value is None: + return datetime.now().strftime("%Y-%m-%d") + if isinstance(value, (int, float)): + # pi message timestamps are milliseconds; other clients use seconds. + if value > 10_000_000_000: + value = value / 1000 + return datetime.fromtimestamp(value).strftime("%Y-%m-%d") + text = str(value) + try: + if text.endswith("Z"): + dt = datetime.fromisoformat(text[:-1] + "+00:00") + else: + dt = datetime.fromisoformat(text) + if dt.tzinfo is not None: + dt = dt.astimezone() + return dt.strftime("%Y-%m-%d") + except Exception: + return datetime.now().strftime("%Y-%m-%d") + + +def runtime_env() -> dict[str, str]: + home = str(Path.home()) + path_parts = [ + os.environ.get("PATH", ""), + f"{home}/.local/bin", + f"{home}/.npm-global/bin", + f"{home}/.local/share/mise/shims", + ] + env = os.environ.copy() + env["PATH"] = os.pathsep.join(part for part in path_parts if part) + return env + + +ENV = runtime_env() + + +def find_command(name: str) -> str | None: + return shutil.which(name, path=ENV.get("PATH")) + + +def empty_bucket() -> dict[str, int]: + return { + "inputTokens": 0, + "outputTokens": 0, + "cacheReadInputTokens": 0, + "cacheCreationInputTokens": 0, + } + + +def empty_stats() -> dict[str, Any]: + return { + "todayPrompts": 0, + "todaySessions": 0, + "todayTotalTokens": 0, + "todayTokensByModel": {}, + "recentDays": [], + "totalPrompts": 0, + "totalSessions": 0, + "activeDays": 0, + "activeDates": [], + "modelUsage": {}, + } + + +def base_record(**overrides: Any) -> dict[str, Any]: + record: dict[str, Any] = { + "schemaVersion": 1, + "id": AGENT_ID, + "name": AGENT_NAME, + "updatedAt": datetime.now(timezone.utc).isoformat(), + "ready": True, + "hasLocalStats": True, + # Prompt and session counts are real: every pi session message carries + # them, so today's tooltip may show "N prompts · M sessions". + "hasPromptStats": True, + "tierLabel": "Prepaid", + "usageStatusText": "", + "authHelpText": "", + "limits": [], + } + record.update(empty_stats()) + record.update(overrides) + return record + + +def pi_session_roots() -> list[Path]: + return [ + Path.home() / ".pi" / "agent" / "sessions", + Path.home() / ".omp" / "agent" / "sessions", + ] + + +def scan_pi_sessions() -> dict[str, Any]: + """Aggregate pi/omp session messages that ran on the deepseek provider.""" + now = datetime.now() + today = now.strftime("%Y-%m-%d") + recent_dates = [ + (now - timedelta(days=offset)).strftime("%Y-%m-%d") + for offset in range(6, -1, -1) + ] + recent = {day: 0 for day in recent_dates} + + today_prompts = 0 + today_total_tokens = 0 + today_sessions: set[str] = set() + today_tokens_by_model: dict[str, int] = {} + total_prompts = 0 + total_sessions: set[str] = set() + active_days: set[str] = set() + model_usage: dict[str, dict[str, int]] = {} + seen_messages: set[str] = set() + + rg = find_command("rg") or "rg" + for root in pi_session_roots(): + if not root.exists(): + continue + try: + proc = subprocess.Popen( + [rg, "--json", "-e", r'"provider"\s*:\s*"deepseek"', str(root)], + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + errors="replace", + env=ENV, + ) + except FileNotFoundError: + continue + + assert proc.stdout is not None + for raw in proc.stdout: + try: + event = json.loads(raw) + if event.get("type") != "match": + continue + line = event.get("data", {}).get("lines", {}).get("text", "") + path = event.get("data", {}).get("path", {}).get("text", "pi-session") + entry = json.loads(line) + except Exception: + continue + + if entry.get("type") != "message": + continue + message_key = path + ":" + str(entry.get("id") or "") + if message_key in seen_messages: + continue + seen_messages.add(message_key) + message = entry.get("message") or {} + if message.get("role") != "assistant": + continue + if str(message.get("provider") or "") != "deepseek": + continue + + usage = message.get("usage") or {} + if not usage: + continue + total = number(usage.get("totalTokens")) + input_tokens = number(usage.get("input")) + output_tokens = number(usage.get("output")) + cache_read = number(usage.get("cacheRead")) + cache_write = number(usage.get("cacheWrite")) + if total and not (input_tokens or output_tokens or cache_read or cache_write): + input_tokens = total + if not (input_tokens or output_tokens or cache_read or cache_write): + continue + + day = local_day(entry.get("timestamp") or message.get("timestamp")) + model = model_name(message.get("model")) + bucket = model_usage.setdefault(model, empty_bucket()) + bucket["inputTokens"] += input_tokens + bucket["outputTokens"] += output_tokens + bucket["cacheReadInputTokens"] += cache_read + bucket["cacheCreationInputTokens"] += cache_write + + total_tokens = input_tokens + output_tokens + cache_read + cache_write + total_prompts += 1 + total_sessions.add(path) + active_days.add(day) + if day in recent: + recent[day] += total_tokens + if day == today: + today_prompts += 1 + today_sessions.add(path) + today_total_tokens += total_tokens + today_tokens_by_model[model] = today_tokens_by_model.get(model, 0) + total_tokens + + try: + proc.wait(timeout=1) + except Exception: + proc.kill() + + return { + "todayPrompts": today_prompts, + "todaySessions": len(today_sessions), + "todayTotalTokens": today_total_tokens, + "todayTokensByModel": today_tokens_by_model, + "recentDays": [{"date": day, "messageCount": recent[day]} for day in recent_dates], + "totalPrompts": total_prompts, + "totalSessions": len(total_sessions), + "activeDays": len(active_days), + "activeDates": sorted(active_days), + "modelUsage": model_usage, + } + + +def opencode_db_path() -> Path: + data_home = Path(os.environ.get("XDG_DATA_HOME") or (Path.home() / ".local" / "share")) + return data_home / "opencode" / "opencode.db" + + +def scan_opencode_sessions() -> tuple[dict[str, Any], bool]: + """opencode messages on a deepseek provider, mirroring the codex scan. + + Returns the stats and whether the scan ran to completion: a scan cut short + by a database error still contributes what it read, but a caller that caches + must not treat it as the whole story. Read-only: opencode may be writing + right now. + """ + db = opencode_db_path() + if not db.is_file(): + return empty_stats(), True + + import sqlite3 + + now = datetime.now() + today = now.strftime("%Y-%m-%d") + recent_dates = [ + (now - timedelta(days=offset)).strftime("%Y-%m-%d") + for offset in range(6, -1, -1) + ] + recent = {day: 0 for day in recent_dates} + + today_prompts = 0 + today_total_tokens = 0 + today_sessions: set[str] = set() + today_tokens_by_model: dict[str, int] = {} + total_prompts = 0 + total_sessions: set[str] = set() + active_days: set[str] = set() + model_usage: dict[str, dict[str, int]] = {} + + try: + conn = sqlite3.connect(db.resolve().as_uri() + "?mode=ro", uri=True, timeout=2) + except sqlite3.Error: + return empty_stats(), False + try: + conn.execute("PRAGMA query_only = ON") + # The json_extract conditions are the authority; the LIKE gates are pure + # acceleration that skip rows whose JSON cannot contain the pair, the same + # trade the codex collector makes on this table. + for session_id, raw in conn.execute( + "SELECT session_id, data FROM message" + " WHERE data LIKE '%\"role\"%:%\"assistant\"%'" + " AND data LIKE '%\"providerID\"%:%\"deepseek\"%'" + " AND CASE WHEN json_valid(data) THEN json_extract(data, '$.role') END = 'assistant'" + " AND CASE WHEN json_valid(data) THEN json_extract(data, '$.providerID') END = 'deepseek'" + ): + try: + entry = json.loads(raw) + if not isinstance(entry, dict) or entry.get("role") != "assistant": + continue + if str(entry.get("providerID") or "") != "deepseek": + continue + tokens = entry.get("tokens") or {} + cache = tokens.get("cache") or {} + input_tokens = number(tokens.get("input")) + # opencode keeps thinking tokens out of output; both are generated. + output_tokens = number(tokens.get("output")) + number(tokens.get("reasoning")) + cache_read = number(cache.get("read")) + cache_write = number(cache.get("write")) + if not (input_tokens or output_tokens or cache_read or cache_write): + continue + day = local_day((entry.get("time") or {}).get("created")) + model = model_name(str(entry.get("modelID") or "").rstrip("/").split("/")[-1]) + bucket = model_usage.setdefault(model, empty_bucket()) + bucket["inputTokens"] += input_tokens + bucket["outputTokens"] += output_tokens + bucket["cacheReadInputTokens"] += cache_read + bucket["cacheCreationInputTokens"] += cache_write + total_tokens = input_tokens + output_tokens + cache_read + cache_write + total_prompts += 1 + total_sessions.add("opencode:" + str(session_id)) + active_days.add(day) + if day in recent: + recent[day] += total_tokens + if day == today: + today_prompts += 1 + today_sessions.add("opencode:" + str(session_id)) + today_total_tokens += total_tokens + today_tokens_by_model[model] = today_tokens_by_model.get(model, 0) + total_tokens + except Exception: + continue + except sqlite3.Error: + return empty_stats(), False + finally: + conn.close() + + return { + "todayPrompts": today_prompts, + "todaySessions": len(today_sessions), + "todayTotalTokens": today_total_tokens, + "todayTokensByModel": today_tokens_by_model, + "recentDays": [{"date": day, "messageCount": recent[day]} for day in recent_dates], + "totalPrompts": total_prompts, + "totalSessions": len(total_sessions), + "activeDays": len(active_days), + "activeDates": sorted(active_days), + "modelUsage": model_usage, + }, True + + +def merge_stats(first: dict[str, Any], second: dict[str, Any]) -> dict[str, Any]: + """Sum two stats dicts. recentDays arrive as parallel per-day lists.""" + + def day_map(recent_days: list[dict[str, Any]]) -> dict[str, int]: + out: dict[str, int] = {} + for day in recent_days: + out[str(day.get("date") or "")] = number(day.get("messageCount")) + return out + + days = day_map(first.get("recentDays") or []) + for date, count in day_map(second.get("recentDays") or []).items(): + days[date] = days.get(date, 0) + count + + def add_models(target: dict[str, dict[str, int]], source: Any) -> None: + for model, bucket in (source or {}).items(): + if not isinstance(bucket, dict): + continue + slot = target.setdefault(str(model), empty_bucket()) + slot["inputTokens"] += number(bucket.get("inputTokens")) + slot["outputTokens"] += number(bucket.get("outputTokens")) + slot["cacheReadInputTokens"] += number(bucket.get("cacheReadInputTokens")) + slot["cacheCreationInputTokens"] += number(bucket.get("cacheCreationInputTokens")) + + def add_today(target: dict[str, int], source: Any) -> None: + for model, tokens in (source or {}).items(): + target[str(model)] = target.get(str(model), 0) + number(tokens) + + model_usage: dict[str, dict[str, int]] = {} + add_models(model_usage, first.get("modelUsage")) + add_models(model_usage, second.get("modelUsage")) + + today_by_model: dict[str, int] = {} + add_today(today_by_model, first.get("todayTokensByModel")) + add_today(today_by_model, second.get("todayTokensByModel")) + + return { + "todayPrompts": number(first.get("todayPrompts")) + number(second.get("todayPrompts")), + "todaySessions": number(first.get("todaySessions")) + number(second.get("todaySessions")), + "todayTotalTokens": number(first.get("todayTotalTokens")) + number(second.get("todayTotalTokens")), + "todayTokensByModel": today_by_model, + "recentDays": [{"date": day, "messageCount": days[day]} for day in sorted(days)], + "totalPrompts": number(first.get("totalPrompts")) + number(second.get("totalPrompts")), + "totalSessions": number(first.get("totalSessions")) + number(second.get("totalSessions")), + "activeDays": number(first.get("activeDays")) + number(second.get("activeDays")), + "activeDates": sorted(set(first.get("activeDates") or []) | set(second.get("activeDates") or [])), + "modelUsage": model_usage, + } + + +# -------------------------------------------------------------- credentials + +def read_key_from_dict(parsed: Any, provider: str) -> str: + if not isinstance(parsed, dict): + return "" + entry = parsed.get(provider) + if not isinstance(entry, dict): + return "" + return str(entry.get("key") or entry.get("apiKey") or entry.get("api_key") or "").strip() + + +def read_json_key(path: Path, provider: str) -> str: + try: + return read_key_from_dict(json.loads(path.read_text()), provider) + except (OSError, json.JSONDecodeError): + return "" + + +def api_key() -> str: + env_key = str(os.environ.get("DEEPSEEK_API_KEY", "")).strip() + if env_key: + return env_key + key = read_json_key(Path.home() / ".pi" / "agent" / "auth.json", "deepseek") + if key: + return key + data_home = Path(os.environ.get("XDG_DATA_HOME") or (Path.home() / ".local" / "share")) + return read_json_key(data_home / "opencode" / "auth.json", "deepseek") + + +def config_path() -> Path: + config_home = Path(os.environ.get("XDG_CONFIG_HOME") or (Path.home() / ".config")) + return config_home / "omarchy" / "agents" / "deepseek.json" + + +def read_config() -> dict[str, Any]: + try: + parsed = json.loads(config_path().read_text()) + return parsed if isinstance(parsed, dict) else {} + except (OSError, json.JSONDecodeError): + return {} + + +# ----------------------------------------------------------------- balance + +def fetch_balance(api_base_url: str, key: str) -> dict[str, Any]: + """GET /user/balance: total_balance, granted_balance, topped_up_balance.""" + url = api_base_url.rstrip("/") + "/user/balance" + request = urllib.request.Request( + url, + headers={"Authorization": "Bearer " + key, "Accept": "application/json"}, + ) + try: + with urllib.request.urlopen(request, timeout=15) as response: + payload = json.load(response) + except urllib.error.HTTPError as error: + raise DeepSeekError(f"DeepSeek balance endpoint returned HTTP {error.code}") + except urllib.error.URLError as error: + raise DeepSeekError("Could not reach the DeepSeek API") from error + except (json.JSONDecodeError, TimeoutError) as error: + raise DeepSeekError("DeepSeek returned an invalid balance response") from error + + if not isinstance(payload, dict): + raise DeepSeekError("DeepSeek returned an invalid balance response") + infos = payload.get("balance_infos") + if not isinstance(infos, list) or not infos or not isinstance(infos[0], dict): + raise DeepSeekError("DeepSeek balance response carried no balance_infos") + first = infos[0] + + def money(value: Any) -> Decimal: + try: + return Decimal(str(value or "0")) + except InvalidOperation: + return Decimal("0") + + return { + "remaining": money(first.get("total_balance")), + "granted": money(first.get("granted_balance")), + "toppedUp": money(first.get("topped_up_balance")), + "currency": str(first.get("currency") or "CNY"), + } + + +def balance_record(key: str, api_base_url: str) -> dict[str, Any] | None: + config = read_config() + try: + funded = Decimal(str(config.get("fundedAmount") or "0")) + except InvalidOperation: + funded = Decimal("0") + if not funded.is_finite() or funded < 0: + funded = Decimal("0") + + try: + live = fetch_balance(api_base_url, key) + except DeepSeekError as error: + print(f"omarchy-agent-usage-deepseek: balance unavailable ({error})", file=sys.stderr) + return None + + remaining = live["remaining"] + spent = max(Decimal("0"), funded - remaining) if funded > 0 else Decimal("0") + return { + "remaining": float(max(Decimal("0"), remaining)), + "funded": float(funded), + "spent": float(spent), + "currency": str(live["currency"] or "CNY"), + # funded/spent derive from the configured fundedAmount, not DeepSeek's + # own ledger, so they are labelled estimates; remaining is live. + "estimated": funded > 0, + } + + +def scan(api_base_url: str) -> dict[str, Any]: + stats = merge_stats(scan_pi_sessions(), scan_opencode_sessions()[0]) + record = base_record() + record.update(stats) + + key = api_key() + if key: + record["balance"] = balance_record(key, api_base_url) + else: + print("omarchy-agent-usage-deepseek: no DeepSeek key found; balance omitted", file=sys.stderr) + return record + + +def main() -> int: + parser = argparse.ArgumentParser(description="Print the DeepSeek usage record as JSON") + # Stats are local file scans and the balance is one API call; there is no + # cache to force past and no faster limits-only path. The flags exist so + # every collector accepts the same invocation. + parser.add_argument("--force", action="store_true") + parser.add_argument("--limits-only", action="store_true") + parser.add_argument("--api-base-url", default=os.environ.get("DEEPSEEK_API_BASE_URL", API_BASE_URL)) + args = parser.parse_args() + + try: + record = scan(args.api_base_url) + except Exception as error: + record = base_record(usageStatusText="DeepSeek unavailable", authHelpText="DeepSeek usage scan failed") + print(f"omarchy-agent-usage-deepseek: {type(error).__name__}: {error}", file=sys.stderr) + print(json.dumps(record, separators=(",", ":"))) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/shell/plugins/agents/README.md b/shell/plugins/agents/README.md index 51ad4e43c2d..c0d2b175229 100644 --- a/shell/plugins/agents/README.md +++ b/shell/plugins/agents/README.md @@ -55,6 +55,7 @@ light surfaces — and the bar glyph stands in when there is none. | `claude` | Anthropic's OAuth usage endpoint (5-hour session + 7-day weekly) | `~/.claude/projects` transcripts, opencode sessions on an Anthropic provider, plus `stats-cache.json` and `history.jsonl` as fallback | | `codex` | The Codex app-server RPC | native Codex CLI session files (plus pi and opencode sessions) | | `fireworks` | Estimated prepaid balance: configured funding minus rated account costs | Fireworks billing API, grouped by day and model for the last 30 days | +| `deepseek` | Live prepaid balance from `GET https://api.deepseek.com/user/balance`, with the optional `fundedAmount` estimate for the funded-versus-spent line | pi/omp session files whose assistant messages ran on the deepseek provider, plus opencode sessions on a deepseek provider | Claude limits need a signed-in CLI; without credentials the panel says so and falls back to local stats only. A non-default Claude directory is honored via @@ -62,7 +63,12 @@ falls back to local stats only. A non-default Claude directory is honored via `FIREWORKS_API_KEY` and `FIREWORKS_ACCOUNT_ID` first, then `~/.fireworks/auth.ini` (which `firectl set-api-key` creates), then the key opencode stores in `~/.local/share/opencode/auth.json` when Fireworks is -signed in there. +signed in there. DeepSeek reads `DEEPSEEK_API_KEY` first, then the key pi +stores in `~/.pi/agent/auth.json`, then the key opencode stores in its own +`auth.json` when DeepSeek is signed in there. Its `fundedAmount` estimate +lives in `~/.config/omarchy/agents/deepseek.json`, the same shape the +fireworks collector uses; without it the panel shows the live remaining +credit and skips the funded-versus-spent line. ### Fireworks balance diff --git a/shell/plugins/agents/assets/deepseek.svg b/shell/plugins/agents/assets/deepseek.svg new file mode 100644 index 00000000000..57f9b482977 --- /dev/null +++ b/shell/plugins/agents/assets/deepseek.svg @@ -0,0 +1,4 @@ + + DeepSeek + + diff --git a/test/shell.d/agent-usage-deepseek-scanner-test.sh b/test/shell.d/agent-usage-deepseek-scanner-test.sh new file mode 100755 index 00000000000..1b645af0a09 --- /dev/null +++ b/test/shell.d/agent-usage-deepseek-scanner-test.sh @@ -0,0 +1,201 @@ +#!/bin/bash + +source "$(dirname "$0")/base-test.sh" + +require_command jq +require_command python3 + +TEST_HOME=$(mktemp -d) +trap 'rm -rf "$TEST_HOME"' EXIT + +timestamp="$(date +%Y-%m-%d)T12:00:00Z" + +# ---------------------------------------------------------------- pi and omp + +PI_HOME=$(mktemp -d) +trap 'rm -rf "$TEST_HOME" "$PI_HOME"' EXIT +mkdir -p "$PI_HOME/.pi/agent/sessions/project" "$PI_HOME/.omp/agent/sessions/project" + +cat >"$PI_HOME/.pi/agent/sessions/project/pi.jsonl" <"$PI_HOME/.omp/agent/sessions/project/omp.jsonl" </dev/null) + +[[ $(jq -r '.todayTotalTokens' <<<"$result") == "49" ]] || + fail "DeepSeek collector counts usage from pi and omp sessions" "$result" +pass "DeepSeek collector counts usage from pi and omp sessions" + +[[ $(jq -c '.modelUsage' <<<"$result") == '{"deepseek-pi":{"inputTokens":10,"outputTokens":4,"cacheReadInputTokens":3,"cacheCreationInputTokens":2},"deepseek-omp":{"inputTokens":20,"outputTokens":5,"cacheReadInputTokens":4,"cacheCreationInputTokens":1}}' ]] || + fail "DeepSeek collector filters pi and omp sessions to DeepSeek providers" "$result" +pass "DeepSeek collector filters pi and omp sessions to DeepSeek providers" + +[[ $(jq -r '"\(.todayPrompts):\(.todaySessions)"' <<<"$result") == "2:2" ]] || + fail "DeepSeek collector counts each message once per session" "$result" +pass "DeepSeek collector counts each message once per session" + +# Without a key the record must still be complete and hidden-by-default: +# the panel shows local stats and simply omits the balance section. +[[ $(jq -r '.id + ":" + (.ready|tostring) + ":" + (.hasPromptStats|tostring) + ":" + (has("balance")|tostring)' <<<"$result") == "deepseek:true:true:false" ]] || + fail "DeepSeek collector prints a valid record without a key" "$result" +pass "DeepSeek collector prints a valid record without a key" + +# ---------------------------------------------------------------- opencode + +OPENCODE_HOME=$(mktemp -d) +trap 'rm -rf "$TEST_HOME" "$PI_HOME" "$OPENCODE_HOME"' EXIT + +python3 - "$OPENCODE_HOME/.local/share/opencode/opencode.db" <<'PY' +import json +import sqlite3 +import sys +import time +from pathlib import Path + +db = Path(sys.argv[1]) +db.parent.mkdir(parents=True, exist_ok=True) +conn = sqlite3.connect(db) +conn.execute("CREATE TABLE message (id text PRIMARY KEY, session_id text NOT NULL, time_created integer NOT NULL, time_updated integer NOT NULL, data text NOT NULL)") +now_ms = int(time.time() * 1000) + +def message(id, provider, model, input=0, output=0, reasoning=0, read=0, write=0): + return (id, "ses_1", now_ms, now_ms, json.dumps({ + "role": "assistant", + "providerID": provider, + "modelID": model, + "tokens": {"input": input, "output": output, "reasoning": reasoning, "cache": {"read": read, "write": write}}, + "time": {"created": now_ms}, + })) + +conn.executemany( + "INSERT INTO message VALUES (?, ?, ?, ?, ?)", + [ + message("m-1", "deepseek", "deepseek-model", input=100, output=20, reasoning=5, read=30, write=2), + message("m-2", "openai", "gpt-model", input=999, output=999), + message("m-3", "anthropic", "claude-model", input=999, output=999), + ], +) +conn.commit() +conn.close() +PY + +result=$(HOME="$OPENCODE_HOME" XDG_DATA_HOME="$OPENCODE_HOME/.local/share" \ + "$ROOT/bin/omarchy-agent-usage-deepseek" 2>/dev/null) + +[[ $(jq -r '.todayTotalTokens' <<<"$result") == "157" ]] || + fail "DeepSeek collector counts opencode messages on a DeepSeek provider" "$result" +pass "DeepSeek collector counts opencode messages on a DeepSeek provider" + +[[ $(jq -c '.modelUsage["deepseek-model"]' <<<"$result") == '{"inputTokens":100,"outputTokens":25,"cacheReadInputTokens":30,"cacheCreationInputTokens":2}' ]] || + fail "DeepSeek collector keeps reasoning with output and cache apart in opencode" "$result" +pass "DeepSeek collector keeps reasoning with output and cache apart in opencode" + +# ---------------------------------------------------------------- balance + +result=$(python3 - "$ROOT/bin/omarchy-agent-usage-deepseek" "$TEST_HOME/.config" <<'PY' +import importlib.machinery +import importlib.util +import json +import os +import sys +from decimal import Decimal +from pathlib import Path + +collector_path = str(Path(sys.argv[1])) +os.environ["XDG_CONFIG_HOME"] = sys.argv[2] + +loader = importlib.machinery.SourceFileLoader("deepseek_collector", collector_path) +spec = importlib.util.spec_from_loader(loader.name, loader) +scanner = importlib.util.module_from_spec(spec) +loader.exec_module(scanner) +real_fetch_balance = scanner.fetch_balance + +config_dir = Path(sys.argv[2]) / "omarchy" / "agents" +config_dir.mkdir(parents=True, exist_ok=True) +(config_dir / "deepseek.json").write_text(json.dumps({"fundedAmount": 20})) + +scanner.fetch_balance = lambda base_url, key: { + "remaining": Decimal("8.15"), + "granted": Decimal("0"), + "toppedUp": Decimal("8.15"), + "currency": "USD", +} +balance = scanner.balance_record("test-key", "https://example.invalid") +summary = {"withConfig": balance} + +scanner.fetch_balance = lambda base_url, key: (_ for _ in ()).throw( + scanner.DeepSeekError("balance endpoint unavailable") +) +import contextlib +import io +with contextlib.redirect_stderr(io.StringIO()): + summary["onFailure"] = scanner.balance_record("test-key", "https://example.invalid") +scanner.fetch_balance = real_fetch_balance + +# The real request must carry the bearer key and hit /user/balance. +captured = {} + +class FakeResponse: + def __init__(self, payload): + self.payload = payload + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def read(self): + return json.dumps({ + "is_available": True, + "balance_infos": [{ + "currency": "CNY", + "total_balance": "42.50", + "granted_balance": "2.50", + "topped_up_balance": "40.00", + }], + }).encode("utf-8") + +import urllib.request +real_urlopen = urllib.request.urlopen +urllib.request.urlopen = lambda request, timeout=15: ( + captured.update({"url": request.full_url, "auth": request.get_header("Authorization")}) + or FakeResponse(None) +) +try: + live = scanner.fetch_balance("https://api.deepseek.com", "key-123") +finally: + urllib.request.urlopen = real_urlopen + +summary["liveShape"] = { + "remaining": float(live["remaining"]), + "granted": float(live["granted"]), + "toppedUp": float(live["toppedUp"]), + "currency": live["currency"], +} +summary["requestShape"] = {"url": captured.get("url"), "auth": captured.get("auth")} +print(json.dumps(summary, separators=(",", ":"))) +PY +) + +[[ $(jq -c '.withConfig | {remaining, funded, spent, currency, estimated}' <<<"$result") == '{"remaining":8.15,"funded":20.0,"spent":11.85,"currency":"USD","estimated":true}' ]] || + fail "DeepSeek collector derives the estimated ledger from fundedAmount" "$result" +pass "DeepSeek collector derives the estimated ledger from fundedAmount" + +[[ $(jq -r '.onFailure' <<<"$result") == "null" ]] || + fail "DeepSeek collector omits the balance when the endpoint fails" "$result" +pass "DeepSeek collector omits the balance when the endpoint fails" + +[[ $(jq -c '.liveShape' <<<"$result") == '{"remaining":42.5,"granted":2.5,"toppedUp":40.0,"currency":"CNY"}' ]] || + fail "DeepSeek collector parses the live balance response" "$result" +pass "DeepSeek collector parses the live balance response" + +[[ $(jq -c '.requestShape' <<<"$result") == '{"url":"https://api.deepseek.com/user/balance","auth":"Bearer key-123"}' ]] || + fail "DeepSeek collector requests /user/balance with the bearer key" "$result" +pass "DeepSeek collector requests /user/balance with the bearer key" From 4a2a1026baf68b63f006360d0d24def242ec6db1 Mon Sep 17 00:00:00 2001 From: Alexander Holbreich Date: Sun, 27 Sep 2026 16:45:09 +0200 Subject: [PATCH 2/2] Count shared DeepSeek days once and keep refreshes quiet pi and opencode now feed one accumulator, so a day active in both is one active day instead of two. A failed balance lookup lands in the record's status text the way Fireworks reports it, and a machine without a DeepSeek key no longer logs a line to the shell on every refresh. Lists deepseek in the plugin's provider defaults. Co-Authored-By: Claude Opus 5.5 --- bin/omarchy-agent-usage-deepseek | 253 ++++++------------ shell/plugins/agents/manifest.json | 5 +- .../agent-usage-deepseek-scanner-test.sh | 29 +- 3 files changed, 107 insertions(+), 180 deletions(-) diff --git a/bin/omarchy-agent-usage-deepseek b/bin/omarchy-agent-usage-deepseek index 2d2fac5e7f5..71efa7c2667 100755 --- a/bin/omarchy-agent-usage-deepseek +++ b/bin/omarchy-agent-usage-deepseek @@ -34,10 +34,6 @@ from typing import Any AGENT_ID = "deepseek" AGENT_NAME = "DeepSeek" -AUTH_HELP = ( - "Add a DeepSeek key in pi, set DEEPSEEK_API_KEY, " - "or sign in to DeepSeek in opencode." -) API_BASE_URL = "https://api.deepseek.com" @@ -143,6 +139,64 @@ def base_record(**overrides: Any) -> dict[str, Any]: return record +class Stats: + """Token and activity totals shared by every local source. + + Sessions and days are sets, so a day active in both pi and opencode counts + once toward activeDays. + """ + + def __init__(self) -> None: + now = datetime.now() + self.today = now.strftime("%Y-%m-%d") + self.recent_dates = [ + (now - timedelta(days=offset)).strftime("%Y-%m-%d") + for offset in range(6, -1, -1) + ] + self.recent = {day: 0 for day in self.recent_dates} + self.today_prompts = 0 + self.today_total_tokens = 0 + self.today_sessions: set[str] = set() + self.today_tokens_by_model: dict[str, int] = {} + self.total_prompts = 0 + self.total_sessions: set[str] = set() + self.active_days: set[str] = set() + self.model_usage: dict[str, dict[str, int]] = {} + + def add(self, session: str, day: str, model: str, input_tokens: int, output_tokens: int, cache_read: int, cache_write: int) -> None: + bucket = self.model_usage.setdefault(model, empty_bucket()) + bucket["inputTokens"] += input_tokens + bucket["outputTokens"] += output_tokens + bucket["cacheReadInputTokens"] += cache_read + bucket["cacheCreationInputTokens"] += cache_write + + total_tokens = input_tokens + output_tokens + cache_read + cache_write + self.total_prompts += 1 + self.total_sessions.add(session) + self.active_days.add(day) + if day in self.recent: + self.recent[day] += total_tokens + if day == self.today: + self.today_prompts += 1 + self.today_sessions.add(session) + self.today_total_tokens += total_tokens + self.today_tokens_by_model[model] = self.today_tokens_by_model.get(model, 0) + total_tokens + + def record(self) -> dict[str, Any]: + return { + "todayPrompts": self.today_prompts, + "todaySessions": len(self.today_sessions), + "todayTotalTokens": self.today_total_tokens, + "todayTokensByModel": self.today_tokens_by_model, + "recentDays": [{"date": day, "messageCount": self.recent[day]} for day in self.recent_dates], + "totalPrompts": self.total_prompts, + "totalSessions": len(self.total_sessions), + "activeDays": len(self.active_days), + "activeDates": sorted(self.active_days), + "modelUsage": self.model_usage, + } + + def pi_session_roots() -> list[Path]: return [ Path.home() / ".pi" / "agent" / "sessions", @@ -150,24 +204,8 @@ def pi_session_roots() -> list[Path]: ] -def scan_pi_sessions() -> dict[str, Any]: - """Aggregate pi/omp session messages that ran on the deepseek provider.""" - now = datetime.now() - today = now.strftime("%Y-%m-%d") - recent_dates = [ - (now - timedelta(days=offset)).strftime("%Y-%m-%d") - for offset in range(6, -1, -1) - ] - recent = {day: 0 for day in recent_dates} - - today_prompts = 0 - today_total_tokens = 0 - today_sessions: set[str] = set() - today_tokens_by_model: dict[str, int] = {} - total_prompts = 0 - total_sessions: set[str] = set() - active_days: set[str] = set() - model_usage: dict[str, dict[str, int]] = {} +def scan_pi_sessions(stats: Stats) -> None: + """Add pi/omp session messages that ran on the deepseek provider.""" seen_messages: set[str] = set() rg = find_command("rg") or "rg" @@ -224,84 +262,35 @@ def scan_pi_sessions() -> dict[str, Any]: continue day = local_day(entry.get("timestamp") or message.get("timestamp")) - model = model_name(message.get("model")) - bucket = model_usage.setdefault(model, empty_bucket()) - bucket["inputTokens"] += input_tokens - bucket["outputTokens"] += output_tokens - bucket["cacheReadInputTokens"] += cache_read - bucket["cacheCreationInputTokens"] += cache_write - - total_tokens = input_tokens + output_tokens + cache_read + cache_write - total_prompts += 1 - total_sessions.add(path) - active_days.add(day) - if day in recent: - recent[day] += total_tokens - if day == today: - today_prompts += 1 - today_sessions.add(path) - today_total_tokens += total_tokens - today_tokens_by_model[model] = today_tokens_by_model.get(model, 0) + total_tokens + stats.add(path, day, model_name(message.get("model")), input_tokens, output_tokens, cache_read, cache_write) try: proc.wait(timeout=1) except Exception: proc.kill() - return { - "todayPrompts": today_prompts, - "todaySessions": len(today_sessions), - "todayTotalTokens": today_total_tokens, - "todayTokensByModel": today_tokens_by_model, - "recentDays": [{"date": day, "messageCount": recent[day]} for day in recent_dates], - "totalPrompts": total_prompts, - "totalSessions": len(total_sessions), - "activeDays": len(active_days), - "activeDates": sorted(active_days), - "modelUsage": model_usage, - } - def opencode_db_path() -> Path: data_home = Path(os.environ.get("XDG_DATA_HOME") or (Path.home() / ".local" / "share")) return data_home / "opencode" / "opencode.db" -def scan_opencode_sessions() -> tuple[dict[str, Any], bool]: - """opencode messages on a deepseek provider, mirroring the codex scan. +def scan_opencode_sessions(stats: Stats) -> None: + """Add opencode messages on a deepseek provider, mirroring the codex scan. - Returns the stats and whether the scan ran to completion: a scan cut short - by a database error still contributes what it read, but a caller that caches - must not treat it as the whole story. Read-only: opencode may be writing - right now. + Read-only: opencode may be writing right now. A scan cut short by a + database error keeps what it already read. """ db = opencode_db_path() if not db.is_file(): - return empty_stats(), True + return import sqlite3 - now = datetime.now() - today = now.strftime("%Y-%m-%d") - recent_dates = [ - (now - timedelta(days=offset)).strftime("%Y-%m-%d") - for offset in range(6, -1, -1) - ] - recent = {day: 0 for day in recent_dates} - - today_prompts = 0 - today_total_tokens = 0 - today_sessions: set[str] = set() - today_tokens_by_model: dict[str, int] = {} - total_prompts = 0 - total_sessions: set[str] = set() - active_days: set[str] = set() - model_usage: dict[str, dict[str, int]] = {} - try: conn = sqlite3.connect(db.resolve().as_uri() + "?mode=ro", uri=True, timeout=2) except sqlite3.Error: - return empty_stats(), False + return try: conn.execute("PRAGMA query_only = ON") # The json_extract conditions are the authority; the LIKE gates are pure @@ -331,91 +320,14 @@ def scan_opencode_sessions() -> tuple[dict[str, Any], bool]: continue day = local_day((entry.get("time") or {}).get("created")) model = model_name(str(entry.get("modelID") or "").rstrip("/").split("/")[-1]) - bucket = model_usage.setdefault(model, empty_bucket()) - bucket["inputTokens"] += input_tokens - bucket["outputTokens"] += output_tokens - bucket["cacheReadInputTokens"] += cache_read - bucket["cacheCreationInputTokens"] += cache_write - total_tokens = input_tokens + output_tokens + cache_read + cache_write - total_prompts += 1 - total_sessions.add("opencode:" + str(session_id)) - active_days.add(day) - if day in recent: - recent[day] += total_tokens - if day == today: - today_prompts += 1 - today_sessions.add("opencode:" + str(session_id)) - today_total_tokens += total_tokens - today_tokens_by_model[model] = today_tokens_by_model.get(model, 0) + total_tokens + stats.add("opencode:" + str(session_id), day, model, input_tokens, output_tokens, cache_read, cache_write) except Exception: continue except sqlite3.Error: - return empty_stats(), False + pass finally: conn.close() - return { - "todayPrompts": today_prompts, - "todaySessions": len(today_sessions), - "todayTotalTokens": today_total_tokens, - "todayTokensByModel": today_tokens_by_model, - "recentDays": [{"date": day, "messageCount": recent[day]} for day in recent_dates], - "totalPrompts": total_prompts, - "totalSessions": len(total_sessions), - "activeDays": len(active_days), - "activeDates": sorted(active_days), - "modelUsage": model_usage, - }, True - - -def merge_stats(first: dict[str, Any], second: dict[str, Any]) -> dict[str, Any]: - """Sum two stats dicts. recentDays arrive as parallel per-day lists.""" - - def day_map(recent_days: list[dict[str, Any]]) -> dict[str, int]: - out: dict[str, int] = {} - for day in recent_days: - out[str(day.get("date") or "")] = number(day.get("messageCount")) - return out - - days = day_map(first.get("recentDays") or []) - for date, count in day_map(second.get("recentDays") or []).items(): - days[date] = days.get(date, 0) + count - - def add_models(target: dict[str, dict[str, int]], source: Any) -> None: - for model, bucket in (source or {}).items(): - if not isinstance(bucket, dict): - continue - slot = target.setdefault(str(model), empty_bucket()) - slot["inputTokens"] += number(bucket.get("inputTokens")) - slot["outputTokens"] += number(bucket.get("outputTokens")) - slot["cacheReadInputTokens"] += number(bucket.get("cacheReadInputTokens")) - slot["cacheCreationInputTokens"] += number(bucket.get("cacheCreationInputTokens")) - - def add_today(target: dict[str, int], source: Any) -> None: - for model, tokens in (source or {}).items(): - target[str(model)] = target.get(str(model), 0) + number(tokens) - - model_usage: dict[str, dict[str, int]] = {} - add_models(model_usage, first.get("modelUsage")) - add_models(model_usage, second.get("modelUsage")) - - today_by_model: dict[str, int] = {} - add_today(today_by_model, first.get("todayTokensByModel")) - add_today(today_by_model, second.get("todayTokensByModel")) - - return { - "todayPrompts": number(first.get("todayPrompts")) + number(second.get("todayPrompts")), - "todaySessions": number(first.get("todaySessions")) + number(second.get("todaySessions")), - "todayTotalTokens": number(first.get("todayTotalTokens")) + number(second.get("todayTotalTokens")), - "todayTokensByModel": today_by_model, - "recentDays": [{"date": day, "messageCount": days[day]} for day in sorted(days)], - "totalPrompts": number(first.get("totalPrompts")) + number(second.get("totalPrompts")), - "totalSessions": number(first.get("totalSessions")) + number(second.get("totalSessions")), - "activeDays": number(first.get("activeDays")) + number(second.get("activeDays")), - "activeDates": sorted(set(first.get("activeDates") or []) | set(second.get("activeDates") or [])), - "modelUsage": model_usage, - } - # -------------------------------------------------------------- credentials @@ -499,7 +411,7 @@ def fetch_balance(api_base_url: str, key: str) -> dict[str, Any]: } -def balance_record(key: str, api_base_url: str) -> dict[str, Any] | None: +def balance_record(key: str, api_base_url: str) -> dict[str, Any]: config = read_config() try: funded = Decimal(str(config.get("fundedAmount") or "0")) @@ -508,12 +420,7 @@ def balance_record(key: str, api_base_url: str) -> dict[str, Any] | None: if not funded.is_finite() or funded < 0: funded = Decimal("0") - try: - live = fetch_balance(api_base_url, key) - except DeepSeekError as error: - print(f"omarchy-agent-usage-deepseek: balance unavailable ({error})", file=sys.stderr) - return None - + live = fetch_balance(api_base_url, key) remaining = live["remaining"] spent = max(Decimal("0"), funded - remaining) if funded > 0 else Decimal("0") return { @@ -528,15 +435,21 @@ def balance_record(key: str, api_base_url: str) -> dict[str, Any] | None: def scan(api_base_url: str) -> dict[str, Any]: - stats = merge_stats(scan_pi_sessions(), scan_opencode_sessions()[0]) + stats = Stats() + scan_pi_sessions(stats) + scan_opencode_sessions(stats) record = base_record() - record.update(stats) + record.update(stats.record()) + # Without a key the record carries local stats only; the panel leaves the + # balance section out rather than reporting an error. key = api_key() if key: - record["balance"] = balance_record(key, api_base_url) - else: - print("omarchy-agent-usage-deepseek: no DeepSeek key found; balance omitted", file=sys.stderr) + try: + record["balance"] = balance_record(key, api_base_url) + except DeepSeekError as error: + record["usageStatusText"] = "Balance unavailable" + record["authHelpText"] = str(error) return record @@ -554,7 +467,7 @@ def main() -> int: record = scan(args.api_base_url) except Exception as error: record = base_record(usageStatusText="DeepSeek unavailable", authHelpText="DeepSeek usage scan failed") - print(f"omarchy-agent-usage-deepseek: {type(error).__name__}: {error}", file=sys.stderr) + print(f"omarchy-agent-usage-deepseek: {type(error).__name__}", file=sys.stderr) print(json.dumps(record, separators=(",", ":"))) return 0 diff --git a/shell/plugins/agents/manifest.json b/shell/plugins/agents/manifest.json index 1743a798214..1e287586a9d 100644 --- a/shell/plugins/agents/manifest.json +++ b/shell/plugins/agents/manifest.json @@ -5,7 +5,7 @@ "version": "1.0.0", "author": "Omarchy", "license": "MIT", - "description": "Claude Code, Codex, and Fireworks usage, limits, and pace in a native Omarchy bar panel.", + "description": "Claude Code, Codex, Fireworks, and DeepSeek usage, limits, and pace in a native Omarchy bar panel.", "kinds": ["bar-widget"], "activation": "on-demand", "entryPoints": { @@ -21,7 +21,8 @@ "providers": { "claude": { "enabled": true }, "codex": { "enabled": true }, - "fireworks": { "enabled": true } + "fireworks": { "enabled": true }, + "deepseek": { "enabled": true } }, "refreshIntervalSec": 900, "syncMode": "Off", diff --git a/test/shell.d/agent-usage-deepseek-scanner-test.sh b/test/shell.d/agent-usage-deepseek-scanner-test.sh index 1b645af0a09..b003f82a60b 100755 --- a/test/shell.d/agent-usage-deepseek-scanner-test.sh +++ b/test/shell.d/agent-usage-deepseek-scanner-test.sh @@ -96,9 +96,19 @@ pass "DeepSeek collector counts opencode messages on a DeepSeek provider" fail "DeepSeek collector keeps reasoning with output and cache apart in opencode" "$result" pass "DeepSeek collector keeps reasoning with output and cache apart in opencode" +# A day active in both pi and opencode is still one active day. +mkdir -p "$OPENCODE_HOME/.pi/agent/sessions/project" +cp "$PI_HOME/.pi/agent/sessions/project/pi.jsonl" "$OPENCODE_HOME/.pi/agent/sessions/project/" +result=$(HOME="$OPENCODE_HOME" XDG_DATA_HOME="$OPENCODE_HOME/.local/share" \ + "$ROOT/bin/omarchy-agent-usage-deepseek" 2>/dev/null) + +[[ $(jq -r '"\(.todayTotalTokens):\(.totalSessions):\(.activeDays)"' <<<"$result") == "176:2:1" ]] || + fail "DeepSeek collector counts a day shared by pi and opencode once" "$result" +pass "DeepSeek collector counts a day shared by pi and opencode once" + # ---------------------------------------------------------------- balance -result=$(python3 - "$ROOT/bin/omarchy-agent-usage-deepseek" "$TEST_HOME/.config" <<'PY' +result=$(HOME="$TEST_HOME" XDG_DATA_HOME="$TEST_HOME/.local/share" python3 - "$ROOT/bin/omarchy-agent-usage-deepseek" "$TEST_HOME/.config" <<'PY' import importlib.machinery import importlib.util import json @@ -132,10 +142,13 @@ summary = {"withConfig": balance} scanner.fetch_balance = lambda base_url, key: (_ for _ in ()).throw( scanner.DeepSeekError("balance endpoint unavailable") ) -import contextlib -import io -with contextlib.redirect_stderr(io.StringIO()): - summary["onFailure"] = scanner.balance_record("test-key", "https://example.invalid") +scanner.api_key = lambda: "test-key" +record = scanner.scan("https://example.invalid") +summary["onFailure"] = { + "hasBalance": "balance" in record, + "status": record["usageStatusText"], + "help": record["authHelpText"], +} scanner.fetch_balance = real_fetch_balance # The real request must carry the bearer key and hit /user/balance. @@ -188,9 +201,9 @@ PY fail "DeepSeek collector derives the estimated ledger from fundedAmount" "$result" pass "DeepSeek collector derives the estimated ledger from fundedAmount" -[[ $(jq -r '.onFailure' <<<"$result") == "null" ]] || - fail "DeepSeek collector omits the balance when the endpoint fails" "$result" -pass "DeepSeek collector omits the balance when the endpoint fails" +[[ $(jq -c '.onFailure' <<<"$result") == '{"hasBalance":false,"status":"Balance unavailable","help":"balance endpoint unavailable"}' ]] || + fail "DeepSeek collector reports a failed balance lookup in the record" "$result" +pass "DeepSeek collector reports a failed balance lookup in the record" [[ $(jq -c '.liveShape' <<<"$result") == '{"remaining":42.5,"granted":2.5,"toppedUp":40.0,"currency":"CNY"}' ]] || fail "DeepSeek collector parses the live balance response" "$result"