diff --git a/CLAUDE.md b/CLAUDE.md index 0994ae6b..f7346be0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -55,7 +55,7 @@ Claude Code ──stdin──> usage_statusline.py (hook) ──write──> ~/. - **Claude Code side**: `usage_statusline.py` is installed into `~/.claude/usage-statusline.py` by `setup_hook.py` and wired into `~/.claude/settings.json`'s `statusLine`. Every time Claude Code refreshes its status line, it pipes the session JSON to the hook on stdin; the hook atomically writes it to `~/.claude/usage-status.json`. The UI reads that file — never the network. - **Codex side**: no hook is possible (Codex CLI has no equivalent), so `codex_loader.py` scans `~/.codex/sessions/**/*.jsonl` and pulls `rate_limits` straight from the conversation logs. -- **Read priority** in `usage_client.py`: `usage-status.json` → `usag-status.json` (v0.1.x legacy) → `tt-status.json` (compat fallback for users migrating from the third-party tool `stormzhang/token-tracker`; **NOT an in-repo module — no `token-tracker` directory or source exists anywhere on this machine**). +- **Read priority** in `usage_client.py`: the newest usable data from `usage-status.json` → `usag-status.json` (v0.1.x legacy) → `tt-status.json` (compat fallback for users migrating from the third-party tool `stormzhang/token-tracker`) or Claude Code's `~/.claude.json` `cachedUsageUtilization` fallback; **no `token-tracker` module or source exists in this repository**. ### Module map diff --git a/menubar_state.py b/menubar_state.py index 535b94d4..66ca076b 100644 --- a/menubar_state.py +++ b/menubar_state.py @@ -133,6 +133,16 @@ class HistorySourceScan: codex_paths: tuple[Path, ...] +def history_cache_needs_reload( + previous_fingerprint: tuple[tuple[str, int, float], ...] | None, + current_fingerprint: tuple[tuple[str, int, float], ...], + *, + has_cached_result: bool, +) -> bool: + """Decide whether Windows history projections need to be rebuilt.""" + return not has_cached_result or previous_fingerprint != current_fingerprint + + def _jsonl_paths(root: Path) -> tuple[Path, ...]: if not root.exists(): return () diff --git a/session_hooks.py b/session_hooks.py index 013b484b..c6a34411 100644 --- a/session_hooks.py +++ b/session_hooks.py @@ -125,7 +125,9 @@ def _migrate_bundled_python_commands_if_needed( if not isinstance(hook, dict): continue command = hook.get("command") - if not isinstance(command, str) or not _uses_bundled_app_python(command): + if not isinstance(command, str) or not any( + marker in command for marker in _RESUME_MARKERS + ): continue if command != new_command: hook["command"] = new_command @@ -134,6 +136,55 @@ def _migrate_bundled_python_commands_if_needed( if resume_changed: details.append("resume") + new_command = _terse_command() + terse_changed = False + for entry in entries: + if not isinstance(entry, dict) or not _is_terse_entry(entry): + continue + hooks = entry.get("hooks") + if not isinstance(hooks, list): + continue + for hook in hooks: + if not isinstance(hook, dict): + continue + command = hook.get("command") + if not isinstance(command, str) or not any( + marker in command for marker in _TERSE_MARKERS + ): + continue + if command != new_command: + hook["command"] = new_command + changed = True + terse_changed = True + if terse_changed: + details.append("terse") + + hooks_data = data.get("hooks") + prompt_entries = hooks_data.get("UserPromptSubmit") if isinstance(hooks_data, dict) else None + if isinstance(prompt_entries, list): + new_command = _terse_reminder_command() + reminder_changed = False + for entry in prompt_entries: + if not isinstance(entry, dict) or not _is_terse_reminder_entry(entry): + continue + hooks = entry.get("hooks") + if not isinstance(hooks, list): + continue + for hook in hooks: + if not isinstance(hook, dict): + continue + command = hook.get("command") + if not isinstance(command, str) or not any( + marker in command for marker in _TERSE_REMINDER_MARKERS + ): + continue + if command != new_command: + hook["command"] = new_command + changed = True + reminder_changed = True + if reminder_changed: + details.append("terse_reminder") + if not changed: return _save_settings(data) @@ -182,20 +233,17 @@ def _resolve_terse_reminder_source() -> Path: def _resume_command() -> str: python = _find_system_python() - source = _resolve_resume_source() - return f"{_shell_arg(python)} {_shell_arg(str(source))}" + return f"{_shell_arg(python)} {_shell_arg(str(RESUME_HOOK_TARGET))}" def _terse_command() -> str: python = _find_system_python() - source = _resolve_terse_source() - return f"{_shell_arg(python)} {_shell_arg(str(source))}" + return f"{_shell_arg(python)} {_shell_arg(str(TERSE_HOOK_TARGET))}" def _terse_reminder_command() -> str: python = _find_system_python() - source = _resolve_terse_reminder_source() - return f"{_shell_arg(python)} {_shell_arg(str(source))}" + return f"{_shell_arg(python)} {_shell_arg(str(TERSE_REMINDER_HOOK_TARGET))}" def _copy_resume_script() -> None: @@ -873,7 +921,6 @@ def _migrate_resume_command_if_needed() -> None: entries = _session_start_list(settings) if not entries: return - old_target = str(RESUME_HOOK_TARGET) new_command = _resume_command() changed = False for entry in entries: @@ -886,7 +933,11 @@ def _migrate_resume_command_if_needed() -> None: if not isinstance(hook, dict): continue command = hook.get("command") - if not isinstance(command, str) or old_target not in command: + if ( + not isinstance(command, str) + or not any(marker in command for marker in _RESUME_MARKERS) + or command == new_command + ): continue hook["command"] = new_command changed = True @@ -895,7 +946,7 @@ def _migrate_resume_command_if_needed() -> None: _save_settings(settings) _append_self_heal_log( "migrate_resume_command", - f"{RESUME_HOOK_TARGET} -> {_resolve_resume_source()}", + f"{_resolve_resume_source()} -> {RESUME_HOOK_TARGET}", ) diff --git a/tests/test_menubar_state.py b/tests/test_menubar_state.py index 33638ea7..cda504b8 100644 --- a/tests/test_menubar_state.py +++ b/tests/test_menubar_state.py @@ -169,6 +169,20 @@ def test_file_event_refresh_decision_merges_into_existing_trailing() -> None: assert decision.trailing_delay is None +def test_history_cache_reload_decision() -> None: + fingerprint = (("history", 1, 10.0),) + + assert menubar_state.history_cache_needs_reload( + None, fingerprint, has_cached_result=False + ) + assert not menubar_state.history_cache_needs_reload( + fingerprint, fingerprint, has_cached_result=True + ) + assert menubar_state.history_cache_needs_reload( + fingerprint, (("history", 2, 11.0),), has_cached_result=True + ) + + def test_project_rows_for_windows_matches_window_boundaries() -> None: now = datetime.now(UTC).replace(microsecond=0) diff --git a/tests/test_session_resume.py b/tests/test_session_resume.py index 6d33fcaa..45c3e6cc 100644 --- a/tests/test_session_resume.py +++ b/tests/test_session_resume.py @@ -187,6 +187,15 @@ def _write_diagnosis_state(path: Path, *, fingerprint: str, reminded_at: datetim ) +@pytest.mark.parametrize("value", [".", "..."]) +def test_clean_request_rejects_punctuation_only(value: str) -> None: + assert mod._clean_request(value) == "" + + +def test_clean_request_keeps_structured_text() -> None: + assert mod._clean_request("fix menubar.py") == "fix menubar.py" + + def test_build_prompt_reads_previous_session( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: diff --git a/tests/test_setup_hook.py b/tests/test_setup_hook.py index b6f32ba7..937cabe4 100644 --- a/tests/test_setup_hook.py +++ b/tests/test_setup_hook.py @@ -511,6 +511,8 @@ def test_self_heal_migrates_bundled_python_commands( resume_source = tmp_path / "usage_session_resume.py" resume_source.write_text('__version__ = "1.0"\n', encoding="utf-8") monkeypatch.setattr(session_hooks, "_resolve_resume_source", lambda: resume_source) + resume_target = tmp_path / ".claude" / "usage-session-resume.py" + monkeypatch.setattr(session_hooks, "RESUME_HOOK_TARGET", resume_target) settings.write_text( json.dumps( { @@ -544,7 +546,7 @@ def test_self_heal_migrates_bundled_python_commands( data = json.loads(settings.read_text(encoding="utf-8")) assert data["statusLine"]["command"] == expected_statusline_command(hook_target) hooks = data["hooks"]["SessionStart"][0]["hooks"] - assert hooks[0]["command"] == expected_statusline_command(resume_source) + assert hooks[0]["command"] == expected_statusline_command(resume_target) migrate_entries = [ entry for entry in data["usage"]["selfHealLog"] @@ -562,9 +564,9 @@ def test_self_heal_keeps_correct_python_commands_unchanged( ) -> None: settings = setup_paths.settings hook_target = setup_paths.hook_target - resume_source = tmp_path / "usage_session_resume.py" - resume_source.write_text('__version__ = "1.0"\n', encoding="utf-8") - monkeypatch.setattr(session_hooks, "_resolve_resume_source", lambda: resume_source) + resume_target = tmp_path / ".claude" / "usage-session-resume.py" + monkeypatch.setattr(session_hooks, "RESUME_HOOK_TARGET", resume_target) + resume_command = session_hooks._resume_command() settings.write_text( json.dumps( { @@ -576,9 +578,7 @@ def test_self_heal_keeps_correct_python_commands_unchanged( "SessionStart": [ { "matcher": session_hooks.RESUME_MATCHER, - "hooks": [ - {"type": "command", "command": f"/usr/bin/python3 {resume_source}"} - ], + "hooks": [{"type": "command", "command": resume_command}], } ] }, @@ -591,8 +591,5 @@ def test_self_heal_keeps_correct_python_commands_unchanged( data = json.loads(settings.read_text(encoding="utf-8")) assert data["statusLine"]["command"] == f"/usr/bin/python3 {hook_target}" - assert ( - data["hooks"]["SessionStart"][0]["hooks"][0]["command"] - == f"/usr/bin/python3 {resume_source}" - ) + assert data["hooks"]["SessionStart"][0]["hooks"][0]["command"] == resume_command assert "usage" not in data diff --git a/tests/test_setup_hook_resume.py b/tests/test_setup_hook_resume.py index 2cc7ddf1..73650bde 100644 --- a/tests/test_setup_hook_resume.py +++ b/tests/test_setup_hook_resume.py @@ -47,8 +47,7 @@ def test_enable_registers_hook_and_writes_sidecar( assert isinstance(first_hook, dict) command = first_hook["command"] assert isinstance(command, str) - assert str(resume_target) not in command - assert resume_paths.source.as_posix() in command + assert resume_target.as_posix() in command # Sidecar carries the i18n-sourced prompt template for every shipped language. bundle = json.loads(sidecar.read_text(encoding="utf-8")) assert {"zh-TW", "en", "ja", "ko", "zh-CN"} <= set(bundle) @@ -89,7 +88,7 @@ def test_enable_preserves_existing_hooks( data = json.loads(settings.read_text(encoding="utf-8")) commands = [h["command"] for e in data["hooks"]["SessionStart"] for h in e["hooks"]] assert "other" in commands - assert any("usage_session_resume" in c for c in commands) + assert any("usage-session-resume" in c for c in commands) assert data["hooks"]["PreToolUse"][0]["hooks"][0]["command"] == "guard" @@ -142,11 +141,11 @@ def test_self_heal_restores_missing_script_when_enabled( detail = data["usage"]["selfHealLog"][-1]["detail"] assert data["usage"]["selfHealLog"][-1]["action"] == "restore_resume_hook" assert "missing=script,sidecar" in detail - assert "registered=source" in detail + assert "registered=target" in detail assert "recent_claude_entries=" in detail -def test_self_heal_migrates_existing_target_command( +def test_self_heal_normalizes_existing_target_command( resume_paths: ResumeHookPaths, ) -> None: settings = resume_paths.settings @@ -166,7 +165,7 @@ def test_self_heal_migrates_existing_target_command( "hooks": [ { "type": "command", - "command": f"/usr/bin/python3 {resume_target}", + "command": f"/usr/bin/python3 {source}", "timeout": 3, }, {"type": "command", "command": "other"}, @@ -189,8 +188,7 @@ def test_self_heal_migrates_existing_target_command( assert session_entry["custom"] == "keep" assert migrated_hook["type"] == "command" assert migrated_hook["timeout"] == 3 - assert str(resume_target) not in migrated_hook["command"] - assert source.as_posix() in migrated_hook["command"] + assert resume_target.as_posix() in migrated_hook["command"] assert session_entry["hooks"][1]["command"] == "other" assert data["hooks"]["PreToolUse"][0]["hooks"][0]["command"] == "guard" # The stale "1.2" target also triggers a version update in the same pass, so the diff --git a/tests/test_setup_hook_terse.py b/tests/test_setup_hook_terse.py index 391fadac..b726c015 100644 --- a/tests/test_setup_hook_terse.py +++ b/tests/test_setup_hook_terse.py @@ -55,8 +55,7 @@ def test_enable_registers_hook_and_writes_sidecar(terse_paths: TerseHookPaths) - assert isinstance(first_hook, dict) command = first_hook["command"] assert isinstance(command, str) - assert str(terse_paths.terse_target) not in command - assert terse_paths.source.as_posix() in command + assert terse_paths.terse_target.as_posix() in command bundle = json.loads(terse_paths.sidecar.read_text(encoding="utf-8")) assert {"zh-TW", "en", "ja", "ko", "zh-CN"} <= set(bundle) assert "Terse mode is on for this entire conversation" in bundle["en"]["instruction"] @@ -88,7 +87,7 @@ def test_enable_preserves_existing_hooks(terse_paths: TerseHookPaths) -> None: data = json.loads(settings.read_text(encoding="utf-8")) commands = [h["command"] for e in data["hooks"]["SessionStart"] for h in e["hooks"]] assert "other" in commands - assert any("usage_terse_mode" in c for c in commands) + assert any("usage-terse-mode" in c for c in commands) assert data["hooks"]["PreToolUse"][0]["hooks"][0]["command"] == "guard" @@ -288,8 +287,7 @@ def test_enable_registers_reminder_hook(terse_paths: TerseHookPaths) -> None: assert isinstance(first_hook, dict) command = first_hook["command"] assert isinstance(command, str) - assert str(terse_paths.terse_reminder_target) not in command - assert terse_paths.reminder_source.as_posix() in command + assert terse_paths.terse_reminder_target.as_posix() in command def test_enable_reminder_is_idempotent(terse_paths: TerseHookPaths) -> None: diff --git a/tests/test_usage_client.py b/tests/test_usage_client.py index 48bb0d2c..076b9d4d 100644 --- a/tests/test_usage_client.py +++ b/tests/test_usage_client.py @@ -11,6 +11,7 @@ import json import logging import os +from datetime import datetime from pathlib import Path from typing import Any @@ -24,8 +25,32 @@ @pytest.fixture(autouse=True) -def reset_recent_activity_cache(monkeypatch: pytest.MonkeyPatch) -> None: +def isolate_claude_files(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: monkeypatch.setattr(usage_client, "_recent_activity_cache", None) + monkeypatch.setattr(usage_client, "CLAUDE_JSON_FILE", str(tmp_path / ".claude.json")) + + +def _write_claude_json(path: Path, fetched_at: float) -> None: + path.write_text( + json.dumps( + { + "cachedUsageUtilization": { + "fetchedAtMs": fetched_at * 1000, + "utilization": { + "five_hour": { + "utilization": 3, + "resets_at": "2026-07-16T08:29:59.915566+08:00", + }, + "seven_day": { + "utilization": 99, + "resets_at": "2026-07-17T04:59:59.915591+08:00", + }, + }, + } + } + ), + encoding="utf-8", + ) def test_read_status_file_returns_none_when_both_paths_missing( @@ -272,6 +297,96 @@ def test_fetch_once_without_status_file_returns_non_success( assert outcome.state is usage_client.PollState.TOKEN_ERROR +def test_fetch_once_uses_claude_json_when_status_is_missing( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fetched_at = 1_784_144_611.575 + claude_json_path = tmp_path / ".claude.json" + monkeypatch.setattr(usage_client, "STATUS_FILE", str(tmp_path / "usage-status.json")) + monkeypatch.setattr(usage_client, "LEGACY_STATUS_FILE", str(tmp_path / "legacy.json")) + monkeypatch.setattr(usage_client, "TT_STATUS_FILE", str(tmp_path / "tt-status.json")) + monkeypatch.setattr(usage_client, "CLAUDE_JSON_FILE", str(claude_json_path)) + monkeypatch.setattr("usage_client.time.time", lambda: fetched_at + 1) + _write_claude_json(claude_json_path, fetched_at) + + outcome = asyncio.run(usage_client.ClaudeUsageClient(mock=False).fetch_once()) + + assert outcome.state is usage_client.PollState.SUCCESS + assert outcome.snapshot is not None + assert outcome.snapshot.data_source == "claude-json" + assert outcome.snapshot.current_percent == 3 + assert outcome.snapshot.weekly_percent == 99 + assert outcome.snapshot.current_reset_at == pytest.approx( + datetime.fromisoformat("2026-07-16T08:29:59.915566+08:00").timestamp() + ) + assert outcome.snapshot.weekly_reset_at == pytest.approx( + datetime.fromisoformat("2026-07-17T04:59:59.915591+08:00").timestamp() + ) + + +@pytest.mark.parametrize("status_age", [-1.0, 1.0]) +def test_fetch_once_chooses_newest_complete_source( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, status_age: float +) -> None: + fetched_at = 1_784_144_611.575 + status_path = tmp_path / "usage-status.json" + claude_json_path = tmp_path / ".claude.json" + monkeypatch.setattr(usage_client, "STATUS_FILE", str(status_path)) + monkeypatch.setattr(usage_client, "LEGACY_STATUS_FILE", str(tmp_path / "legacy.json")) + monkeypatch.setattr(usage_client, "TT_STATUS_FILE", str(tmp_path / "tt-status.json")) + monkeypatch.setattr(usage_client, "CLAUDE_JSON_FILE", str(claude_json_path)) + monkeypatch.setattr("usage_client.time.time", lambda: fetched_at + 2) + _write_complete_status(status_path, fetched_at + status_age) + _write_claude_json(claude_json_path, fetched_at) + + outcome = asyncio.run(usage_client.ClaudeUsageClient(mock=False).fetch_once()) + + assert outcome.snapshot is not None + if status_age >= 0: + assert outcome.snapshot.data_source == "hook" + assert outcome.snapshot.current_percent == 12 + else: + assert outcome.snapshot.data_source == "claude-json" + assert outcome.snapshot.current_percent == 3 + + +@pytest.mark.parametrize( + "contents", + ["{bad json", "{}", '{"cachedUsageUtilization": {}}'], +) +def test_invalid_claude_json_preserves_missing_status_error( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, contents: str +) -> None: + claude_json_path = tmp_path / ".claude.json" + monkeypatch.setattr(usage_client, "STATUS_FILE", str(tmp_path / "usage-status.json")) + monkeypatch.setattr(usage_client, "LEGACY_STATUS_FILE", str(tmp_path / "legacy.json")) + monkeypatch.setattr(usage_client, "TT_STATUS_FILE", str(tmp_path / "tt-status.json")) + monkeypatch.setattr(usage_client, "CLAUDE_JSON_FILE", str(claude_json_path)) + claude_json_path.write_text(contents, encoding="utf-8") + + outcome = asyncio.run(usage_client.ClaudeUsageClient(mock=False).fetch_once()) + + assert outcome.state is usage_client.PollState.TOKEN_ERROR + + +def test_invalid_claude_json_preserves_incomplete_status_loading( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + status_path = tmp_path / "usage-status.json" + claude_json_path = tmp_path / ".claude.json" + monkeypatch.setattr(usage_client, "STATUS_FILE", str(status_path)) + monkeypatch.setattr(usage_client, "LEGACY_STATUS_FILE", str(tmp_path / "legacy.json")) + monkeypatch.setattr(usage_client, "TT_STATUS_FILE", str(tmp_path / "tt-status.json")) + monkeypatch.setattr(usage_client, "CLAUDE_JSON_FILE", str(claude_json_path)) + status_path.write_text('{"foo": "bar"}', encoding="utf-8") + claude_json_path.write_text("{}", encoding="utf-8") + + outcome = asyncio.run(usage_client.ClaudeUsageClient(mock=False).fetch_once()) + + assert outcome.state is usage_client.PollState.LOADING + assert outcome.message == "awaiting_rate_limits" + + def test_fetch_once_returns_awaiting_rate_limits_when_status_has_no_limits( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: @@ -340,6 +455,35 @@ def counting_open(*args: Any, **kwargs: Any) -> Any: assert open_calls == 1 +def test_fetch_once_reuses_claude_json_snapshot_until_mtime_changes( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + claude_json_path = tmp_path / ".claude.json" + monkeypatch.setattr(usage_client, "STATUS_FILE", str(tmp_path / "missing.json")) + monkeypatch.setattr(usage_client, "LEGACY_STATUS_FILE", str(tmp_path / "legacy.json")) + monkeypatch.setattr(usage_client, "TT_STATUS_FILE", str(tmp_path / "tt.json")) + monkeypatch.setattr(usage_client, "CLAUDE_JSON_FILE", str(claude_json_path)) + claude_json_path.write_text("{}", encoding="utf-8") + calls = 0 + original = usage_client._read_claude_json_snapshot + + def counting_read() -> usage_client.UsageSnapshot | None: + nonlocal calls + calls += 1 + return original() + + monkeypatch.setattr(usage_client, "_read_claude_json_snapshot", counting_read) + client = usage_client.ClaudeUsageClient(mock=False) + + asyncio.run(client.fetch_once()) + asyncio.run(client.fetch_once()) + current = claude_json_path.stat().st_mtime + os.utime(claude_json_path, (current + 1, current + 1)) + asyncio.run(client.fetch_once()) + + assert calls == 2 + + def test_fetch_once_recomputes_stale_state_when_status_mtime_is_unchanged( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: @@ -570,3 +714,34 @@ def rglob(self, _pattern: str) -> list[FakePath]: now + usage_client.RECENT_ACTIVITY_CACHE_TTL_SECONDS ) is True assert projects_dir.calls == 2 + + +def test_cached_claude_json_rezeroes_after_reset_passes( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fetched_at = 1_784_144_611.575 + five_reset = datetime.fromisoformat("2026-07-16T08:29:59.915566+08:00").timestamp() + seven_reset = datetime.fromisoformat("2026-07-17T04:59:59.915591+08:00").timestamp() + claude_json_path = tmp_path / ".claude.json" + monkeypatch.setattr(usage_client, "STATUS_FILE", str(tmp_path / "usage-status.json")) + monkeypatch.setattr(usage_client, "LEGACY_STATUS_FILE", str(tmp_path / "legacy.json")) + monkeypatch.setattr(usage_client, "TT_STATUS_FILE", str(tmp_path / "tt-status.json")) + monkeypatch.setattr(usage_client, "CLAUDE_JSON_FILE", str(claude_json_path)) + fake_now = fetched_at + 1 + monkeypatch.setattr("usage_client.time.time", lambda: fake_now) + _write_claude_json(claude_json_path, fetched_at) + + client = usage_client.ClaudeUsageClient(mock=False) + first = asyncio.run(client.fetch_once()) + assert first.snapshot is not None + assert first.snapshot.current_percent == 3 + assert first.snapshot.weekly_percent == 99 + + # File untouched, but the five-hour window resets: the cache hit must re-derive + # expiry-sensitive fields instead of replaying the stale parse. + fake_now = five_reset + 10 + assert fake_now < seven_reset + second = asyncio.run(client.fetch_once()) + assert second.snapshot is not None + assert second.snapshot.current_percent == 0 + assert second.snapshot.weekly_percent == 99 diff --git a/tests/test_wintray.py b/tests/test_wintray.py index a9016e40..a7958923 100644 --- a/tests/test_wintray.py +++ b/tests/test_wintray.py @@ -487,7 +487,9 @@ def test_show_panel_places_window_before_showing( controller = wintray._WindowsTrayController(mock=True, interval=60) calls: list[str] = [] monkeypatch.setattr(controller, "_place_window", lambda: calls.append("place")) - monkeypatch.setattr(controller, "inject_state", lambda: calls.append("inject")) + monkeypatch.setattr( + controller, "inject_state", lambda *, force=False: calls.append(f"inject:{force}") + ) monkeypatch.setattr(controller, "refresh", lambda: calls.append("refresh")) controller.window = SimpleNamespace( show=lambda: calls.append("show"), hide=lambda: calls.append("hide") @@ -496,7 +498,79 @@ def test_show_panel_places_window_before_showing( controller.show_panel() assert controller.visible is True - assert calls == ["place", "show", "inject", "refresh"] + assert calls == ["place", "show", "inject:True", "refresh"] + + +def test_tray_update_skips_unchanged_values(monkeypatch: pytest.MonkeyPatch) -> None: + controller = wintray._WindowsTrayController(mock=True, interval=60) + controller.latest_state = _state() + icon = SimpleNamespace(icon=None, title=None) + controller.icon = icon + images: list[float | None] = [] + + def fake_draw_tray_icon(percent: float | None) -> object: + images.append(percent) + return object() + + monkeypatch.setattr(wintray, "draw_tray_icon", fake_draw_tray_icon) + + controller._update_tray() + first_image = icon.icon + controller._update_tray() + controller.latest_state.claude_session.percent = 26.0 + controller._update_tray() + + assert images == [25.0, 26.0] + assert icon.icon is not first_image + + +def test_inject_state_skips_duplicate_but_forces_after_panel_reopens( + monkeypatch: pytest.MonkeyPatch, +) -> None: + controller = wintray._WindowsTrayController(mock=True, interval=60) + injected: list[str] = [] + controller.window = SimpleNamespace( + evaluate_js=injected.append, + show=lambda: None, + hide=lambda: None, + ) + monkeypatch.setattr(controller, "_place_window", lambda: None) + monkeypatch.setattr(controller, "refresh", lambda: None) + + controller.inject_state() + controller.inject_state() + controller.show_panel() + controller.on_loaded() + controller.show_panel() + controller.show_panel() + + assert len(injected) == 4 + + +def test_build_state_reuses_history_until_fingerprint_changes( + monkeypatch: pytest.MonkeyPatch, +) -> None: + controller = wintray._WindowsTrayController(mock=True, interval=60) + fingerprints = iter([(("history", 1, 10.0),), (("history", 1, 10.0),), (("history", 2, 11.0),)]) + monkeypatch.setattr( + menubar_state, + "history_source_scan", + lambda: menubar_state.HistorySourceScan(next(fingerprints), (), ()), + ) + calls: list[int] = [] + original = controller._load_entries + + def counting_load_entries(scan: menubar_state.HistorySourceScan) -> wintray._RefreshData: + calls.append(1) + return original(scan) + + monkeypatch.setattr(controller, "_load_entries", counting_load_entries) + + controller._build_state() + controller._build_state() + controller._build_state() + + assert calls == [1, 1] def test_hide_section_updates_preferences_and_visible_panel( diff --git a/usage_client.py b/usage_client.py index 85df97df..e204d112 100644 --- a/usage_client.py +++ b/usage_client.py @@ -12,6 +12,7 @@ import os import time from dataclasses import dataclass +from datetime import datetime from enum import StrEnum from pathlib import Path from typing import Any @@ -25,6 +26,7 @@ STATUS_FILE = os.path.expanduser("~/.claude/usage-status.json") LEGACY_STATUS_FILE = os.path.expanduser("~/.claude/usag-status.json") TT_STATUS_FILE = os.path.expanduser("~/.claude/tt-status.json") +CLAUDE_JSON_FILE = os.path.expanduser("~/.claude.json") CLAUDE_PROJECTS_DIR = Path(os.path.expanduser("~/.claude/projects")) # Stale files only affect hints; quota values still render. @@ -102,6 +104,21 @@ def _as_finite_float(value: Any) -> float | None: return numeric if math.isfinite(numeric) else None +def _iso_timestamp(value: Any) -> float | None: + if not isinstance(value, str): + return None + try: + parsed = datetime.fromisoformat(value) + except ValueError: + return None + if parsed.tzinfo is None: + return None + try: + return parsed.timestamp() + except (OSError, OverflowError, ValueError): + return None + + def _read_status_file() -> tuple[dict[str, Any], str, float] | None: """Read the first available status JSON, preferring usage-owned files.""" for path in (STATUS_FILE, LEGACY_STATUS_FILE, TT_STATUS_FILE): @@ -138,6 +155,82 @@ def _source_from_path(source_path: str) -> str: return "hook" +def _read_claude_json_snapshot() -> UsageSnapshot | None: + """Read Claude Code's own cached quota utilization as a fallback.""" + try: + os.stat(CLAUDE_JSON_FILE) + with open(CLAUDE_JSON_FILE, encoding="utf-8") as f: + data = json.load(f) + except (OSError, json.JSONDecodeError, UnicodeDecodeError): + return None + if not isinstance(data, dict): + return None + + cached = _as_dict(data.get("cachedUsageUtilization")) + fetched_at_ms = _as_finite_float(cached.get("fetchedAtMs")) + if fetched_at_ms is None: + return None + utilization = _as_dict(cached.get("utilization")) + five = _as_dict(utilization.get("five_hour")) + seven = _as_dict(utilization.get("seven_day")) + five_raw = five.get("utilization") + seven_raw = seven.get("utilization") + if five_raw is None and seven_raw is None: + return None + + five_pct = _pct(five_raw) if five_raw is not None else None + seven_pct = _pct(seven_raw) if seven_raw is not None else None + if five_pct is None and seven_pct is None: + return None + + now = time.time() + five_reset = _iso_timestamp(five.get("resets_at")) if five else None + seven_reset = _iso_timestamp(seven.get("resets_at")) if seven else None + if five and five_reset is None: + return None + if seven and seven_reset is None: + return None + five_reset = five_reset if five_reset is not None else now + seven_reset = seven_reset if seven_reset is not None else now + if five_pct is not None and five_reset < now: + five_pct = 0 + if seven_pct is not None and seven_reset < now: + seven_pct = 0 + + polled_at = fetched_at_ms / 1000 + return UsageSnapshot( + current_percent=five_pct, + current_reset_at=five_reset, + weekly_percent=seven_pct, + weekly_reset_at=seven_reset, + current_status="", + polled_at=polled_at, + is_stale=(now - polled_at) > STALE_SECONDS, + data_source="claude-json", + ) + + +def _time_adjusted(snapshot: UsageSnapshot) -> UsageSnapshot: + """Re-derive expiry-sensitive fields of a cached snapshot at the current time.""" + now = time.time() + five_pct = snapshot.current_percent + if five_pct is not None and snapshot.current_reset_at < now: + five_pct = 0 + seven_pct = snapshot.weekly_percent + if seven_pct is not None and snapshot.weekly_reset_at < now: + seven_pct = 0 + return UsageSnapshot( + current_percent=five_pct, + current_reset_at=snapshot.current_reset_at, + weekly_percent=seven_pct, + weekly_reset_at=snapshot.weekly_reset_at, + current_status=snapshot.current_status, + polled_at=snapshot.polled_at, + is_stale=(now - snapshot.polled_at) > STALE_SECONDS, + data_source=snapshot.data_source, + ) + + def _has_recent_claude_project_activity(now: float) -> bool: global _recent_activity_cache @@ -242,6 +335,10 @@ def __init__(self, *, interval_seconds: int = 60, mock: bool = False) -> None: self._cached_data: dict[str, Any] | None = None self._cached_path: str | None = None self._cached_mtime: float | None = None + self._claude_json_cached_path: str | None = None + self._claude_json_cached_mtime: float | None = None + self._claude_json_cached_snapshot: UsageSnapshot | None = None + self._claude_json_cache_valid = False async def aclose(self) -> None: return None @@ -250,6 +347,8 @@ async def fetch_once(self) -> PollOutcome: if self.mock: return self._mock_outcome() + claude_json_snapshot = self._read_claude_json_snapshot_cached() + if ( (stat_result := _status_file_stat()) is not None and self._cached_data is not None @@ -265,6 +364,8 @@ async def fetch_once(self) -> PollOutcome: self._cached_data = None self._cached_path = None self._cached_mtime = None + if claude_json_snapshot is not None: + return self._success_outcome(claude_json_snapshot) message_key = "usage_status_missing" if current_hook_state() in { "us-direct", @@ -281,6 +382,14 @@ async def fetch_once(self) -> PollOutcome: self._cached_path = source_path self._cached_mtime = mtime + status_polled_at = _as_finite_float(data.get("_received_at_ts")) + if claude_json_snapshot is not None and ( + not _has_complete_rate_limits(data) + or status_polled_at is None + or status_polled_at < claude_json_snapshot.polled_at + ): + return self._success_outcome(claude_json_snapshot) + if not _has_complete_rate_limits(data): outcome = PollOutcome( state=PollState.LOADING, @@ -302,10 +411,48 @@ async def fetch_once(self) -> PollOutcome: self._last_outcome = outcome return outcome + return self._success_outcome(snapshot, mtime=mtime, source_path=source_path) + + def _read_claude_json_snapshot_cached(self) -> UsageSnapshot | None: + try: + mtime = os.stat(CLAUDE_JSON_FILE).st_mtime + except OSError: + self._claude_json_cache_valid = False + self._claude_json_cached_path = None + self._claude_json_cached_mtime = None + self._claude_json_cached_snapshot = None + return None + if ( + self._claude_json_cache_valid + and self._claude_json_cached_path == CLAUDE_JSON_FILE + and self._claude_json_cached_mtime == mtime + ): + # The file may sit unchanged across a quota reset, so the expiry-derived + # fields must be recomputed on every hit — only the parse is cached. + if self._claude_json_cached_snapshot is None: + return None + return _time_adjusted(self._claude_json_cached_snapshot) + snapshot = _read_claude_json_snapshot() + self._claude_json_cache_valid = True + self._claude_json_cached_path = CLAUDE_JSON_FILE + self._claude_json_cached_mtime = mtime + self._claude_json_cached_snapshot = snapshot + return snapshot + + def _success_outcome( + self, + snapshot: UsageSnapshot, + *, + mtime: float | None = None, + source_path: str | None = None, + ) -> PollOutcome: now = time.time() message = _hook_broken_message(now, snapshot.polled_at) if snapshot.is_stale: - source_tag = "tt-status" if snapshot.data_source == "tt-fallback" else "usage" + source_tag = { + "tt-fallback": "tt-status", + "claude-json": "claude.json", + }.get(snapshot.data_source, "usage") mins = int((now - snapshot.polled_at) / 60) message = message or f"⚠ {source_tag} stale {mins}m" diff --git a/usage_session_resume.py b/usage_session_resume.py index 934a4a38..c3e4002b 100644 --- a/usage_session_resume.py +++ b/usage_session_resume.py @@ -527,9 +527,12 @@ def _clean_request(value: object) -> str: text = _IMAGE_MARKER.sub("", text).strip() if not text: return "" - if starts_with_image and _substantive_len(text) < _MIN_SUBSTANTIVE_CHARS: + substantive_len = _substantive_len(text) + if substantive_len == 0: return "" - if _substantive_len(text) < _MIN_SUBSTANTIVE_CHARS and not _has_structural_signal(text): + if starts_with_image and substantive_len < _MIN_SUBSTANTIVE_CHARS: + return "" + if substantive_len < _MIN_SUBSTANTIVE_CHARS and not _has_structural_signal(text): return "" if len(text) > _MAX_REQUEST_CHARS: text = text[: _MAX_REQUEST_CHARS - 1].rstrip() + "…" diff --git a/wintray.py b/wintray.py index 9285785e..8164ff31 100644 --- a/wintray.py +++ b/wintray.py @@ -8,6 +8,7 @@ import logging import os import threading +import time import tomllib import webbrowser from dataclasses import dataclass @@ -468,6 +469,13 @@ def __init__(self, mock: bool, interval: int) -> None: self.stopping = threading.Event() self.refresh_lock = threading.Lock() self._quota_notifier = QuotaNotifier(_quota_notification_thresholds()) + self.usage_client = ClaudeUsageClient(mock=mock) + self._last_tray_percent: float | None = None + self._last_tray_tooltip: str | None = None + self._last_injected_state: str | None = None + self._history_fingerprint: tuple[tuple[str, int, float], ...] | None = None + self._cached_history: _RefreshData | None = None + self._cached_projects: tuple[list[tuple[str, int, float | None]], ...] | None = None def _empty_state(self) -> menubar_state.PopoverState: missing = menubar_state._missing_row @@ -526,7 +534,7 @@ def on_loaded(self) -> None: # only re-applies after a visible panel switch reloads the document. if self.visible: self._place_window() - self.inject_state() + self.inject_state(force=True) def _working_area(self) -> tuple[int, int, int, int] | None: """Return the primary monitor work area (without taskbar).""" @@ -642,22 +650,32 @@ def refresh(self) -> None: threading.Thread(target=self._refresh_worker, daemon=True).start() def _refresh_worker(self) -> None: + debug_timing = os.environ.get("USAGE_DEBUG") == "1" + + def measure(stage: str, started_at: float) -> None: + if debug_timing: + elapsed_ms = (time.monotonic() - started_at) * 1000 + logger.debug("refresh_timing stage=%s elapsed_ms=%.1f", stage, elapsed_ms) + try: - self.latest_state = self._build_state() + self.latest_state = self._build_state(measure=measure, debug_timing=debug_timing) self._process_quota_notifications(self.latest_state) + started_at = time.monotonic() if debug_timing else 0.0 self._update_tray() + measure("update_tray", started_at) if self.visible: + started_at = time.monotonic() if debug_timing else 0.0 self.inject_state() + measure("inject_state", started_at) except Exception: if os.environ.get("USAGE_DEBUG") == "1": logger.warning("Windows tray refresh failed", exc_info=True) finally: self.refresh_lock.release() - def _load_entries(self) -> _RefreshData: + def _load_entries(self, scan: menubar_state.HistorySourceScan) -> _RefreshData: if self.mock: return _RefreshData([], None) - scan = menubar_state.history_source_scan() entries: list[UsageEntry] = [] error_key = None try: @@ -674,21 +692,50 @@ def _load_entries(self) -> _RefreshData: error_key = "history_load_error_parse" return _RefreshData(entries, error_key) - def _build_state(self) -> menubar_state.PopoverState: + def _build_state( + self, + *, + measure: Any = lambda _stage, _started_at: None, + debug_timing: bool = False, + ) -> menubar_state.PopoverState: + started_at = time.monotonic() if debug_timing else 0.0 codex_rows, _codex_pct, _model, codex_stale = menubar_state.codex_rows( mock=self.mock, language=self.language, burn_rate_trackers=self.burn_rate_trackers, ) + measure("codex_load", started_at) + started_at = time.monotonic() if debug_timing else 0.0 agy_result = menubar_agy.load_refresh_result(self.language) agy = agy_result.projection or menubar_agy.fallback_projection(self.language) - history = self._load_entries() - projects = ( - _mock_projects() - if self.mock - else menubar_state.project_rows_for_windows(history.entries) - ) + measure("agy_load", started_at) + started_at = time.monotonic() if debug_timing else 0.0 + scan = menubar_state.history_source_scan() + if menubar_state.history_cache_needs_reload( + self._history_fingerprint, + scan.fingerprint, + has_cached_result=( + self._cached_history is not None and self._cached_projects is not None + ), + ): + self._cached_history = self._load_entries(scan) + self._cached_projects = ( + _mock_projects() + if self.mock + else menubar_state.project_rows_for_windows(self._cached_history.entries) + ) + # A load error may be transient (e.g. a file locked mid-write); keep the + # fingerprint unset so the next poll retries instead of pinning the error. + self._history_fingerprint = ( + scan.fingerprint if self._cached_history.history_error_key is None else None + ) + history = self._cached_history + projects = self._cached_projects + assert history is not None and projects is not None + measure("history_load", started_at) + started_at = time.monotonic() if debug_timing else 0.0 outcome = asyncio.run(self._fetch()) + measure("fetch", started_at) return menubar_state.build_popover_state( outcome=outcome, codex_rows=codex_rows, @@ -720,25 +767,30 @@ def _build_state(self) -> menubar_state.PopoverState: ) async def _fetch(self) -> Any: - client = ClaudeUsageClient(mock=self.mock) - try: - return await client.fetch_once() - finally: - await client.aclose() + return await self.usage_client.fetch_once() def _update_tray(self) -> None: if self.icon is None: return - self.icon.icon = draw_tray_icon(self.latest_state.claude_session.percent) - self.icon.title = build_tooltip(self.latest_state) + percent = self.latest_state.claude_session.percent + tooltip = build_tooltip(self.latest_state) + if percent == self._last_tray_percent and tooltip == self._last_tray_tooltip: + return + self.icon.icon = draw_tray_icon(percent) + self.icon.title = tooltip + self._last_tray_percent = percent + self._last_tray_tooltip = tooltip - def inject_state(self) -> None: + def inject_state(self, *, force: bool = False) -> None: if self.window is None: return encoded = json.dumps( _state_payload(self.latest_state), ensure_ascii=False, separators=(",", ":") ) + if not force and encoded == self._last_injected_state: + return self.window.evaluate_js(f"window.usageApplyState({encoded})") + self._last_injected_state = encoded def show_panel(self, _icon: Any = None, _item: Any = None) -> None: if self.visible: @@ -750,7 +802,7 @@ def show_panel(self, _icon: Any = None, _item: Any = None) -> None: self.visible = True self._place_window() self.window.show() - self.inject_state() + self.inject_state(force=True) self.refresh() def switch_panel(self, panel_id: str) -> None: