Session Monitor

#!/usr/bin/env python3
"""Session Monitor keeper — runs INSIDE each session container.

The keeper-fed model: instead of the monitor pulling from claude.ai (cookie +
Cloudflare + a laptop pup tab), each container's keeper reads its OWN Claude Code
sessions from local files and REPORTS them up to the shared monitor service —
tagged with the owner identity. No claude.ai, no cookie, no Cloudflare, and it
works while the laptop is off (the container is up).

This file is M1: the read+report loop. M2 adds supervising `claude remote-control`
+ revive; the monitor service stores reports per-owner (multi-tenant).
"""
import os, json, glob, time, urllib.request, socket, shutil, subprocess, threading, re, base64, hashlib

HOME = os.path.expanduser("~")
SESSIONS_DIR = os.path.join(HOME, ".claude", "sessions")
PROJECTS_DIR = os.path.join(HOME, ".claude", "projects")
DEFAULT_NEW_CWD = os.environ.get("KEEPER_NEW_CWD", os.path.join(HOME, "project"))  # start new sessions here
CONFIG = os.path.join(HOME, ".claude.json")

# Where to report. The shared monitor service URL (defaults to the local server
# for the single-container demo; set SM_SERVICE_URL to the service container).
SERVICE = os.environ.get("SM_SERVICE_URL", "http://localhost:8846").rstrip("/")
SM_TOKEN = os.environ.get("SM_TOKEN", "")          # shared secret for the public service
REPORT_EVERY = int(os.environ.get("KEEPER_INTERVAL", "60"))
FULL_MAX = int(os.environ.get("KEEPER_FULL_MAX", "1500"))   # cap the on-demand full transcript (bounds payload/RAM)

# The running keeper's own code version. Bump on every keeper.py change that a
# container should pick up. Written to ~/.session-keeper/keeper.version on startup
# and sent in each report, so BOTH the CLI (`keeper status`) and the board can tell
# when a container is running stale keeper code — no more "healthy" while ancient.
# Keep in lockstep with the package version.
KEEPER_VERSION = "0.9.0"
VERSION_FILE = os.path.join(HOME, ".session-keeper", "keeper.version")


def _headers(extra=None):
    # Cloudflare's WAF 403s the default Python-urllib UA — set a real one.
    h = {"content-type": "application/json", "User-Agent": "session-keeper/1.0 (+adom)"}
    if SM_TOKEN:
        h["X-SM-Token"] = SM_TOKEN
    if extra:
        h.update(extra)
    return h
RC_MARKERS = ("remote-control", "remote_control", "environment_secret",
              "claudeusercontent", "remote-control-sdk")


# --- image attachments ---------------------------------------------------------
# The service stores images (phone upload / paste, and agent-produced images) by an
# unguessable id. For a delivered command we GET each id into a local inbox and hand
# the agent the ABSOLUTE PATHS in the prompt (its Read tool has native vision). We
# surface images back to the board too: the injected prompt carries a marker whose
# paths embed the id, so the transcript's user turn is self-describing on re-scan
# (survives a keeper restart with no extra state); agent images are found by path in
# assistant text, uploaded, and attached to that message.
INBOX = os.path.join(HOME, ".session-keeper", "inbox")
IMG_EXT = {"image/png": "png", "image/jpeg": "jpg", "image/webp": "webp", "image/gif": "gif"}
MEDIA_BY_EXT = {"png": "image/png", "jpg": "image/jpeg", "jpeg": "image/jpeg",
                "webp": "image/webp", "gif": "image/gif"}
IMG_MAX_BYTES = 12 * 1024 * 1024
INBOX_MAX = 200                                          # cap inbox files (prune oldest)
_IMG_MARKER = re.compile(r"\[The user attached (?:image|file)\(s\)\. View them with the Read tool: ([^\]]+)\]")
_IMG_PATH_RE = re.compile(r"(/[^\s'\"`)]+?\.(?:png|jpe?g|webp|gif))", re.IGNORECASE)


def _safe_iid(iid):
    return bool(iid) and bool(re.match(r"^[A-Za-z0-9_-]{16,64}$", iid))


def _prune_inbox():
    try:
        files = [os.path.join(INBOX, f) for f in os.listdir(INBOX)]
        if len(files) <= INBOX_MAX:
            return
        files.sort(key=lambda p: os.path.getmtime(p))
        for p in files[:len(files) - INBOX_MAX]:
            try:
                os.remove(p)
            except Exception:
                pass
    except Exception:
        pass


def fetch_command_images(image_ids):
    """Download a command's image ids into the local inbox; return (paths, ok) where
    ok is False if ANY requested id failed to download (so the caller can retry via
    reclaim rather than silently injecting text-only). Filenames = sanitized id only."""
    paths, ok = [], True
    if not image_ids:
        return paths, ok
    os.makedirs(INBOX, exist_ok=True)
    for iid in list(image_ids)[:8]:
        if not _safe_iid(iid):
            ok = False
            continue
        try:
            req = urllib.request.Request(SERVICE + "/api/img/" + iid, headers=_headers())
            with urllib.request.urlopen(req, timeout=30) as r:
                data = r.read()
                ctype = (r.headers.get("Content-Type", "") or "").split(";")[0].strip()
                disp = r.headers.get("Content-Disposition", "") or ""
            m_fn = re.search(r'filename="([^"\r\n]+)"', disp)
            if m_fn:                                     # general file: keep the real name (agent sees it)
                safe = "".join(c for c in os.path.basename(m_fn.group(1)) if c.isalnum() or c in "._-")[:80]
                path = os.path.join(INBOX, iid + "__" + (safe or "file.bin"))
            else:                                        # image: id + ext from the content type
                ext = IMG_EXT.get(ctype, "png")
                path = os.path.join(INBOX, iid + "." + ext)
            with open(path, "wb") as f:
                f.write(data)
            paths.append(path)
        except Exception:
            ok = False
    _prune_inbox()
    return paths, ok


IMG_EXTS = {"png", "jpg", "jpeg", "webp", "gif"}


def _text_with_images(text, paths):
    if not paths:
        return text or ""
    kind = "image" if all(p.rsplit(".", 1)[-1].lower() in IMG_EXTS for p in paths) else "file"
    marker = ("[The user attached %s(s). View them with the Read tool: " % kind
              + ", ".join(paths) + "]")
    return ((text or "").strip() + "\n\n" + marker).strip()


def _extract_user_images(text):
    """Parse the injected attachment marker back into image urls (the inbox path's
    basename IS the image id). Hardened against spoofing: every path must live in OUR
    inbox and carry a valid-id basename — a human typing the marker phrase can't
    produce inbox paths, so their text is left untouched. Returns (urls, clean_text)."""
    m = _IMG_MARKER.search(text or "")
    if not m:
        return [], [], text
    urls, files = [], []
    for p in m.group(1).split(","):
        p = p.strip()
        if os.path.dirname(p) != INBOX:                  # not a path WE wrote → treat as ordinary text
            return [], [], text
        base = os.path.basename(p)
        if "__" in base:                                 # general file: <id>__<name>
            iid, fname = base.split("__", 1)
            if not _safe_iid(iid):
                return [], [], text
            files.append({"url": "api/img/" + iid, "name": fname})
        else:                                            # image: <id>.<ext>
            iid = base.rsplit(".", 1)[0]
            if not _safe_iid(iid):
                return [], [], text
            urls.append("api/img/" + iid)
    if not urls and not files:
        return [], [], text
    clean = _IMG_MARKER.sub("", text).strip()
    return urls, files, clean


# agent-produced images: upload a local file ONCE (cached by path+mtime+size) -> url
_UPLOAD_CACHE = os.path.join(HOME, ".session-keeper", "uploaded-imgs.json")
_uploaded = None


def _uploaded_map():
    global _uploaded
    if _uploaded is None:
        try:
            _uploaded = json.load(open(_UPLOAD_CACHE))
        except Exception:
            _uploaded = {}
    return _uploaded


def _save_uploaded():
    try:
        os.makedirs(os.path.dirname(_UPLOAD_CACHE), exist_ok=True)
        tmp = _UPLOAD_CACHE + ".tmp"
        with open(tmp, "w") as f:
            json.dump(dict(list(_uploaded_map().items())[-1000:]), f)
        os.replace(tmp, _UPLOAD_CACHE)
    except Exception:
        pass


def _upload_local_image(path):
    try:
        st = os.stat(path)
    except Exception:
        return None
    if st.st_size <= 0 or st.st_size > IMG_MAX_BYTES:
        return None
    try:
        with open(path, "rb") as _sf:
            _sample = _sf.read(65536)                     # cheap content fingerprint (avoids path|mtime|size collisions)
    except Exception:
        return None
    key = "%s|%d|%d|%s" % (path, int(st.st_mtime), st.st_size, hashlib.sha1(_sample).hexdigest()[:12])
    cache = _uploaded_map()
    if key in cache:
        return cache[key]
    mt = MEDIA_BY_EXT.get(path.rsplit(".", 1)[-1].lower())
    if not mt:
        return None
    try:
        with open(path, "rb") as f:
            raw = f.read()
        body = json.dumps({"media_type": mt, "data_b64": base64.b64encode(raw).decode(),
                           "name": os.path.basename(path)}).encode()
        req = urllib.request.Request(SERVICE + "/api/img", data=body, method="POST", headers=_headers())
        with urllib.request.urlopen(req, timeout=45) as r:
            out = json.loads(r.read())
        url = out.get("url")
        if url:
            cache[key] = url
            _save_uploaded()
        return url
    except Exception:
        return None


def _agent_images_on():
    return os.environ.get("SM_KEEPER_AGENT_IMAGES", "1").strip().lower() not in ("0", "false", "off", "no")


def _within(realp, real_cwd):
    return realp == real_cwd or realp.startswith(real_cwd.rstrip("/") + "/")


def _extract_agent_images(text, cwd=None):
    """Find real image files an assistant message names by ABSOLUTE PATH and upload them
    (bounded + cached). SECURITY: only files INSIDE the session's cwd subtree, never a
    hidden/dot component (.ssh/.aws/.config/.git/...), never a symlink that escapes cwd
    (realpath re-checked), never anything when cwd is unknown or the kill-switch is off.
    This stops an agent (or a prompt-injection) from exfiltrating arbitrary local files."""
    if not _agent_images_on() or not cwd:
        return []
    try:
        real_cwd = os.path.realpath(cwd)
    except Exception:
        return []
    urls, seen = [], set()
    for m in _IMG_PATH_RE.findall(text or ""):
        if m in seen:
            continue
        seen.add(m)
        try:
            rp = os.path.realpath(m)                      # resolves symlinks + .. (traversal can't escape below)
        except Exception:
            continue
        if not _within(rp, real_cwd):                    # escaped cwd (incl. via symlink / ..) → skip
            continue
        # only the REMAINDER relative to cwd may not contain a hidden/dot component — a
        # session whose cwd itself is under a dotfiles dir still renders its in-cwd images.
        rel = rp[len(real_cwd):]
        if any(part.startswith(".") for part in rel.split("/") if part):
            continue
        if not os.path.isfile(rp):
            continue
        u = _upload_local_image(rp)
        if u:
            urls.append(u)
        if len(urls) >= 4:
            break
    return urls


def owner_identity():
    """Tenant key = the Adom/Claude account this container is logged into."""
    try:
        acct = json.load(open(CONFIG)).get("oauthAccount", {})
        return {"email": acct.get("emailAddress"), "name": acct.get("displayName"),
                "org": acct.get("organizationName"), "org_uuid": acct.get("organizationUuid")}
    except Exception:
        return {"email": None}


def machine_id():
    host = os.environ.get("VSCODE_PROXY_URI", "")       # platform-provided; carries the friendly name
    if host:
        host = host.split("//", 1)[-1].split("/", 1)[0]
    return {"id": socket.gethostname(), "proxy_host": host or None,
            "name": os.environ.get("SM_CONTAINER_NAME") or None,   # optional user-set display name
            "cwd": os.environ.get("PWD", "")}


def _pid_alive(pid):
    try:
        os.kill(int(pid), 0)
        return True
    except Exception:
        return False


def _pid_is_claude(pid):
    """Guard against PID REUSE: os.kill(pid,0) only proves *some* process holds that
    pid — after the OS recycles it, a stale sessions/<pid>.json can read as 'alive'
    and wrongly block injection ('held by VS Code') into a session that's actually
    closed. Confirm the pid is really a claude process before trusting it."""
    try:
        with open(f"/proc/{int(pid)}/cmdline", "rb") as f:
            cl = f.read().replace(b"\0", b" ").decode("utf-8", "ignore").lower()
        return "claude" in cl
    except Exception:
        return True                                     # no /proc (non-Linux) → don't over-block


def _claude_alive(pid):
    return _pid_alive(pid) and _pid_is_claude(pid)


# --- delivery idempotency: never inject the same command twice -----------------
# A lost ack leaves a command 'dispatched' server-side, which the service re-serves
# for lost-ack recovery. We record every command id we've already injected locally
# and, on a re-serve, RE-ACK without re-injecting — so a dropped ack (or a keeper
# crash between inject and ack) can never spawn a duplicate agent turn. Persisted so
# a restart mid-flight can't re-inject.
LEDGER = os.path.join(HOME, ".session-keeper", "done-cmds.json")
_done_ids = None


def _done_set():
    global _done_ids
    if _done_ids is None:
        try:
            _done_ids = set(json.load(open(LEDGER)))
        except Exception:
            _done_ids = set()
    return _done_ids


def _mark_injected(cid):
    d = _done_set()
    d.add(cid)
    try:
        os.makedirs(os.path.dirname(LEDGER), exist_ok=True)
        tmp = LEDGER + ".tmp"
        with open(tmp, "w") as f:
            json.dump(sorted(d)[-2000:], f)     # cap the on-disk ledger (backstop)
        os.replace(tmp, LEDGER)
    except Exception:
        pass


# in-flight: session_id -> pid of a keeper-launched turn still running, so we never
# start a second headless turn on a session whose previous turn hasn't finished.
_inflight = {}


def _pid_zombie(pid):
    """A reaped-not-yet/exited child still passes os.kill(pid,0) — check the actual
    process state so a zombie never reads as 'a turn is still running'."""
    try:
        with open(f"/proc/{int(pid)}/stat") as f:
            return f.read().rsplit(")", 1)[1].split()[0] == "Z"
    except Exception:
        return False                                    # no /proc → don't invent zombies


def _inflight_alive(session_id):
    pid = _inflight.get(session_id)
    if pid and _pid_alive(pid) and not _pid_zombie(pid):
        return True
    _inflight.pop(session_id, None)
    return False


def _jsonl_for(session_id):
    hits = glob.glob(os.path.join(PROJECTS_DIR, "**", session_id + ".jsonl"), recursive=True)
    return hits[0] if hits else None


def _is_rc(path):
    """Cheap scan: does this session's transcript carry remote-control markers?"""
    try:
        with open(path, "rb") as f:
            f.seek(0)
            head = f.read(200_000).decode("utf-8", "ignore")
        return any(m in head for m in RC_MARKERS)
    except Exception:
        return False


def _compact_q(qs):
    """Trim an AskUserQuestion `questions` array to a small, board-safe shape."""
    out = []
    for q in (qs or [])[:4]:
        out.append({
            "header": (q.get("header") or "")[:40],
            "question": (q.get("question") or "")[:600],
            "multiSelect": bool(q.get("multiSelect")),
            "options": [{"label": (o.get("label") or "")[:80],
                         "description": (o.get("description") or "")[:200]}
                        for o in (q.get("options") or [])[:6]],
        })
    return out


def _summarize(path, cwd=None, full=False):
    """Pull title + last user/assistant text + timestamps from a transcript.
    Title priority: the user's rename (custom-title) > Claude's ai-title >
    a summary entry > the first user message."""
    title, last_user, last_user_at, last_asst, last_asst_at, last_at = None, "", None, "", None, None
    first_user = custom_title = ai_title = summary_title = None
    interrupted = False                                 # is the tail an unresolved '[Request interrupted]'?
    pending_q = None                                    # an AskUserQuestion awaiting the user (structured)
    last_model = None                                   # the model the most recent assistant turn ran on
    t_cwd = None                                         # session cwd from the transcript (agent-image scoping)
    msgs = []                                           # chronological user/assistant messages (for the convo view)
    open_tool = None                                    # name of a trailing tool_use no result ever came back for
    try:
        lines = open(path, errors="ignore").read().splitlines()
    except Exception:
        return {}
    for ln in lines:
        try:
            o = json.loads(ln)
        except Exception:
            continue
        t, ts = o.get("type"), o.get("timestamp")
        if ts:
            last_at = ts
        if o.get("cwd"):
            t_cwd = o["cwd"]
        if t == "custom-title" and o.get("customTitle"):
            custom_title = o["customTitle"]            # the user's manual rename — last one wins
        elif t == "ai-title" and o.get("aiTitle"):
            ai_title = o["aiTitle"]
        elif t == "summary" and o.get("summary"):
            summary_title = o["summary"]
        msg = o.get("message") or {}
        content = msg.get("content")
        text = ""
        if isinstance(content, str):
            text = content
        elif isinstance(content, list):
            text = "".join(b.get("text", "") for b in content
                           if isinstance(b, dict) and b.get("type") == "text")
        text = text.strip()
        # pending AskUserQuestion: a structured question tool_use with no following
        # user turn. Detect BEFORE the no-text skip — the tool_use carries no text.
        if t == "assistant" and isinstance(content, list):
            _tus = [b2.get("name") for b2 in content if isinstance(b2, dict) and b2.get("type") == "tool_use"]
            if _tus:
                open_tool = _tus[0]                     # trailing tool_use → possibly awaiting approval
            elif text:
                open_tool = None                        # a plain text reply means nothing is pending
            for b in content:
                if isinstance(b, dict) and b.get("type") == "tool_use" and b.get("name") == "AskUserQuestion":
                    q = (b.get("input") or {}).get("questions") or []
                    if q:
                        pending_q = q
        elif t == "user":
            pending_q = None                            # any user turn after the question = answered / moved on
            open_tool = None                            # any user/tool_result entry resolves a pending tool_use
            # the harness auto-rejects a tool call whose permission prompt sat
            # unanswered (~10 min) — same marker as a manual "don't allow". Either
            # way the turn is PAUSED and needs a nudge; flag it like an interrupt.
            if "doesn't want to take this action" in ln:
                interrupted = True
        if not text:
            continue
        if t == "user" and not o.get("isMeta"):        # FIX: was `not str(...)` → always False
            # an interruption marker (VS Code timeout OR a manual stop) — the turn is
            # PAUSED. Flag it, but don't show it as a real prompt; a later real message
            # (below) resolves it. Nothing is ever auto-resumed.
            if "[request interrupted" in text.lower():
                interrupted = True
                continue
            # skip tool-result/command-wrapper noise + Claude Code's system-injected
            # compaction preamble so they're never mistaken for YOUR message.
            if (text.startswith("<") or text.startswith("Caveat:")
                    or text.startswith("This session is being continued from a previous conversation")):
                continue
            uimgs, ufiles, text = _extract_user_images(text)   # attached image urls + file chips; marker stripped
            last_user, last_user_at = text, ts
            interrupted = False                         # a fresh prompt means you've picked it back up
            if first_user is None:
                first_user = text
            msgs.append({"role": "user", "text": text, "at": ts, "images": uimgs or None, "files": ufiles or None})
        elif t == "assistant":
            last_asst, last_asst_at = text, ts
            interrupted = False                         # a completed/continued reply means it resumed
            m = msg.get("model")
            if m and m != "<synthetic>":                # track the model the latest real turn used
                last_model = m
            msgs.append({"role": "assistant", "text": text, "at": ts})
    title = (custom_title or ai_title or summary_title
             or (first_user or "")[:60].strip() or None)

    eff_cwd = cwd or t_cwd                               # scope agent-image extraction to the session's cwd

    def _capl(lst, clen):
        out = []
        for m in lst:
            t = m["text"] or ""
            if len(t) > clen:                            # NEVER cut silently — say so, mid-word cuts read as bugs
                t = t[:clen] + "\n\n… [message truncated]"
            d = {"role": m["role"], "text": t, "at": m["at"]}
            if m.get("files"):
                d["files"] = m["files"]
            imgs = m.get("images")
            if m["role"] == "assistant" and not imgs:
                imgs = _extract_agent_images(m["text"], eff_cwd)   # agent images: ONLY inside cwd
            if imgs:
                d["images"] = imgs                        # render inline on the board (both roles)
            out.append(d)
        return out
    def _cap(n, clen):
        return _capl(msgs[-n:], clen)
    def _recent(cap_turns, clen):
        # Focus the card on the CURRENT exchange: anchor on your last REAL message and
        # show it + everything after (the agent's reply chain). So a session where the
        # agent ran many turns in a row still shows YOUR message, not just agent turns.
        anchor = 0
        for i in range(len(msgs) - 1, -1, -1):
            if msgs[i]["role"] == "user":
                anchor = i
                break
        w = msgs[anchor:]
        if len(w) > cap_turns:                            # huge exchange → keep the anchor + the most recent turns
            w = [w[0]] + w[-(cap_turns - 1):]
        return _capl(w, clen)
    # recent = the convo shown in the OPEN card (audio/text overlay). Anchored on your
    # last message so it's always visible; per-message cap 4000 (700 cut replies short).
    out = {"title": title, "last_user": last_user, "last_user_at": last_user_at,
           "last_agent": last_asst, "last_agent_at": last_asst_at, "last_at": last_at,
           "needs_login": _needs_login(last_asst), "interrupted": interrupted,
           "pending_question": _compact_q(pending_q) if pending_q else None,
           "model": last_model,
           "open_tool": open_tool,
           "recent": _recent(8, 12000), "transcript": _cap(60, 12000)}
    if full:                                             # ENTIRE convo for an open card (on demand only)
        out["transcript_full"] = _cap(FULL_MAX, 12000)
        out["full_oldest_truncated"] = len(msgs) > FULL_MAX
    return out


def _needs_login(la):
    """The container's Claude Code is logged out — a headless `claude -p` returns
    'Not logged in · Please run /login' and that lands as the reply. Flag it so the
    UI can say 'run /login here' instead of showing it as a normal Claude message."""
    la = (la or "").lower()
    return "not logged in" in la and "/login" in la


# Sessions older than this (dead + untouched) are dropped from the board. Alive
# or recently-active sessions are always kept. Override with KEEPER_MAX_AGE_DAYS.
MAX_AGE = int(os.environ.get("KEEPER_MAX_AGE_DAYS", "3")) * 86400

# "Waiting on a tool approval" detection: these tools execute in milliseconds once
# approved — if one is the transcript's dangling tail for a while on an IDE-held
# session, a permission prompt is almost certainly sitting unanswered (they auto-
# reject after ~10 min, killing the turn while the user is away — the exact
# blindness SM exists to prevent). Long-running tools (Bash, Agent, ...) are
# excluded: a quiet tail there usually just means the tool is still working.
QUICK_TOOLS = {"Write", "Edit", "MultiEdit", "NotebookEdit", "TodoWrite"}
APPROVAL_QUIET_SEC = int(os.environ.get("KEEPER_APPROVAL_QUIET_SEC", "90"))


def _pending_approval(summ, held, mt, now):
    return bool(held and summ.get("open_tool") in QUICK_TOOLS
                and (now - mt) > APPROVAL_QUIET_SEC)


MAX_SESSIONS = int(os.environ.get("KEEPER_MAX_SESSIONS", "40"))


def _is_sidechain(path):
    """True for subagent/Task transcripts (every entry carries isSidechain:true) —
    those are ephemeral agent runs, not user sessions, so we keep them off the board."""
    try:
        with open(path, "rb") as f:
            head = f.read(120_000).decode("utf-8", "ignore")
        return '"isSidechain":true' in head or '"isSidechain": true' in head
    except Exception:
        return False


def collect_sessions():
    """Container sessions the keeper can drive. Enumerated from the TRANSCRIPTS
    (not just sessions/<pid>.json) so a session you CLOSE in the IDE stays on the
    board as idle/resumable instead of vanishing — every transcript is reachable
    via `claude --resume -p`. sessions/<pid>.json supplies live-process info."""
    # 1. live-process info per sessionId (pid file exists only while open)
    live = {}
    for sf in glob.glob(os.path.join(SESSIONS_DIR, "*.json")):
        try:
            s = json.load(open(sf))
        except Exception:
            continue
        sid = s.get("sessionId")
        if not sid:
            continue
        a = _claude_alive(s.get("pid"))
        if sid not in live or a:                        # prefer the alive record
            live[sid] = (s, a)

    # 2. recent transcripts (incl. closed sessions) — newest first, capped
    now = time.time()
    cand = {}
    for jp in glob.glob(os.path.join(PROJECTS_DIR, "**", "*.jsonl"), recursive=True):
        sid = os.path.basename(jp)[:-6]                 # strip ".jsonl"
        try:
            mt = os.path.getmtime(jp)
        except Exception:
            continue
        if sid not in live and (now - mt) > MAX_AGE:
            continue                                    # not live + long-stale → skip
        if sid not in cand or mt > cand[sid][1]:
            cand[sid] = (jp, mt)
    ordered = sorted(cand.items(), key=lambda kv: kv[1][1], reverse=True)[:MAX_SESSIONS * 2]

    out = []
    for sid, (jp, mt) in ordered:
        if len(out) >= MAX_SESSIONS:
            break
        if _is_sidechain(jp):                           # subagent/Task run → not a user session
            continue
        s, alive = live.get(sid, ({}, False))
        summ = _summarize(jp, (s or {}).get("cwd"))
        held = bool(alive and s.get("kind") == "interactive")
        recent = (now - mt) < 45
        # live → working/awaiting; closed/no-process → idle (still resumable)
        if held:
            status = "working" if recent else "awaiting"
        else:
            status = "working" if recent else "idle"
        out.append({
            "session_id": sid, "pid": s.get("pid"), "cwd": s.get("cwd"),
            "alive": alive, "status": status,
            "held_live": held,                          # open in IDE → can't deliver headlessly
            "resumable": True,                          # keeper can drive it via --resume -p
            "rc": _is_rc(jp),
            "title": summ.get("title") or sid[:8],
            "last_user_at": summ.get("last_user_at"), "last_agent_at": summ.get("last_agent_at"),
            "last_user": (summ.get("last_user") or "")[:8000],
            "last_agent": (summ.get("last_agent") or "")[:12000],
            "recent": summ.get("recent") or [],         # last ~3 exchanges for the card convo view
            "needs_login": summ.get("needs_login", False),
            "interrupted": summ.get("interrupted", False),
            "pending_question": summ.get("pending_question"),   # structured AskUserQuestion awaiting the user
            "pending_approval": _pending_approval(summ, held, mt, now),  # a tool-approval prompt is sitting unanswered in the IDE
            "model": summ.get("model"),                          # model the latest turn ran on
            "turn_active": _inflight_alive(sid),                 # a keeper-launched turn is running right now (precise)
            "last_at": summ.get("last_at"), "version": s.get("version"),
        })
    out.sort(key=lambda x: x.get("last_at") or "", reverse=True)
    return out


def claude_bin():
    return (os.environ.get("CLAUDE_BIN") or shutil.which("claude")
            or os.path.expanduser("~/.local/share/code-server/extensions/"
               "anthropic.claude-code-2.1.191-linux-x64/resources/native-binary/claude"))


def _session_proc(session_id):
    """The sessions/<pid>.json record for a session (pid, cwd, kind), if any."""
    for sf in glob.glob(os.path.join(SESSIONS_DIR, "*.json")):
        try:
            s = json.load(open(sf))
        except Exception:
            continue
        if s.get("sessionId") == session_id:
            return s
    return None


def held_live(session_id):
    """Is this session currently held by a LIVE interactive process (IDE)?"""
    s = _session_proc(session_id)
    return bool(s and _claude_alive(s.get("pid")) and s.get("kind") == "interactive")


def actively_working(session_id):
    """Is a turn being GENERATED right now? (transcript touched in the last ~12s).
    Delivery defers only during active generation — NOT for the whole time the IDE
    has the session open. (held_live stays true while you're away on your phone, so
    waiting on it queued messages forever — the bug this replaces.)"""
    jp = _jsonl_for(session_id)
    try:
        return bool(jp) and (time.time() - os.path.getmtime(jp)) < 12
    except Exception:
        return False


def _session_cwd(session_id):
    """Recover the session's working dir — from the live proc record, else from
    a `cwd` field inside its transcript (so headless `--resume` runs in-place)."""
    s = _session_proc(session_id) or {}
    if s.get("cwd"):
        return s["cwd"]
    jp = _jsonl_for(session_id)
    if jp:
        for ln in open(jp, errors="ignore"):
            if '"cwd"' in ln:
                try:
                    cwd = json.loads(ln).get("cwd")
                    if cwd:
                        return cwd
                except Exception:
                    pass
    return os.getcwd()


def inject(session_id, text, fork=False, model=None, allow_held=False):
    """Deliver one turn into a session via `claude --resume -p` (claude.ai-free).
    Returns (ok, result_text, effective_session_id).

    NORMAL sends are FIRE-AND-FORGET: we launch the turn detached and NEVER wait-
    then-kill (a real turn can run for minutes; killing it corrupts the user's
    work). We do a brief, non-destructive delivery CHECK (~4s): a healthy turn
    keeps running / writes to the transcript, whereas a logged-out or bad-session
    `claude -p` exits fast and non-zero. That lets us report a real failure instead
    of a false 'delivered'. Still running past the window → we return success and
    leave it detached. BRANCH waits (bounded) to capture the forked session id."""
    cwd = _session_cwd(session_id)
    cmd = [claude_bin(), "--resume", session_id, "-p", text, "--output-format", "json"]
    if model:                                           # per-session model choice (applies to this turn)
        cmd += ["--model", model]
    if fork:
        cmd.append("--fork-session")
        try:                                            # branch: need the new id back
            p = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, timeout=180)
            out = json.loads(p.stdout or "{}")
            return (not out.get("is_error", bool(p.returncode)),
                    out.get("result"), out.get("session_id") or session_id)
        except Exception as e:
            return (False, f"inject failed: {e}", session_id)
    # FINAL SAFETY: never `claude --resume -p` a session that's LIVE in the IDE. A
    # headless resume on a live session forks a turn the editor never sees and can
    # DESYNC/lose the session from the editor's list (what bit John's HD thread). The
    # drain already gates on held_live, but re-check at the last moment to close any
    # detection/timing race. Only force/takeover (deliberate, confirmed) pass allow_held.
    if not allow_held and held_live(session_id):
        return (False, "not delivered — this session is open in VS Code; resuming a live "
                       "session can desync it. Close it there, or use Take over / Fork.", session_id)
    jp = _jsonl_for(session_id)
    try:
        mt0 = os.path.getmtime(jp) if jp else 0
    except Exception:
        mt0 = 0
    try:
        dn = open(os.devnull, "wb")
        p = subprocess.Popen(cmd, cwd=cwd, stdout=dn, stderr=dn, stdin=subprocess.DEVNULL,
                             start_new_session=True)    # detached — runs to completion on its own
    except Exception as e:
        return (False, f"inject failed: {e}", session_id)
    _inflight[session_id] = p.pid                       # guard against overlapping turns
    # REAP the detached child when it exits (a zombie still passes os.kill(pid,0),
    # which pinned turn_active True forever). And the moment the turn ENDS, fast-
    # report this session — so the service sees turn_active=false together with the
    # FINAL reply text, and the "Reply ready" push never quotes a stale message.
    def _reap_and_report(proc=p, s_id=session_id):
        try:
            proc.wait()
        except Exception:
            pass
        try:
            report_one(s_id)
        except Exception:
            pass
    threading.Thread(target=_reap_and_report, daemon=True).start()
    delivered = "delivered — the agent is working; the reply will appear shortly"
    deadline = time.time() + 4
    while time.time() < deadline:
        rc = p.poll()
        if rc is None:                                  # still running → a real turn is underway
            try:
                if jp and os.path.getmtime(jp) > mt0:
                    return (True, delivered, session_id)
            except Exception:
                pass
            time.sleep(0.3)
            continue
        _inflight.pop(session_id, None)                 # it already exited
        if rc == 0:
            return (True, "delivered", session_id)
        return (False, "delivery failed — the container's Claude Code looks logged out "
                       "(run /login there) or the session is unavailable", session_id)
    return (True, delivered, session_id)


def drain_commands(mid):
    """Pull pending sends for this machine and execute them. Returns the 'watch'
    list (sessions a client has OPEN) so the loop can fast-report just those."""
    try:
        req = urllib.request.Request(SERVICE + "/api/keeper/commands?machine=" + mid, headers=_headers())
        with urllib.request.urlopen(req, timeout=15) as r:
            resp = json.loads(r.read())
        cmds = resp.get("commands", [])
        watch = resp.get("watch", [])
        fetch = resp.get("fetch", [])
    except Exception:
        return [], []
    for c in cmds:
        cid = c.get("id")
        sid, mode = c.get("session_id"), c.get("mode") or "auto"
        # EXACTLY-ONCE: if we already injected this command (the service re-served it
        # because our earlier ack was lost), just re-ack — NEVER re-inject.
        if cid in _done_set():
            _ack(mid, cid, "done", "delivered")
            continue
        if mode == "takeover":                          # steal the session from VS Code
            # Kill the live IDE claude process holding this session so it's FREE, then
            # deliver headlessly. Destructive by design (interrupts the IDE turn) — the
            # UI confirms first. After this the session is SM-driven; reopening it in the
            # IDE (claude --resume) hands it back.
            if cid in _done_set():
                _ack(mid, cid, "done", "delivered"); continue
            proc = _session_proc(sid) or {}
            pid = proc.get("pid")
            if pid and _pid_alive(pid):
                try:
                    os.kill(int(pid), 15)               # SIGTERM the IDE-side claude
                except Exception:
                    pass
                for _ in range(20):                     # wait (≤4s) for it to release the session
                    if not _pid_alive(pid):
                        break
                    time.sleep(0.2)
            ok, result, eff = inject(sid, c.get("text") or "", fork=False, model=c.get("model"), allow_held=True)
            if ok:
                _mark_injected(cid)
            _ack(mid, cid, "done" if ok else "error",
                 ("took over — now driving from Session Monitor" if ok else result))
            continue
        if mode == "new_session":                       # start a fresh session (no target sid)
            if cid in _newsession_inflight:
                continue
            _newsession_inflight.add(cid)
            threading.Thread(target=_run_new_session,
                             args=(mid, cid, c.get("text") or "", c.get("cwd"), c.get("model")),
                             daemon=True).start()
            continue
        # A session OPEN in the IDE owns its own stdin (held by VS Code) — a headless
        # `claude --resume -p` forks a turn the live session never sees, or hangs and
        # times out. There is NO claude.ai-free way to inject into the live process
        # (researched). So fail FAST with a clear, actionable status instead of a 300s
        # hang. (Branch forks a copy; force injects anyway; or close the session here.)
        if mode not in ("branch", "force", "takeover") and held_live(sid):
            _held_seen[sid] = time.time()               # remember it was live (for the settle window)
            # Open in VS Code → we can't inject headlessly WHILE it's held (that forks a
            # turn the IDE never sees / hangs). But keep it QUEUED (re-served) and deliver
            # the moment the session is free there — instead of a terminal "blocked".
            _ack(mid, cid, "waiting",
                 "queued — this session is open in VS Code; it delivers once it's free there")
            continue
        # SETTLE: after the editor releases a session (reload/close), give it a few
        # seconds to fully re-register before we `--resume` it — delivering the instant
        # it frees can race the editor's session re-registration and lose the session.
        if mode == "auto" and 0 < (time.time() - _held_seen.get(sid, 0)) < HELD_SETTLE_SEC:
            _ack(mid, cid, "waiting", "delivering shortly — letting the editor finish releasing this session")
            continue
        # don't start a 2nd turn while a previous keeper-launched turn is still running,
        # or while a turn is generating right now (avoid overlapping/corrupt transcripts)
        if mode != "branch" and _inflight_alive(sid):
            _ack(mid, cid, "waiting", "a previous message is still being processed — delivering once it settles")
            continue
        if mode == "auto" and actively_working(sid):
            _ack(mid, cid, "waiting", "agent is mid-response — delivering once it settles")
            continue
        # attach images: download to the inbox and reference by ABSOLUTE PATH so the
        # agent's Read tool (native vision) can view them. Only reached once we're
        # actually going to deliver (past the held/waiting guards).
        img_ids = c.get("images") or []
        img_paths, imgs_ok = fetch_command_images(img_ids)
        if img_ids and not imgs_ok:
            # some attachments didn't download — do NOT inject text-only or ack done.
            # Leave the command 'dispatched' so the server reclaims + retries it (bounded);
            # the idempotency ledger still guarantees exactly-once if a later try succeeds.
            continue
        text = _text_with_images(c.get("text") or "", img_paths)
        if mode == "branch":
            # A fork can take up to 180s; running it inline would starve report() and
            # trip a false "keeper down" alert. Run it off the loop and ack on finish.
            if cid in _branch_inflight:
                continue                                # already forking in a background thread
            _branch_inflight.add(cid)
            threading.Thread(target=_run_branch, args=(mid, cid, sid, text, c.get("model")),
                             daemon=True).start()
            continue
        ok, result, eff = inject(sid, text, fork=False, model=c.get("model"),
                                 allow_held=(mode == "force"))
        if ok:
            _mark_injected(cid)          # record BEFORE the ack so a lost ack can't re-inject
        _ack(mid, cid, "done" if ok else "error", result, branch_id=None)
    return watch, fetch


_newsession_inflight = set()                            # command ids starting a new session in a bg thread


def _all_sids():
    """Every session id we can see from local transcripts (for new-session detection)."""
    return {os.path.basename(jp)[:-6]
            for jp in glob.glob(os.path.join(PROJECTS_DIR, "**", "*.jsonl"), recursive=True)}


def _run_new_session(mid, cid, prompt, cwd, model=None):
    """Start a BRAND-NEW headless session (`claude -p`, no --resume) with an initial
    prompt, DETACHED so a long first turn is never killed. Poll briefly for the new
    session id so the UI can jump to it; if it doesn't appear in the window it still
    lands via the normal report sweep."""
    try:
        cwd = cwd or DEFAULT_NEW_CWD
        if not os.path.isdir(cwd):
            cwd = HOME
        before = _all_sids()
        cmd = [claude_bin(), "-p", prompt]
        if model:
            cmd += ["--model", model]
        try:
            subprocess.Popen(cmd, cwd=cwd, stdin=subprocess.DEVNULL,
                             stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                             start_new_session=True)
        except Exception as e:
            _ack(mid, cid, "error", f"couldn't start a new session: {e}")
            return
        _mark_injected(cid)                             # exactly-once: a re-served cmd never re-launches
        new_id = None
        for _ in range(25):                            # ~25s to spot the new transcript
            time.sleep(1.0)
            fresh = _all_sids() - before
            if fresh:
                # newest by transcript mtime
                new_id = max(fresh, key=lambda x: (os.path.getmtime(_jsonl_for(x)) if _jsonl_for(x) else 0))
                break
        if new_id:
            try:
                report_one(new_id)                     # fast-push so the board shows it immediately
            except Exception:
                pass
        _ack(mid, cid, "done", ("new session started" if new_id else "new session starting"),
             branch_id=new_id)                         # branch_id carries the new id → client opens it
    finally:
        _newsession_inflight.discard(cid)


_held_seen = {}                                         # sid -> ts last seen held_live (settle window)
HELD_SETTLE_SEC = int(os.environ.get("KEEPER_HELD_SETTLE", "8"))
_branch_inflight = set()                                # command ids being forked in a background thread


def _run_branch(mid, cid, sid, text, model=None):
    """Run a --fork-session inject off the main loop, then ack (so a slow fork can't
    starve the keeper's heartbeat). The server keeps the command 'dispatched' and
    won't re-serve within the reclaim window; the ledger blocks any late re-serve."""
    try:
        ok, result, eff = inject(sid, text, fork=True, model=model)
        if ok:
            _mark_injected(cid)
        _ack(mid, cid, "done" if ok else "error", result, branch_id=(eff if ok else None))
    finally:
        _branch_inflight.discard(cid)


def _ack(mid, cid, status, result, branch_id=None):
    body = json.dumps({"machine": mid, "id": cid, "status": status,
                       "result": (result or "")[:4000], "branch_id": branch_id}).encode()
    try:
        urllib.request.urlopen(urllib.request.Request(
            SERVICE + "/api/keeper/ack", data=body, method="POST",
            headers=_headers()), timeout=15).read()
    except Exception:
        pass


def _logged_out():
    """True when this container has NO usable Anthropic auth (needs /login) — read
    locally every cycle, zero API calls. We flag ONLY when none of the auth sources
    exist (no oauthAccount, no credentials token, no API-key env), so an expired-but-
    refreshable token — which Claude Code refreshes on its own — is never mis-flagged."""
    try:
        if (json.load(open(CONFIG)).get("oauthAccount") or {}).get("emailAddress"):
            return False
    except Exception:
        pass
    try:
        cred = json.load(open(os.path.join(HOME, ".claude", ".credentials.json")))
        if (cred.get("claudeAiOauth") or {}).get("accessToken"):
            return False
    except Exception:
        pass
    return not os.environ.get("ANTHROPIC_API_KEY")


def report():
    payload = {"owner": owner_identity(), "machine": machine_id(),
               "reported_at": int(time.time()), "keeper_version": KEEPER_VERSION,
               "logged_out": _logged_out(),               # container-level Anthropic auth state (proactive)
               "sessions": collect_sessions()}
    data = json.dumps(payload).encode()
    req = urllib.request.Request(SERVICE + "/api/keeper/report", data=data,
                                 method="POST", headers=_headers())
    with urllib.request.urlopen(req, timeout=20) as r:
        return json.loads(r.read()), len(payload["sessions"])


def collect_one(sid):
    """Build the board entry for a SINGLE session (cheap — one transcript)."""
    jp = _jsonl_for(sid)
    if not jp:
        return None
    s = _session_proc(sid) or {}
    alive = _claude_alive(s.get("pid"))
    try:
        mt = os.path.getmtime(jp)
    except Exception:
        mt = time.time()
    summ = _summarize(jp, (s or {}).get("cwd") or _session_cwd(sid))
    held = bool(alive and s.get("kind") == "interactive")
    recent = (time.time() - mt) < 45
    status = ("working" if recent else "awaiting") if held else ("working" if recent else "idle")
    return {
        "session_id": sid, "pid": s.get("pid"), "cwd": s.get("cwd"),
        "alive": alive, "status": status, "held_live": held, "resumable": True,
        "rc": _is_rc(jp), "title": summ.get("title") or sid[:8],
        "last_user_at": summ.get("last_user_at"), "last_agent_at": summ.get("last_agent_at"),
        "last_user": (summ.get("last_user") or "")[:8000], "last_agent": (summ.get("last_agent") or "")[:12000],
        "recent": summ.get("recent") or [],             # card convo view (last ~3 exchanges)
        "needs_login": summ.get("needs_login", False),
        "interrupted": summ.get("interrupted", False),
        "pending_question": summ.get("pending_question"),
        "pending_approval": _pending_approval(summ, held, mt, time.time()),
        "model": summ.get("model"),
        "turn_active": _inflight_alive(sid),
        "transcript": summ.get("transcript") or [],     # fuller convo for the Detail page (open card only)
        "last_at": summ.get("last_at"), "version": s.get("version"),
    }


_transcript_sent = {}          # sid -> ts of last full-transcript send (throttle)


def send_transcript(sid):
    """Build ONE session's ENTIRE transcript and push it to the service (on demand,
    for an open card). Throttled per-session so repeated 'fetch' asks don't rebuild it
    every second on a huge session."""
    now = time.time()
    if now - _transcript_sent.get(sid, 0) < 6:
        return
    _transcript_sent[sid] = now
    jp = _jsonl_for(sid)
    if not jp:
        return
    s = _session_proc(sid) or {}
    summ = _summarize(jp, (s or {}).get("cwd") or _session_cwd(sid), full=True)
    body = json.dumps({"owner": owner_identity(), "machine": machine_id(),
                       "session_id": sid, "msgs": summ.get("transcript_full") or [],
                       "oldest_truncated": summ.get("full_oldest_truncated", False)}).encode()
    try:
        urllib.request.urlopen(urllib.request.Request(
            SERVICE + "/api/keeper/transcript", data=body, method="POST", headers=_headers()), timeout=15).read()
    except Exception:
        _transcript_sent.pop(sid, None)                  # let the next fetch retry


def report_one(sid):
    """Fast push of one open card's session so it can feel live."""
    e = collect_one(sid)
    if not e:
        return
    body = json.dumps({"owner": owner_identity(), "machine": machine_id(), "session": e}).encode()
    try:
        urllib.request.urlopen(urllib.request.Request(
            SERVICE + "/api/keeper/report-one", data=body, method="POST", headers=_headers()), timeout=10).read()
    except Exception:
        pass


if __name__ == "__main__":
    try:                                               # stamp the running code version so `keeper status` can spot staleness
        with open(VERSION_FILE, "w") as _vf:
            _vf.write(KEEPER_VERSION)
    except Exception:
        pass
    o = owner_identity(); m = machine_id()
    print(f"keeper: v{KEEPER_VERSION} owner={o.get('email')} machine={m.get('proxy_host') or m.get('id')} → {SERVICE}", flush=True)
    once = "--once" in os.sys.argv
    mid = m.get("id") or m.get("proxy_host")
    last_full = 0.0
    report_backoff = 0.0        # grows on repeated report failures so a down service / WAF 403 isn't hammered
    fails = 0
    while True:
        now = time.time()
        try:
            if now - last_full >= REPORT_EVERY + report_backoff:   # full sweep: heartbeat + all sessions
                res, n = report()
                last_full = now; report_backoff = 0.0; fails = 0
                print(f"reported {n} rc session(s) → {res.get('ok')}", flush=True)
        except Exception as e:
            last_full = now                            # advance even on failure → no hot-spin retry loop
            fails += 1
            report_backoff = min(120.0, 2.0 ** min(fails, 7))
            print(f"report failed ({fails}, backoff {int(report_backoff)}s): {e}", flush=True)
        watch, fetch = [], []
        try:
            watch, fetch = drain_commands(mid)         # queued sends + open cards + full-transcript asks
        except Exception as e:
            print(f"drain failed: {e}", flush=True)
        for sid in watch:                              # fast-push open cards (1 transcript each)
            try:
                report_one(sid)
            except Exception:
                pass
        for sid in fetch:                              # ENTIRE convo for a just-opened card (off the loop)
            threading.Thread(target=send_transcript, args=(sid,), daemon=True).start()
        if once:
            break
        time.sleep(0.5 if watch else 2.5)              # ACTIVE: 0.5s fast-push while a card is open; else relaxed