diff --git a/bin/omarchy-agent-usage-deepseek b/bin/omarchy-agent-usage-deepseek
new file mode 100755
index 00000000000..71efa7c2667
--- /dev/null
+++ b/bin/omarchy-agent-usage-deepseek
@@ -0,0 +1,476 @@
+#!/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"
+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
+
+
+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",
+ Path.home() / ".omp" / "agent" / "sessions",
+ ]
+
+
+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"
+ 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"))
+ 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()
+
+
+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(stats: Stats) -> None:
+ """Add opencode messages on a deepseek provider, mirroring the codex scan.
+
+ 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
+
+ import sqlite3
+
+ try:
+ conn = sqlite3.connect(db.resolve().as_uri() + "?mode=ro", uri=True, timeout=2)
+ except sqlite3.Error:
+ return
+ 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])
+ stats.add("opencode:" + str(session_id), day, model, input_tokens, output_tokens, cache_read, cache_write)
+ except Exception:
+ continue
+ except sqlite3.Error:
+ pass
+ finally:
+ conn.close()
+
+
+# -------------------------------------------------------------- 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]:
+ 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")
+
+ live = fetch_balance(api_base_url, key)
+ 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 = Stats()
+ scan_pi_sessions(stats)
+ scan_opencode_sessions(stats)
+ record = base_record()
+ 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:
+ try:
+ record["balance"] = balance_record(key, api_base_url)
+ except DeepSeekError as error:
+ record["usageStatusText"] = "Balance unavailable"
+ record["authHelpText"] = str(error)
+ 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__}", 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 @@
+
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
new file mode 100755
index 00000000000..b003f82a60b
--- /dev/null
+++ b/test/shell.d/agent-usage-deepseek-scanner-test.sh
@@ -0,0 +1,214 @@
+#!/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"
+
+# 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=$(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
+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")
+)
+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.
+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 -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"
+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"