Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
209 changes: 209 additions & 0 deletions Scripts/mimo-usage.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,209 @@
#!/usr/bin/env python3
"""
mimo-usage — local token usage tracker for cc-mimo

Scans ~/.claude-envs/mimo/.claude/projects/**/*.jsonl session files,
sums input/output/cache tokens per time window (today/week/all),
writes to ~/.codexbar/mimo-local-usage.json, and prints a human-readable
summary by default.

Usage:
mimo-usage # show summary (also refreshes cache)
mimo-usage --update # refresh cache only, no output (for LaunchAgent/wrapper)
mimo-usage --json # JSON output
mimo-usage --short # 1-line status (for status line / widget)
"""
import json
import os
import sys
import time
from pathlib import Path
from datetime import datetime, timedelta, timezone

MIMO_HOME = Path.home() / ".claude-envs" / "mimo"
PROJECTS_DIR = MIMO_HOME / ".claude" / "projects"
CACHE_PATH = Path.home() / ".codexbar" / "mimo-local-usage.json"


def parse_session_usage(jsonl_path: Path):
"""Yield (timestamp_iso, usage_dict) for each assistant message with usage."""
try:
with jsonl_path.open() as f:
for line in f:
try:
d = json.loads(line)
ts = d.get("timestamp")
msg = d.get("message")
if not isinstance(msg, dict):
continue
usage = msg.get("usage")
if not isinstance(usage, dict):
continue
if not ts:
continue
yield ts, usage
except (json.JSONDecodeError, ValueError):
continue
except (OSError, IOError):
return


def aggregate_usage():
"""Scan all mimo session jsonls and return windowed token sums."""
now = datetime.now(timezone.utc)
today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
# Week starts on Monday 00:00 UTC
week_start = today_start - timedelta(days=today_start.weekday())

windows = {
"today": {"input": 0, "output": 0, "cache_read": 0, "cache_create": 0, "messages": 0},
"week": {"input": 0, "output": 0, "cache_read": 0, "cache_create": 0, "messages": 0},
"all_time": {"input": 0, "output": 0, "cache_read": 0, "cache_create": 0, "messages": 0},
}
sessions_scanned = 0
last_activity = None

if not PROJECTS_DIR.exists():
return windows, sessions_scanned, last_activity

for jsonl in PROJECTS_DIR.rglob("*.jsonl"):
sessions_scanned += 1
for ts_str, usage in parse_session_usage(jsonl):
try:
# Parse ISO timestamp (may end with Z)
ts = datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
if ts.tzinfo is None:
ts = ts.replace(tzinfo=timezone.utc)
except (ValueError, TypeError):
continue

input_t = int(usage.get("input_tokens", 0) or 0)
output_t = int(usage.get("output_tokens", 0) or 0)
cache_read_t = int(usage.get("cache_read_input_tokens", 0) or 0)
cache_create_t = int(usage.get("cache_creation_input_tokens", 0) or 0)

if last_activity is None or ts > last_activity:
last_activity = ts

# all_time
w = windows["all_time"]
w["input"] += input_t
w["output"] += output_t
w["cache_read"] += cache_read_t
w["cache_create"] += cache_create_t
w["messages"] += 1

if ts >= week_start:
w = windows["week"]
w["input"] += input_t
w["output"] += output_t
w["cache_read"] += cache_read_t
w["cache_create"] += cache_create_t
w["messages"] += 1

if ts >= today_start:
w = windows["today"]
w["input"] += input_t
w["output"] += output_t
w["cache_read"] += cache_read_t
w["cache_create"] += cache_create_t
w["messages"] += 1

return windows, sessions_scanned, last_activity


def write_cache(windows, sessions_scanned, last_activity):
CACHE_PATH.parent.mkdir(parents=True, exist_ok=True)
payload = {
"updated_at": datetime.now(timezone.utc).isoformat(),
"last_activity": last_activity.isoformat() if last_activity else None,
"sessions_scanned": sessions_scanned,
"windows": windows,
"source": "local-jsonl-scan",
"note": "Local token accounting from cc-mimo session jsonl. Not a quota; mimo platform.xiaomimimo.com SSO cookie required for real quota.",
}
tmp = CACHE_PATH.with_suffix(".json.tmp")
tmp.write_text(json.dumps(payload, indent=2))
tmp.replace(CACHE_PATH)
return payload


def fmt_tokens(n: int) -> str:
if n >= 1_000_000:
return f"{n / 1_000_000:.1f}M"
if n >= 1_000:
return f"{n / 1_000:.1f}k"
return str(n)


def short_status(payload):
"""1-line status line."""
w = payload["windows"]["week"]
total = w["input"] + w["output"] + w["cache_read"]
return f"mimo: {fmt_tokens(total)} tok this week ({w['messages']} msg)"


def human_summary(payload):
"""Multi-line human-readable summary."""
last = payload.get("last_activity")
if last:
try:
last_dt = datetime.fromisoformat(last)
ago = datetime.now(timezone.utc) - last_dt
if ago.total_seconds() < 60:
ago_str = "just now"
elif ago.total_seconds() < 3600:
ago_str = f"{int(ago.total_seconds() / 60)}m ago"
elif ago.total_seconds() < 86400:
ago_str = f"{int(ago.total_seconds() / 3600)}h ago"
else:
ago_str = f"{ago.days}d ago"
except (ValueError, TypeError):
ago_str = last
else:
ago_str = "never"

lines = [
"== MiMo (local tracker) ==",
f"Sessions scanned: {payload['sessions_scanned']}",
f"Last activity: {ago_str}",
"",
]
for window_name, label in [("today", "Today"), ("week", "This week"), ("all_time", "All time")]:
w = payload["windows"][window_name]
in_t = fmt_tokens(w["input"])
out_t = fmt_tokens(w["output"])
cr_t = fmt_tokens(w["cache_read"])
cc_t = fmt_tokens(w["cache_create"])
total = w["input"] + w["output"] + w["cache_read"]
lines.append(f"{label:>10}: {fmt_tokens(total):>8} total | in={in_t} out={out_t} cache_r={cr_t} cache_c={cc_t} | msg={w['messages']}")
lines.append("")
lines.append("Note: this is local accounting from cc-mimo session jsonl.")
lines.append("Real platform quota requires Chrome cookie (cookieSource=manual).")
return "\n".join(lines)


def main():
args = sys.argv[1:]
quiet = "--update" in args
json_out = "--json" in args
short = "--short" in args

windows, sessions_scanned, last_activity = aggregate_usage()
payload = write_cache(windows, sessions_scanned, last_activity)

if quiet:
return 0
if json_out:
print(json.dumps(payload, indent=2))
return 0
if short:
print(short_status(payload))
return 0

print(human_summary(payload))
return 0


if __name__ == "__main__":
sys.exit(main())
8 changes: 8 additions & 0 deletions Sources/CodexBar/MenuCardView.swift
Original file line number Diff line number Diff line change
Expand Up @@ -864,6 +864,14 @@ extension UsageMenuCardView.Model {
}

if input.provider == .mimo, input.snapshot != nil {
// Local fallback (cc-mimo session jsonl scan) has no platform billing —
// suppress the misleading "balance updates / daily billing" footer.
let resolvedSource = input.sourceLabel?
.trimmingCharacters(in: .whitespacesAndNewlines)
.lowercased()
if resolvedSource == "local" {
return []
}
return [
L("Balance updates in near-real time (up to 5 min lag)"),
L("Daily billing data finalizes at 07:00 UTC"),
Expand Down
91 changes: 91 additions & 0 deletions Sources/CodexBarCore/Providers/MiMo/MiMoLocalUsageFallback.swift
Original file line number Diff line number Diff line change
@@ -0,0 +1,91 @@
import Foundation
#if canImport(FoundationNetworking)
import FoundationNetworking
#endif

/// Local tracker fallback for MiMo when Xiaomi platform.xiaomimimo.com cookie is unavailable.
///
/// Reads the JSON cache produced by `Scripts/mimo-usage.py` which scans
/// `~/.claude-envs/mimo/.claude/projects/**/*.jsonl` and aggregates token usage
/// per time window. This is local accounting only — not real platform quota —
/// but gives users a useful view when SSO cookie access is blocked (keychain,
/// Chrome session-cookie expiry, etc.).
///
/// **Implicit opt-in**: this fallback only triggers when the cache file exists;
/// users who do not run `Scripts/mimo-usage.py` see no behavior change.
///
/// See `docs/mimo.md` "Local fallback (opt-in)" for setup instructions.
public enum MiMoLocalUsageFallback {
public static func defaultCachePath() -> String {
"\(NSHomeDirectory())/.codexbar/mimo-local-usage.json"
}

public static func snapshot(now: Date = Date()) -> MiMoUsageSnapshot? {
self.snapshot(cachePath: self.defaultCachePath(), now: now)
}

public static func snapshot(cachePath: String, now: Date) -> MiMoUsageSnapshot? {
let url = URL(fileURLWithPath: cachePath)
guard let data = try? Data(contentsOf: url) else { return nil }
guard let json = try? JSONSerialization.jsonObject(with: data) as? [String: Any] else { return nil }

let windows = json["windows"] as? [String: Any] ?? [:]
let week = windows["week"] as? [String: Any] ?? [:]
let today = windows["today"] as? [String: Any] ?? [:]
let allTime = windows["all_time"] as? [String: Any] ?? [:]
let sessionsScanned = json["sessions_scanned"] as? Int ?? 0

let weekInput = Self.intValue(week["input"])
let weekOutput = Self.intValue(week["output"])
let weekCacheRead = Self.intValue(week["cache_read"])
let weekTotal = weekInput + weekOutput + weekCacheRead
Comment on lines +38 to +41
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Include cache-creation tokens in local totals

For sessions that contain cache_creation_input_tokens, the tracker writes cache_create into each window, but the fallback snapshot only sums input/output/cache_read for today/week/all-time. Those cache-creation tokens are billable local activity and are displayed by the script summary, so omitting them here underreports the MiMo card totals and progress for any cached-prompt session that creates cache entries.

Useful? React with 👍 / 👎.


let todayInput = Self.intValue(today["input"])
let todayOutput = Self.intValue(today["output"])
let todayCacheRead = Self.intValue(today["cache_read"])
let todayTotal = todayInput + todayOutput + todayCacheRead

let allInput = Self.intValue(allTime["input"])
let allOutput = Self.intValue(allTime["output"])
let allCacheRead = Self.intValue(allTime["cache_read"])
let allTotal = allInput + allOutput + allCacheRead

// planCode shows today/week/lifetime/sessions in the loginMethod row.
var parts: [String] = []
if todayTotal > 0 { parts.append("\(Self.fmtTokens(todayTotal)) today") }
if weekTotal > 0 { parts.append("\(Self.fmtTokens(weekTotal)) week") }
if allTotal > 0 { parts.append("\(Self.fmtTokens(allTotal)) total") }
parts.append("\(sessionsScanned) sessions")
let planCode = parts.joined(separator: " · ")

// Progress bar: weekly usage vs lifetime baseline (this-week vs all-time activity ratio).
// Idle (week=0) → bar empty, lifetime as baseline so the bar is meaningful once user resumes cc-mimo.
let tokenLimit = max(allTotal, weekTotal + 1)
let tokenUsed = weekTotal
let tokenPercent = tokenLimit > 0 ? Double(tokenUsed) / Double(tokenLimit) : 0

return MiMoUsageSnapshot(
balance: 0,
currency: "",
planCode: planCode,
planPeriodEnd: nil,
planExpired: false,
tokenUsed: tokenUsed,
tokenLimit: tokenLimit,
tokenPercent: tokenPercent,
updatedAt: now)
}

private static func fmtTokens(_ n: Int) -> String {
if n >= 1_000_000 { return String(format: "%.1fM", Double(n) / 1_000_000) }
if n >= 1000 { return String(format: "%.1fk", Double(n) / 1000) }
return "\(n)"
}

private static func intValue(_ raw: Any?) -> Int {
if let i = raw as? Int { return i }
if let d = raw as? Double { return Int(d) }
if let s = raw as? String, let i = Int(s) { return i }
return 0
}
}
35 changes: 34 additions & 1 deletion Sources/CodexBarCore/Providers/MiMo/MiMoProviderDescriptor.swift
Original file line number Diff line number Diff line change
Expand Up @@ -74,13 +74,46 @@ struct MiMoWebFetchStrategy: ProviderFetchStrategy {
{
return true
}
return MiMoCookieImporter.hasSession(browserDetection: context.browserDetection)
if MiMoCookieImporter.hasSession(browserDetection: context.browserDetection) {
return true
}
// Local tracker fallback file exists → strategy is available even without
// cookie/browser session. This keeps the provider switcher from saying
// "No available fetch strategy for mimo" when only the local fallback is
// configured (see MiMoLocalUsageFallback + `Scripts/mimo-usage.py`).
return MiMoLocalUsageFallback.snapshot() != nil
#else
return false
#endif
}

func fetch(_ context: ProviderFetchContext) async throws -> ProviderFetchResult {
do {
return try await self.fetchFromWeb(context)
} catch {
// Local tracker fallback: when platform cookie/SSO is unavailable
// (missingCookie / invalidCookie / loginRequired / invalidCredentials),
// fall back to local jsonl token accounting via ~/.codexbar/mimo-local-usage.json
// (populated by `~/bin/mimo-usage`, scanning ~/.claude-envs/mimo session jsonl).
if Self.shouldFallbackToLocal(error: error),
let local = MiMoLocalUsageFallback.snapshot()
{
return self.makeResult(usage: local.toUsageSnapshot(), sourceLabel: "local")
Comment on lines +98 to +101
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Let the local cache make MiMo fetching available

When MiMo is in auto mode with no cached/importable browser session, isAvailable() returns false, and ProviderFetchPipeline.fetch skips fetch() entirely before this fallback branch can run. That means the documented opt-in path for users with only ~/.codexbar/mimo-local-usage.json and no platform cookies still ends in noAvailableStrategy instead of showing local usage; the strategy needs to report available when the local cache exists or expose a separate local strategy.

Useful? React with 👍 / 👎.

}
throw error
}
}

private static func shouldFallbackToLocal(error: Error) -> Bool {
if error is MiMoSettingsError { return true }
guard let mimoError = error as? MiMoUsageError else { return false }
switch mimoError {
case .invalidCredentials, .loginRequired: return true
case .parseFailed, .networkError: return false
}
}

private func fetchFromWeb(_ context: ProviderFetchContext) async throws -> ProviderFetchResult {
guard context.settings?.mimo?.cookieSource != .off else {
throw MiMoSettingsError.missingCookie
}
Expand Down
Loading