ai-agent/sidecar/routes_accounts.py

"""
``/health`` and the account routes: ``/accounts`` (login health) and the
headless login / logout flow under ``/accounts/{account}/…``.
"""

import re

from fastapi import APIRouter, Header, HTTPException
from pydantic import BaseModel

import claude_accounts
import claude_cli
import pairing
from claude_accounts import (_account_env, _auth_status, _login_account,
                             _login_error)
from config import (ACCOUNTS, CLAUDE_BIN, DEFAULT_ACCOUNT, DEFAULT_CWD,
                    DEFAULT_MODEL, PERMISSION_MODE, TOKEN)
from validate import _auth

router = APIRouter()

@router.get("/health")
def health() -> dict:
    """Open (no bearer): the deploy probe and the hub's worker poller read it.
    The worker fields say who this runner is — ``paired`` is whether a hub
    holds its token (SIDECAR_TOKEN counts, so the host sidecar reads paired)."""
    return {"ok": True, "cwd": DEFAULT_CWD, "claude": CLAUDE_BIN,
            "defaultModel": DEFAULT_MODEL, "accounts": list(ACCOUNTS),
            "defaultAccount": DEFAULT_ACCOUNT,
            "permissionMode": PERMISSION_MODE,
            **pairing.identity(), "paired": bool(TOKEN or pairing.token())}


@router.get("/accounts")
def accounts(authorization: str | None = Header(default=None)) -> dict:
    """Each account's login health (who it is, which org, which plan)."""
    _auth(authorization)
    return {"accounts": [{**_auth_status(a),
                          "pendingLogin": claude_cli.pending_login(a)}
                         for a in ACCOUNTS],
            "default": DEFAULT_ACCOUNT}


# ---- login / logout ---------------------------------------------------------
# `claude auth login` driven headlessly — see claude_cli.py for the flow. Each
# account logs in under its own env (_account_env), so the credentials land in
# that account's config dir; the default account writes the host user's own
# ~/.claude login.

class LoginBody(BaseModel):
    email: str | None = None          # pre-fills the sign-in page (--email)


class LoginCodeBody(BaseModel):
    code: str                         # the `code#state` the sign-in page shows


@router.post("/accounts/{account}/login")
def login_start(account: str, body: LoginBody | None = None,
                authorization: str | None = Header(default=None)) -> dict:
    """Start a subscription login: returns the sign-in URL to open (on any
    device); the code it ends with goes to ``/login/code``."""
    _auth(authorization)
    account = _login_account(account)
    email = ((body.email if body else None) or "").strip() or None
    if email and not re.fullmatch(r"[^@\s]+@[^@\s]+", email):
        raise HTTPException(400, "invalid email")
    try:
        return claude_cli.start_login(account, [CLAUDE_BIN],
                                      _account_env(account), email)
    except claude_cli.LoginError as e:
        raise _login_error(e, account)


@router.post("/accounts/{account}/login/code")
def login_code(account: str, body: LoginCodeBody,
               authorization: str | None = Header(default=None)) -> dict:
    """Finish the pending login with the pasted code; answers the fresh
    ``auth status`` of the account."""
    _auth(authorization)
    account = _login_account(account)
    try:
        claude_cli.submit_code(account, body.code)
    except claude_cli.LoginError as e:
        raise _login_error(e, account)
    claude_accounts._account_status_cache.pop(account, None)
    return _auth_status(account)


@router.delete("/accounts/{account}/login")
def login_cancel(account: str,
                 authorization: str | None = Header(default=None)) -> dict:
    _auth(authorization)
    return {"cancelled": claude_cli.cancel_login(_login_account(account))}


@router.post("/accounts/{account}/logout")
def logout(account: str,
           authorization: str | None = Header(default=None)) -> dict:
    """`claude auth logout` for the account. New launches on it fail with a
    typed ``auth`` error until it logs back in; runs already in flight may
    keep going on the access token they hold in memory."""
    _auth(authorization)
    account = _login_account(account)
    claude_cli.cancel_login(account)
    try:
        msg = claude_cli.logout([CLAUDE_BIN], _account_env(account))
    except claude_cli.LoginError as e:
        raise _login_error(e, account)
    claude_accounts._account_status_cache.pop(account, None)
    return {**_auth_status(account), "message": msg}