Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
2 changes: 1 addition & 1 deletion CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
10 changes: 10 additions & 0 deletions menubar_state.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 ()
Expand Down
62 changes: 55 additions & 7 deletions session_hooks.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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)
Expand Down Expand Up @@ -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:
Expand Down
14 changes: 14 additions & 0 deletions tests/test_menubar_state.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down
9 changes: 9 additions & 0 deletions tests/test_session_resume.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
19 changes: 8 additions & 11 deletions tests/test_setup_hook.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
{
Expand Down Expand Up @@ -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"]
Expand All @@ -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(
{
Expand All @@ -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}],
}
]
},
Expand All @@ -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
12 changes: 5 additions & 7 deletions tests/test_setup_hook_resume.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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"


Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
8 changes: 3 additions & 5 deletions tests/test_setup_hook_terse.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"]
Expand Down Expand Up @@ -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"


Expand Down Expand Up @@ -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:
Expand Down
Loading
Loading