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

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


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;
# bare LM Studio ids route to the local EVOX2 provider — see runner/cli.mjs).
PI_DEFAULT_MODEL = os.environ.get("SIDECAR_PI_MODEL", "qwen/qwen3.6-35b-a3b")
# 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.
CLAUDE_PROJECTS = pathlib.Path(os.environ.get(
    "CLAUDE_PROJECTS_DIR", pathlib.Path.home() / ".claude" / "projects"))
# 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)


def _proc_state(pid: int) -> str | None:
    """Process state letter from /proc/pid/stat ('Z' = zombie, 'S'/'R' = live)."""
    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 | None:
    """Process start-time in clock ticks since boot (field 22 of /proc/pid/stat).

    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 ')'."""
    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


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


def _auth(authorization: str | None) -> None:
    if not TOKEN:
        raise HTTPException(500, "sidecar not configured (no SIDECAR_TOKEN)")
    expected = f"Bearer {TOKEN}"
    if authorization != expected:
        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 _launch(args: list[str], sid: str, cwd: str, *,
            model: str | None = None, append: bool = False) -> 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={**os.environ, "CLAUDE_CODE_ENTRYPOINT": "ai-agent-sidecar"},
        )
    _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,
        }))
    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)
        raise HTTPException(502, f"claude exited immediately (code {rc}): "
                                 f"{_log_tail(log_path, err_from)}")
    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_tail(log_path: pathlib.Path, since: int = 0, limit: int = 400) -> str:
    """The last of what this run wrote to its log — the message to show the user."""
    try:
        with open(log_path, "rb") as f:
            f.seek(since)
            text = f.read().decode(errors="replace")
    except OSError:
        return "(no output)"
    lines = [ln for ln in text.strip().splitlines() if ln.strip()]
    return (lines[-1][:limit] if lines else "(no output)")


@app.get("/health")
def health() -> dict:
    return {"ok": True, "cwd": DEFAULT_CWD, "claude": CLAUDE_BIN,
            "defaultModel": DEFAULT_MODEL}


@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"),
                })
    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 = body.cwd or DEFAULT_CWD
    if not pathlib.Path(cwd).is_dir():
        raise HTTPException(400, f"cwd does not exist: {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
    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)
    if REMOTE_CONTROL:
        args.append("--remote-control")
    return _launch(args, sid, cwd, model=model)


@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 = body.cwd or DEFAULT_CWD
    if not pathlib.Path(cwd).is_dir():
        raise HTTPException(400, f"cwd does not exist: {cwd}")

    # 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)
    if REMOTE_CONTROL:
        args.append("--remote-control")
    return _launch(args, sid, cwd, model=model, append=True)


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 = body.cwd or DEFAULT_CWD
    if not pathlib.Path(cwd).is_dir():
        raise HTTPException(400, f"cwd does not exist: {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)

    src = next(CLAUDE_PROJECTS.glob(f"*/{src_sid}.jsonl"), None)
    if src is None:
        raise HTTPException(404, f"source session {src_sid} not found under "
                                 f"{CLAUDE_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]
    if REMOTE_CONTROL:
        args.append("--remote-control")
    try:
        return _launch(args, new_sid, cwd, model=model)
    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}