diff --git a/CHANGELOG.md b/CHANGELOG.md index acda16b..1ce94ba 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,48 @@ All notable changes to this project are documented here. The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and the project aims to follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.3.0] - 2026-07-25 + +### Added +- Optional hot-switch mode: saved Claude config directories become credential + profiles for one main session directory, with lock-safe Keychain/config swaps, + current-token sync-back, `cctop switch [NAME]`, and the TUI `x` key. +- A headless `cctop autoswitch` supervisor keeps automatic rotation running + independently of the TUI and terminal job control. +- Automatic rotation to the healthy Claude profile with the most headroom at + 1% remaining (configurable), including delegated refresh of expired saved + access tokens without a login/logout cycle. Candidates must have strictly + more headroom than the active profile, preventing ping-pong while still using + every account that can make progress. + +### Changed +- `get_token` no longer falls back to the default account's Keychain service, so + a logged-out config dir reads as having no token instead of borrowing the + default account's. The default-profile Keychain and identity-path rules now + live in one place (`authctl`) instead of being restated per call site. +- A rejected live credential now triggers delegated refresh and then immediate + profile recovery, even while usage reads are rate-limited. The dead main + credential is never synced over its saved profile, and stale usage data can + no longer select a locally expired target. +- Automatic rotation now accepts an above-threshold profile when it has + strictly more headroom than the active one, so a fully exhausted login can + hand work to a still-usable account without introducing switch ping-pong. +- The TUI and autoswitch supervisor now share last-good usage metadata. During + HTTP 429 backoff, new cctop processes keep showing cached percentages, reset + times, and reading age instead of replacing useful data with an error. +- Saved-profile expiry is checked before the usage request, so a logged-out + account is refreshed or reported as `needs re-login` instead of being + misclassified when the usage endpoint returns a fleet-wide 429 first. +- A recent but partial shared cache no longer suppresses accounts without a + successful reading; every configured profile remains visible. +- A successful fresh profile whose usage windows have not started now ranks as + 0% used instead of being excluded from automatic switching as unavailable. + +### Preserved +- Original per-config-directory session monitoring and account provisioning + remain the default behavior when `hot_switch` is disabled; the login profiles + cctop snapshots for itself are hidden from account discovery while it is off. + ## [0.2.0] - 2026-07-19 ### Added @@ -52,5 +94,6 @@ follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html). - `LICENSE` (MIT), `SECURITY.md`, and packaging metadata (distribution `cctop-tui`; the command stays `cctop`). +[0.3.0]: https://github.com/ryanirl/cctop/releases/tag/v0.3.0 [0.2.0]: https://github.com/ryanirl/cctop/releases/tag/v0.2.0 [0.1.0]: https://github.com/ryanirl/cctop/releases/tag/v0.1.0 diff --git a/README.md b/README.md index 1b6424f..ef77038 100644 --- a/README.md +++ b/README.md @@ -14,9 +14,9 @@ those up too. Everything is read from files the tools already write plus a couple of free, read-only usage reads; nothing here ever spends message quota. cctop touches -your credentials only to make those reads, never logs or transmits a token, and -is read-only except for two explicit, additive account actions. See -[SECURITY.md](SECURITY.md) for the full trust statement. +your credentials only for those reads and optional local hot switching, never +logs or transmits a token, and otherwise changes state only through explicit +account actions. See [SECURITY.md](SECURITY.md) for the full trust statement. ![cctop](https://raw.githubusercontent.com/ryanirl/cctop/main/docs/hero.png) @@ -66,14 +66,15 @@ cctop --no-limits # skip the usage fetch (no network, session table only) cctop setup # pick a provider; hand off to its agent to help configure cctop accounts # list discovered accounts (read-only) +cctop switch [NAME] # hot-switch main Claude sessions (or choose the best) cctop doctor # read-only self-check (platform, binaries, token/expiry) cctop config init # write a starter ~/.config/cctop/config.toml (optional) cctop add-account # provision a new account (dry-run; see Accounts below) ``` -TUI keys: `r` refresh now · `R` refresh token · `a` add account · `s` stats · -`,` settings · `q` quit. The footer shows a live countdown to the next auto -refresh; `r` refreshes immediately and resets it. +TUI keys: `r` refresh now · `R` refresh token · `x` hot-switch account · `a` +add account · `s` stats · `,` settings · `q` quit. The footer shows a live +countdown to the next auto refresh; `r` refreshes immediately and resets it. ### Token refresh (`R`) @@ -82,10 +83,9 @@ lazily when you *use* an account, so an account you are merely monitoring drifts past expiry and the usage read starts failing (`token expired`). Press `R` and cctop asks the tool that owns the credential to renew it: it runs `claude mcp list` under each account's config dir (a quota-free command whose -startup renews and rewrites the Keychain record). **cctop never writes a -credential itself** — it only triggers the owner binary and reads the result — -and it is a no-op on tokens that are still valid. An account whose refresh token -is itself dead reports "needs re-login" (only a fresh `/login` can fix that). +startup renews and rewrites the Keychain record). This delegated refresh is a +no-op on tokens that are still valid. An account whose refresh token is itself +dead reports "needs re-login" (only a fresh `/login` can fix that). ### Accounts @@ -102,6 +102,64 @@ existing config, and signs you in. It is **strictly additive** — it never deletes, overwrites, or modifies existing config, credentials, or sessions, and only writes a shell alias if you explicitly ask for one. +### Hot-switch mode + +The original separate-session mode remains the default. To keep one main +`~/.claude` session history and use the other config directories only as saved +login profiles, enable: + +```toml +[settings] +hot_switch = true +main_config_dir = "~/.claude" +auto_switch_remaining_percent = 1 +``` + +Press `x` (or run `cctop switch NAME`) to swap the healthiest saved login into +the main Keychain/config store. No aliases or account mapping are required: +cctop identifies logins by their Claude organization identity, collapses +duplicate login directories automatically, and labels them from their saved +email address. If the active login exists only in the mutable main store, cctop +automatically preserves it under `~/.config/cctop/profiles/` before switching. +Before each swap, cctop copies the main credential back to its active saved +profile so refresh-token rotation is retained. Claude Code's own advisory locks +make the update safe around running sessions; they pick up the new login after +the macOS Keychain cache refreshes. + +While the TUI is running, cctop automatically chooses the healthy Claude +profile with the most headroom when the active profile reaches 1% remaining +(99% used). This threshold is configurable with +`auto_switch_remaining_percent`; a profile that is itself past the threshold is +chosen only when it still has strictly more headroom than the active profile. +That lets a 93%-used login take over from one at 100% without allowing two +equally exhausted logins to ping-pong. Disable `hot_switch` at any time to return +to the original per-directory session view — the saved profiles cctop manages +for itself stay out of the account list while it is off. + +For rotation that does not depend on keeping the TUI in the foreground, run +`cctop autoswitch`. It polls immediately and then at `limits_refresh_seconds`, +refreshes dormant saved logins, and applies the same rotation policy as the TUI. +It is suitable for a user service such as macOS launchd. Use +`cctop autoswitch --once` to verify the configured fleet without starting the +long-running supervisor. Successful usage readings are shared through a local +metadata-only cache, so another cctop process still shows the last percentages, +reset times, and reading age when Anthropic temporarily returns HTTP 429. +Locally expired profiles are refreshed before the usage request; a dead refresh +token is shown as `needs re-login` rather than being masked by a fleet-wide 429. +An incomplete cache never hides accounts that lack a successful reading. + +Keeping the saved profile in sync is what avoids re-logins: an access token +lasts ~12-15h, but the refresh token behind it is what actually keeps an +account alive, and it rotates as Claude Code uses it. cctop copies the current +main credential back to its saved profile before normal swaps and limits polls +while that credential is valid, so a profile is never reactivated with a +superseded refresh token. Only a refresh token that is genuinely dead needs +`/login` again, and nothing local can substitute for that OAuth round trip. If +the live main credential is rejected, +cctop first asks Claude Code to refresh it; when that refresh token is dead, +cctop preserves the saved copy and immediately activates a ready profile. This +recovery does not depend on the usage endpoint being available. + ### Configuration (optional) cctop needs no configuration. If you want to rename, hide, reorder, or add diff --git a/SECURITY.md b/SECURITY.md index c69cdd7..c2e309e 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -16,8 +16,8 @@ not do, because a tool that sits next to your credentials should be auditable. | macOS Keychain item `Claude Code-credentials-`, or `~/.claude*/.credentials.json` | the OAuth token used to authenticate the two GETs above | | `~/.codex/auth.json` | the token used to authenticate the Codex usage GET | -All of it is read-only. The two usage GETs are plain reads that consume no -message quota. +The two usage GETs are plain reads that consume no message quota. Credential +writes happen only when hot-switch mode is enabled, as described below. ## What cctop does with your token @@ -28,13 +28,14 @@ message quota. account. - The token is placed in the `Authorization` header of the usage GET and used nowhere else. -- The token is **never** logged, printed, written to disk by cctop, included in - `--json` output, or transmitted anywhere except the provider's own API - (`api.anthropic.com` / `chatgpt.com`) that issued it. +- The token is **never** logged, printed, included in `--json` output, or + transmitted anywhere except the provider's own API (`api.anthropic.com` / + `chatgpt.com`) that issued it. In hot-switch mode, complete credential + records move only between their existing local Keychain/file stores. ## The only things cctop ever writes -cctop is read-only except for two explicit, additive actions you invoke: +cctop is read-only except for these account-management actions: 1. **Account provisioning** (`add-account`, the `a` key): creates a new `~/.claude-N` directory; with `--alias` appends one shell-rc line (after @@ -46,9 +47,21 @@ cctop is read-only except for two explicit, additive actions you invoke: **Claude Code binary itself** renews its own token at startup (a quota-free, non-interactive command). cctop does not write the credential; it delegates to the tool that owns it. +3. **Hot switching** (`hot_switch = true`, the `x` key, `cctop switch`, and the + configured automatic threshold): copies a complete saved credential record + into the main Claude Code Keychain store and replaces only the + `oauthAccount` field in the main config. Before switching away, the current + main credential is synced back to its matching saved profile so a rotated + refresh token is preserved. If the active login has no stable saved profile, + cctop creates one automatically under `~/.config/cctop/profiles/`; + its credential remains in a per-profile Keychain item (or the existing + file-backed format), while the identity file contains only Claude's account + metadata. Writes are performed under Claude Code's own advisory + credential/config locks; file-backed credentials use atomic replace, and + Keychain payloads are sent through `security -i` stdin rather than exposed + as plaintext process arguments. -There is deliberately **no** delete, logout, or credential-writing path in -cctop's own code. +There is deliberately **no** delete or logout path. ## Reporting diff --git a/pyproject.toml b/pyproject.toml index ccaf4eb..81f0ae6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "cctop-tui" -version = "0.2.0" +version = "0.3.0" description = "A live terminal monitor for multiple Claude Code and OpenAI Codex accounts, in the spirit of nvtop/htop." readme = "README.md" requires-python = ">=3.11" diff --git a/src/cctop/app.py b/src/cctop/app.py index 74932e3..edf417b 100644 --- a/src/cctop/app.py +++ b/src/cctop/app.py @@ -19,6 +19,8 @@ from __future__ import annotations from datetime import date, datetime, timezone +from pathlib import Path +from typing import TYPE_CHECKING from rich.console import Group from rich.table import Table as RichTable @@ -29,6 +31,7 @@ from textual.timer import Timer from textual.widgets import DataTable, Rule, Static +from . import config as config_module from .cli import ( _format_age, _format_context, @@ -43,6 +46,9 @@ from .models import AccountLimits, LimitWindow, SessionState from .monitor import FleetMonitor +if TYPE_CHECKING: + from .switcher import SwitchResult + TEAL = "#20B2AA" DIM = "grey37" MUTED = "grey50" @@ -108,6 +114,10 @@ def _account_block( if account.source != "api": lines.append(Text(account.error or "no limit data", style=MUTED)) else: + if account.error: + age = _format_age(account.fetched_at, now) + header.append(f" cached {age} ago · ", style=MUTED) + header.append(account.error, style=MUTED) lines.extend(_gauge_line(window, now, bar_width, label_width) for window in account.windows) return Group(*lines) @@ -212,6 +222,7 @@ class CctopApp(App): ("q", "quit", "Quit"), ("r", "refresh_limits", "Refresh limits"), ("R", "refresh_token", "Refresh token"), + ("x", "hot_switch", "Switch account"), ("a", "add_account", "Add account"), ("s", "stats", "Stats"), ("comma", "settings", "Settings"), @@ -222,11 +233,23 @@ def __init__( accounts: list[Account], limits_interval: float = 180.0, heatmap_weeks: int = 26, + main_config_dir: Path | None = None, + auto_switch_remaining_percent: float = 1.0, + hot_switch: bool = False, ) -> None: super().__init__() - self.monitor = FleetMonitor(accounts) + self.monitor = FleetMonitor( + accounts, + main_config_dir=main_config_dir, + auto_switch_remaining_percent=auto_switch_remaining_percent, + hot_switch=hot_switch, + limits_cache_path=config_module.config_dir() / "limits-cache.json", + ) self._limits_interval = limits_interval self._heatmap_weeks = heatmap_weeks + self._main_config_dir = main_config_dir or Path.home() / ".claude" + self._auto_switch_remaining_percent = auto_switch_remaining_percent + self._hot_switch_enabled = hot_switch self._limits_timer: Timer | None = None self._states_by_id: dict[str, SessionState] = {} self._selected_session_id: str | None = None @@ -400,7 +423,10 @@ def _render_footer(self, states: list[SessionState], now: datetime) -> None: "default", ), (f" usage {age} · {nxt}", MUTED), - (" r refresh · R token · a add · s stats · , settings · q quit", MUTED), + ( + " r refresh · R token · x switch · a add · s stats · , settings · q quit", + MUTED, + ), ) self.query_one("#footer", Static).update(footer) @@ -410,7 +436,10 @@ def _render_footer(self, states: list[SessionState], now: datetime) -> None: def _tick_limits(self) -> None: now = datetime.now(timezone.utc) limits = self.monitor.poll_limits(now) + result = self.monitor.maybe_auto_switch() self.call_from_thread(self._render_limits, limits, now) + if result is not None: + self.call_from_thread(self._show_switch_result, result, "auto switch") def action_stats(self) -> None: from .stats_screen import StatsScreen @@ -423,7 +452,16 @@ def action_settings(self) -> None: from .settings_screen import SettingsScreen, build_rows rows = build_rows(discover_accounts(), config_module.load_config()) - self.push_screen(SettingsScreen(self._limits_interval, self._heatmap_weeks, rows)) + self.push_screen( + SettingsScreen( + self._limits_interval, + self._heatmap_weeks, + self._main_config_dir, + self._auto_switch_remaining_percent, + self._hot_switch_enabled, + rows, + ) + ) def apply_settings(self) -> None: """Reload the config file and apply it live (called by the settings screen). @@ -435,18 +473,63 @@ def apply_settings(self) -> None: from . import config as config_module from .collect import resolve_accounts + from .switcher import ensure_stable_accounts config = config_module.load_config() self._limits_interval = config.limits_refresh_seconds(180.0) self._heatmap_weeks = config.heatmap_weeks(26) + self._main_config_dir = config.main_config_dir() + self._auto_switch_remaining_percent = config.auto_switch_remaining_percent() + self._hot_switch_enabled = config.hot_switch() self.monitor.limits_interval = timedelta(seconds=self._limits_interval) - self.monitor.accounts = resolve_accounts(config) + accounts = resolve_accounts(config) + if self._hot_switch_enabled: + accounts = ensure_stable_accounts(accounts, self._main_config_dir) + self.monitor.accounts = accounts + self.monitor.main_config_dir = self._main_config_dir + self.monitor.auto_switch_remaining_percent = self._auto_switch_remaining_percent + self.monitor.hot_switch = self._hot_switch_enabled self._reschedule_limits_timer() self.action_refresh_limits() self._tick_sessions() self._tick_stats() + def action_hot_switch(self) -> None: + """Switch the running main Claude sessions to the healthiest profile.""" + if not self._hot_switch_enabled: + self.notify( + "enable hot_switch in settings first", + title="account switch", + severity="warning", + ) + return + self._hot_switch() + + @work(thread=True, exclusive=True, group="switch") + def _hot_switch(self) -> None: + result = self.monitor.switch_best() + if result is None: + self.call_from_thread( + lambda: self.notify( + "no healthy alternate Claude profile", + title="account switch", + severity="warning", + ) + ) + return + self.call_from_thread(self._show_switch_result, result, "account switch") + + def _show_switch_result(self, result: SwitchResult, title: str) -> None: + self.notify( + result.message, + title=title, + severity="information" if result.ok else "warning", + timeout=7, + ) + if result.ok: + self._tick_sessions() + def action_add_account(self) -> None: """Provision an account and sign in, without leaving cctop. @@ -457,8 +540,6 @@ def action_add_account(self) -> None: cloned unless typed in, the clone is an allowlist (never credentials or state), and the login is scoped to the one new config dir. """ - from pathlib import Path - from rich.console import Console from . import manage @@ -539,7 +620,10 @@ def _refresh_token(self) -> None: def _force_limits(self) -> None: now = datetime.now(timezone.utc) limits = self.monitor.force_refresh_limits(now) + result = self.monitor.maybe_auto_switch() self.call_from_thread(self._render_limits, limits, now) + if result is not None: + self.call_from_thread(self._show_switch_result, result, "auto switch") # Reset the periodic timer so the next auto-refresh is a full interval # away, keeping the footer countdown honest after a manual refresh. self.call_from_thread(self._reschedule_limits_timer) diff --git a/src/cctop/authctl.py b/src/cctop/authctl.py index 14f3a56..99f3af9 100644 --- a/src/cctop/authctl.py +++ b/src/cctop/authctl.py @@ -1,13 +1,10 @@ -"""Delegated auth control: let the Claude Code binary refresh its own token. +"""Claude credential storage and delegated token refresh. -cctop never rewrites a credential itself. A Claude Code OAuth access token lives -only ~12-15h, and the CLI refreshes it lazily when you use that account, so an -account you are merely monitoring drifts past expiry and the free usage GET -starts returning 401. The safe fix is to ask the tool that *owns* the credential -to renew it: running a lightweight full `claude` command under the account's -CLAUDE_CONFIG_DIR runs Claude Code's startup auth path, which refreshes and -persists the Keychain record itself (the same thing that happens when you open -the account normally). cctop only triggers it and then observes the result. +Saved account directories remain the durable login profiles. Switching copies +one complete credential record into the single main Claude Code store; before +the next switch, the possibly-refreshed main record is copied back to the active +profile. Credentials stay in their existing file or macOS Keychain and are +never printed. Which command matters, verified empirically: `claude mcp list` refreshes an expired token (it needs the auth context, so startup renews it), while @@ -15,9 +12,7 @@ before the refresh). Both are quota-free; we use `mcp list` to refresh and `auth status --json` to read login identity. -Everything here is read-only from cctop's side: it invokes headless commands and -reads the credential's expiry for feedback; it never writes the credential, and -there is deliberately no logout/delete path. +There is deliberately no login, logout, or delete path. """ from __future__ import annotations @@ -38,43 +33,158 @@ _STATUS_ARGS = ("auth", "status", "--json") _COMMAND_TIMEOUT = 30 +_USER_BINARY = Path.home() / ".local" / "bin" / "claude" _BINARY_FALLBACK = Path("/usr/local/lib/node_modules/@anthropic-ai/claude-code/bin/claude.exe") _KEYCHAIN_SERVICE = "Claude Code-credentials" +_KEYCHAIN_ACCOUNT = "user" +_SECURITY_TIMEOUT = 5 + + +class CredentialError(RuntimeError): + """A credential store could not be read or updated safely.""" def find_claude_binary() -> str | None: """The real Claude Code executable, or None if it cannot be located. - Prefers PATH (a `claude` symlink into the npm install); falls back to the - known native-build location so the refresh works even when PATH is bare. + Prefers PATH (a `claude` symlink into the native install), then the standard + user-local link, then the legacy global npm location. The explicit + user-local path matters for launchd, whose deliberately small PATH omits + ``~/.local/bin``. """ found = shutil.which("claude") if found: return found + if _USER_BINARY.exists(): + return str(_USER_BINARY) return str(_BINARY_FALLBACK) if _BINARY_FALLBACK.exists() else None -def _keychain_service(config_dir: Path) -> str: +def keychain_service(config_dir: Path) -> str: + """The Keychain service name Claude Code uses for a profile on macOS. + + Verified empirically: the default ``~/.claude`` profile owns the bare + service name, and every other CLAUDE_CONFIG_DIR gets + ``Claude Code-credentials-``, which + is how a second account stays distinct from the default in one Keychain. + """ + if config_dir == Path.home() / ".claude": + return _KEYCHAIN_SERVICE digest = hashlib.sha256(str(config_dir).encode()).hexdigest()[:8] return f"{_KEYCHAIN_SERVICE}-{digest}" -def read_expiry(config_dir: Path) -> datetime | None: - """The access token's expiry from the account's Keychain record (read-only). +def identity_path(config_dir: Path) -> Path: + """Claude's identity/config file for a profile. - Used to show a TTL and to tell whether a refresh actually renewed the token. - Returns None when the record or its expiresAt field is unavailable. + The default profile is the historical special case at ``~/.claude.json``; + custom ``CLAUDE_CONFIG_DIR`` profiles keep it inside their directory. """ - command = ["security", "find-generic-password", "-s", _keychain_service(config_dir), "-w"] + if config_dir == Path.home() / ".claude": + return Path.home() / ".claude.json" + return config_dir / ".claude.json" + + +def read_identity(config_dir: Path) -> dict: try: - result = subprocess.run(command, capture_output=True, text=True, timeout=5) + value = json.loads(identity_path(config_dir).read_text()) + except (OSError, json.JSONDecodeError): + return {} + return value if isinstance(value, dict) else {} + + +def _keychain_read(service: str) -> str | None: + command = ["security", "find-generic-password", "-s", service, "-w"] + try: + result = subprocess.run(command, capture_output=True, text=True, timeout=_SECURITY_TIMEOUT) except (OSError, subprocess.SubprocessError): return None + return result.stdout.strip() if result.returncode == 0 and result.stdout.strip() else None + + +def _quote_security(value: str) -> str: + """Quote a non-secret value for security(1)'s interactive command parser.""" + return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"' + + +def _keychain_write(service: str, credential: str) -> None: + """Update a Keychain item without putting plaintext credential data in argv.""" + credential_hex = credential.encode("utf-8").hex() + command = ( + "add-generic-password -U " + f"-a {_quote_security(_KEYCHAIN_ACCOUNT)} " + f"-s {_quote_security(service)} -X {credential_hex}\n" + ) + try: + result = subprocess.run( + ["security", "-i"], + input=command, + capture_output=True, + text=True, + timeout=_SECURITY_TIMEOUT, + ) + except (OSError, subprocess.SubprocessError) as error: + raise CredentialError(f"Keychain update failed: {error}") from error if result.returncode != 0: - return None + detail = result.stderr.strip() or f"exit {result.returncode}" + raise CredentialError(f"Keychain update failed: {detail}") + + +def read_credentials(config_dir: Path) -> str: + """Read one profile's complete credential record, without fallback.""" + credentials_file = config_dir / ".credentials.json" try: - oauth = json.loads(result.stdout.strip()).get("claudeAiOauth") or {} - except json.JSONDecodeError: + raw = credentials_file.read_text().strip() + except OSError: + raw = "" + if raw: + try: + payload = json.loads(raw) + except json.JSONDecodeError as error: + raise CredentialError(f"invalid credentials in {credentials_file}") from error + if isinstance(payload.get("claudeAiOauth"), dict): + return raw + + keychain_raw = _keychain_read(keychain_service(config_dir)) + if keychain_raw is None: + raise CredentialError(f"no credential found for {config_dir}") + try: + payload = json.loads(keychain_raw) + except json.JSONDecodeError as error: + raise CredentialError(f"invalid Keychain credential for {config_dir}") from error + if not isinstance(payload.get("claudeAiOauth"), dict): + raise CredentialError(f"credential for {config_dir} has no OAuth record") + return keychain_raw + + +def write_credentials(config_dir: Path, credential: str) -> None: + """Replace one profile's credential in its existing storage backend.""" + try: + payload = json.loads(credential) + except json.JSONDecodeError as error: + raise CredentialError("refusing to write an invalid credential record") from error + if not isinstance(payload.get("claudeAiOauth"), dict): + raise CredentialError("refusing to write a credential without OAuth data") + + credentials_file = config_dir / ".credentials.json" + if credentials_file.exists(): + temporary = credentials_file.with_suffix(".json.cctop-new") + temporary.write_text(credential) + os.chmod(temporary, 0o600) + os.replace(temporary, credentials_file) + return + _keychain_write(keychain_service(config_dir), credential) + + +def read_expiry(config_dir: Path) -> datetime | None: + """The access token's expiry from the account's Keychain record (read-only). + + Used to show a TTL and to tell whether a refresh actually renewed the token. + Returns None when the record or its expiresAt field is unavailable. + """ + try: + oauth = json.loads(read_credentials(config_dir)).get("claudeAiOauth") or {} + except (CredentialError, json.JSONDecodeError): return None expires_at = oauth.get("expiresAt") if not isinstance(expires_at, (int, float)): @@ -90,21 +200,11 @@ def has_credentials(config_dir: Path) -> bool: (logged-out) dir correctly reads as having no credential rather than borrowing the default account's token. """ - credentials_file = config_dir / ".credentials.json" - try: - record = json.loads(credentials_file.read_text()) - oauth = record.get("claudeAiOauth") - if isinstance(oauth, dict) and oauth.get("accessToken"): - return True - except (OSError, json.JSONDecodeError): - pass - - command = ["security", "find-generic-password", "-s", _keychain_service(config_dir), "-w"] try: - result = subprocess.run(command, capture_output=True, text=True, timeout=5) - except (OSError, subprocess.SubprocessError): + oauth = json.loads(read_credentials(config_dir)).get("claudeAiOauth") + except (CredentialError, json.JSONDecodeError): return False - return result.returncode == 0 and bool(result.stdout.strip()) + return isinstance(oauth, dict) and bool(oauth.get("accessToken")) @dataclass(frozen=True) @@ -212,7 +312,7 @@ def refresh(account: str, config_dir: Path, now: datetime | None = None) -> Refr after = read_expiry(config_dir) if after is None or after <= now: - return RefreshResult(account, False, "needs re-login - run the account and /login", after) + return RefreshResult(account, False, "needs re-login - refresh token itself is dead", after) renewed = before is None or after > before verb = "refreshed" if renewed else "already valid" diff --git a/src/cctop/autoswitch.py b/src/cctop/autoswitch.py new file mode 100644 index 0000000..e1a5574 --- /dev/null +++ b/src/cctop/autoswitch.py @@ -0,0 +1,97 @@ +"""A headless hot-switch supervisor independent of the Textual TUI.""" + +from __future__ import annotations + +import time +from collections.abc import Callable +from dataclasses import dataclass +from datetime import datetime, timezone + +from . import config as config_module +from .collect import AccountLimits, default_accounts +from .monitor import FleetMonitor +from .switcher import SwitchResult, active_account, ensure_stable_accounts + + +@dataclass(frozen=True) +class AutoswitchCycle: + active: str | None + limits: list[AccountLimits] + switch: SwitchResult | None + + +class AutoswitchService: + """Poll limits and rotate logins even when no interactive display is open.""" + + def __init__(self, monitor: FleetMonitor, interval_seconds: float) -> None: + self.monitor = monitor + self.interval_seconds = interval_seconds + + @classmethod + def from_config(cls, config: config_module.Config | None = None) -> AutoswitchService: + config = config or config_module.load_config() + if not config.hot_switch(): + raise ValueError("hot switching is disabled") + + main_config_dir = config.main_config_dir() + accounts = ensure_stable_accounts(default_accounts(), main_config_dir) + claude_accounts = [account for account in accounts if account.provider == "claude"] + if not claude_accounts: + raise ValueError("no Claude profiles found") + + monitor = FleetMonitor( + claude_accounts, + main_config_dir=main_config_dir, + auto_switch_remaining_percent=config.auto_switch_remaining_percent(), + hot_switch=True, + limits_cache_path=config_module.config_dir() / "limits-cache.json", + ) + return cls(monitor, config.limits_refresh_seconds(180.0)) + + def poll(self, now: datetime | None = None) -> AutoswitchCycle: + now = now or datetime.now(timezone.utc) + limits = self.monitor.poll_limits(now, force=True) + switch = self.monitor.maybe_auto_switch() + current = active_account(self.monitor.accounts, self.monitor.main_config_dir) + return AutoswitchCycle(current.name if current is not None else None, limits, switch) + + def run( + self, + *, + once: bool = False, + emit: Callable[[str], None] = print, + wait: Callable[[float], None] = time.sleep, + ) -> AutoswitchCycle: + """Run immediately, then continue at the configured interval.""" + previous_summary: str | None = None + while True: + cycle = self.poll() + summary = _cycle_summary(cycle) + if summary != previous_summary or cycle.switch is not None: + emit(summary) + previous_summary = summary + if once: + return cycle + wait(self.interval_seconds) + + +def _binding_percent(limits: AccountLimits) -> float | None: + if limits.source != "api": + return None + values = [window.percent for window in limits.windows if window.has_data] + return max(values, default=0.0) + + +def _cycle_summary(cycle: AutoswitchCycle) -> str: + readings = [] + for item in cycle.limits: + percent = _binding_percent(item) + if percent is not None: + suffix = f" (cached: {item.error})" if item.error else "" + readings.append(f"{item.account}={percent:g}%{suffix}") + else: + readings.append(f"{item.account}=unavailable ({item.error or 'no usage data'})") + status = ( + cycle.switch.message if cycle.switch is not None else f"active {cycle.active or 'unknown'}" + ) + return f"{status}; " + ", ".join(readings) diff --git a/src/cctop/cli.py b/src/cctop/cli.py index fd75ad7..c10c260 100644 --- a/src/cctop/cli.py +++ b/src/cctop/cli.py @@ -148,7 +148,8 @@ def _render_limits(limits: list[AccountLimits], now: datetime, console: Console) gauges = " ".join(_render_gauge(w, now) for w in account.windows) age = _format_age(account.fetched_at, now) - console.print(f"{header} {gauges} [grey50]{age} ago[/grey50]") + cached = f" · cached: {account.error}" if account.error else "" + console.print(f"{header} {gauges} [grey50]{age} ago{cached}[/grey50]") def _render_table(snapshot: FleetSnapshot, multi_account: bool) -> Table: @@ -248,30 +249,65 @@ def _snapshot_to_dict(snapshot: FleetSnapshot) -> dict: } +def _session_dir( + account: Account, + hot_switch: bool, + main_config_dir: Path, + active: Account | None, +) -> Path | None: + """Where this account's live sessions actually run, or None if it has none. + + In hot-switch mode every Claude session runs in the one main dir, so only + the active profile owns any; another profile's leftover registry would + otherwise be counted as that profile's own live sessions. + """ + if hot_switch and account.provider == "claude": + return main_config_dir if account == active else None + return account.config_dir + + def _cmd_accounts() -> None: - """List the discovered accounts (read-only): tier and login status.""" + """List discovered profiles, login state, and the hot-switch active profile.""" + from . import config as config_module from . import usage from .registry import read_registry + from .switcher import active_account + + config = config_module.load_config() + hot_switch = config.hot_switch() + main_config_dir = config.main_config_dir() console = Console() table = Table(title="cctop accounts", title_style="bold", expand=False) + if hot_switch: + table.add_column("active", justify="center") table.add_column("acct", style="magenta") - table.add_column("config dir", style="grey70") + table.add_column("login dir", style="grey70") table.add_column("tier") table.add_column("token", justify="center") table.add_column("live", justify="right") - for account in default_accounts(): - tier = _tier_label(usage.read_tier(account.config_dir)) or "-" - has_token = usage.get_token(account.config_dir) is not None - live = sum(1 for _ in read_registry(account.config_dir)) - table.add_row( + accounts = default_accounts() + current = active_account(accounts, main_config_dir) if hot_switch else None + for account in accounts: + if account.provider == "codex": + tier = "-" + has_token = (account.config_dir / "auth.json").exists() + else: + tier = _tier_label(usage.read_tier(account.auth_dir)) or "-" + has_token = usage.get_token(account.auth_dir) is not None + session_dir = _session_dir(account, hot_switch, main_config_dir, current) + live = sum(1 for _ in read_registry(session_dir)) if session_dir else 0 + row = [ account.name, - str(account.config_dir).replace(str(Path.home()), "~"), + str(account.auth_dir).replace(str(Path.home()), "~"), tier, "[green]yes[/green]" if has_token else "[grey50]no[/grey50]", str(live), - ) + ] + if hot_switch: + row.insert(0, "[green]*[/green]" if account == current else "") + table.add_row(*row) console.print(table) @@ -287,10 +323,16 @@ def _cmd_doctor(accounts: list[Account] | None = None) -> None: import sys from . import authctl, usage + from . import config as config_module from .codex_usage import CODEX_DIR from .registry import read_registry + from .switcher import active_account accounts = accounts if accounts is not None else default_accounts() + config = config_module.load_config() + hot_switch = config.hot_switch() + main_config_dir = config.main_config_dir() + current = active_account(accounts, main_config_dir) if hot_switch else None now = datetime.now(timezone.utc) console = Console() @@ -306,11 +348,13 @@ def _cmd_doctor(accounts: list[Account] | None = None) -> None: console.print(f"codex {codex_note} ({str(CODEX_DIR).replace(str(Path.home()), '~')})\n") table = Table(title="accounts", title_style="bold", expand=False) - for column in ("acct", "provider", "config dir", "token", "expires", "tier", "live"): + for column in ("acct", "provider", "login dir", "token", "expires", "tier", "live"): table.add_column(column) for account in accounts: - config_display = str(account.config_dir).replace(str(Path.home()), "~") + # The login dir, not the session dir: it is where the token, expiry, and + # tier reported on this row were actually read from. + config_display = str(account.auth_dir).replace(str(Path.home()), "~") if account.provider == "codex": has_token = (CODEX_DIR / "auth.json").exists() table.add_row( @@ -324,11 +368,12 @@ def _cmd_doctor(accounts: list[Account] | None = None) -> None: ) continue - has_token = authctl.has_credentials(account.config_dir) - expiry = authctl.read_expiry(account.config_dir) + has_token = authctl.has_credentials(account.auth_dir) + expiry = authctl.read_expiry(account.auth_dir) expires = _format_reset(expiry, now) if expiry else "-" - tier = _tier_label(usage.read_tier(account.config_dir)) or "-" - live = sum(1 for _ in read_registry(account.config_dir)) + tier = _tier_label(usage.read_tier(account.auth_dir)) or "-" + session_dir = _session_dir(account, hot_switch, main_config_dir, current) + live = sum(1 for _ in read_registry(session_dir)) if session_dir else 0 table.add_row( account.name, "claude", @@ -518,9 +563,13 @@ def _config_template() -> str: "", "[settings]", "# limits_refresh_seconds = 180 # how often to refetch usage limits", + "# hot_switch = false # true: all Claude sessions use one main dir", + '# main_config_dir = "~/.claude"', + "# auto_switch_remaining_percent = 1 # rotate at 1% remaining", "", "# Accounts cctop auto-detected. Rename via `name`, hide with `hidden = true`,", "# reorder by moving blocks, or add your own block pointing at any config dir.", + "# Hot-switch login stores are matched automatically by account identity.", ] for account in discover_accounts(): directory = str(account.config_dir).replace(home, "~", 1) @@ -535,6 +584,80 @@ def _config_template() -> str: return "\n".join(lines) + "\n" +def _cmd_switch(argv: list[str]) -> None: + """Hot-switch the main Claude sessions to a saved profile. + + `cctop switch NAME` activates that profile; with no name, the healthy + profile with the most headroom. Both change live credential state, so an + unrecognized argument must not be silently read as a profile name. + """ + from . import config as config_module + from .collect import account_limits + from .switcher import best_account, ensure_stable_accounts, switch_account + + console = Console() + if len(argv) > 1 or (argv and argv[0].startswith("-")): + console.print("usage: [bold]cctop switch [NAME][/bold]") + console.print( + "[grey50]with no NAME, picks the healthy profile with the most headroom[/grey50]" + ) + return + + config = config_module.load_config() + if not config.hot_switch(): + console.print( + "[red]hot switching is disabled.[/red] Set " + "[bold]hot_switch = true[/bold] in cctop settings." + ) + return + + accounts = ensure_stable_accounts(default_accounts(), config.main_config_dir()) + claude_accounts = [account for account in accounts if account.provider == "claude"] + spec = argv[0] if argv else "" + if spec: + target = next((account for account in claude_accounts if account.name == spec), None) + if target is None: + console.print(f"[red]unknown Claude profile: {spec}[/red]") + return + else: + limits = [account_limits(account) for account in claude_accounts] + target = best_account(accounts, limits, config.main_config_dir()) + if target is None: + console.print("[red]no healthy alternate Claude profile[/red]") + return + + result = switch_account(accounts, target, config.main_config_dir()) + style = "green" if result.ok else "red" + console.print(f"[{style}]{result.message}[/{style}]") + + +def _cmd_autoswitch(argv: list[str]) -> None: + """Run hot switching without coupling it to an interactive TUI.""" + from .autoswitch import AutoswitchService + + parser = argparse.ArgumentParser( + prog="cctop autoswitch", + description="Continuously refresh usage and hot-switch Claude profiles.", + ) + parser.add_argument( + "--once", + action="store_true", + help="Poll and apply the switch policy once, then exit.", + ) + args = parser.parse_args(argv) + console = Console() + try: + service = AutoswitchService.from_config() + except ValueError as error: + console.print(f"[red]{error}[/red]") + return + + try: + service.run(once=args.once, emit=console.print) + except KeyboardInterrupt: + console.print("[grey50]autoswitch stopped[/grey50]") + + def _cmd_config(argv: list[str]) -> None: """`cctop config init [--force]` writes a starter config; `config path` prints it. @@ -611,7 +734,8 @@ def _confirm(console: Console, prompt: str) -> bool: dir = "~/.claude-work" # the account's CLAUDE_CONFIG_DIR (or ~/.codex) provider = "claude" # or "codex" hidden = false # true to hide it - and [settings] supports limits_refresh_seconds and heatmap_weeks. + and [settings] supports limits_refresh_seconds, heatmap_weeks, hot_switch, + main_config_dir, and auto_switch_remaining_percent. 4. Only edit that config file, and be strictly additive: never delete or overwrite my credentials, sessions, or other config. Confirm before writing. 5. When done, tell me to run `cctop`. @@ -713,6 +837,12 @@ def main() -> None: if argv[:1] == ["add-account"]: _cmd_add_account(argv[1:]) return + if argv[:1] == ["switch"]: + _cmd_switch(argv[1:]) + return + if argv[:1] == ["autoswitch"]: + _cmd_autoswitch(argv[1:]) + return if argv[:1] == ["doctor"]: _cmd_doctor() return @@ -763,17 +893,26 @@ def main() -> None: args = parser.parse_args() accounts = _resolve_accounts(args) + from . import config as config_module + + cfg = config_module.load_config() + configured_hot_switch = cfg.hot_switch() and not args.account and args.config_dir is None + if configured_hot_switch: + from .switcher import ensure_stable_accounts + + accounts = ensure_stable_accounts(accounts, cfg.main_config_dir()) if not args.once and not args.json: # Default: launch the live TUI, which polls with its own throttles. - from . import config as config_module from .app import CctopApp - cfg = config_module.load_config() CctopApp( accounts, limits_interval=cfg.limits_refresh_seconds(180.0), heatmap_weeks=cfg.heatmap_weeks(26), + main_config_dir=cfg.main_config_dir(), + auto_switch_remaining_percent=cfg.auto_switch_remaining_percent(), + hot_switch=configured_hot_switch, ).run() return @@ -783,6 +922,8 @@ def main() -> None: now=now, include_dead=args.include_dead, with_limits=not args.no_limits, + hot_switch=configured_hot_switch, + main_config_dir=cfg.main_config_dir(), ) if args.json: diff --git a/src/cctop/collect.py b/src/cctop/collect.py index 3d6409f..8744dfc 100644 --- a/src/cctop/collect.py +++ b/src/cctop/collect.py @@ -1,10 +1,10 @@ """Assemble a FleetSnapshot across one or more accounts. -Each account is a (name, config_dir) pair: cc-0 -> ~/.claude, cc-1 -> -~/.claude-1, matching the user's shell aliases. The registry gives each -account's live process table and status; the transcript gives model, tokens, -cost, and context size; limits.py supplies the real usage-limit gauges. All of -it is combined into immutable records tagged by account. +Each account has a session/config directory and may have a separate stable +credential directory for hot switching. The registry gives each account's live +process table and status; the transcript gives model, tokens, cost, and context +size; limits.py supplies the real usage-limit gauges. All of it is combined +into immutable records tagged by account. """ from __future__ import annotations @@ -41,15 +41,22 @@ @dataclass(frozen=True) class Account: - """A named account backed by its own config directory. + """A named account backed by session state and a saved login. `provider` selects how sessions and limits are read: "claude" (the default) - or "codex". + or "codex". `credential_dir`, when set, keeps a stable saved login separate + from the mutable main hot-switch session directory. """ name: str config_dir: Path provider: str = "claude" + credential_dir: Path | None = None + + @property + def auth_dir(self) -> Path: + """Stable saved-login directory, separate from live session state.""" + return self.credential_dir or self.config_dir def default_config_dir() -> Path: @@ -74,7 +81,7 @@ def _provider_for(config_dir: Path) -> str: return "codex" if "codex" in config_dir.name.lower() else "claude" -def discover_accounts() -> list[Account]: +def discover_accounts(include_managed: bool = False) -> list[Account]: """Every account auto-detected under the home dir. ~/.claude is cc-0; any ~/.claude- that looks like a real config dir @@ -82,6 +89,11 @@ def discover_accounts() -> list[Account]: ~/.claude-work become cc-work), so it is not limited to the numeric shell- alias convention. ~/.codex is added as cx-0. Discovering by glob means a new account appears automatically, with no hardcoded list to maintain. + + `include_managed` additionally surfaces cctop's own snapshotted login + profiles. They are only meaningful in hot-switch mode, where deduplication + folds each one back into the account it was snapshotted from; anywhere else + they would double every such account, so they stay hidden by default. """ home = Path.home() found: list[tuple[tuple[int, str], Account]] = [] @@ -97,6 +109,18 @@ def discover_accounts() -> list[Account]: order = (int(suffix), "") if suffix.isdigit() else (10_000, suffix) found.append((order, Account(f"cc-{suffix}", path))) + # Profiles snapshotted automatically from the mutable main login live under + # cctop's own config directory, outside ~/.claude-* so Claude never treats + # them as session homes. They re-enter discovery here on later runs. + if include_managed: + from . import config as config_module + + managed_root = config_module.config_dir() / "profiles" + if managed_root.is_dir(): + for path in managed_root.iterdir(): + if path.is_dir() and _looks_like_config_dir(path): + found.append(((20_000, path.name), Account(f"cc-{path.name}", path))) + found.sort(key=lambda item: item[0]) accounts = [account for _, account in found] or [Account("default", default_config_dir())] @@ -118,11 +142,12 @@ def resolve_accounts(config=None) -> list[Account]: if config is None: config = config_module.load_config() + hot_switch = config.hot_switch() overrides = {override.dir: override for override in config.accounts} result: list[Account] = [] seen: set[Path] = set() - for account in discover_accounts(): + for account in discover_accounts(include_managed=hot_switch): seen.add(account.config_dir) override = overrides.get(account.config_dir) if override is not None and override.hidden: @@ -132,6 +157,7 @@ def resolve_accounts(config=None) -> list[Account]: override.name or account.name, account.config_dir, override.provider or account.provider, + override.switch_dir, ) result.append(account) @@ -143,12 +169,90 @@ def resolve_accounts(config=None) -> list[Account]: override.name or override.dir.name.lstrip("."), override.dir, override.provider or _provider_for(override.dir), + override.switch_dir, ) ) + if hot_switch: + return _deduplicate_hot_accounts(result, config.main_config_dir()) return result +def _automatic_name(account: Account) -> str: + """An identity-derived label, falling back to the discovered path label.""" + from . import authctl + + oauth = authctl.read_identity(account.auth_dir).get("oauthAccount") + if isinstance(oauth, dict): + email = oauth.get("emailAddress") + if isinstance(email, str) and email: + return email + return account.name + + +def _identity_key(account: Account) -> tuple[str, str]: + """Stable dedup key for one saved login; unknown identities stay separate.""" + from . import authctl + + oauth = authctl.read_identity(account.auth_dir).get("oauthAccount") + if isinstance(oauth, dict): + org = oauth.get("organizationUuid") + if isinstance(org, str) and org: + return ("org", org) + email = oauth.get("emailAddress") + if isinstance(email, str) and email: + return ("email", email.lower()) + return ("path", str(account.auth_dir)) + + +def _deduplicate_hot_accounts(accounts: list[Account], main_config_dir: Path) -> list[Account]: + """Collapse duplicate login dirs while retaining the one main session home.""" + from . import authctl + + groups: dict[tuple[str, str], list[Account]] = {} + order: list[tuple[str, str]] = [] + codex: list[Account] = [] + for account in accounts: + if account.provider != "claude": + codex.append(account) + continue + key = _identity_key(account) + if key not in groups: + groups[key] = [] + order.append(key) + groups[key].append(account) + + result: list[Account] = [] + for key in order: + group = groups[key] + session = next( + (account for account in group if account.config_dir == main_config_dir), + group[0], + ) + explicit = next((account for account in group if account.credential_dir), None) + stable = next( + ( + account + for account in group + if account.auth_dir != main_config_dir and authctl.has_credentials(account.auth_dir) + ), + None, + ) + source = explicit or stable or session + configured_name = next( + ( + account.name + for account in group + if not account.name.startswith("cc-") and account.name != "default" + ), + None, + ) + name = configured_name or _automatic_name(source) + result.append(Account(name, session.config_dir, "claude", source.auth_dir)) + + return [*result, *codex] + + # The default account list applies the config file over auto-detection. default_accounts = resolve_accounts @@ -257,7 +361,7 @@ def account_limits(account: Account) -> AccountLimits: from . import usage - return usage.fetch_account_limits(account.name, account.config_dir) + return usage.fetch_account_limits(account.name, account.auth_dir) def build_snapshot( @@ -265,6 +369,8 @@ def build_snapshot( now: datetime | None = None, include_dead: bool = False, with_limits: bool = True, + hot_switch: bool = False, + main_config_dir: Path | None = None, ) -> FleetSnapshot: """Merge every account's registry, transcripts, and limits into a snapshot. @@ -283,9 +389,18 @@ def build_snapshot( for account in accounts: if account.provider == "codex": states.extend(codex_session_states(account, now)) - continue - for session in read_registry(account.config_dir): - states.append(_state_for_session(session, account, now)) + elif not hot_switch: + for session in read_registry(account.config_dir): + states.append(_state_for_session(session, account, now)) + + if hot_switch: + from .switcher import active_account + + main = main_config_dir or Path.home() / ".claude" + active = active_account(accounts, main) + session_account = Account(active.name if active else "main", main) + for session in read_registry(main): + states.append(_state_for_session(session, session_account, now)) if not include_dead: states = [state for state in states if state.alive] diff --git a/src/cctop/config.py b/src/cctop/config.py index ab30ffd..3767ca7 100644 --- a/src/cctop/config.py +++ b/src/cctop/config.py @@ -34,6 +34,7 @@ class AccountOverride: dir: Path name: str | None = None provider: str | None = None + switch_dir: Path | None = None hidden: bool = False @@ -52,6 +53,29 @@ def heatmap_weeks(self, default: int) -> int: value = self.settings.get("heatmap_weeks") return int(value) if isinstance(value, int) and value > 0 else default + def main_config_dir(self, default: Path | None = None) -> Path: + """The one Claude state directory whose live sessions are shared. + + Account directories are credential profiles. Claude itself always runs + against this main directory; cctop swaps a selected profile into it. + """ + value = self.settings.get("main_config_dir") + if isinstance(value, str) and value.strip(): + return Path(value).expanduser() + return default or Path.home() / ".claude" + + def auto_switch_remaining_percent(self, default: float = 1.0) -> float: + """Headroom at which the active Claude profile is rotated.""" + value = self.settings.get("auto_switch_remaining_percent") + if isinstance(value, (int, float)) and 0 <= float(value) < 100: + return float(value) + return default + + def hot_switch(self, default: bool = False) -> bool: + """Use one main Claude session directory and swap saved logins into it.""" + value = self.settings.get("hot_switch") + return value if isinstance(value, bool) else default + def load_config(path: Path | None = None) -> Config: """Load the config file, or an empty Config when it is absent/unparseable.""" @@ -71,6 +95,9 @@ def load_config(path: Path | None = None) -> Config: dir=Path(str(entry["dir"])).expanduser(), name=entry.get("name"), provider=entry.get("provider"), + switch_dir=( + Path(str(entry["switch_dir"])).expanduser() if entry.get("switch_dir") else None + ), hidden=bool(entry.get("hidden", False)), ) ) @@ -93,7 +120,13 @@ def save_config(settings: dict, accounts: list[AccountOverride], path: Path | No path = path or config_path() lines = ["# cctop config, written by the settings screen.", "", "[settings]"] for key, value in settings.items(): - lines.append(f"{key} = {value}") + if isinstance(value, bool): + lines.append(f"{key} = {'true' if value else 'false'}") + elif isinstance(value, str): + escaped = value.replace("\\", "\\\\").replace('"', '\\"') + lines.append(f'{key} = "{escaped}"') + else: + lines.append(f"{key} = {value}") for override in accounts: lines += [ "", @@ -103,6 +136,8 @@ def save_config(settings: dict, accounts: list[AccountOverride], path: Path | No ] if override.provider: lines.append(f'provider = "{override.provider}"') + if override.switch_dir: + lines.append(f'switch_dir = "{_home_relative(override.switch_dir)}"') if override.hidden: lines.append("hidden = true") path.parent.mkdir(parents=True, exist_ok=True) diff --git a/src/cctop/limits_cache.py b/src/cctop/limits_cache.py new file mode 100644 index 0000000..315da63 --- /dev/null +++ b/src/cctop/limits_cache.py @@ -0,0 +1,99 @@ +"""Shared last-good usage readings for the TUI and autoswitch supervisor.""" + +from __future__ import annotations + +import json +import os +from datetime import datetime +from pathlib import Path + +from .models import AccountLimits, LimitWindow + + +def load(path: Path) -> dict[str, AccountLimits]: + """Read cached API results, ignoring an absent or invalid cache.""" + try: + payload = json.loads(path.read_text()) + except (OSError, json.JSONDecodeError): + return {} + if not isinstance(payload, dict) or payload.get("version") != 1: + return {} + + results: dict[str, AccountLimits] = {} + for entry in payload.get("accounts", []): + if not isinstance(entry, dict): + continue + name = entry.get("account") + fetched_at = _datetime(entry.get("fetched_at")) + if not isinstance(name, str) or fetched_at is None: + continue + windows = [] + for item in entry.get("windows", []): + if not isinstance(item, dict): + continue + try: + windows.append( + LimitWindow( + kind=str(item["kind"]), + label=str(item["label"]), + percent=float(item["percent"]), + resets_at=_datetime(item.get("resets_at")), + severity=str(item["severity"]), + is_active=bool(item["is_active"]), + has_data=bool(item["has_data"]), + ) + ) + except (KeyError, TypeError, ValueError): + continue + results[name] = AccountLimits( + name, + entry.get("tier") if isinstance(entry.get("tier"), str) else None, + windows, + "api", + fetched_at, + ) + return results + + +def save(path: Path, limits: dict[str, AccountLimits]) -> None: + """Atomically persist only successful, non-secret usage metadata.""" + accounts = [] + for item in limits.values(): + if item.source != "api" or item.fetched_at is None: + continue + accounts.append( + { + "account": item.account, + "tier": item.tier, + "fetched_at": item.fetched_at.isoformat(), + "windows": [ + { + "kind": window.kind, + "label": window.label, + "percent": window.percent, + "resets_at": ( + window.resets_at.isoformat() if window.resets_at is not None else None + ), + "severity": window.severity, + "is_active": window.is_active, + "has_data": window.has_data, + } + for window in item.windows + ], + } + ) + + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.{os.getpid()}.cctop-new") + temporary.write_text(json.dumps({"version": 1, "accounts": accounts}, separators=(",", ":"))) + os.chmod(temporary, 0o600) + os.replace(temporary, path) + + +def _datetime(value: object) -> datetime | None: + if not isinstance(value, str): + return None + try: + return datetime.fromisoformat(value) + except ValueError: + return None diff --git a/src/cctop/locks.py b/src/cctop/locks.py new file mode 100644 index 0000000..cd1d40f --- /dev/null +++ b/src/cctop/locks.py @@ -0,0 +1,66 @@ +"""Claude Code-compatible advisory locks for credential/config switching.""" + +from __future__ import annotations + +import os +import random +import threading +import time +from contextlib import contextmanager +from pathlib import Path + +STALE_SECONDS = 10.0 +TOUCH_SECONDS = 3.0 +TIMEOUT_SECONDS = 9.0 + + +@contextmanager +def claude_lock(target: Path, timeout: float = TIMEOUT_SECONDS): + """Hold the same ``.lock`` directory protocol Claude Code uses.""" + lock_dir = target.parent / f"{target.name}.lock" + lock_dir.parent.mkdir(parents=True, exist_ok=True) + started = time.monotonic() + while True: + try: + os.mkdir(lock_dir) + break + except FileExistsError: + pass + if time.monotonic() - started > timeout: + raise TimeoutError(f"timed out waiting for {lock_dir}") + try: + stale = time.time() - lock_dir.stat().st_mtime > STALE_SECONDS + except FileNotFoundError: + continue + if stale: + try: + os.rmdir(lock_dir) + continue + except OSError: + # A lock dir we cannot reap (not empty, not ours to remove) must + # not turn the wait into a spin; back off like a live holder. + pass + time.sleep(0.25 + random.random() * 0.25) + + stopped = threading.Event() + + def touch() -> None: + while not stopped.wait(TOUCH_SECONDS): + try: + os.utime(lock_dir) + except OSError: + return + + thread = threading.Thread(target=touch, daemon=True) + thread.start() + try: + yield + finally: + stopped.set() + thread.join(timeout=1) + try: + os.rmdir(lock_dir) + except OSError: + # Releasing is best effort: the work inside already succeeded or + # raised, and a stuck lock dir is reaped by the next waiter. + pass diff --git a/src/cctop/monitor.py b/src/cctop/monitor.py index 4168d73..e453cf5 100644 --- a/src/cctop/monitor.py +++ b/src/cctop/monitor.py @@ -9,15 +9,24 @@ from __future__ import annotations +from dataclasses import replace from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import TYPE_CHECKING from . import authctl from .authctl import RefreshResult from .collect import Account, _effective_status, account_limits, codex_session_states +from .limits_cache import load as load_limits_cache +from .limits_cache import save as save_limits_cache from .models import AccountLimits, FleetSnapshot, SessionState, UsageTotals from .registry import process_alive, read_registry from .transcript import TranscriptTailer, find_transcript +if TYPE_CHECKING: + # Imported lazily at runtime: switching is only reachable in hot-switch mode. + from .switcher import SwitchResult + # Usage windows move slowly (a 5h window shifts well under 1%/min, weekly ones # barely at all), so poll a few minutes apart rather than every minute: fresh # enough, and far below whatever rate the usage endpoint enforces. @@ -29,6 +38,7 @@ # here, double each consecutive failure, cap so it always recovers eventually. _BACKOFF_BASE = timedelta(seconds=60) _BACKOFF_CAP = timedelta(minutes=15) +_AUTH_RETRY = timedelta(minutes=15) _EMPTY_TOTALS = UsageTotals(0, 0, 0, 0, 0.0) @@ -40,22 +50,69 @@ def __init__( self, accounts: list[Account], limits_interval: timedelta = DEFAULT_LIMITS_INTERVAL, + main_config_dir: Path | None = None, + auto_switch_remaining_percent: float = 1.0, + hot_switch: bool = False, + limits_cache_path: Path | None = None, ) -> None: self.accounts = accounts self.limits_interval = limits_interval + self.main_config_dir = main_config_dir or Path.home() / ".claude" + self.auto_switch_remaining_percent = auto_switch_remaining_percent + self.hot_switch = hot_switch + self._limits_cache_path = limits_cache_path self._tailers: dict[str, TranscriptTailer] = {} - self.limits: list[AccountLimits] = [] - self.limits_fetched_at: datetime | None = None + cached = load_limits_cache(limits_cache_path) if limits_cache_path is not None else {} + account_names = {account.name for account in accounts} + self._good_limits = {name: item for name, item in cached.items() if name in account_names} + self.limits = list(self._good_limits.values()) + fetched = [item.fetched_at for item in self.limits if item.fetched_at is not None] + self.limits_fetched_at = max(fetched) if fetched else None self._codex_cache: dict[str, list[SessionState]] = {} self._codex_polled_at: datetime | None = None # Per-account rate-limit state: the last successful fetch (so a transient # failure can keep showing real numbers), when the account may be fetched # again, and the current backoff length. - self._good_limits: dict[str, AccountLimits] = {} self._cooldown_until: dict[str, datetime] = {} self._backoff: dict[str, timedelta] = {} + self._transient_error: dict[str, str] = {} + self._auth_retry_after: dict[str, datetime] = {} + self._auth_error: dict[str, str] = {} + self._main_auth_retry_after: datetime | None = None + self._main_auth_error: str | None = None + self._main_auth_checked_at: datetime | None = None + + def _reload_limits_cache(self) -> None: + """Adopt newer readings written by another cctop process.""" + if self._limits_cache_path is None: + return + shared = load_limits_cache(self._limits_cache_path) + account_names = {account.name for account in self.accounts} + changed = False + for name, item in shared.items(): + if name not in account_names: + continue + current = self._good_limits.get(name) + if current is None or ( + item.fetched_at is not None + and (current.fetched_at is None or item.fetched_at > current.fetched_at) + ): + self._good_limits[name] = item + changed = True + if not changed: + return + self.limits = [ + self._good_limits[account.name] + for account in self.accounts + if account.name in self._good_limits + ] + fetched = [item.fetched_at for item in self.limits if item.fetched_at is not None] + if fetched: + newest = max(fetched) + if self.limits_fetched_at is None or newest > self.limits_fetched_at: + self.limits_fetched_at = newest def _tailer_for(self, config_dir, session_id: str) -> TranscriptTailer | None: """The live tailer for a session, created (and its file located) once.""" @@ -69,31 +126,32 @@ def _tailer_for(self, config_dir, session_id: str) -> TranscriptTailer | None: return tailer def poll_sessions(self, now: datetime | None = None) -> list[SessionState]: - """Rebuild the live session list, advancing each tailer by new bytes.""" + """Rebuild the live session list, advancing each tailer by new bytes. + + Claude sessions come from each account's own config dir, or -- in + hot-switch mode, where every session runs in one directory -- from the + main dir alone, attributed to whichever profile is currently active. + """ now = now or datetime.now(timezone.utc) codex_due = self._codex_polled_at is None or now - self._codex_polled_at >= CODEX_INTERVAL states: list[SessionState] = [] seen: set[str] = set() - for account in self.accounts: - if account.provider == "codex": - if codex_due: - self._codex_cache[account.name] = codex_session_states(account, now) - states.extend(self._codex_cache.get(account.name, [])) - continue - for session in read_registry(account.config_dir): + + def append_claude(config_dir: Path, account_name: str) -> None: + for session in read_registry(config_dir): if not process_alive(session.pid): continue seen.add(session.session_id) status = _effective_status(session, True, now) - tailer = self._tailer_for(account.config_dir, session.session_id) + tailer = self._tailer_for(config_dir, session.session_id) if tailer is None: states.append( SessionState( session=session, - account=account.name, + account=account_name, alive=True, status=status, model=None, @@ -108,7 +166,7 @@ def poll_sessions(self, now: datetime | None = None) -> list[SessionState]: states.append( SessionState( session=session, - account=account.name, + account=account_name, alive=True, status=status, model=tailer.model, @@ -118,6 +176,20 @@ def poll_sessions(self, now: datetime | None = None) -> list[SessionState]: ) ) + for account in self.accounts: + if account.provider == "codex": + if codex_due: + self._codex_cache[account.name] = codex_session_states(account, now) + states.extend(self._codex_cache.get(account.name, [])) + elif not self.hot_switch: + append_claude(account.config_dir, account.name) + + if self.hot_switch: + from .switcher import active_account + + active = active_account(self.accounts, self.main_config_dir) + append_claude(self.main_config_dir, active.name if active is not None else "main") + if codex_due: self._codex_polled_at = now @@ -129,7 +201,48 @@ def poll_sessions(self, now: datetime | None = None) -> list[SessionState]: states.sort(key=lambda state: state.last_activity or now, reverse=True) return states + def maybe_auto_switch(self) -> SwitchResult | None: + """Apply the configured remaining-headroom rotation policy to cached limits.""" + if not self.hot_switch: + return None + from .switcher import auto_switch, recover_auth + + if self._main_auth_error is not None: + result = recover_auth( + self.accounts, + self.limits, + self.main_config_dir, + self._main_auth_error, + self._main_auth_checked_at, + ) + if result.ok: + self._main_auth_error = None + self._main_auth_retry_after = None + return result + + return auto_switch( + self.accounts, + self.limits, + self.main_config_dir, + self.auto_switch_remaining_percent, + ) + + def switch_best(self) -> SwitchResult | None: + """Hot-switch to the currently healthiest saved Claude profile.""" + if not self.hot_switch: + return None + from .switcher import best_account, switch_account + + target = best_account(self.accounts, self.limits, self.main_config_dir) + if target is None: + return None + return switch_account(self.accounts, target, self.main_config_dir) + def limits_due(self, now: datetime) -> bool: + configured = {account.name for account in self.accounts} + represented = {item.account for item in self.limits} + if represented != configured: + return True if self.limits_fetched_at is None: return True return now - self.limits_fetched_at >= self.limits_interval @@ -144,21 +257,83 @@ def _fetch_limits_one(self, account: Account, now: datetime) -> AccountLimits: a real error (token expired) is surfaced as-is. """ name = account.name - cooldown = self._cooldown_until.get(name) good = self._good_limits.get(name) + + # A locally expired saved credential is actionable before any network + # request. Anthropic may return a fleet-wide 429 before authentication, + # which would otherwise disguise a logged-out profile as merely rate + # limited and waste another request every time its cooldown expires. + if self.hot_switch and account.provider == "claude": + expires_at = authctl.read_expiry(account.auth_dir) + if expires_at is None or expires_at <= now: + auth_retry = self._auth_retry_after.get(name) + if auth_retry is None or now >= auth_retry: + refreshed = authctl.refresh(name, account.auth_dir, now) + if refreshed.ok: + self._auth_retry_after.pop(name, None) + self._auth_error.pop(name, None) + self._cooldown_until.pop(name, None) + self._backoff.pop(name, None) + else: + self._auth_retry_after[name] = now + _AUTH_RETRY + self._auth_error[name] = refreshed.message + error = self._auth_error.get(name) + if error is not None: + self._good_limits.pop(name, None) + self._transient_error.pop(name, None) + return AccountLimits( + name, + good.tier if good is not None else None, + [], + "none", + None, + error=error, + ) + else: + self._auth_retry_after.pop(name, None) + self._auth_error.pop(name, None) + + cooldown = self._cooldown_until.get(name) if cooldown is not None and now < cooldown: if good is not None: - return good + return replace( + good, + error=self._transient_error.get(name, "rate limited, retrying"), + retriable=True, + ) return AccountLimits( name, None, [], "none", None, error="rate limited, retrying", retriable=True ) result = account_limits(account) + # In hot-switch mode every saved login must remain ready to activate. + # Let Claude Code refresh an expired access token itself; this needs no + # login unless the saved refresh token is genuinely dead. Bound retries + # so a dead login never spawns a command on every limits poll. + auth_retry = self._auth_retry_after.get(name) + expired = result.error is not None and result.error.startswith("token expired") + if ( + self.hot_switch + and account.provider == "claude" + and expired + and (auth_retry is None or now >= auth_retry) + ): + refreshed = authctl.refresh(account.name, account.auth_dir, now) + if refreshed.ok: + self._auth_retry_after.pop(name, None) + self._auth_error.pop(name, None) + result = account_limits(account) + else: + self._auth_retry_after[name] = now + _AUTH_RETRY + self._auth_error[name] = refreshed.message + result = replace(result, error=refreshed.message) + if result.source == "api": self._good_limits[name] = result self._backoff.pop(name, None) self._cooldown_until.pop(name, None) + self._transient_error.pop(name, None) return result if result.retriable: @@ -169,13 +344,55 @@ def _fetch_limits_one(self, account: Account, now: datetime) -> AccountLimits: delay = _BACKOFF_BASE if previous is None else min(previous * 2, _BACKOFF_CAP) self._backoff[name] = delay self._cooldown_until[name] = now + delay - return good if good is not None else result + self._transient_error[name] = result.error or "usage temporarily unavailable" + return ( + replace(good, error=self._transient_error[name], retriable=True) + if good is not None + else result + ) # Non-retriable (token expired, wrong credential): a real, actionable # state. Drop any stale-good so the UI shows what the user must fix. self._good_limits.pop(name, None) + self._transient_error.pop(name, None) return result + def _prepare_main_auth(self, now: datetime) -> None: + """Refresh the live credential or mark it for immediate failover. + + Saved-profile usage probes do not prove that the mutable main store is + usable. Check that store directly before syncing it back: if Claude + cannot refresh it, preserving it would overwrite the saved profile with + the credential that just produced ``Login expired``. + """ + from .switcher import active_account, sync_active_profile + + self._main_auth_checked_at = now + expires_at = authctl.read_expiry(self.main_config_dir) + if expires_at is not None and expires_at > now: + self._main_auth_error = None + self._main_auth_retry_after = None + sync_active_profile(self.accounts, self.main_config_dir) + return + + if self._main_auth_retry_after is not None and now < self._main_auth_retry_after: + return + + current = active_account(self.accounts, self.main_config_dir) + refreshed = authctl.refresh( + current.name if current is not None else "main", + self.main_config_dir, + now, + ) + if refreshed.ok: + self._main_auth_error = None + self._main_auth_retry_after = None + sync_active_profile(self.accounts, self.main_config_dir) + return + + self._main_auth_error = refreshed.message + self._main_auth_retry_after = now + _AUTH_RETRY + def poll_limits( self, now: datetime | None = None, @@ -189,11 +406,17 @@ def poll_limits( forced, so a forced refresh never hammers a rate-limited account. """ now = now or datetime.now(timezone.utc) + self._reload_limits_cache() if not force and not self.limits_due(now): return self.limits + if self.hot_switch: + self._prepare_main_auth(now) + self.limits = [self._fetch_limits_one(account, now) for account in self.accounts] self.limits_fetched_at = now + if self._limits_cache_path is not None: + save_limits_cache(self._limits_cache_path, self._good_limits) return self.limits def force_refresh_limits(self, now: datetime | None = None) -> list[AccountLimits]: @@ -223,7 +446,7 @@ def refresh_credentials( now = now or datetime.now(timezone.utc) results = [ - authctl.refresh(account.name, account.config_dir, now) + authctl.refresh(account.name, account.auth_dir, now) for account in self.accounts if account.provider == "claude" ] diff --git a/src/cctop/settings_screen.py b/src/cctop/settings_screen.py index 4ae8672..ea45232 100644 --- a/src/cctop/settings_screen.py +++ b/src/cctop/settings_screen.py @@ -32,6 +32,7 @@ class AccountRow: dir: Path name: str provider: str + switch_dir: Path | None hidden: bool @@ -48,6 +49,7 @@ def build_rows(detected: list[Account], config: Config) -> list[AccountRow]: dir=account.config_dir, name=override.name if override and override.name else account.name, provider=account.provider, + switch_dir=override.switch_dir if override else None, hidden=bool(override.hidden) if override else False, ) ) @@ -58,6 +60,7 @@ def build_rows(detected: list[Account], config: Config) -> list[AccountRow]: dir=override.dir, name=override.name or override.dir.name.lstrip("."), provider=override.provider or "claude", + switch_dir=override.switch_dir, hidden=override.hidden, ) ) @@ -97,10 +100,21 @@ class SettingsScreen(ModalScreen): BINDINGS = [("escape", "cancel", "Close")] - def __init__(self, interval: float, weeks: int, rows: list[AccountRow]) -> None: + def __init__( + self, + interval: float, + weeks: int, + main_config_dir: Path, + auto_switch_remaining_percent: float, + hot_switch: bool, + rows: list[AccountRow], + ) -> None: super().__init__() self._interval = interval self._weeks = weeks + self._main_config_dir = main_config_dir + self._auto_switch_remaining_percent = auto_switch_remaining_percent + self._hot_switch = hot_switch self._rows = rows def compose(self) -> ComposeResult: @@ -111,12 +125,27 @@ def compose(self) -> ComposeResult: yield Input(value=str(int(self._interval)), id="interval", type="integer") yield Label("Heatmap weeks") yield Input(value=str(self._weeks), id="weeks", type="integer") + yield Label("Hot switch one main Claude session") + yield Switch(value=self._hot_switch, id="hot-switch") + yield Label("Main Claude config directory") + yield Input(value=_home(self._main_config_dir), id="main-config-dir") + yield Label("Auto-switch when remaining (%)") + yield Input( + value=str(self._auto_switch_remaining_percent), + id="remaining", + type="number", + ) yield Label("Accounts (toggle to show, edit the name)") for index, row in enumerate(self._rows): with Horizontal(classes="acct-row"): yield Switch(value=not row.hidden, id=f"show-{index}") yield Input(value=row.name, id=f"name-{index}") yield Label(_home(row.dir), classes="acct-dir") + if row.switch_dir: + yield Label( + f"login: {_home(row.switch_dir)}", + classes="acct-dir", + ) with Horizontal(id="settings-actions"): yield Button("Save", variant="success", id="save") yield Button("Cancel", id="cancel") @@ -141,16 +170,31 @@ def _int(self, widget_id: str, default: int, minimum: int) -> int: return default def _save(self) -> None: + try: + remaining = float(self.query_one("#remaining", Input).value) + except ValueError: + remaining = 1.0 + remaining = min(99.0, max(0.0, remaining)) settings = { "limits_refresh_seconds": self._int("interval", 180, 10), "heatmap_weeks": self._int("weeks", 26, 1), + "hot_switch": self.query_one("#hot-switch", Switch).value, + "main_config_dir": self.query_one("#main-config-dir", Input).value.strip() + or "~/.claude", + "auto_switch_remaining_percent": remaining, } accounts = [] for index, row in enumerate(self._rows): name = self.query_one(f"#name-{index}", Input).value.strip() or row.name hidden = not self.query_one(f"#show-{index}", Switch).value accounts.append( - AccountOverride(dir=row.dir, name=name, provider=row.provider, hidden=hidden) + AccountOverride( + dir=row.dir, + name=name, + provider=row.provider, + switch_dir=row.switch_dir, + hidden=hidden, + ) ) config_module.save_config(settings, accounts) self.app.apply_settings() # type: ignore[attr-defined] diff --git a/src/cctop/switcher.py b/src/cctop/switcher.py new file mode 100644 index 0000000..a91251d --- /dev/null +++ b/src/cctop/switcher.py @@ -0,0 +1,351 @@ +"""Hot-switch saved Claude logins into one main session directory.""" + +from __future__ import annotations + +import json +import os +import re +from contextlib import ExitStack +from dataclasses import dataclass, replace +from datetime import datetime, timezone +from pathlib import Path + +from . import authctl +from .collect import Account +from .locks import claude_lock +from .models import AccountLimits + +# Usage a profile must be under to be worth switching into on demand. +HEALTHY_PERCENT = 99.0 + + +@dataclass(frozen=True) +class SwitchResult: + ok: bool + previous: str | None + active: str | None + message: str + + +def _org_id(config_dir: Path) -> str | None: + value = authctl.read_identity(config_dir).get("oauthAccount") + if not isinstance(value, dict): + return None + org = value.get("organizationUuid") + return org if isinstance(org, str) and org else None + + +def active_account(accounts: list[Account], main_config_dir: Path) -> Account | None: + """Match the main Claude identity to a stable saved profile.""" + active_org = _org_id(main_config_dir) + if active_org is None: + return None + return next( + ( + account + for account in accounts + if account.provider == "claude" and _org_id(account.auth_dir) == active_org + ), + None, + ) + + +def _stage_main_identity(main_config_dir: Path, source_config_dir: Path) -> Path: + """Prepare the post-switch main identity file, without publishing it yet. + + Staged before any credential moves so a source with no usable identity + aborts the switch while nothing has changed, and published by a single + rename after the credentials land. Credential and identity must never + disagree: a main store holding one account's token while the identity names + another would make the next sync copy that token over the other account's + saved profile, destroying a login. + """ + source_oauth = authctl.read_identity(source_config_dir).get("oauthAccount") + if not isinstance(source_oauth, dict) or not source_oauth.get("organizationUuid"): + raise authctl.CredentialError(f"{source_config_dir} has no saved account identity") + + path = authctl.identity_path(main_config_dir) + try: + main = json.loads(path.read_text()) + except (OSError, json.JSONDecodeError): + main = {} + if not isinstance(main, dict): + main = {} + main["oauthAccount"] = source_oauth + temporary = path.with_suffix(path.suffix + ".cctop-new") + temporary.write_text(json.dumps(main, separators=(",", ":"))) + os.chmod(temporary, 0o600) + return temporary + + +def _write_credential(config_dir: Path, credential: str) -> None: + """Install a credential record, skipping the write when it is already there. + + A Keychain-backed profile costs a ``security`` subprocess per write and the + active-profile sync runs on every limits poll, so no-op writes are dropped. + """ + try: + if authctl.read_credentials(config_dir) == credential: + return + except authctl.CredentialError: + pass + authctl.write_credentials(config_dir, credential) + + +def _managed_profile_name(identity: dict) -> str: + oauth = identity.get("oauthAccount") + if not isinstance(oauth, dict): + raise authctl.CredentialError("main Claude profile has no saved account identity") + org = oauth.get("organizationUuid") + if not isinstance(org, str) or not org: + raise authctl.CredentialError("main Claude profile has no organization identity") + cleaned = re.sub(r"[^A-Za-z0-9._-]", "-", org).strip(".-") + if not cleaned: + raise authctl.CredentialError("main Claude profile has an invalid organization identity") + return cleaned[:128] + + +def _write_managed_identity(profile_dir: Path, identity: dict) -> None: + oauth = identity.get("oauthAccount") + if not isinstance(oauth, dict): + raise authctl.CredentialError("main Claude profile has no saved account identity") + path = authctl.identity_path(profile_dir) + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(path.suffix + ".cctop-new") + temporary.write_text(json.dumps({"oauthAccount": oauth}, separators=(",", ":"))) + os.chmod(temporary, 0o600) + os.replace(temporary, path) + + +def ensure_stable_accounts(accounts: list[Account], main_config_dir: Path) -> list[Account]: + """Give the active login a durable, automatically managed credential store. + + A login found only in the mutable main Claude directory would otherwise be + lost on the first switch. Snapshot it once under cctop's config directory, + keyed by its organization identity, then use that profile automatically. + """ + from . import config as config_module + + current = active_account(accounts, main_config_dir) + if current is None or current.auth_dir != main_config_dir: + return accounts + + identity = authctl.read_identity(main_config_dir) + profile_dir = config_module.config_dir() / "profiles" / _managed_profile_name(identity) + with ExitStack() as stack: + for target in sorted({main_config_dir, profile_dir}, key=str): + stack.enter_context(claude_lock(target)) + stack.enter_context(claude_lock(authctl.identity_path(main_config_dir))) + credential = authctl.read_credentials(main_config_dir) + _write_managed_identity(profile_dir, identity) + if (main_config_dir / ".credentials.json").exists(): + managed_file = profile_dir / ".credentials.json" + if not managed_file.exists(): + managed_file.touch(mode=0o600) + _write_credential(profile_dir, credential) + + return [ + replace(account, credential_dir=profile_dir) if account == current else account + for account in accounts + ] + + +def switch_account( + accounts: list[Account], + target: Account, + main_config_dir: Path, + *, + sync_previous: bool = True, +) -> SwitchResult: + """Atomically activate ``target`` while preserving the current login. + + The main credential may have rotated while Claude was using it. It is + synced back to the matching saved profile before the target is installed. + Claude's own credential and config locks close races with a running process. + """ + if target.provider != "claude": + return SwitchResult(False, None, None, "Codex profiles are not switchable") + previous = active_account(accounts, main_config_dir) + if previous is not None and previous.auth_dir == target.auth_dir and sync_previous: + return SwitchResult(True, previous.name, target.name, f"{target.name} already active") + if previous is not None and previous.auth_dir == main_config_dir and sync_previous: + return SwitchResult( + False, + previous.name, + previous.name, + "active profile uses the mutable main directory; point it at a saved login dir", + ) + try: + lock_targets = {main_config_dir, target.auth_dir} + if previous is not None: + lock_targets.add(previous.auth_dir) + with ExitStack() as stack: + for config_dir in sorted(lock_targets, key=str): + stack.enter_context(claude_lock(config_dir)) + stack.enter_context(claude_lock(authctl.identity_path(main_config_dir))) + target_credential = authctl.read_credentials(target.auth_dir) + staged_identity = _stage_main_identity(main_config_dir, target.auth_dir) + try: + if previous is not None and sync_previous: + _write_credential(previous.auth_dir, authctl.read_credentials(main_config_dir)) + _write_credential(main_config_dir, target_credential) + os.replace(staged_identity, authctl.identity_path(main_config_dir)) + finally: + staged_identity.unlink(missing_ok=True) + except (authctl.CredentialError, OSError, TimeoutError) as error: + return SwitchResult( + False, + previous.name if previous else None, + previous.name if previous else None, + str(error), + ) + return SwitchResult( + True, + previous.name if previous else None, + target.name, + f"switched to {target.name}; running Claude sessions pick it up shortly", + ) + + +def sync_active_profile(accounts: list[Account], main_config_dir: Path) -> SwitchResult | None: + """Persist any token rotation from the main store into its saved profile.""" + current = active_account(accounts, main_config_dir) + if current is None or current.auth_dir == main_config_dir: + return None + try: + with ExitStack() as stack: + for config_dir in sorted({main_config_dir, current.auth_dir}, key=str): + stack.enter_context(claude_lock(config_dir)) + _write_credential(current.auth_dir, authctl.read_credentials(main_config_dir)) + except (authctl.CredentialError, OSError, TimeoutError) as error: + return SwitchResult(False, current.name, current.name, str(error)) + return SwitchResult(True, current.name, current.name, f"synced {current.name}") + + +def _binding_percent(limits: AccountLimits) -> float | None: + if limits.source != "api": + return None + values = [window.percent for window in limits.windows if window.has_data] + # A successful API response with no started windows is a fresh account at + # 0%, not an unknown account. parse_windows marks Anthropic's `0%` entries + # without reset timestamps as `has_data=False` so the UI can say "no usage + # yet"; candidate selection must still recognize that as maximum headroom. + return max(values, default=0.0) + + +def best_account( + accounts: list[Account], + limits: list[AccountLimits], + main_config_dir: Path, + max_percent: float = HEALTHY_PERCENT, + now: datetime | None = None, +) -> Account | None: + """The non-active Claude profile with the most limit headroom. + + ``max_percent`` is the usage a candidate must be under to count as healthy. + """ + now = now or datetime.now(timezone.utc) + current = active_account(accounts, main_config_dir) + by_name = {item.account: item for item in limits} + candidates: list[tuple[float, Account]] = [] + for account in accounts: + if account.provider != "claude" or account == current: + continue + expires_at = authctl.read_expiry(account.auth_dir) + if expires_at is None or expires_at <= now: + continue + limits_for_account = by_name.get( + account.name, + AccountLimits("", None, [], "none", None), + ) + percent = _binding_percent(limits_for_account) + if percent is not None and percent < max_percent: + candidates.append((percent, account)) + return min(candidates, key=lambda item: item[0])[1] if candidates else None + + +def auth_recovery_account( + accounts: list[Account], + limits: list[AccountLimits], + main_config_dir: Path, + now: datetime | None = None, +) -> Account | None: + """Choose a saved profile that can restore a rejected live credential. + + A successful usage reading is strongest evidence and remains ranked by + headroom. When the usage endpoint itself is unavailable (for example 429), + a locally unexpired credential is still sufficient to restore Claude Code. + Known-expired and known-exhausted profiles are never selected. + """ + now = now or datetime.now(timezone.utc) + by_name = {item.account: item for item in limits} + candidates: list[tuple[int, float, Account]] = [] + for account in accounts: + if account.provider != "claude" or account.auth_dir == main_config_dir: + continue + expires_at = authctl.read_expiry(account.auth_dir) + if expires_at is None or expires_at <= now: + continue + reading = by_name.get(account.name) + percent = _binding_percent(reading) if reading is not None else None + if percent is not None: + if percent >= HEALTHY_PERCENT: + continue + candidates.append((0, percent, account)) + continue + if reading is not None and not reading.retriable: + continue + candidates.append((1, -expires_at.timestamp(), account)) + return min(candidates, key=lambda item: (item[0], item[1]))[2] if candidates else None + + +def recover_auth( + accounts: list[Account], + limits: list[AccountLimits], + main_config_dir: Path, + error: str, + now: datetime | None = None, +) -> SwitchResult: + """Restore a live login without preserving the credential proven dead.""" + current = active_account(accounts, main_config_dir) + target = auth_recovery_account(accounts, limits, main_config_dir, now) + if target is None: + return SwitchResult( + False, + current.name if current else None, + current.name if current else None, + f"active login unavailable ({error}); no ready profile", + ) + return switch_account(accounts, target, main_config_dir, sync_previous=False) + + +def auto_switch( + accounts: list[Account], + limits: list[AccountLimits], + main_config_dir: Path, + remaining_percent: float, +) -> SwitchResult | None: + """Rotate when the active profile has at most ``remaining_percent`` left. + + A candidate must have strictly more headroom than the current profile. + Requiring it to clear the original trigger would strand a fully exhausted + account when, for example, a 93%-used profile can still finish the work. + Strict improvement prevents ping-pong while using every available account. + """ + current = active_account(accounts, main_config_dir) + if current is None: + return None + trigger = 100.0 - remaining_percent + current_limits = next((item for item in limits if item.account == current.name), None) + used = _binding_percent(current_limits) if current_limits is not None else None + if used is None or used < trigger: + return None + target = best_account(accounts, limits, main_config_dir, min(HEALTHY_PERCENT, used)) + if target is None: + return SwitchResult( + False, + current.name, + current.name, + f"{remaining_percent:g}% remaining; no healthy profile", + ) + return switch_account(accounts, target, main_config_dir) diff --git a/src/cctop/usage.py b/src/cctop/usage.py index 51f6f1f..fe8d239 100644 --- a/src/cctop/usage.py +++ b/src/cctop/usage.py @@ -14,7 +14,6 @@ from __future__ import annotations -import hashlib import json import subprocess import urllib.error @@ -22,21 +21,17 @@ from datetime import datetime, timezone from pathlib import Path +from . import authctl from .models import AccountLimits, LimitWindow API_BASE = "https://api.anthropic.com" USAGE_PATH = "/api/oauth/usage" OAUTH_BETA = "oauth-2025-04-20" -KEYCHAIN_SERVICE = "Claude Code-credentials" def _oauth_account(config_dir: Path) -> dict: """The oauthAccount block from an account's .claude.json (identifiers only).""" - try: - record = json.loads((config_dir / ".claude.json").read_text()) - except (OSError, json.JSONDecodeError): - return {} - account = record.get("oauthAccount") + account = authctl.read_identity(config_dir).get("oauthAccount") return account if isinstance(account, dict) else {} @@ -46,17 +41,6 @@ def read_tier(config_dir: Path) -> str | None: return tier if isinstance(tier, str) else None -def _keychain_service(config_dir: Path) -> str: - """The per-config-dir Keychain service name Claude Code uses on macOS. - - Verified empirically: `Claude Code-credentials-`. This is how a second account (a different - CLAUDE_CONFIG_DIR) is kept distinct from the default in one Keychain. - """ - digest = hashlib.sha256(str(config_dir).encode()).hexdigest()[:8] - return f"{KEYCHAIN_SERVICE}-{digest}" - - def _token_from_credentials_file(config_dir: Path) -> str | None: try: record = json.loads((config_dir / ".credentials.json").read_text()) @@ -91,18 +75,15 @@ def get_token(config_dir: Path) -> str | None: """Resolve this account's OAuth token, keyed to its config dir (not the default). - Prefers an on-disk credentials file; otherwise looks up the per-config-dir - Keychain service, then the legacy default service. The default is last so a - single-account setup still works but never shadows a second account with the - first account's token. + Prefers an on-disk credentials file; otherwise looks up the Keychain service + that belongs to this config dir. There is deliberately no fallback to + another account's service, so a logged-out dir reads as having no token + rather than borrowing the default account's. """ token = _token_from_credentials_file(config_dir) if token: return token - token = _keychain_lookup(_keychain_service(config_dir)) - if token: - return token - return _keychain_lookup(KEYCHAIN_SERVICE) + return _keychain_lookup(authctl.keychain_service(config_dir)) def _get(url: str, token: str) -> tuple[int | None, dict[str, str], str]: diff --git a/tests/test_autoswitch.py b/tests/test_autoswitch.py new file mode 100644 index 0000000..3636291 --- /dev/null +++ b/tests/test_autoswitch.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from pathlib import Path + +import pytest + +import cctop.autoswitch as autoswitch_mod +from cctop.autoswitch import AutoswitchService +from cctop.collect import Account +from cctop.config import Config +from cctop.models import AccountLimits, LimitWindow +from cctop.switcher import SwitchResult + +T0 = datetime(2026, 7, 25, 12, tzinfo=timezone.utc) + + +class FakeMonitor: + def __init__(self) -> None: + self.accounts = [Account("work", Path("/saved"))] + self.main_config_dir = Path("/main") + self.calls = [] + + def poll_limits(self, now: datetime, force: bool = False) -> list[AccountLimits]: + self.calls.append((now, force)) + return [AccountLimits("work", "max", [], "none", None, error="token expired")] + + def maybe_auto_switch(self) -> SwitchResult: + return SwitchResult(True, "work", "other", "switched to other") + + +def test_service_polls_before_applying_switch_policy(monkeypatch) -> None: + monitor = FakeMonitor() + monkeypatch.setattr( + autoswitch_mod, + "active_account", + lambda accounts, main: Account("other", Path("/other")), + ) + + cycle = AutoswitchService(monitor, 60).poll(T0) # type: ignore[arg-type] + + assert monitor.calls == [(T0, True)] + assert cycle.active == "other" + assert cycle.switch is not None and cycle.switch.ok is True + + +def test_once_does_not_wait(monkeypatch) -> None: + monitor = FakeMonitor() + monkeypatch.setattr(autoswitch_mod, "active_account", lambda accounts, main: None) + waits = [] + output = [] + + AutoswitchService(monitor, 60).run( # type: ignore[arg-type] + once=True, + emit=output.append, + wait=waits.append, + ) + + assert waits == [] + assert output == ["switched to other; work=unavailable (token expired)"] + + +def test_service_requires_hot_switch() -> None: + with pytest.raises(ValueError, match="disabled"): + AutoswitchService.from_config(Config(settings={"hot_switch": False})) + + +def test_summary_reports_successful_no_usage_profile_as_zero() -> None: + limits = AccountLimits( + "fresh", + "max", + [LimitWindow("session", "5h", 0, None, "normal", True, has_data=False)], + "api", + T0, + ) + cycle = autoswitch_mod.AutoswitchCycle("work", [limits], None) + + assert autoswitch_mod._cycle_summary(cycle) == "active work; fresh=0%" diff --git a/tests/test_backoff.py b/tests/test_backoff.py index 2327dac..881c973 100644 --- a/tests/test_backoff.py +++ b/tests/test_backoff.py @@ -11,9 +11,11 @@ from pathlib import Path import cctop.monitor as monitor_mod +from cctop.authctl import RefreshResult from cctop.collect import Account from cctop.models import AccountLimits, LimitWindow from cctop.monitor import _BACKOFF_BASE, FleetMonitor +from cctop.switcher import SwitchResult T0 = datetime(2026, 7, 17, 12, 0, 0, tzinfo=timezone.utc) @@ -119,3 +121,87 @@ def test_token_expired_drops_stale_good(monkeypatch) -> None: assert result[0].source == "none" assert result[0].error == "token expired" assert "cc-0" not in monitor._good_limits + + +def test_hot_switch_mode_auto_refreshes_saved_login(monkeypatch) -> None: + monitor, calls = _monitor(monkeypatch, [_expired(), _good(8.0)]) + monitor.hot_switch = True + refreshes = [] + + def refresh(account: str, config_dir: Path, now: datetime) -> RefreshResult: + refreshes.append((account, config_dir, now)) + return RefreshResult(account, True, "refreshed", now + timedelta(hours=12)) + + monkeypatch.setattr(monitor_mod.authctl, "refresh", refresh) + monkeypatch.setattr( + monitor_mod.authctl, + "read_expiry", + lambda config_dir: T0 + timedelta(hours=1), + ) + monkeypatch.setattr("cctop.switcher.sync_active_profile", lambda accounts, main: None) + + result = monitor.poll_limits(T0, force=True) + + assert result[0].source == "api" + assert calls[0] == 2 + assert len(refreshes) == 1 + + +def test_dead_main_credential_is_not_synced_and_requests_failover(monkeypatch) -> None: + account = Account("cc-0", Path("/saved"), "claude") + monitor = FleetMonitor( + [account], + main_config_dir=Path("/main"), + hot_switch=True, + ) + monkeypatch.setattr(monitor_mod.authctl, "read_expiry", lambda config_dir: None) + monkeypatch.setattr( + monitor_mod.authctl, + "refresh", + lambda name, config_dir, now: RefreshResult(name, False, "needs re-login", None), + ) + syncs = [] + monkeypatch.setattr( + "cctop.switcher.sync_active_profile", + lambda accounts, main: syncs.append((accounts, main)), + ) + monkeypatch.setattr(monitor_mod, "account_limits", lambda account: _rate_limited()) + recovery = [] + + def recover(accounts, limits, main, error, now): + recovery.append((accounts, limits, main, error, now)) + return SwitchResult(False, "cc-0", "cc-0", error) + + monitor.poll_limits(T0, force=True) + monkeypatch.setattr("cctop.switcher.recover_auth", recover) + result = monitor.maybe_auto_switch() + + assert syncs == [] + assert result == SwitchResult(False, "cc-0", "cc-0", "needs re-login") + assert recovery[0][3:] == ("needs re-login", T0) + + +def test_hot_switch_reports_dead_local_login_without_usage_request(monkeypatch) -> None: + monitor, calls = _monitor(monkeypatch, [_rate_limited()]) + monitor.hot_switch = True + monkeypatch.setattr( + monitor_mod.authctl, + "read_expiry", + lambda config_dir: ( + T0 + timedelta(hours=1) + if config_dir == monitor.main_config_dir + else T0 - timedelta(minutes=1) + ), + ) + monkeypatch.setattr( + monitor_mod.authctl, + "refresh", + lambda name, config_dir, now: RefreshResult(name, False, "needs re-login", None), + ) + monkeypatch.setattr("cctop.switcher.sync_active_profile", lambda accounts, main: None) + + result = monitor.poll_limits(T0, force=True) + + assert calls[0] == 0 + assert result[0].error == "needs re-login" + assert result[0].retriable is False diff --git a/tests/test_config.py b/tests/test_config.py index ecbb3ed..91347d8 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -20,13 +20,21 @@ def test_load_config_parses_settings_and_accounts(tmp_path: Path) -> None: path.write_text( "[settings]\n" "limits_refresh_seconds = 300\n\n" - '[[account]]\nname = "work"\ndir = "~/.claude"\n\n' + 'main_config_dir = "~/.claude-main"\n' + "hot_switch = true\n" + "auto_switch_remaining_percent = 1\n\n" + '[[account]]\nname = "work"\ndir = "~/.claude"\n' + 'switch_dir = "~/.claude-work-saved"\n\n' '[[account]]\ndir = "~/.claude-9"\nhidden = true\n' ) config = cfg.load_config(path) assert config.limits_refresh_seconds(180.0) == 300.0 + assert config.main_config_dir() == Path.home() / ".claude-main" + assert config.hot_switch() is True + assert config.auto_switch_remaining_percent() == 1.0 assert len(config.accounts) == 2 assert config.accounts[0].name == "work" + assert config.accounts[0].switch_dir == Path.home() / ".claude-work-saved" assert config.accounts[1].hidden is True @@ -46,7 +54,7 @@ def test_limits_refresh_default_fallback(tmp_path: Path) -> None: # -- resolve: config overrides layered over auto-detection --------------------- -def _detected() -> list[Account]: +def _detected(include_managed: bool = False) -> list[Account]: home = Path.home() return [ Account("cc-0", home / ".claude", "claude"), @@ -60,7 +68,11 @@ def test_resolve_renames_hides_and_adds(monkeypatch) -> None: home = Path.home() config = cfg.Config( accounts=[ - cfg.AccountOverride(dir=home / ".claude", name="work"), # rename + cfg.AccountOverride( # rename, with a saved login dir + dir=home / ".claude", + name="work", + switch_dir=home / ".claude-work-saved", + ), cfg.AccountOverride(dir=home / ".claude-9", hidden=True), # hide cfg.AccountOverride(dir=Path("/tmp/extra"), name="extra"), # add ] @@ -68,6 +80,7 @@ def test_resolve_renames_hides_and_adds(monkeypatch) -> None: result = collect.resolve_accounts(config) names = [a.name for a in result] assert names == ["work", "cx-0", "extra"] # renamed, hidden dropped, addition appended + assert result[0].auth_dir == home / ".claude-work-saved" assert result[-1].config_dir == Path("/tmp/extra") @@ -77,6 +90,42 @@ def test_resolve_no_config_is_pure_detection(monkeypatch) -> None: assert [a.name for a in result] == ["cc-0", "cc-9", "cx-0"] +def test_hot_switch_matches_duplicate_logins_without_account_mapping( + tmp_path: Path, monkeypatch +) -> None: + main = tmp_path / ".claude" + saved = tmp_path / ".claude-saved" + other = tmp_path / ".claude-other" + for path, org, email in ( + (main, "org-a", "a@example.com"), + (saved, "org-a", "a@example.com"), + (other, "org-b", "b@example.com"), + ): + path.mkdir() + (path / ".claude.json").write_text( + '{"oauthAccount":{"organizationUuid":"' + org + '","emailAddress":"' + email + '"}}' + ) + (path / ".credentials.json").write_text('{"claudeAiOauth":{"accessToken":"token"}}') + monkeypatch.setattr( + collect, + "discover_accounts", + lambda include_managed=False: [ + Account("cc-0", main), + Account("cc-saved", saved), + Account("cc-other", other), + ], + ) + + result = collect.resolve_accounts( + cfg.Config(settings={"hot_switch": True, "main_config_dir": str(main)}) + ) + + assert result == [ + Account("a@example.com", main, credential_dir=saved), + Account("b@example.com", other, credential_dir=other), + ] + + # -- relaxed discovery --------------------------------------------------------- @@ -89,6 +138,22 @@ def test_discover_catches_named_dirs(tmp_path: Path, monkeypatch) -> None: assert "cc-0" in names and "cc-1" in names and "cc-work" in names +def test_managed_profiles_stay_hidden_outside_hot_switch(tmp_path: Path, monkeypatch) -> None: + """Snapshotted logins would otherwise double the account they came from.""" + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config")) + (tmp_path / ".claude" / "sessions").mkdir(parents=True) + managed = cfg.config_dir() / "profiles" / "org-a" + managed.mkdir(parents=True) + (managed / ".claude.json").write_text('{"oauthAccount":{"organizationUuid":"org-a"}}') + + assert [a.name for a in collect.discover_accounts()] == ["cc-0"] + assert [a.config_dir for a in collect.discover_accounts(include_managed=True)] == [ + tmp_path / ".claude", + managed, + ] + + # -- config init --------------------------------------------------------------- @@ -109,14 +174,31 @@ def test_config_init_writes_and_refuses_overwrite(tmp_path: Path, monkeypatch, c def test_save_config_round_trips(tmp_path: Path) -> None: path = tmp_path / "config.toml" accounts = [ - cfg.AccountOverride(dir=Path("/x/.claude"), name="work", provider="claude"), + cfg.AccountOverride( + dir=Path("/x/.claude"), + name="work", + provider="claude", + switch_dir=Path("/x/.claude-work-saved"), + ), cfg.AccountOverride(dir=Path("/x/.claude-1"), name="alt", provider="claude", hidden=True), ] - cfg.save_config({"limits_refresh_seconds": 90, "heatmap_weeks": 10}, accounts, path) + cfg.save_config( + { + "limits_refresh_seconds": 90, + "heatmap_weeks": 10, + "main_config_dir": "~/.claude", + "hot_switch": True, + }, + accounts, + path, + ) loaded = cfg.load_config(path) assert loaded.limits_refresh_seconds(180.0) == 90.0 assert loaded.heatmap_weeks(26) == 10 + assert loaded.main_config_dir() == Path.home() / ".claude" + assert loaded.hot_switch() is True + assert loaded.accounts[0].switch_dir == Path("/x/.claude-work-saved") assert {o.name: o.hidden for o in loaded.accounts} == {"work": False, "alt": True} diff --git a/tests/test_limits_cache.py b/tests/test_limits_cache.py new file mode 100644 index 0000000..ee74a72 --- /dev/null +++ b/tests/test_limits_cache.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import cctop.monitor as monitor_mod +from cctop.collect import Account +from cctop.limits_cache import load, save +from cctop.models import AccountLimits, LimitWindow +from cctop.monitor import FleetMonitor + +T0 = datetime(2026, 7, 26, 12, tzinfo=timezone.utc) + + +def _reading(percent: float = 93.0) -> AccountLimits: + return AccountLimits( + "work", + "max", + [LimitWindow("session", "5h", percent, T0 + timedelta(hours=2), "warning", True)], + "api", + T0, + ) + + +def test_cache_round_trips_percent_and_reset(tmp_path: Path) -> None: + path = tmp_path / "limits-cache.json" + + save(path, {"work": _reading()}) + cached = load(path)["work"] + + assert cached.windows[0].percent == 93.0 + assert cached.windows[0].resets_at == T0 + timedelta(hours=2) + assert path.stat().st_mode & 0o777 == 0o600 + + +def test_new_monitor_shows_shared_cache_during_429(tmp_path: Path, monkeypatch) -> None: + path = tmp_path / "limits-cache.json" + save(path, {"work": _reading()}) + monkeypatch.setattr( + monitor_mod, + "account_limits", + lambda account: AccountLimits( + account.name, + "max", + [], + "none", + None, + error="usage: rate limited (429)", + retriable=True, + ), + ) + monitor = FleetMonitor( + [Account("work", Path("/work"))], + limits_cache_path=path, + ) + + result = monitor.poll_limits(T0 + timedelta(minutes=1), force=True) + + assert result[0].windows[0].percent == 93.0 + assert result[0].windows[0].resets_at == T0 + timedelta(hours=2) + assert result[0].error == "usage: rate limited (429)" + + +def test_running_monitor_adopts_newer_shared_reading_without_network( + tmp_path: Path, monkeypatch +) -> None: + path = tmp_path / "limits-cache.json" + calls = [] + monitor = FleetMonitor( + [Account("work", Path("/work"))], + limits_interval=timedelta(minutes=3), + limits_cache_path=path, + ) + save(path, {"work": _reading(81.0)}) + monkeypatch.setattr(monitor_mod, "account_limits", lambda account: calls.append(account)) + + result = monitor.poll_limits(T0 + timedelta(minutes=1)) + + assert result[0].windows[0].percent == 81.0 + assert calls == [] + + +def test_recent_partial_cache_does_not_hide_uncached_account(tmp_path: Path) -> None: + path = tmp_path / "limits-cache.json" + save(path, {"work": _reading()}) + monitor = FleetMonitor( + [ + Account("work", Path("/work")), + Account("needs-login", Path("/needs-login")), + ], + limits_interval=timedelta(minutes=3), + limits_cache_path=path, + ) + + assert monitor.limits_due(T0 + timedelta(minutes=1)) is True diff --git a/tests/test_switcher.py b/tests/test_switcher.py new file mode 100644 index 0000000..9a44e82 --- /dev/null +++ b/tests/test_switcher.py @@ -0,0 +1,340 @@ +"""Hot switching preserves saved logins and rotates only at the configured edge.""" + +from __future__ import annotations + +import json +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from cctop.collect import Account +from cctop.models import AccountLimits, LimitWindow +from cctop.switcher import ( + active_account, + auth_recovery_account, + auto_switch, + best_account, + ensure_stable_accounts, + recover_auth, + switch_account, + sync_active_profile, +) + + +def _profile( + path: Path, + org: str, + token: str, + expires_at: datetime | None = None, +) -> None: + expires_at = expires_at or datetime(2030, 1, 1, tzinfo=timezone.utc) + path.mkdir() + (path / ".claude.json").write_text( + json.dumps({"oauthAccount": {"organizationUuid": org, "emailAddress": f"{org}@x"}}) + ) + (path / ".credentials.json").write_text( + json.dumps( + { + "claudeAiOauth": { + "accessToken": token, + "refreshToken": f"refresh-{token}", + "expiresAt": int(expires_at.timestamp() * 1000), + } + } + ) + ) + + +def _limits(name: str, percent: float) -> AccountLimits: + return AccountLimits( + name, + "max", + [LimitWindow("session", "5h", percent, None, "normal", True)], + "api", + datetime.now(timezone.utc), + ) + + +def test_switch_syncs_rotated_main_credential_back_to_previous_profile(tmp_path: Path) -> None: + main = tmp_path / "main" + previous = tmp_path / "saved-a" + target = tmp_path / "saved-b" + _profile(main, "org-a", "rotated-a") + _profile(previous, "org-a", "stale-a") + _profile(target, "org-b", "token-b") + (main / ".claude.json").write_text( + json.dumps({"keep": True, "oauthAccount": {"organizationUuid": "org-a"}}) + ) + accounts = [Account("a", previous), Account("b", target)] + + result = switch_account(accounts, accounts[1], main) + + assert result.ok is True + assert ( + json.loads((previous / ".credentials.json").read_text())["claudeAiOauth"]["accessToken"] + == "rotated-a" + ) + assert ( + json.loads((main / ".credentials.json").read_text())["claudeAiOauth"]["accessToken"] + == "token-b" + ) + identity = json.loads((main / ".claude.json").read_text()) + assert identity["keep"] is True + assert identity["oauthAccount"]["organizationUuid"] == "org-b" + assert active_account(accounts, main) == accounts[1] + + +def test_sync_active_profile_persists_main_token_rotation(tmp_path: Path) -> None: + main = tmp_path / "main" + saved = tmp_path / "saved" + _profile(main, "org-a", "rotated-a") + _profile(saved, "org-a", "stale-a") + accounts = [Account("a", saved)] + + result = sync_active_profile(accounts, main) + + assert result is not None and result.ok is True + assert ( + json.loads((saved / ".credentials.json").read_text())["claudeAiOauth"]["accessToken"] + == "rotated-a" + ) + + +def test_switch_refuses_mutable_main_as_only_saved_profile(tmp_path: Path) -> None: + main = tmp_path / "main" + target = tmp_path / "saved-b" + _profile(main, "org-a", "token-a") + _profile(target, "org-b", "token-b") + accounts = [Account("a", main), Account("b", target)] + + result = switch_account(accounts, accounts[1], main) + + assert result.ok is False + assert "mutable main directory" in result.message + assert active_account(accounts, main) == accounts[0] + + +def test_main_only_login_gets_automatic_managed_profile(tmp_path: Path, monkeypatch) -> None: + main = tmp_path / "main" + _profile(main, "org-a", "token-a") + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config")) + accounts = [Account("org-a@x", main)] + + prepared = ensure_stable_accounts(accounts, main) + + managed = tmp_path / "config" / "cctop" / "profiles" / "org-a" + assert prepared == [Account("org-a@x", main, credential_dir=managed)] + assert ( + json.loads((managed / ".credentials.json").read_text())["claudeAiOauth"]["accessToken"] + == "token-a" + ) + assert ( + json.loads((managed / ".claude.json").read_text())["oauthAccount"]["organizationUuid"] + == "org-a" + ) + + +def test_best_account_chooses_most_headroom_and_skips_active(tmp_path: Path) -> None: + main = tmp_path / "main" + a, b, c = tmp_path / "a", tmp_path / "b", tmp_path / "c" + _profile(main, "org-a", "main-a") + _profile(a, "org-a", "a") + _profile(b, "org-b", "b") + _profile(c, "org-c", "c") + accounts = [Account("a", a), Account("b", b), Account("c", c)] + + chosen = best_account(accounts, [_limits("a", 99), _limits("b", 40), _limits("c", 10)], main) + + assert chosen == accounts[2] + + +def test_best_account_rejects_stale_usage_for_an_expired_profile(tmp_path: Path) -> None: + main = tmp_path / "main" + active = tmp_path / "active" + expired = tmp_path / "expired" + ready = tmp_path / "ready" + now = datetime.now(timezone.utc) + _profile(main, "org-a", "main-a") + _profile(active, "org-a", "active-a") + _profile(expired, "org-b", "expired-b", now - timedelta(minutes=1)) + _profile(ready, "org-c", "ready-c", now + timedelta(hours=6)) + accounts = [ + Account("active", active), + Account("expired", expired), + Account("ready", ready), + ] + + chosen = best_account( + accounts, + [_limits("active", 92), _limits("expired", 10), _limits("ready", 20)], + main, + now=now, + ) + + assert chosen == accounts[2] + + +def test_best_account_selects_successful_profile_with_no_usage_yet(tmp_path: Path) -> None: + main = tmp_path / "main" + active = tmp_path / "active" + fresh = tmp_path / "fresh" + _profile(main, "org-a", "main-a") + _profile(active, "org-a", "saved-a") + _profile(fresh, "org-b", "fresh-b") + accounts = [Account("active", active), Account("fresh", fresh)] + no_usage_yet = AccountLimits( + "fresh", + "max", + [LimitWindow("session", "5h", 0, None, "normal", True, has_data=False)], + "api", + datetime.now(timezone.utc), + ) + + chosen = best_account(accounts, [_limits("active", 90), no_usage_yet], main) + + assert chosen == accounts[1] + + +def test_auto_switch_fires_at_one_percent_remaining(tmp_path: Path) -> None: + main = tmp_path / "main" + a, b = tmp_path / "a", tmp_path / "b" + _profile(main, "org-a", "rotated-a") + _profile(a, "org-a", "saved-a") + _profile(b, "org-b", "token-b") + accounts = [Account("a", a), Account("b", b)] + + assert auto_switch(accounts, [_limits("a", 98.9), _limits("b", 10)], main, 1.0) is None + result = auto_switch(accounts, [_limits("a", 99.0), _limits("b", 10)], main, 1.0) + + assert result is not None and result.ok is True + assert result.active == "b" + + +def test_auto_switch_refuses_a_target_with_less_headroom(tmp_path: Path) -> None: + """A worse candidate would re-trigger rotation and is not an escape.""" + main = tmp_path / "main" + a, b = tmp_path / "a", tmp_path / "b" + _profile(main, "org-a", "rotated-a") + _profile(a, "org-a", "saved-a") + _profile(b, "org-b", "token-b") + accounts = [Account("a", a), Account("b", b)] + + result = auto_switch(accounts, [_limits("a", 92), _limits("b", 95)], main, 10.0) + + assert result is not None and result.ok is False + assert "no healthy profile" in result.message + assert active_account(accounts, main) == accounts[0] + + +def test_exhausted_account_switches_to_imperfect_but_better_profile(tmp_path: Path) -> None: + main = tmp_path / "main" + exhausted = tmp_path / "exhausted" + better = tmp_path / "better" + _profile(main, "org-a", "main-a") + _profile(exhausted, "org-a", "saved-a") + _profile(better, "org-b", "token-b") + accounts = [Account("exhausted", exhausted), Account("better", better)] + + result = auto_switch( + accounts, + [_limits("exhausted", 100), _limits("better", 93)], + main, + 10.0, + ) + + assert result is not None and result.ok is True + assert result.active == "better" + + +def test_switch_leaves_credential_and_identity_agreeing_when_it_fails(tmp_path: Path) -> None: + """A target with no identity must not strand the target's token in main.""" + main = tmp_path / "main" + previous = tmp_path / "saved-a" + target = tmp_path / "saved-b" + _profile(main, "org-a", "rotated-a") + _profile(previous, "org-a", "stale-a") + _profile(target, "org-b", "token-b") + (target / ".claude.json").write_text(json.dumps({"oauthAccount": {}})) + accounts = [Account("a", previous), Account("b", target)] + + result = switch_account(accounts, accounts[1], main) + + assert result.ok is False + assert active_account(accounts, main) == accounts[0] + # Main still holds account a's token, matching the identity it still claims, + # so the next sync cannot copy b's token over a's saved profile. + assert ( + json.loads((main / ".credentials.json").read_text())["claudeAiOauth"]["accessToken"] + == "rotated-a" + ) + assert not list(main.glob("*.cctop-new")) + + +def test_sync_skips_the_write_when_the_profile_is_already_current(tmp_path: Path) -> None: + main = tmp_path / "main" + saved = tmp_path / "saved" + _profile(main, "org-a", "token-a") + _profile(saved, "org-a", "token-a") + credentials = saved / ".credentials.json" + before = credentials.stat().st_mtime_ns + + assert sync_active_profile([Account("a", saved)], main) is not None + assert credentials.stat().st_mtime_ns == before + + +def test_auth_recovery_uses_longest_lived_profile_when_usage_is_rate_limited( + tmp_path: Path, +) -> None: + main = tmp_path / "main" + short = tmp_path / "short" + long = tmp_path / "long" + now = datetime.now(timezone.utc) + _profile(main, "org-a", "dead", now - timedelta(minutes=1)) + _profile(short, "org-b", "short", now + timedelta(minutes=10)) + _profile(long, "org-c", "long", now + timedelta(hours=6)) + accounts = [Account("short", short), Account("long", long)] + unavailable = [ + AccountLimits( + account.name, + "max", + [], + "none", + None, + error="usage: rate limited (429)", + retriable=True, + ) + for account in accounts + ] + + assert auth_recovery_account(accounts, unavailable, main, now) == accounts[1] + + +def test_auth_recovery_does_not_copy_dead_main_over_saved_profile(tmp_path: Path) -> None: + main = tmp_path / "main" + saved = tmp_path / "saved" + now = datetime.now(timezone.utc) + _profile(main, "org-a", "dead-main", now - timedelta(minutes=1)) + _profile(saved, "org-a", "valid-saved", now + timedelta(hours=6)) + accounts = [Account("a", saved)] + unavailable = [ + AccountLimits( + "a", + "max", + [], + "none", + None, + error="usage: rate limited (429)", + retriable=True, + ) + ] + + result = recover_auth(accounts, unavailable, main, "needs re-login", now) + + assert result.ok is True + assert ( + json.loads((main / ".credentials.json").read_text())["claudeAiOauth"]["accessToken"] + == "valid-saved" + ) + assert ( + json.loads((saved / ".credentials.json").read_text())["claudeAiOauth"]["accessToken"] + == "valid-saved" + ) diff --git a/uv.lock b/uv.lock index 1a1fad9..d7fcee2 100644 --- a/uv.lock +++ b/uv.lock @@ -48,7 +48,7 @@ wheels = [ [[package]] name = "cctop-tui" -version = "0.1.0" +version = "0.3.0" source = { editable = "." } dependencies = [ { name = "rich" },