ai-agent/sidecar/launch.py

"""
Launching and stopping a run: the detached ``claude -p`` / runner process, its
pid record and log, the launch probe, and the claude-only argv extras
(``--settings``, ``--remote-control``).
"""

import json
import os
import pathlib
import shlex
import signal
import subprocess
import sys
import time

from fastapi import HTTPException

import claude_cli
import pids
import shared_skills
from claude_accounts import _run_env
from config import (CLAUDE_BIN, DEFAULT_ACCOUNT, DEFAULT_CWD, DEFAULT_MODEL,
                    DISALLOWED_TOOLS, LAUNCH_PROBE_S, LOG_DIR, PERMISSION_MODE,
                    REMOTE_CONTROL, STOP_GUARD, STOP_GUARD_ENABLED,
                    STOP_GUARD_TIMEOUT_S, STOP_WAIT_S)

def _claude_settings() -> dict:
    """The settings every claude run is launched with (``--settings`` merges
    them over the user's own): the inbox opt-in that lets `/message` reach a
    live run, and the Stop guard hook (see the constants above)."""
    s: dict = {"crossSessionInbound": "accept"}
    if STOP_GUARD_ENABLED and STOP_GUARD.is_file():
        cmd = f"{shlex.quote(sys.executable)} {shlex.quote(str(STOP_GUARD))}"
        s["hooks"] = {"Stop": [{"hooks": [{
            "type": "command", "command": cmd,
            "timeout": STOP_GUARD_TIMEOUT_S}]}]}
    return s


def _claude_settings_args(scope: dict | None = None) -> list[str]:
    s = _claude_settings()
    if scope:
        s.update(_scope_settings(scope))
    return ["--settings", json.dumps(s, separators=(",", ":"))]


# ── per-run scope ────────────────────────────────────────────────────────────
# A scoped run (the hub's guests — backend users/runs.py) can't run on
# bypassPermissions: under it, allow rules do nothing. It runs on `dontAsk`
# instead — anything a rule doesn't allow is refused without a prompt — with
# edits allowed only under ``writeDirs``, these tools allowed outright, and the
# ``denyRead`` reads refused (Read rules also cover cat/head/sed… in Bash).
# Bash itself is unconfined unless ``sandbox`` (Claude Code's bubblewrap
# sandbox, which needs bwrap + socat on the host) is on.
SCOPE_TOOLS = ("Bash", "WebFetch", "WebSearch", "Agent", "Task", "Skill",
               "Monitor", "ScheduleWakeup", "TaskStop", "TaskOutput")
_SCOPE_KEYS = {"writeDirs", "denyRead", "systemPrompt", "remoteControl", "sandbox",
               "skills"}


def _abs_rule(path: str) -> str:
    """A path as a permission-rule pattern: ``~`` expanded (the run's home is
    this process's), then the rule syntax's ``//`` prefix for an absolute
    path. A trailing ``/**`` glob survives ``abspath``."""
    p = os.path.expanduser(str(path).strip())
    glob = p.endswith("/**")
    p = os.path.abspath(p[:-3] if glob else p) + ("/**" if glob else "")
    return "/" + p


def valid_scope(scope: dict | None) -> dict | None:
    if scope is None:
        return None
    if not isinstance(scope, dict) or set(scope) - _SCOPE_KEYS:
        raise HTTPException(400, f"bad scope keys: {sorted(set(scope or {}) - _SCOPE_KEYS)}")
    for k in ("writeDirs", "denyRead", "skills"):
        v = scope.get(k) or []
        if not isinstance(v, list) or not all(isinstance(x, str) and x.strip() for x in v):
            raise HTTPException(400, f"scope.{k} must be a list of strings")
    return scope


def _scope_settings(scope: dict) -> dict:
    writes = [_abs_rule(d).rstrip("/") for d in scope.get("writeDirs") or []]
    out: dict = {"permissions": {
        "allow": [f"Edit({w}/**)" for w in writes] + list(SCOPE_TOOLS),
        "deny": [f"Read({_abs_rule(p)})" for p in scope.get("denyRead") or []],
    }}
    if scope.get("sandbox"):
        out["sandbox"] = {
            "enabled": True, "failIfUnavailable": True,
            "allowUnsandboxedCommands": False, "autoAllowBashIfSandboxed": True,
            "filesystem": {"allowWrite": [os.path.expanduser(d)
                                          for d in scope.get("writeDirs") or []]},
        }
    return out


def scope_args(scope: dict | None) -> list[str]:
    """The flags a scope adds on top of :func:`_claude_args`'s."""
    if not scope:
        return []
    args = ["--permission-prompts", "none"]
    if scope.get("systemPrompt"):
        args += ["--append-system-prompt", str(scope["systemPrompt"])[:8000]]
    # The owner's shared skills, as a generated `shared` plugin.
    plugin = shared_skills.plugin_dir(scope.get("skills"))
    if plugin is not None:
        args += ["--plugin-dir", str(plugin)]
    return args


def _claude_args(prompt: str, flag: str, sid: str,
                 scope: dict | None = None) -> list[str]:
    """What every claude run shares — the prompt, how it is keyed to its
    session (``--session-id`` for a spawn, ``--resume`` for a continue or a
    fork), the headless output + permission flags, the disallowed tools and
    the ``--settings`` payload (stop guard, inbox opt-in, and a scoped run's
    permission rules). One builder so no launch path can miss one of them."""
    args = [CLAUDE_BIN, "-p", prompt, flag, sid,
            "--output-format", "json",
            "--permission-mode", "dontAsk" if scope else PERMISSION_MODE]
    if DISALLOWED_TOOLS:
        args += ["--disallowed-tools", " ".join(DISALLOWED_TOOLS)]
    return args + _claude_settings_args(scope) + scope_args(scope)


def _remote_control(account: str, scope: dict | None = None) -> list[str]:
    """`--remote-control` for default-account runs only: a work run would list
    itself in the employer org's claude.ai session list. A scoped run that
    says ``remoteControl: false`` (a guest's) stays out of it too."""
    if scope and scope.get("remoteControl") is False:
        return []
    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),
        )
    pids._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:
        pids._pidfile(sid).write_text(json.dumps({
            "pid": proc.pid,
            "startedAt": int(time.time()),
            "starttime": pids._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:
        pids._procs.pop(sid, None)
        pids._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 pids._alive(rec):
        time.sleep(0.15)
    if pids._alive(rec):
        try:
            os.killpg(pid, signal.SIGKILL)
        except OSError:
            pass
        deadline = time.monotonic() + 2.0
        while time.monotonic() < deadline and pids._alive(rec):
            time.sleep(0.15)
        if pids._alive(rec):
            raise HTTPException(409, "could not stop the running session")
    pids._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 ""