ai-agent/backend/accounts.py

"""
Claude accounts — which subscription a conversation runs (and is billed) on.

The homelab runs Claude Code under two logins: the owner's **personal** Max
account (the CLI's default config dir, ``~/.claude``) and a **work** seat (Orus)
whose config dir is ``~/.claude-work``. Claude Code moves the whole
account-shaped state with ``CLAUDE_CONFIG_DIR`` — credentials, ``.claude.json``,
``settings.json`` and the ``projects/`` transcripts — so an account *is* a
config dir, and the sidecar picks one per run by exporting that variable.

The profiles are **hard-coded here**, not user settings: the backend has to
decide a conversation's account for its own server-side totals (activity,
projects, skills, agents), so it can't wait on a client preference. Settings
only owns the presentation (plan name, monthly price, whether an account counts
in the aggregated totals of the dashboards).

A conversation's account resolves in this order (``resolve``):

1. **explicit** — ``meta.account``, stamped by ``/api/spawn`` (and carried over
   by resume/fork/cron), i.e. the account the run was actually launched on;
2. **source** — ``meta.accountSource``, stamped by the indexer when it imports
   a transcript out of a non-default account's ``projects/`` dir (catches a work
   session started from a terminal, in any cwd);
3. **cwd** — the working-folder rule: a session that ran inside one of the
   account's repos (or a git worktree of one) is that account's;
4. the default account.

No DB migration or re-parse is needed: every summary already carries its
``cwd``, and subagents share their parent's ``sessionId`` (so its metadata)
and ``cwd``.
"""

from __future__ import annotations

import os
import pathlib

DEFAULT = "personal"

# ``configDir`` None ⇒ the CLI's own default (``~/.claude`` + ``~/.claude.json``).
# Never set CLAUDE_CONFIG_DIR to ``~/.claude`` for it: with the override the CLI
# keeps ``.claude.json`` *inside* the dir, so the personal login would look new.
# ``cwd`` is where a new run of that account starts when the composer didn't
# choose (``~`` is expanded by the host sidecar, not here — the backend runs in
# a container whose home isn't the host's). ``projects`` are the ``~/projects``
# slugs whose conversations belong to the account (a project tag routes a spawn,
# and the cwd rule matches the same names, worktrees included).
ACCOUNTS: list[dict] = [
    {
        "id": "personal",
        "label": "Personal",
        "description": "Gabriel's own Claude Max subscription — every lab session.",
        "color": "sky",
        "configDir": None,
        "cwd": None,
        "projects": [],
        "excludeFromTotals": False,
        "default": True,
    },
    {
        "id": "work",
        "label": "Work",
        "description": "The Orus Claude seat — sessions in orus-monorepo.",
        "color": "amber",
        "configDir": "~/.claude-work",
        "cwd": "~/projects/orus-monorepo",
        "projects": ["orus-monorepo"],
        # Work spend is the employer's seat, not the owner's money: it stays
        # out of every aggregated total unless a view asks for it.
        "excludeFromTotals": True,
        "default": False,
    },
]

_BY_ID = {a["id"]: a for a in ACCOUNTS}

# Where each non-default account's live transcripts are mounted in the
# container (compose: ``${HOME}/.claude-work/projects:/transcripts-work:ro``).
# The default account's source is the backend's ``TRANSCRIPTS_DIR``.
_TRANSCRIPTS_ENV = {"work": "WORK_TRANSCRIPTS_DIR"}


def ids() -> list[str]:
    return [a["id"] for a in ACCOUNTS]


def get(account_id: str | None) -> dict | None:
    return _BY_ID.get((account_id or "").strip().lower())


def valid(account_id: str | None) -> str | None:
    """The canonical id for ``account_id``, or None when it names no account."""
    a = get(account_id)
    return a["id"] if a else None


def counts_in_totals(account_id: str | None) -> bool:
    a = get(account_id) or _BY_ID[DEFAULT]
    return not a.get("excludeFromTotals")


def transcripts_dirs() -> dict[str, pathlib.Path]:
    """account id → its live transcripts dir, for the non-default accounts whose
    mount env is set (an unset/empty env ⇒ that account isn't ingested)."""
    out: dict[str, pathlib.Path] = {}
    for acct, env in _TRANSCRIPTS_ENV.items():
        raw = os.environ.get(env, "")
        if raw:
            out[acct] = pathlib.Path(raw).resolve()
    return out


def _segments(path: str) -> list[str]:
    return [p for p in (path or "").replace("\\", "/").split("/") if p]


def account_for_cwd(cwd: str | None) -> str | None:
    """The account whose repo ``cwd`` sits in, else None.

    Matches a path segment equal to one of an account's project slugs
    (``…/projects/orus-monorepo/apps/web``) and the worktree dirs
    ``scripts/new-worktree.sh -p <slug>`` makes (``…/worktrees/<slug>-<name>``).
    A bare prefix is not enough — ``orus-monorepo-notes`` outside ``worktrees/``
    is a different directory."""
    segs = _segments(cwd or "")
    if not segs:
        return None
    for a in ACCOUNTS:
        for slug in a.get("projects") or []:
            for i, seg in enumerate(segs):
                if seg == slug:
                    return a["id"]
                if (seg.startswith(slug + "-") and i > 0
                        and segs[i - 1] == "worktrees"):
                    return a["id"]
    return None


def account_for_projects(projects: list[str] | None) -> str | None:
    """The account a spawn tagged with ``projects`` routes to (the first
    non-default account claiming one of them), else None."""
    for p in projects or []:
        slug = (p or "").strip().split(":")[-1]  # tolerate "project:<slug>"
        for a in ACCOUNTS:
            if slug and slug in (a.get("projects") or []):
                return a["id"]
    return None


def resolve(summary: dict | None, meta: dict | None) -> str:
    """A conversation's account: explicit stamp → import source → cwd → default."""
    m = meta or {}
    for key in ("account", "accountSource"):
        v = valid(m.get(key))
        if v:
            return v
    return account_for_cwd((summary or {}).get("cwd")) or DEFAULT


def selector(param: str | None):
    """A predicate over account ids for an ``?account=`` query param.

    * unset / ``""`` / ``totals`` — the aggregated-totals view: every account
      except the ones flagged ``excludeFromTotals`` (the default everywhere);
    * ``all`` — every account;
    * an account id — that account only.
    Unknown ids select nothing (an empty page beats silently widening)."""
    p = (param or "").strip().lower()
    if p in ("", "totals"):
        return counts_in_totals
    if p == "all":
        return lambda _a: True
    return lambda a: (a or DEFAULT) == p


def public() -> list[dict]:
    """The profiles as ``/api/accounts`` serves them (no host paths)."""
    return [{"id": a["id"], "label": a["label"],
             "description": a["description"], "color": a["color"],
             "projects": list(a.get("projects") or []),
             "excludeFromTotals": bool(a.get("excludeFromTotals")),
             "default": bool(a.get("default"))} for a in ACCOUNTS]


def sidecar_payload(account_id: str) -> dict:
    """What the sidecar needs to launch on an account: its id and the default
    cwd for a fresh run (host-relative, ``~``-expanded over there)."""
    a = get(account_id) or _BY_ID[DEFAULT]
    return {"account": a["id"], "cwd": a.get("cwd")}