-
Notifications
You must be signed in to change notification settings - Fork 1.1k
feat(mimo): opt-in local tracker fallback for MiMo card #1284
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| 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()) |
| 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 | ||
|
|
||
| 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 | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
When MiMo is in auto mode with no cached/importable browser session, 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 | ||
| } | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
For sessions that contain
cache_creation_input_tokens, the tracker writescache_createinto 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 👍 / 👎.