from __future__ import annotations import json import os from datetime import datetime from pathlib import Path import pytest import runtime_paths import statusline from health_client import AccountSnapshot, HealthStatus from health_store import HealthSnapshotStore def write_codex_health( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, slug: str, snapshot: AccountSnapshot, ) -> None: runtime = tmp_path / "runtime" runtime.mkdir(exist_ok=True) monkeypatch.setenv("SYSTRAY_RUNTIME_DIR", str(runtime)) HealthSnapshotStore(runtime / "health_cache.json", stale_after_s=0).write({slug: snapshot}) def test_configured_effort_prefers_live_payload_over_settings_default( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: fake_home = tmp_path / "home" (fake_home / ".claude").mkdir(parents=True) (fake_home / ".claude" / "settings.json").write_text( json.dumps({"effortLevel": "low"}), encoding="utf-8" ) monkeypatch.setattr(statusline.Path, "home", classmethod(lambda cls: fake_home)) assert statusline._configured_effort({"effort": {"level": "xhigh"}}) == "xhigh" assert statusline._configured_effort({}) == "low" def test_ingest_claude_usage_writes_account_local_live_limits(tmp_path: Path) -> None: account_home = tmp_path / "claude-accounts" / "multideal" / "CLAUDE_HOME" account_home.mkdir(parents=True) assert statusline.ingest_claude_usage( { "rate_limits": { "five_hour": {"used_percentage": 19, "resets_at": 1_710_003_600}, "seven_day": {"used_percentage": 37, "resets_at": 1_710_604_800}, } }, { "SYSTRAY_CLAUDE_ACCOUNT_SLUG": "multideal", "SYSTRAY_CLAUDE_ACCOUNT_HOME": str(account_home), }, ) assert json.loads((account_home / "rate-limits-cache.json").read_text()) == { "five_hour": {"percent": 19, "resets_at": 1_710_003_600}, "seven_day": {"percent": 37, "resets_at": 1_710_604_800}, } def test_ingest_claude_usage_derives_account_from_existing_config_dir(tmp_path: Path) -> None: account_home = tmp_path / "claude-accounts" / "multideal" / "CLAUDE_HOME" account_home.mkdir(parents=True) assert statusline.ingest_claude_usage( {"rate_limits": {"five_hour": {"used_percentage": 23, "resets_at": 99}}}, {"CLAUDE_CONFIG_DIR": str(account_home)}, ) assert json.loads((account_home / "rate-limits-cache.json").read_text()) == { "five_hour": {"percent": 23, "resets_at": 99} } def test_session_pin_survives_default_switch_and_renders_cached_codex_usage( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: state_dir = tmp_path / "state" account_home = tmp_path / "accounts" / "rafa" / "CODEX_HOME" account_home.mkdir(parents=True) payload = { "session_id": "session-a", "model": {"display_name": "GPT-5.6 Terra"}, "workspace": {"current_dir": "/home/user/Projects/overdeck"}, "context_window": {"used_percentage": 42}, } pin = statusline.pin_session( payload, { "SYSTRAY_CODEX_ACCOUNT_SLUG": "rafa", "SYSTRAY_CODEX_ACCOUNT_HOME": str(account_home), }, state_dir, now=1_710_000_000.0, ) write_codex_health( tmp_path, monkeypatch, "rafa", AccountSnapshot( HealthStatus.OK, 10, 20, primary_reset_at=1_710_003_600.0, secondary_reset_at=1_710_604_800.0, checked_at=1_710_000_000.0, ), ) assert pin.account_slug == "rafa" rendered = statusline.render(payload, state_dir, now=1_710_000_000.0) assert "GPT-5.6 Terra" in rendered assert "ctx:" in rendered assert "42%" in rendered assert "codex:rafa" in rendered assert "5h:" in rendered assert "10%" in rendered assert "7d:" in rendered assert "20%" in rendered assert "rate_limits" not in rendered assert "\033[01;32mGPT-5.6 Terra" in rendered assert "\033[00;35m5h:" in rendered assert "\033[00;32m10%\033[00m" in rendered def test_render_omits_unavailable_five_hour_window( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: state_dir = tmp_path / "state" account_home = tmp_path / "accounts" / "rafa" / "CODEX_HOME" account_home.mkdir(parents=True) statusline.write_session_pin( state_dir, statusline.SessionPin("session-a", "rafa", account_home, 1_710_000_000.0), ) write_codex_health( tmp_path, monkeypatch, "rafa", AccountSnapshot( HealthStatus.OK, None, 20, secondary_reset_at=1_710_604_800.0, checked_at=1_710_000_000.0, ), ) rendered = statusline.render({"session_id": "session-a"}, state_dir, now=1_710_000_000.0) codex_line = rendered.splitlines()[1] assert "5h:" not in codex_line assert "7d:" in codex_line assert "20%" in codex_line def test_render_replaces_missing_pin_with_current_valid_default( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: state_dir = tmp_path / "state" current_home = tmp_path / "accounts" / "multideal" / "CODEX_HOME" current_home.mkdir(parents=True) payload = {"session_id": "session-a"} statusline.write_session_pin( state_dir, statusline.SessionPin("session-a", "new-account", tmp_path / "missing", 1_710_000_000.0), ) monkeypatch.setattr( statusline, "_default_pin", lambda session_id, now: statusline.SessionPin(session_id, "multideal", current_home, now), ) write_codex_health( tmp_path, monkeypatch, "multideal", AccountSnapshot( HealthStatus.OK, 4, 8, primary_reset_at=1_710_003_600.0, secondary_reset_at=1_710_604_800.0, checked_at=1_710_000_001.0, ), ) rendered = statusline.render(payload, state_dir, now=1_710_000_001.0, environment={}) assert rendered.splitlines()[1].startswith("codex:multideal") assert "4%" in rendered.splitlines()[1] assert "new-account" not in rendered replacement = statusline.SessionPin( "session-a", "multideal", current_home, 1_710_000_001.0 ) assert statusline.read_session_pin(state_dir, "session-a") == replacement def test_render_hides_missing_pin_slug_when_no_valid_default( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: state_dir = tmp_path / "state" payload = {"session_id": "session-a"} statusline.write_session_pin( state_dir, statusline.SessionPin("session-a", "new-account", tmp_path / "missing", 1_710_000_000.0), ) monkeypatch.setattr(statusline, "_default_pin", lambda session_id, now: None) rendered = statusline.render(payload, state_dir, now=1_710_000_001.0, environment={}) assert rendered.splitlines()[1] == "codex:unavailable (pinned account missing)" assert "new-account" not in rendered def test_start_session_pins_the_running_account(tmp_path: Path) -> None: state_dir = tmp_path / "state" account_home = tmp_path / "accounts" / "rafa" / "CODEX_HOME" account_home.mkdir(parents=True) payload = {"session_id": "session-a"} pin = statusline.start_session( payload, {"SYSTRAY_CODEX_ACCOUNT_SLUG": "rafa", "SYSTRAY_CODEX_ACCOUNT_HOME": str(account_home)}, state_dir, now=1_710_000_000.0, ) assert pin is not None assert statusline.read_session_pin(state_dir, "session-a") == pin def test_render_restores_provider_neutral_session_metadata_and_colors( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: state_dir = tmp_path / "state" account_home = tmp_path / "accounts" / "rafa" / "CODEX_HOME" account_home.mkdir(parents=True) transcript = tmp_path / "session.jsonl" records = [ { "type": "user", "promptId": "prompt-1", "timestamp": "2024-03-09T16:00:00Z", "message": {"content": "first"}, }, { "type": "assistant", "timestamp": "2024-03-09T16:00:01Z", "message": {"content": [{"type": "text", "text": "answer"}]}, }, { "type": "user", "promptId": "prompt-1", "timestamp": "2024-03-09T16:00:02Z", "message": {"content": "duplicate journal record"}, }, { "type": "user", "toolUseResult": {"ok": True}, "timestamp": "2024-03-09T16:30:00Z", "message": {"content": [{"type": "tool_result"}]}, }, { "type": "user", "promptId": "prompt-2", "timestamp": "2024-03-09T17:01:00Z", "message": {"content": "second"}, }, { "type": "assistant", "timestamp": "2024-03-09T17:01:06Z", "message": {"content": [{"type": "text", "text": "answer"}]}, }, ] transcript.write_text("".join(json.dumps(record) + "\n" for record in records), encoding="utf-8") statusline.write_session_pin( state_dir, statusline.SessionPin("session-a", "rafa", account_home, 1_710_000_000.0), ) payload = { "session_id": "session-a", "transcript_path": str(transcript), "model": {"display_name": "GPT-5.6 Terra"}, "workspace": {"current_dir": "/home/user/Projects/overdeck"}, "context_window": {"used_percentage": 42}, } monkeypatch.setattr(statusline, "_configured_effort", lambda payload: "med") rendered = statusline.render(payload, state_dir, now=1_710_003_736.0) first_line = rendered.splitlines()[0] assert first_line.startswith( "\033[01;32mGPT-5.6 Terra(med)\033[00m:" "\033[01;34m~/Projects/overdeck\033[00m" ) assert "\033[00;36m2t\033[00m" in first_line assert "[FT]" not in first_line expected_activity = datetime.fromtimestamp(1_710_003_666.0).strftime("%H:%M:%S %d/%m") assert f"\033[00;33m{expected_activity}\033[00m" in first_line assert "\033[02m1h1m\033[00m" in first_line assert "\033[00;33mctx:\033[01;33m42%\033[00m" in first_line assert "$" not in rendered assert "FT:" not in rendered assert " I:" not in rendered assert " O:" not in rendered assert " C:" not in rendered def test_render_marks_old_activity_red_and_ignores_malformed_transcript_lines( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: state_dir = tmp_path / "state" account_home = tmp_path / "CODEX_HOME" account_home.mkdir() transcript = tmp_path / "session.jsonl" transcript.write_text( "not-json\n" + json.dumps( { "type": "assistant", "timestamp": "2024-03-09T16:00:00Z", "message": {"content": "answer"}, } ) + "\n", encoding="utf-8", ) statusline.write_session_pin( state_dir, statusline.SessionPin("session-a", "rafa", account_home, 1_710_000_000.0), ) monkeypatch.setattr(statusline, "_configured_effort", lambda payload: "") rendered = statusline.render( {"session_id": "session-a", "transcript_path": str(transcript)}, state_dir, now=1_710_003_601.0, ) expected_activity = datetime.fromtimestamp(1_710_000_000.0).strftime("%H:%M:%S %d/%m") assert f"\033[01;31m{expected_activity}\033[00m" in rendered assert "[FT]" not in rendered assert "0t" not in rendered def test_session_pin_follows_the_account_the_resumed_session_runs_on(tmp_path: Path) -> None: state_dir = tmp_path / "state" payload = {"session_id": "session-a"} homes = {} for slug in ("multideal", "zync"): home = tmp_path / "accounts" / slug / "CODEX_HOME" home.mkdir(parents=True) homes[slug] = home first = statusline.pin_session( payload, { "SYSTRAY_CODEX_ACCOUNT_SLUG": "multideal", "SYSTRAY_CODEX_ACCOUNT_HOME": str(homes["multideal"]), }, state_dir, now=1_710_000_000.0, ) resumed = statusline.pin_session( payload, { "SYSTRAY_CODEX_ACCOUNT_SLUG": "zync", "SYSTRAY_CODEX_ACCOUNT_HOME": str(homes["zync"]), }, state_dir, now=1_710_000_100.0, ) unlaunched = statusline.pin_session(payload, {}, state_dir, now=1_710_000_200.0) assert first.account_slug == "multideal" assert resumed.account_slug == "zync" assert resumed.account_home == homes["zync"] assert unlaunched.account_slug == "zync" def _claude_account_home(tmp_path: Path, slug: str = "multideal") -> Path: account_home = tmp_path / "claude-accounts" / slug / "CLAUDE_HOME" account_home.mkdir(parents=True) return account_home def test_render_shows_pinned_claude_account_usage(tmp_path: Path) -> None: state_dir = tmp_path / "state" account_home = _claude_account_home(tmp_path) (account_home / "rate-limits-cache.json").write_text( json.dumps( { "five_hour": {"utilization": 12, "resets_at": 1_710_003_600}, "seven_day": {"utilization": 44, "resets_at": 1_710_604_800}, } ), encoding="utf-8", ) os.utime(account_home / "rate-limits-cache.json", (1_710_000_000, 1_710_000_000)) rendered = statusline.render( {"session_id": "session-a"}, state_dir, now=1_710_000_000.0, environment={ "SYSTRAY_CLAUDE_ACCOUNT_SLUG": "multideal", "SYSTRAY_CLAUDE_ACCOUNT_HOME": str(account_home), }, ) claude_line = rendered.splitlines()[2] assert claude_line.startswith("claude:multideal ") assert "5h:" in claude_line assert "12%" in claude_line assert "7d:" in claude_line assert "44%" in claude_line assert "stale" not in claude_line def test_render_derives_claude_account_from_config_dir_and_marks_stale(tmp_path: Path) -> None: account_home = _claude_account_home(tmp_path) (account_home / "rate-limits-cache.json").write_text( json.dumps({"five_hour": {"percent": 7, "resets_at": 1_710_003_600}}), encoding="utf-8" ) os.utime(account_home / "rate-limits-cache.json", (1_710_000_000, 1_710_000_000)) rendered = statusline.render( {"session_id": "session-a"}, tmp_path / "state", now=1_710_000_000.0 + statusline.STALE_AFTER_SECONDS + 1, environment={"CLAUDE_CONFIG_DIR": str(account_home)}, ) claude_line = rendered.splitlines()[2] assert claude_line.startswith("claude:multideal ") assert "7%" in claude_line assert "stale" in claude_line assert "7d:" not in claude_line def test_render_reports_missing_claude_cache_and_account(tmp_path: Path) -> None: account_home = _claude_account_home(tmp_path) without_cache = statusline.render( {"session_id": "session-a"}, tmp_path / "state", now=1_710_000_000.0, environment={"CLAUDE_CONFIG_DIR": str(account_home)}, ) without_account = statusline.render( {"session_id": "session-a"}, tmp_path / "state", now=1_710_000_000.0, environment={}, ) assert without_cache.splitlines()[2] == "claude:multideal usage:unavailable (no cache)" assert without_account.splitlines()[2] == "claude:unavailable (no account)" def test_claude_account_resolves_home_from_slug(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: account_home = _claude_account_home(tmp_path, "zync") monkeypatch.setenv(runtime_paths.RUNTIME_DIR_ENV, str(tmp_path)) assert statusline.claude_account({"SYSTRAY_CLAUDE_ACCOUNT_SLUG": "zync"}) == ( "zync", account_home, ) def _pinned_session(tmp_path: Path) -> tuple[Path, dict[str, object]]: state_dir = tmp_path / "state" account_home = tmp_path / "accounts" / "zync" / "CODEX_HOME" account_home.mkdir(parents=True) statusline.write_session_pin( state_dir, statusline.SessionPin("session-a", "zync", account_home, 1_710_000_000.0), ) return state_dir, {"session_id": "session-a"} def test_render_follows_the_indicator_health_cache_between_renders( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: state_dir, payload = _pinned_session(tmp_path) write_codex_health( tmp_path, monkeypatch, "zync", AccountSnapshot( HealthStatus.UNKNOWN, None, None, checked_at=1_710_000_000.0, detail=None ), ) blocked = statusline.render(payload, state_dir, now=1_710_000_010.0, environment={}) assert blocked.splitlines()[1] == "codex:zync usage:unavailable (unknown)" write_codex_health( tmp_path, monkeypatch, "zync", AccountSnapshot( HealthStatus.OK, None, 56, secondary_reset_at=1_710_600_000.0, checked_at=1_710_000_060.0, ), ) healed = statusline.render(payload, state_dir, now=1_710_000_070.0, environment={}) assert "56%" in healed.splitlines()[1] assert "unavailable" not in healed.splitlines()[1] def test_render_marks_a_health_cache_the_indicator_stopped_updating( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: state_dir, payload = _pinned_session(tmp_path) write_codex_health( tmp_path, monkeypatch, "zync", AccountSnapshot( HealthStatus.OK, None, 56, secondary_reset_at=1_710_600_000.0, checked_at=1_710_000_000.0, ), ) rendered = statusline.render(payload, state_dir, now=1_710_003_600.0, environment={}) assert "stale" in rendered.splitlines()[1] def test_render_reports_an_account_the_health_cache_never_saw( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: state_dir, payload = _pinned_session(tmp_path) write_codex_health( tmp_path, monkeypatch, "multideal", AccountSnapshot(HealthStatus.OK, 1, 2, checked_at=1_710_000_000.0), ) rendered = statusline.render(payload, state_dir, now=1_710_000_010.0, environment={}) assert rendered.splitlines()[1] == "codex:zync usage:unavailable (no health cache)" def test_render_shows_limits_the_indicator_retained_through_an_unknown_probe( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: state_dir, payload = _pinned_session(tmp_path) write_codex_health( tmp_path, monkeypatch, "zync", AccountSnapshot( HealthStatus.UNKNOWN, None, 100, secondary_reset_at=1_710_600_000.0, checked_at=1_710_000_000.0, ), ) rendered = statusline.render(payload, state_dir, now=1_710_000_010.0, environment={}) assert "100%" in rendered.splitlines()[1] assert "unavailable" not in rendered.splitlines()[1] def test_render_reports_a_broken_account_even_with_retained_limits( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: state_dir, payload = _pinned_session(tmp_path) write_codex_health( tmp_path, monkeypatch, "zync", AccountSnapshot( HealthStatus.BROKEN, None, 100, secondary_reset_at=1_710_600_000.0, checked_at=1_710_000_000.0, detail="signed out", ), ) rendered = statusline.render(payload, state_dir, now=1_710_000_010.0, environment={}) assert rendered.splitlines()[1] == "codex:zync usage:unavailable (signed out)"