ai-agent/sidecar/sidecar.py

"""
claude-sidecar — a tiny host-side FastAPI wrapper that launches `claude -p`.

The ai-agent viewer runs in a Docker container and therefore can't reach the
host's `claude` CLI (auth, hooks, skills all live in the host user's
`~/.claude`). This sidecar runs natively on the host as the same user, so a
session it spawns is *exactly* like one started from a terminal — same
credentials, same RTK/vault hooks, same CLAUDE.md/skills.

Flow:
  ai-agent backend  ──POST /spawn {prompt, sessionId}──▶  this sidecar
                    ◀──────── {sessionId, pid} ─────────
The sidecar fires `claude -p <prompt> --session-id <uuid> --output-format json`
as a detached background process (its own session/process group) and returns as
soon as the run is known to have *started* (it watches the child for a couple of
seconds and reports a 502 if it dies on the spot — a resume whose transcript
can't be found, a bad cwd, a broken CLI). Claude writes its transcript to
`~/.claude/projects/-home-gabrielvidal-homelab/<uuid>.jsonl`, which the ai-agent
container already watches read-only — so the new conversation shows up in the
viewer within a couple of seconds and the UI redirects to it.

Restart-proofing (why the runs survive the sidecar restarting).
  Spawned runs are *daemons*: `start_new_session=True` detaches them into their
  own session/process group, and the systemd unit runs with `KillMode=process`
  so a `systemctl restart claude-sidecar` (i.e. what happens when Claude edits
  *this very sidecar* and redeploys it) only signals uvicorn — the in-flight
  `claude -p` children keep running. Their identity is persisted to disk in a
  per-session pidfile (`logs/<uuid>.pid`, JSON with pid + `/proc` start-time),
  so a freshly restarted sidecar re-discovers the survivors on startup and can
  still list and interrupt them. Startup also GCs pidfiles whose process is gone
  or whose PID has been recycled (start-time mismatch).

Auth is a shared bearer token (SIDECAR_TOKEN) so nothing on the LAN can drive
`claude` on the host but the ai-agent backend.

Accounts. A claude run launches on one Claude login: the default account (the
CLI's own `~/.claude`) or one declared in SIDECAR_ACCOUNTS (`work=~/.claude-work`),
whose dir is exported as CLAUDE_CONFIG_DIR for that run. Resume and fork must
name the account the session started on — its transcript only exists under
that account's `projects/`. A non-default account with no login is a 409, never
a fallback, and ANTHROPIC_API_KEY is stripped once accounts are declared.
"""

import hmac
import json
import logging
import os
import pathlib
import re
import signal
import subprocess
import time
import uuid as uuidlib

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

import claude_cli
import feed
import outbox
import pairing
import skills_list


class _QuietSessionsPoll(logging.Filter):
    """Drop ``/sessions`` access-log lines: the ai-agent backend polls that
    endpoint every ~2s for run-exit detection, which would otherwise flood the
    journal with tens of thousands of identical lines a day."""

    def filter(self, record: logging.LogRecord) -> bool:
        return "/sessions" not in record.getMessage()


logging.getLogger("uvicorn.access").addFilter(_QuietSessionsPoll())

TOKEN = os.environ.get("SIDECAR_TOKEN", "")
CLAUDE_BIN = os.environ.get("CLAUDE_BIN", "claude")
# The pi-harness runner (runner/cli.mjs): wraps `pi --mode
# json` and mirrors the session as a Claude-Code-schema transcript the viewer
# watches. Launched with the exact same flag shape as `claude`.
RUNNER_BIN = os.environ.get(
    "RUNNER_BIN",
    str(pathlib.Path(__file__).resolve().parent.parent / "runner" / "cli.mjs"))
DEFAULT_CWD = os.environ.get("SIDECAR_CWD", str(pathlib.Path.home() / "homelab"))
DEFAULT_MODEL = os.environ.get("SIDECAR_MODEL", "opus")
# What a pi-harness run uses when no model tag was picked (an OpenRouter id;
# the runner resolves each id's provider from ~/.pi/agent/models.json, so a
# locally-served id may carry a vendor prefix too — see runner/cli.mjs).
PI_DEFAULT_MODEL = os.environ.get("SIDECAR_PI_MODEL", "qwen/qwen3.8-27b")
# What `claude --model` accepts: a family alias (`opus`) or a full model id
# (`claude-opus-4-8`, `claude-haiku-4-5-20251001`). The viewer picks from the
# live Anthropic model list, so we don't pin an allow-list here — we only reject
# shapes that aren't a model at all, since the value goes onto a command line.
MODEL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
# pi model ids additionally carry a vendor prefix and optional thinking suffix
# (`qwen/qwen3.6-35b-a3b`, `…:thinking`).
PI_MODEL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/:-]{0,127}$")
# Turning thinking off is the single thinking control the composer exposes, and
# each harness spells it its own way: the claude CLI takes `--thinking disabled`
# (its scale is enabled|adaptive|disabled), pi takes `--thinking off` (its scale
# is off|minimal|low|medium|high|xhigh|max). Thinking *on* emits no flag at all,
# so each CLI keeps whatever its own default level is — we only ever say "off".
THINKING_OFF = {"claude": "disabled", "pi": "off"}
# The claude CLI's `--effort <level>` — how hard the model works a turn. It is
# the *claude-only* replacement for the thinking toggle: the composer shows an
# effort select for claude runs and keeps the on/off toggle for pi, which has no
# equivalent flag. Unset emits nothing, so the CLI keeps its own default.
EFFORT_LEVELS = ("low", "medium", "high", "xhigh", "max")
# Headless sessions can't answer permission prompts, so they run with
# permissions bypassed (same as `--dangerously-skip-permissions`). This is the
# owner's own box acting on its own repo; keep it behind the bearer token.
PERMISSION_MODE = os.environ.get("SIDECAR_PERMISSION_MODE", "bypassPermissions")
# Tools a sidecar-spawned claude run must not use. AskUserQuestion is the
# interactive terminal chooser — headless runs have no terminal to answer it
# on, and the ai-agent's own ask channel is the `ask-form` skill (a form the
# PWA renders inline in the conversation). Space-separated env to override;
# set SIDECAR_DISALLOWED_TOOLS="" to disable.
DISALLOWED_TOOLS = os.environ.get(
    "SIDECAR_DISALLOWED_TOOLS", "AskUserQuestion").split()
# Enable Claude Code Remote Control by default so every sidecar-spawned run
# shows up in the Claude app and can be driven from the phone. `claude
# --remote-control` with no name auto-generates one from the hostname prefix.
# Set SIDECAR_REMOTE_CONTROL=0 to opt out.
REMOTE_CONTROL = os.environ.get("SIDECAR_REMOTE_CONTROL", "1") not in ("0", "false", "")
LOG_DIR = pathlib.Path(
    os.environ.get("SIDECAR_LOG_DIR", pathlib.Path(__file__).parent / "logs"))
# Where the claude CLI keeps its session transcripts (one dir per cwd slug).
# /fork does its transcript surgery here — it's the host user's own tree.
# This is the *default* account's tree; other accounts keep theirs under their
# own config dir (see ACCOUNTS).
CLAUDE_PROJECTS = pathlib.Path(os.environ.get(
    "CLAUDE_PROJECTS_DIR", pathlib.Path.home() / ".claude" / "projects"))


def _parse_accounts(raw: str) -> dict[str, pathlib.Path | None]:
    """``SIDECAR_ACCOUNTS`` → {account id: config dir}.

    Format: comma-separated ``id=dir`` pairs (``work=~/.claude-work``). The
    default account (``SIDECAR_DEFAULT_ACCOUNT``, ``personal``) is always
    present and maps to None: it runs on the CLI's own default config — never
    on ``CLAUDE_CONFIG_DIR=~/.claude``, which would move ``.claude.json`` into
    the dir and make the logged-in account look brand new."""
    out: dict[str, pathlib.Path | None] = {}
    for part in (raw or "").split(","):
        if "=" not in part:
            continue
        acct, _, d = part.partition("=")
        acct, d = acct.strip().lower(), d.strip()
        if re.fullmatch(r"[a-z][a-z0-9_-]{0,31}", acct) and d:
            out[acct] = pathlib.Path(os.path.expanduser(d))
    return out


# Claude accounts a run can be launched on. An account *is* a config dir:
# CLAUDE_CONFIG_DIR moves the login, .claude.json, settings and the projects/
# transcripts together, so exporting it per run is the whole switch.
DEFAULT_ACCOUNT = os.environ.get("SIDECAR_DEFAULT_ACCOUNT", "personal")
_ACCOUNTS_RAW = os.environ.get("SIDECAR_ACCOUNTS", "")
ACCOUNTS: dict[str, pathlib.Path | None] = {
    DEFAULT_ACCOUNT: None, **_parse_accounts(_ACCOUNTS_RAW)}
ACCOUNTS[DEFAULT_ACCOUNT] = None
# With accounts declared, every run authenticates with a subscription login —
# so an ANTHROPIC_API_KEY that leaks into the sidecar's env would silently
# override BOTH accounts and bill the API key instead. Strip it from every run.
# (Unset — the standalone in-container runner — the key may be the only auth,
# so it passes through unchanged.)
STRIP_API_KEY = bool(_ACCOUNTS_RAW.strip())
# `claude auth status` per account for GET /accounts — cached, it spawns a CLI.
ACCOUNT_STATUS_TTL_S = float(os.environ.get("SIDECAR_ACCOUNT_STATUS_TTL_S", "60"))
# How long a freshly launched run is watched before we answer the caller. A
# `claude` that can't start at all (bad `--resume` id, bad cwd, broken install)
# exits non-zero within ~2.5s; anything still alive at the end of the window is
# a real run. Without this the launch is fire-and-forget and a dead-on-arrival
# run looks exactly like a healthy one — the viewer then waits forever for turns
# that never come ("sending…" with no error).
LAUNCH_PROBE_S = float(os.environ.get("SIDECAR_LAUNCH_PROBE_S", "3.5"))
# How long a force-resume waits for the SIGINT'd run to die before escalating
# to SIGKILL. `claude` handles SIGINT by writing the interrupt marker and
# exiting — usually well under a second. The total force-resume worst case
# (STOP_WAIT_S + 2s kill wait + LAUNCH_PROBE_S) must stay under the backend's
# 15s sidecar call timeout.
STOP_WAIT_S = float(os.environ.get("SIDECAR_STOP_WAIT_S", "6"))

app = FastAPI(title="claude-sidecar", docs_url=None, redoc_url=None)


# ---- pid tracking ---------------------------------------------------------
# Each running session is persisted to `logs/<sid>.pid` as a small JSON record
# so the sidecar can re-discover, list and interrupt runs that outlived it. The
# `/proc` start-time pins the record to *this* process, so a recycled PID (a new
# unrelated process that happens to reuse the number) is never mistaken for a
# live session — critical before we send it a signal.


# The runs we launched ourselves, kept only so we can *reap* them. A `claude`
# started with Popen stays our child even though it leads its own session, so
# when it exits it lingers as a zombie until the parent waits on it — and a
# zombie still answers `os.kill(pid, 0)` and still has a `/proc/<pid>/stat`. So
# a finished run kept looking "alive", `/resume` took the idempotency shortcut
# below, launched nothing, and the prompt was silently dropped. We poll (reap)
# them before every liveness check, and treat state `Z` as dead as a belt-and-
# braces guard (a survivor of a previous sidecar instance is reparented to init,
# which reaps it for us).
_procs: dict[str, subprocess.Popen] = {}


def _reap() -> None:
    """Wait on any of our children that have exited, so no zombie lingers."""
    for sid, proc in list(_procs.items()):
        if proc.poll() is not None:  # returncode set ⇒ child reaped
            _procs.pop(sid, None)


# Linux reads /proc directly (no dependency — the host sidecar's venv has only
# fastapi + uvicorn); anywhere else (the macOS worker) falls back to psutil,
# which the worker installer puts in its venv. Both answer the same two
# questions: is the process a zombie, and what is its start-time fingerprint.
_HAVE_PROC = pathlib.Path("/proc/self/stat").exists()


def _psutil_proc(pid: int):
    try:
        import psutil  # noqa: PLC0415 — optional, macOS worker only
        return psutil.Process(pid)
    except Exception:  # ImportError, NoSuchProcess, AccessDenied
        return None


def _proc_state(pid: int) -> str | None:
    """Process state letter ('Z' = zombie, 'S'/'R' = live), from /proc/pid/stat
    on Linux, psutil elsewhere."""
    if not _HAVE_PROC:
        p = _psutil_proc(pid)
        try:
            return None if p is None else (
                "Z" if p.status() == "zombie" else "R")
        except Exception:
            return None
    try:
        stat = pathlib.Path(f"/proc/{pid}/stat").read_text()
        return stat[stat.rindex(")") + 1:].split()[0]
    except (OSError, ValueError, IndexError):
        return None


def _proc_starttime(pid: int) -> int | float | None:
    """Process start-time fingerprint: clock ticks since boot (field 22 of
    /proc/pid/stat) on Linux, psutil's ``create_time()`` (epoch seconds,
    rounded) elsewhere.

    Unique per (pid, boot), so it fingerprints the exact process and lets us
    detect PID reuse. The `comm` field (field 2) can contain spaces and parens,
    so we parse everything after the final ')'."""
    if not _HAVE_PROC:
        p = _psutil_proc(pid)
        try:
            return None if p is None else round(p.create_time(), 2)
        except Exception:
            return None
    try:
        stat = pathlib.Path(f"/proc/{pid}/stat").read_text()
        after = stat[stat.rindex(")") + 1:].split()
        return int(after[19])  # field 22 overall (fields 3.. after comm)
    except (OSError, ValueError, IndexError):
        return None


def _pidfile(sid: str) -> pathlib.Path:
    return LOG_DIR / f"{sid}.pid"


def _read_record(sid: str) -> dict | None:
    """Load a session's pid record, tolerating the legacy plain-integer format."""
    try:
        raw = _pidfile(sid).read_text().strip()
    except OSError:
        return None
    if not raw:
        return None
    try:
        rec = json.loads(raw)
        if isinstance(rec, dict) and rec.get("pid"):
            return rec
    except ValueError:
        pass
    try:  # legacy pidfile: bare pid, no start-time guard
        return {"pid": int(raw)}
    except ValueError:
        return None


def _alive(rec: dict) -> bool:
    """True if the recorded process is still *running* — same process, not a zombie.

    Reaps our finished children first: an exited-but-unwaited child would
    otherwise still pass every check below and be mistaken for a live run."""
    pid = rec.get("pid")
    if not pid:
        return False
    _reap()
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True  # exists but owned by someone else — still 'alive'
    except OSError:
        return False
    if _proc_state(pid) == "Z":
        return False  # exited, just not reaped yet
    want = rec.get("starttime")
    if want is not None:
        now = _proc_starttime(pid)
        if now is not None and now != want:
            return False  # PID was recycled into a different process
    return True


def _live_record(sid: str) -> dict | None:
    """The session's pid record if its run is still going; else None (and the
    stale pidfile is dropped, so finished runs don't accumulate on disk)."""
    rec = _read_record(sid)
    if rec is None:
        return None
    if _alive(rec):
        return rec
    _pidfile(sid).unlink(missing_ok=True)
    return None


@app.on_event("startup")
def _reconcile_pidfiles() -> None:
    """On (re)start, GC pidfiles whose process is gone and count the survivors.

    The spawned runs are daemons, so after a sidecar restart the disk pidfiles
    are our only record of what's still running — reconcile them into a clean
    view and drop the dead ones so /interrupt never signals a stale PID."""
    if not LOG_DIR.is_dir():
        return
    survived = 0
    for p in sorted(LOG_DIR.glob("*.pid")):
        rec = _read_record(p.stem)
        if rec is None or not _alive(rec):
            p.unlink(missing_ok=True)
        else:
            survived += 1
    print(f"[sidecar] startup reconcile: {survived} session(s) survived restart",
          flush=True)


class SpawnBody(BaseModel):
    prompt: str
    sessionId: str | None = None      # caller may pre-pick the id (ai-agent does)
    model: str | None = None
    cwd: str | None = None
    harness: str | None = None        # "claude" (default) | "pi"
    thinking: bool | None = None      # False ⇒ run with thinking off; None = on
    effort: str | None = None         # claude only; None ⇒ the CLI's default
    account: str | None = None        # claude only; None ⇒ DEFAULT_ACCOUNT


class ResumeBody(BaseModel):
    sessionId: str                    # the existing session to continue
    prompt: str                       # the new user turn to send
    model: str | None = None
    cwd: str | None = None            # must match the session's original cwd
    harness: str | None = None        # "claude" (default) | "pi"
    thinking: bool | None = None      # False ⇒ run with thinking off; None = on
    effort: str | None = None         # claude only; None ⇒ what it last ran with
    # The account the session started on: its transcript only exists in that
    # account's projects/, so a resume on any other one can't find it.
    account: str | None = None
    # force=True: a run that's still live doesn't answer `reused` — it is
    # stopped first (SIGINT its group, wait for it to die, escalate to SIGKILL)
    # and the resume then launches with the new prompt. This is how "send a
    # message to a running conversation" restarts the session on the new turn.
    force: bool = False


class ForkBody(BaseModel):
    sessionId: str                    # fresh uuid for the fork (backend-minted)
    srcSessionId: str                 # the session to branch off
    prompt: str                       # the edited message — the fork's next turn
    cutLine: int                      # raw JSONL line index of the fork point
    cutText: str | None = None        # collapsed prefix of that message (canary)
    cutUserOrd: int | None = None     # its 0-based ordinal among user msgs (pi)
    model: str | None = None
    cwd: str | None = None            # must match the source session's cwd
    harness: str | None = None        # "claude" (default) | "pi"
    account: str | None = None        # the source session's account


def _auth(authorization: str | None) -> None:
    """Bearer check. The token is SIDECAR_TOKEN (the host sidecar, the
    in-container runner) or, on a paired worker, the one minted at pairing."""
    token = TOKEN or pairing.token()
    if not token:
        raise HTTPException(500, "sidecar not configured (no SIDECAR_TOKEN, "
                                 "not paired)")
    if not hmac.compare_digest(authorization or "", f"Bearer {token}"):
        raise HTTPException(401, "unauthorized")


def _valid_uuid(sid: str) -> str:
    sid = (sid or "").strip()
    try:
        uuidlib.UUID(sid)  # claude requires a valid UUID for --session-id/--resume
    except ValueError:
        raise HTTPException(400, "sessionId must be a valid UUID")
    return sid


def _valid_model(model: str | None, harness: str = "claude") -> str | None:
    """Sanity-check a `--model` value (an alias or a model id). None = unset."""
    model = (model or "").strip()
    if not model:
        return None
    rx = PI_MODEL_RE if harness == "pi" else MODEL_RE
    if not rx.match(model):
        raise HTTPException(400, f"invalid model: {model!r}")
    return model


def _valid_harness(harness: str | None) -> str:
    h = (harness or "claude").strip().lower()
    if h not in ("claude", "pi"):
        raise HTTPException(400, f"invalid harness: {harness!r}")
    return h


def _thinking_args(thinking: bool | None, harness: str) -> list[str]:
    """The `--thinking` flag for a run, in the harness's own vocabulary.

    Only *off* is expressible: thinking on (or unset) emits nothing, leaving each
    CLI on its own default level. Both harnesses take the same flag name, so this
    is a straight append onto either argv."""
    if thinking is False:
        return ["--thinking", THINKING_OFF[harness]]
    return []


def _effort_args(effort: str | None, harness: str) -> list[str]:
    """The `--effort` flag for a run. Claude-only, and validated as an
    allow-list (unlike `--model`, the CLI's scale is a closed set), so a bogus
    value is a 400 here rather than a run that dies on an unknown flag value.
    Unset — or any pi run — emits nothing and the CLI keeps its default."""
    effort = (effort or "").strip().lower()
    if not effort or harness != "claude":
        return []
    if effort not in EFFORT_LEVELS:
        raise HTTPException(400, f"invalid effort: {effort!r}")
    return ["--effort", effort]


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))


# The `notify` CLI (cli/notify.py — push / ask / form through the hub) ships
# with this repo so every runner has it: appended to each run's PATH (the lab
# host's own skill aliases come first and resolve to the same file).
CLI_DIR = pathlib.Path(__file__).resolve().parent.parent / "cli"
# A paired remote worker has no hub URL or credential, so its runs reach the
# hub through *this* sidecar's outbox (outbox.py): the hub pulls it. The host
# sidecar (SIDECAR_TOKEN set) is next to the hub — its runs talk to it directly.
IS_WORKER = not TOKEN
OUTBOX = outbox.Outbox(pairing.state_dir() / "outbox.json")


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 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"}
    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"


def _remote_control(account: str) -> list[str]:
    """`--remote-control` for default-account runs only: a work run would list
    itself in the employer org's claude.ai session list."""
    return ["--remote-control"] if REMOTE_CONTROL and account == DEFAULT_ACCOUNT \
        else []


def _run_cwd(cwd: str | None) -> str:
    """The run's working dir (``~`` expanded — the backend sends an account's
    default cwd host-relative), which must exist."""
    cwd = os.path.expanduser(cwd) if cwd else DEFAULT_CWD
    if not pathlib.Path(cwd).is_dir():
        raise HTTPException(400, f"cwd does not exist: {cwd}")
    return cwd


def _launch(args: list[str], sid: str, cwd: str, *,
            model: str | None = None, append: bool = False,
            account: str | None = None) -> dict:
    """Fire a detached ``claude -p`` run (a daemon) and persist its pid record.

    The child leads its own session/process group (``start_new_session``) so the
    run survives past this request *and* past the sidecar itself; the systemd
    unit's ``KillMode=process`` keeps it alive across a sidecar restart. We
    persist its identity (pid + ``/proc`` start-time) to ``logs/<sid>.pid`` so a
    restarted sidecar can re-discover, list and ``/interrupt`` it, and never
    signals a recycled PID. ``append`` keeps a resumed run's output alongside the
    original spawn log instead of clobbering it.

    The launch is *checked*, not fire-and-forget: we watch the child for
    ``LAUNCH_PROBE_S`` and raise 502 with the tail of its log if it exits
    non-zero in that window (``No conversation found with session ID: …`` when a
    ``--resume`` can't find its transcript, a broken CLI, …). The caller then
    sees a real error instead of a run that never produces a turn."""
    LOG_DIR.mkdir(parents=True, exist_ok=True)
    log_path = LOG_DIR / f"{sid}.log"
    err_from = log_path.stat().st_size if append and log_path.exists() else 0
    with open(log_path, "ab" if append else "wb") as log:
        proc = subprocess.Popen(
            args, cwd=cwd, stdin=subprocess.DEVNULL, stdout=log, stderr=log,
            start_new_session=True,
            env=_run_env(sid, account),
        )
    _procs[sid] = proc  # kept so we reap it when it exits (see _reap)
    # The child leads its own process group (start_new_session), so pid == pgid.
    try:
        _pidfile(sid).write_text(json.dumps({
            "pid": proc.pid,
            "startedAt": int(time.time()),
            "starttime": _proc_starttime(proc.pid),
            "model": model or DEFAULT_MODEL,
            "cwd": cwd,
            "account": account,
        }))
    except OSError:
        pass

    rc = _probe_launch(proc)
    if rc is not None and rc != 0:
        _procs.pop(sid, None)
        _pidfile(sid).unlink(missing_ok=True)
        msg, kind = claude_cli.run_failure(_log_text(log_path, err_from))
        raise HTTPException(502, claude_cli.error_detail(
            msg, kind=kind, account=account, exitCode=rc))
    return {"sessionId": sid, "pid": proc.pid, "log": str(log_path)}


def _probe_launch(proc: subprocess.Popen) -> int | None:
    """Poll a just-started run for LAUNCH_PROBE_S. Its exit code if it died in
    that window (0 = a legitimately quick, successful run), else None."""
    deadline = time.monotonic() + LAUNCH_PROBE_S
    while time.monotonic() < deadline:
        rc = proc.poll()
        if rc is not None:
            return rc
        time.sleep(0.1)
    return None


def _stop_run(sid: str, rec: dict) -> None:
    """Stop a live run so a forced resume can take over its session.

    SIGINT the process group (like the /interrupt endpoint — claude writes its
    ``[Request interrupted by user]`` marker and exits), wait up to STOP_WAIT_S
    for it to actually die, and escalate to SIGKILL if it won't. Launching the
    ``--resume`` while the old run still lives would append duplicate turns to
    the same transcript, so a run that survives even SIGKILL is a hard error."""
    pid = rec["pid"]
    try:
        os.killpg(pid, signal.SIGINT)
    except ProcessLookupError:
        pass  # died between the liveness check and the signal — that's fine
    except OSError as e:
        raise HTTPException(500, f"could not signal process: {e}")
    deadline = time.monotonic() + STOP_WAIT_S
    while time.monotonic() < deadline and _alive(rec):
        time.sleep(0.15)
    if _alive(rec):
        try:
            os.killpg(pid, signal.SIGKILL)
        except OSError:
            pass
        deadline = time.monotonic() + 2.0
        while time.monotonic() < deadline and _alive(rec):
            time.sleep(0.15)
        if _alive(rec):
            raise HTTPException(409, "could not stop the running session")
    _pidfile(sid).unlink(missing_ok=True)


def _log_text(log_path: pathlib.Path, since: int = 0) -> str:
    """What this run wrote to its log (from ``since``) — parsed by
    ``claude_cli.run_failure`` into the message to show the user."""
    try:
        with open(log_path, "rb") as f:
            f.seek(since)
            return f.read().decode(errors="replace")
    except OSError:
        return ""


@app.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())}


_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


@app.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


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))


@app.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)


@app.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)
    _account_status_cache.pop(account, None)
    return _auth_status(account)


@app.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))}


@app.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)
    _account_status_cache.pop(account, None)
    return {**_auth_status(account), "message": msg}


@app.get("/sessions")
def sessions(authorization: str | None = Header(default=None)) -> dict:
    """List sessions still running (survivors of any sidecar restart included)."""
    _auth(authorization)
    out = []
    if LOG_DIR.is_dir():
        for p in sorted(LOG_DIR.glob("*.pid")):
            sid = p.stem
            rec = _live_record(sid)
            if rec:
                out.append({
                    "sessionId": sid,
                    "pid": rec.get("pid"),
                    "startedAt": rec.get("startedAt"),
                    "model": rec.get("model"),
                    "cwd": rec.get("cwd"),
                    "account": rec.get("account"),
                })
    return {"sessions": out, "count": len(out)}


@app.post("/spawn")
def spawn(body: SpawnBody, authorization: str | None = Header(default=None)) -> dict:
    _auth(authorization)
    prompt = (body.prompt or "").strip()
    if not prompt:
        raise HTTPException(400, "empty prompt")

    sid = _valid_uuid(body.sessionId or str(uuidlib.uuid4()))
    cwd = _run_cwd(body.cwd)

    harness = _valid_harness(body.harness)
    if harness == "pi":
        # pi harness: same flag shape, different binary. The runner wraps
        # `pi --mode json` and mirrors the session into the watched pi
        # transcripts dir (permission scope = the run's cwd by default).
        model = _valid_model(body.model, "pi") or PI_DEFAULT_MODEL
        args = [RUNNER_BIN, "-p", prompt, "--session-id", sid, "--model", model]
        args += _thinking_args(body.thinking, harness)
        return _launch(args, sid, cwd, model=model)

    # The viewer's composer sends the model tag the user picked (a CLI alias like
    # `opus`, or a pinned id like `claude-opus-4-5-20251101`); with none picked we
    # fall back to DEFAULT_MODEL. Either way `claude --model` resolves it.
    model = _valid_model(body.model) or DEFAULT_MODEL
    account = _valid_account(body.account)
    _require_login(account)
    args = [
        CLAUDE_BIN, "-p", prompt,
        "--session-id", sid,
        "--output-format", "json",
        "--permission-mode", PERMISSION_MODE,
        "--model", model,
    ]
    if DISALLOWED_TOOLS:
        args += ["--disallowed-tools", " ".join(DISALLOWED_TOOLS)]
    args += _thinking_args(body.thinking, harness)
    args += _effort_args(body.effort, harness)
    args += _remote_control(account)
    return _launch(args, sid, cwd, model=model, account=account)


@app.post("/resume")
def resume(body: ResumeBody, authorization: str | None = Header(default=None)) -> dict:
    """Continue an existing conversation with a new user turn.

    Runs ``claude -p <prompt> --resume <sessionId>`` (default: reuses the
    original session id, so it appends to the same
    ``~/.claude/projects/<proj>/<sid>.jsonl`` the viewer already watches — the
    new turns stream straight into the open conversation). ``--resume`` is scoped
    to the current directory, so the caller must pass the session's original
    ``cwd``.

    ``model`` switches the model the continuation runs on (the composer sends the
    conversation's current model by default, so nothing changes unless the user
    picks another tag). Omitted ⇒ no ``--model`` flag ⇒ the session keeps its own."""
    _auth(authorization)
    prompt = (body.prompt or "").strip()
    if not prompt:
        raise HTTPException(400, "empty prompt")

    sid = _valid_uuid(body.sessionId)
    harness = _valid_harness(body.harness)
    model = _valid_model(body.model, harness)
    cwd = _run_cwd(body.cwd)
    account = _valid_account(body.account) if harness == "claude" else None
    if account:
        _require_login(account)

    # Idempotency: if a run for this session is *really* still going, a rapid
    # double-submit would launch a SECOND `claude -p --resume` appending duplicate
    # turns to the same transcript (and clobbering the pidfile so only one stays
    # interruptible). We take no new prompt in that case — and say so
    # (`reused`), because the turn was NOT accepted: the caller has to surface
    # that rather than show a message that will never be answered.
    # With force=True the caller wants the new prompt to win instead: stop the
    # live run (it lands its interrupt marker in the transcript), then resume.
    rec = _live_record(sid)
    if rec:
        if not body.force:
            return {"sessionId": sid, "pid": rec.get("pid"),
                    "log": str(LOG_DIR / f"{sid}.log"), "reused": True}
        _stop_run(sid, rec)

    if harness == "pi":
        # The composer sends the conversation's current model by default; with
        # nothing picked the runner's own default keeps the session coherent.
        model = model or PI_DEFAULT_MODEL
        args = [RUNNER_BIN, "-p", prompt, "--resume", sid, "--model", model]
        args += _thinking_args(body.thinking, harness)
        return _launch(args, sid, cwd, model=model, append=True)

    args = [
        CLAUDE_BIN, "-p", prompt,
        "--resume", sid,
        "--output-format", "json",
        "--permission-mode", PERMISSION_MODE,
    ]
    if DISALLOWED_TOOLS:
        args += ["--disallowed-tools", " ".join(DISALLOWED_TOOLS)]
    if model:
        args += ["--model", model]
    args += _thinking_args(body.thinking, harness)
    args += _effort_args(body.effort, harness)
    args += _remote_control(account)
    return _launch(args, sid, cwd, model=model, append=True, account=account)


def _collapse(s: str) -> str:
    return " ".join((s or "").split())


def _record_texts(line: str) -> list[str]:
    """Every text payload in one transcript record (message content as a plain
    string, or the text blocks of a content list) — what the fork-point canary
    compares against."""
    try:
        o = json.loads(line)
    except ValueError:
        return []
    c = (o.get("message") or {}).get("content")
    if isinstance(c, str):
        return [c]
    if isinstance(c, list):
        return [b.get("text", "") for b in c
                if isinstance(b, dict) and b.get("type") == "text"]
    return []


def _fork_transcript(src: pathlib.Path, dst: pathlib.Path, new_sid: str,
                     cut_line: int, cut_text: str | None) -> None:
    """Write the first ``cut_line`` lines of ``src`` as a new session file.

    This is the same shape ``claude --resume --fork-session`` produces — the
    identical records under a new ``sessionId`` (record uuids stay; they're
    only unique within a file's parent chain) — just cut short at the fork
    point instead of cloning the whole session. ``cut_text`` is the message
    the viewer showed at that ordinal: if the record at ``cut_line`` doesn't
    contain it, the archive the backend counted lines in has diverged from
    the live file (it never should — the import is append-only) and we refuse
    rather than fork at the wrong message."""
    try:
        lines = src.read_text(encoding="utf-8", errors="ignore") \
                   .splitlines(keepends=True)
    except OSError as e:
        raise HTTPException(502, f"cannot read source transcript: {e}")
    if not 0 < cut_line < len(lines):
        raise HTTPException(409, f"fork point out of range: line {cut_line} "
                                 f"of {len(lines)}")
    if cut_text:
        want = _collapse(cut_text)
        if not any(_collapse(t)[:len(want)] == want
                   for t in _record_texts(lines[cut_line])):
            raise HTTPException(409, "fork point mismatch — the transcript "
                                     "changed since the viewer loaded it; "
                                     "reload the conversation and retry")
    keep = lines[:cut_line]
    # The records just before a user message are that turn's bookkeeping
    # (queue-operation enqueue/dequeue carrying the message text, last-prompt).
    # They belong to the message being edited away, so a cut at the user
    # record would otherwise leak its text into the fork — trim them.
    def _rec_type(ln: str) -> str | None:
        try:
            return json.loads(ln).get("type")
        except ValueError:
            return None
    while keep and _rec_type(keep[-1]) in ("queue-operation", "last-prompt"):
        keep.pop()
    if not keep:
        raise HTTPException(400, "nothing before the fork point")
    out = []
    for ln in keep:
        s = ln.strip()
        if s.startswith("{"):
            try:
                o = json.loads(s)
                if "sessionId" in o:
                    o["sessionId"] = new_sid
                out.append(json.dumps(o, ensure_ascii=False) + "\n")
                continue
            except ValueError:
                pass
        out.append(ln if ln.endswith("\n") else ln + "\n")
    try:
        dst.write_text("".join(out), encoding="utf-8")
    except OSError as e:
        raise HTTPException(502, f"cannot write forked transcript: {e}")


@app.post("/fork")
def fork(body: ForkBody, authorization: str | None = Header(default=None)) -> dict:
    """Branch a new session off an existing one at a message boundary.

    Claude Code has no message-level fork (``--fork-session`` clones whole
    sessions), so: copy the source session file truncated at the fork point
    under a fresh session id (see ``_fork_transcript``), then ``claude -p
    <prompt> --resume <newSid>`` — claude loads the truncated history and
    appends the edited turn. The pi harness forks natively: pi sessions are
    trees, so the runner branches pi's own session file (``--fork-from``) and
    mirrors the truncated prefix itself."""
    _auth(authorization)
    prompt = (body.prompt or "").strip()
    if not prompt:
        raise HTTPException(400, "empty prompt")
    new_sid = _valid_uuid(body.sessionId)
    src_sid = _valid_uuid(body.srcSessionId)
    harness = _valid_harness(body.harness)
    model = _valid_model(body.model, harness)
    cwd = _run_cwd(body.cwd)

    if harness == "pi":
        model = model or PI_DEFAULT_MODEL
        args = [RUNNER_BIN, "-p", prompt, "--session-id", new_sid,
                "--fork-from", src_sid, "--cut-line", str(body.cutLine),
                "--cut-user-ord", str(body.cutUserOrd or 0), "--model", model]
        if body.cutText:
            args += ["--cut-text", body.cutText]
        return _launch(args, new_sid, cwd, model=model)

    # A fork stays on the source's account: the source transcript lives only
    # in that account's projects/, and so must the fork for --resume to find it.
    account = _valid_account(body.account)
    _require_login(account)
    projects = _account_projects(account)
    src = next(projects.glob(f"*/{src_sid}.jsonl"), None)
    if src is None:
        raise HTTPException(404, f"source session {src_sid} not found under "
                                 f"{projects}")
    dst = src.parent / f"{new_sid}.jsonl"
    _fork_transcript(src, dst, new_sid, body.cutLine, body.cutText)
    args = [
        CLAUDE_BIN, "-p", prompt,
        "--resume", new_sid,
        "--output-format", "json",
        "--permission-mode", PERMISSION_MODE,
    ]
    if DISALLOWED_TOOLS:
        args += ["--disallowed-tools", " ".join(DISALLOWED_TOOLS)]
    if model:
        args += ["--model", model]
    args += _remote_control(account)
    try:
        return _launch(args, new_sid, cwd, model=model, account=account)
    except HTTPException:
        dst.unlink(missing_ok=True)  # no headless truncated session left behind
        raise


class InterruptBody(BaseModel):
    sessionId: str


@app.post("/interrupt")
def interrupt(body: InterruptBody,
             authorization: str | None = Header(default=None)) -> dict:
    """Stop a running session by sending its process group a SIGINT.

    SIGINT is what Ctrl+C delivers to a terminal's foreground process group, so
    a headless ``claude -p`` run handles it the same way: it aborts the current
    turn, writes a ``[Request interrupted by user]`` marker to its transcript,
    and exits. We signal the whole process group (pid == pgid, since the run was
    started with ``start_new_session``) so in-flight tool subprocesses stop too.

    Reads the run's identity from its on-disk pidfile, so this works even for a
    session spawned by a *previous* sidecar instance that has since restarted."""
    _auth(authorization)
    sid = _valid_uuid(body.sessionId)

    # Verify it's still our live process before signalling — never fire a signal
    # at a PID that has already exited (possibly still a zombie), or been
    # recycled into something else. _live_record drops the stale pidfile for us.
    rec = _live_record(sid)
    if rec is None:
        raise HTTPException(404, "no running process for this session")
    pid = rec["pid"]
    pid_path = _pidfile(sid)

    try:
        os.killpg(pid, signal.SIGINT)
    except ProcessLookupError:
        # Raced us to exit between the liveness check and the signal.
        pid_path.unlink(missing_ok=True)
        raise HTTPException(404, "process already exited")
    except OSError as e:
        raise HTTPException(500, f"could not signal process: {e}")

    pid_path.unlink(missing_ok=True)
    return {"sessionId": sid, "pid": pid, "signal": "SIGINT", "ok": True}


# ---- worker endpoints --------------------------------------------------------
# Mounted on every runner (host sidecar, in-container runner, remote worker) —
# they're only *used* when a hub pairs with this runner as a remote worker:
# /feed is how the hub pulls transcripts it has no mount for, /pair how it gets
# the bearer in the first place. See feed.py / pairing.py.
app.include_router(feed.make_router(
    _auth, lambda: _account_projects(DEFAULT_ACCOUNT)))
app.include_router(pairing.make_router(
    _auth, lambda: _auth_status(DEFAULT_ACCOUNT),
    lambda: {"cwd": DEFAULT_CWD, "defaultAccount": DEFAULT_ACCOUNT,
             "permissionMode": PERMISSION_MODE,
             "projectsDir": str(_account_projects(DEFAULT_ACCOUNT))}))
# /outbox: a worker session's notifications / asks / forms, queued for the hub
# to collect (outbox.py). Runs authenticate with the per-worker run token.
app.include_router(outbox.make_router(
    OUTBOX, _auth, lambda: outbox.run_token(pairing.state_dir())))
# /skills: the skills a run here can use (the default account's user-level
# skills + the default cwd's project skills) — what the hub's composer lists
# when a prompt is sent to this worker (skills_list.py).
app.include_router(skills_list.make_router(
    _auth, lambda: skills_list.skill_roots(ACCOUNTS.get(DEFAULT_ACCOUNT),
                                           pathlib.Path(DEFAULT_CWD))))


if __name__ == "__main__":
    # `python sidecar.py` — how the macOS worker's launchd agent starts it:
    # SIDECAR_BIND is the Mac's Tailscale address, so the listener is reachable
    # over the tailnet only (nothing on the office LAN, nothing on loopback).
    import uvicorn

    bind = os.environ.get("SIDECAR_BIND", "127.0.0.1")
    port = int(os.environ.get("SIDECAR_PORT", "8790"))
    if not (TOKEN or pairing.token()) and not pairing._read("pairing-code"):
        print(f"[worker] not paired — pairing code: {pairing.new_code()}",
              flush=True)
    uvicorn.run(app, host=bind, port=port)