ai-agent/sidecar/claude_accounts.py

"""
Claude accounts: which login a run launches on, the environment that selects
it, where its transcripts live, and its ``claude auth status`` (cached).

Owns the mutable ``_account_status_cache`` — the login routes invalidate it as
``claude_accounts._account_status_cache``.
"""

import json
import os
import pathlib
import subprocess
import time

from fastapi import HTTPException

import claude_cli
import outbox
import pairing
from config import (ACCOUNT_STATUS_TTL_S, ACCOUNTS, BG_WAIT_CEILING_MS, LOG_DIR,
                    CLAUDE_BIN, CLAUDE_PROJECTS, CLI_DIR, DEFAULT_ACCOUNT,
                    IS_WORKER, RUN_ENV_STRIP, STRIP_API_KEY)

def _valid_account(account: str | None) -> str:
    """The account a claude run launches on (unset ⇒ DEFAULT_ACCOUNT). An
    unknown id is a 400 — never a quiet fallback onto the default account,
    which would bill the wrong subscription."""
    a = (account or "").strip().lower() or DEFAULT_ACCOUNT
    if a not in ACCOUNTS:
        raise HTTPException(400, f"unknown account: {account!r} "
                                 f"(known: {', '.join(ACCOUNTS)})")
    return a


def _require_login(account: str) -> None:
    """Fail loudly when a non-default account's config dir has no login.

    Without a `.credentials.json` there, `claude -p` would stop on "not logged
    in" — or worse, fall back to an API key. Refusing up front puts the fix in
    the error the composer shows."""
    cfg = ACCOUNTS.get(account)
    if cfg is None:
        return  # the default account — its login is the host user's own
    if not (cfg / ".credentials.json").is_file():
        raise HTTPException(409, claude_cli.error_detail(
            f"account {account!r} is not logged in", kind="auth",
            account=account))


def _outbox_url() -> str:
    bind = os.environ.get("SIDECAR_BIND", "127.0.0.1")
    return f"http://{bind}:{os.environ.get('SIDECAR_PORT', '8790')}"


def _run_env(sid: str, account: str | None) -> dict[str, str]:
    """``_account_env`` plus what the notify CLI needs: the session id (so a
    push links the right conversation without a resolver) and, on a worker,
    the outbox transport."""
    env = _account_env(account)
    env["CLAUDE_SESSION_ID"] = sid
    env["PATH"] = f"{env.get('PATH', '')}:{CLI_DIR}" if env.get("PATH") else str(CLI_DIR)
    if BG_WAIT_CEILING_MS:
        env["CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS"] = BG_WAIT_CEILING_MS
    env["STOP_GUARD_STATE_DIR"] = str(LOG_DIR)  # stop_guard.py's hold marker + mailbox
    if IS_WORKER:
        env["AI_AGENT_OUTBOX_URL"] = _outbox_url()
        env["AI_AGENT_OUTBOX_TOKEN"] = outbox.run_token(pairing.state_dir())
    return env


def _account_env(account: str | None) -> dict[str, str]:
    """The environment a run launches with: the account's config dir exported
    as CLAUDE_CONFIG_DIR (the default account gets it *removed*, so a stray
    value in the unit can't redirect it), and ANTHROPIC_API_KEY stripped when
    accounts are declared (see STRIP_API_KEY). ``account`` None = a pi run,
    which isn't a Claude login at all: the host env passes through as-is."""
    env = {**os.environ, "CLAUDE_CODE_ENTRYPOINT": "ai-agent-sidecar"}
    # The sidecar's own bearer is not the run's business: with it, any session
    # could drive /spawn, /message, /interrupt and the account logins of this
    # very sidecar, outside whatever scope its own launch was given.
    for k in RUN_ENV_STRIP:
        env.pop(k, None)
    if account is None:
        return env
    env.pop("CLAUDE_CONFIG_DIR", None)
    if STRIP_API_KEY:
        env.pop("ANTHROPIC_API_KEY", None)
    cfg = ACCOUNTS.get(account)
    if cfg is not None:
        env["CLAUDE_CONFIG_DIR"] = str(cfg)
    return env


def _account_projects(account: str) -> pathlib.Path:
    """Where an account's transcripts live (one dir per cwd slug)."""
    cfg = ACCOUNTS.get(account)
    return CLAUDE_PROJECTS if cfg is None else cfg / "projects"


_account_status_cache: dict[str, tuple[float, dict]] = {}


def _auth_status(account: str) -> dict:
    """``claude auth status`` run under the account's config dir, reduced to
    the fields the Settings health card shows. Cached ACCOUNT_STATUS_TTL_S."""
    hit = _account_status_cache.get(account)
    if hit and time.monotonic() - hit[0] < ACCOUNT_STATUS_TTL_S:
        return hit[1]
    cfg = ACCOUNTS.get(account)
    out: dict = {"id": account, "configDir": str(cfg) if cfg else None,
                 "default": account == DEFAULT_ACCOUNT, "loggedIn": False}
    try:
        r = subprocess.run([CLAUDE_BIN, "auth", "status"], capture_output=True,
                           text=True, timeout=20, stdin=subprocess.DEVNULL,
                           env=_account_env(account))
        st = json.loads(r.stdout or "{}")
        out.update({k: st.get(k) for k in (
            "loggedIn", "authMethod", "email", "orgName", "subscriptionType",
            "projectsDirectory")})
        out["loggedIn"] = bool(st.get("loggedIn"))
    except (OSError, ValueError, subprocess.TimeoutExpired) as e:
        out["error"] = str(e)[:200]
    _account_status_cache[account] = (time.monotonic(), out)
    return out


def _login_account(account: str) -> str:
    a = (account or "").strip().lower()
    if a not in ACCOUNTS:
        raise HTTPException(404, f"unknown account: {account!r}")
    return a


def _login_error(e: "claude_cli.LoginError", account: str) -> HTTPException:
    return HTTPException(e.status, claude_cli.error_detail(
        e.message, kind=e.kind, account=account))