#!/usr/bin/env -S uv run --no-project --python >=3.12 python
"""Read-only fleet reporter: one row per agent session on THIS host.

Step 1 of docs/backlog/fleet-coordination-via-precis.md. No DB, no precis
writes: it reads tmux, git worktrees, Claude transcripts
(`~/.claude/projects/<cwd with non-alphanumerics -> '-'>/*.jsonl`) and Codex
rollouts (`$CODEX_HOME/sessions/**/rollout-*.jsonl`), then prints a compact
fisheye (EXCEPTIONS / READY / SHIPPING / WAITING / AGENTS / QUOTA) or
`--json`.

State per row: working | idle | waiting | dead | unknown. `dead` means the
worktree lock's pid is gone, or no tmux pane exists while the transcript says
working and has been silent for DEAD_SILENT_MIN (a session outside tmux that
is actively writing is therefore not flagged). Usage: scripts/fleet-report
[--repo PATH]... [--json] [--quiet-min N] [--since-min N].

Step 2 collector: `--push` POSTs the report to the local precis HTTP MCP as
`put(kind='fleet', mode='report')` (URL from PRECIS_FLEET_MCP_URL, default
http://127.0.0.1:8765/mcp; bearer from PRECIS_MCP_TOKEN). `--remote HOST`
(repeatable) runs this script on HOST over ssh and pushes that host's report.
Without those flags it stays read-only.

Session-hook collector: `--hook` reports only the calling session as a
one-row partial report (the session's own state and context fill, not a
host scan), for harness lifecycle hooks (SessionStart/Stop/SessionEnd;
an uncommitted .devin/hooks.v1.json wires Devin, a
.claude/settings.json entry wires Claude). `--hook --push` sends it;
hook mode never breaks the session (failures go to stderr, exit 0;
PRECIS_FLEET_HOOK=0 disables).
"""

from __future__ import annotations

import argparse
import json
import os
import re
import socket
import subprocess
import sys
import time
import urllib.error
import urllib.request
from datetime import UTC, datetime
from pathlib import Path
from typing import IO

# Regexes matched (case-insensitive, any line) against the last lines of a
# pane. One hit marks the session `waiting` (an approval/permission dialog).
APPROVAL_PATTERNS = (
    r"do you want to (proceed|make this edit|create|allow)",
    r"\b1\.\s*yes\b.*\n?.*\b(2\.|3\.)\s*(yes|no)\b",
    r"would you like to run the following command",
    r"press enter to confirm or esc to cancel",
    r"\[y/n\]|\(y/n\)",
    r"allow (this|once|always)|approve (this|once)",
    r"permission (required|request)",
)
# Codex's Plan-mode request_user_input modal, as seen on melchior 2026-10-09:
# "Question 1/1 (1 unanswered)" … "tab to add notes | enter to submit answer".
QUESTION_PATTERNS = (
    r"Question \d+/\d+ \(\d+ unanswered\)",
    r"enter to submit answer",
)
AGENT_COMMANDS = ("claude", "codex", "devin", "node")
CLAUDE_WINDOW_DEFAULT = 200_000
#: Context window per Claude model id, from the harness's own model table
#: (claude-code 2.1.x embeds ``context:{window:…}`` per model): opus-4.7+,
#: sonnet-5+, fable-5+ and mythos-5 run a native 1M window; haiku-4-5 and
#: older ids are 200k. A trailing ``[1m]`` selects the 1M window on models
#: that support it; an unlisted id falls back to --claude-window.
CLAUDE_MODEL_WINDOWS = {
    "claude-fable-5": 1_000_000,
    "claude-fable-5-1": 1_000_000,
    "claude-mythos-5": 1_000_000,
    "claude-mythos-5-1": 1_000_000,
    "claude-opus-4-7": 1_000_000,
    "claude-opus-4-8": 1_000_000,
    "claude-opus-5": 1_000_000,
    "claude-opus-5-5": 1_000_000,
    "claude-sonnet-5": 1_000_000,
    "claude-sonnet-5-5": 1_000_000,
    "claude-haiku-4-5": 200_000,
    # Bare tier names in transcripts resolve to the current tier model.
    "fable": 1_000_000,
    "opus": 1_000_000,
    "sonnet": 1_000_000,
    "haiku": 200_000,
}
DEAD_SILENT_MIN = 5.0
RECENT_WORKING_S = 120
REPORT_VERSION = 2
DEFAULT_MCP_URL = "http://127.0.0.1:8765/mcp"
TAIL_BYTES = 262_144


def now_ts() -> float:
    return time.time()


def iso(ts: float | None) -> str | None:
    if ts is None:
        return None
    return datetime.fromtimestamp(ts, UTC).strftime("%Y-%m-%dT%H:%M:%SZ")


def parse_iso(text: str | None) -> float | None:
    if not text:
        return None
    try:
        return datetime.fromisoformat(text.replace("Z", "+00:00")).timestamp()
    except ValueError:
        return None


def run(*args: str, cwd: str | Path | None = None) -> str | None:
    try:
        r = subprocess.run(
            args, cwd=cwd, capture_output=True, text=True, encoding="utf-8", check=False
        )
    except (OSError, ValueError):
        return None
    return r.stdout if r.returncode == 0 else None


def tail_lines(path: Path, nbytes: int = TAIL_BYTES) -> list[str]:
    """Complete lines from the last `nbytes` of a file (first partial dropped)."""
    try:
        with path.open("rb") as f:
            size = f.seek(0, os.SEEK_END)
            f.seek(max(0, size - nbytes))
            data = f.read()
            partial = size > nbytes
    except OSError:
        return []
    lines = data.decode("utf-8", "replace").splitlines()
    return lines[1:] if partial else lines


def json_tail(path: Path) -> list[dict]:
    out = []
    for line in tail_lines(path):
        try:
            d = json.loads(line)
        except ValueError:
            continue
        if isinstance(d, dict):
            out.append(d)
    return out


# ---------------------------------------------------------------- tmux / git


def tmux_panes() -> list[dict]:
    fmt = "\t".join(
        f"#{{{k}}}"
        for k in (
            "session_name",
            "window_index",
            "window_name",
            "pane_id",
            "pane_pid",
            "pane_current_command",
            "pane_current_path",
            "window_activity",
        )
    )
    out = run("tmux", "list-panes", "-a", "-F", fmt)
    panes = []
    for line in (out or "").splitlines():
        p = line.split("\t")
        if len(p) != 8:
            continue
        panes.append(
            {
                "session": p[0],
                "window": p[1],
                "window_name": p[2],
                "pane_id": p[3],
                "pane_pid": p[4],
                "command": p[5],
                "path": p[6],
                "activity": float(p[7]) if p[7].isdigit() else None,
            }
        )
    return panes


def pane_screen(pane_id: str) -> str:
    return run("tmux", "capture-pane", "-p", "-t", pane_id, "-S", "-15") or ""


def pane_approval(text: str) -> bool:
    return any(re.search(p, text, re.I) for p in APPROVAL_PATTERNS)


def claude_window(model: str | None, override: int | None = None) -> int:
    """Context window for a transcript's model id; --claude-window wins."""
    if override:
        return override
    m = (model or "").strip().lower()
    if m.endswith("[1m]"):
        return 1_000_000
    m = re.sub(r"-\d{8}$", "", m.split("@", 1)[0])  # dated variant suffix
    return CLAUDE_MODEL_WINDOWS.get(m, CLAUDE_WINDOW_DEFAULT)


def pane_ctx_pct(text: str) -> float | None:
    """Context fill % from a live pane footer; None when it shows none.

    Devin prints ``Context: 212k / 262k tokens (81%)``, Claude ``Opus 5.5
    19% ctx`` (both are % used) and Codex ``Context 87% left`` (% left).
    The last match wins: the footer is the bottom-most line on screen.
    """
    for pat, used in (
        (r"Context:[^\n()]*\((\d+)%\)", True),
        (r"(\d+)%\s*ctx\b", True),
        (r"Context\s+(\d+)%\s*left", False),
    ):
        hits = list(re.finditer(pat, text, re.I))
        if hits:
            pct = float(hits[-1].group(1))
            return pct if used else 100 - pct
    return None


def codex_footer(text: str) -> tuple[str | None, float | None]:
    """State and context fill from the Codex TUI status line.

    A fresh Codex writes no rollout until its first turn, and a rollout lags
    a live turn; the footer (`… · Context 87% left · Ready`) is current.
    """
    state = None
    lines = [ln for ln in text.splitlines() if "Context" in ln and "left" in ln]
    if lines:
        last = lines[-1].rsplit("·", 1)[-1].strip().lower()
        state = "idle" if last.startswith("ready") else "working" if last else None
    return state, pane_ctx_pct(text)


def git(path: str | Path, *args: str) -> str | None:
    out = run("git", "-C", str(path), *args)
    return out.strip() if out is not None else None


def pid_alive(pid: int) -> bool:
    """kill -0; EPERM means the process exists (mirrors scripts/inflight)."""
    try:
        os.kill(pid, 0)
    except PermissionError:
        return True
    except OSError:
        return False
    return True


def list_worktrees(repo: str | Path) -> list[dict]:
    out = run("git", "-C", str(repo), "worktree", "list", "--porcelain")
    trees, cur = [], None
    for line in (out or "").splitlines() + [""]:
        if not line:
            if cur:
                trees.append(cur)
            cur = None
        elif line.startswith("worktree "):
            cur = {"path": line[9:], "branch": "", "locked": None}
        elif cur is not None and line.startswith("branch "):
            cur["branch"] = line[7:].removeprefix("refs/heads/")
        elif cur is not None and line.startswith("locked"):
            cur["locked"] = line[6:].strip()
    return trees


def worktree_info(path: str, entry: dict | None = None) -> dict | None:
    top = git(path, "rev-parse", "--show-toplevel")
    if not top:
        return None
    common = git(top, "rev-parse", "--path-format=absolute", "--git-common-dir")
    project = Path(common).parent.name if common else Path(top).name
    branch = (
        (entry or {}).get("branch")
        or git(top, "rev-parse", "--abbrev-ref", "HEAD")
        or ""
    )
    locked = (entry or {}).get("locked")
    if entry is None:
        for t in list_worktrees(top):
            if os.path.realpath(t["path"]) == os.path.realpath(top):
                locked = t["locked"]
    status = git(top, "status", "--porcelain") or ""
    ab = git(top, "rev-list", "--left-right", "--count", "main...HEAD") or ""
    behind, ahead = (
        (int(x) for x in ab.split()) if len(ab.split()) == 2 else (None, None)
    )
    purpose = ""
    try:
        for line in (
            (Path(top) / ".claude" / "purpose").read_text(encoding="utf-8").splitlines()
        ):
            if line.strip():
                purpose = line.strip()
                break
    except OSError:
        pass
    pid = None
    m = re.search(r"pid[\s:=]+(\d+)", locked or "")
    if m:
        pid = int(m.group(1))
    return {
        "path": os.path.realpath(top),
        "project": project,
        "tree": Path(top).name,
        "branch": branch,
        "dirty": len([ln for ln in status.splitlines() if ln.strip()]),
        "ahead": ahead,
        "behind": behind,
        "purpose": purpose,
        "lock": locked,
        "lock_pid": pid,
        "lock_alive": pid_alive(pid) if pid else None,
    }


# ---------------------------------------------------------------- transcripts


def claude_dir(cwd: str) -> Path:
    return Path.home() / ".claude" / "projects" / re.sub(r"[^A-Za-z0-9]", "-", cwd)


def newest(paths) -> Path | None:
    best, best_m = None, -1.0
    for p in paths:
        try:
            m = p.stat().st_mtime
        except OSError:
            continue
        if m > best_m:
            best, best_m = p, m
    return best


def claude_session(cwd: str, window: int | None) -> dict | None:
    """State from the newest transcript for this cwd (None when absent)."""
    f = newest(claude_dir(cwd).glob("*.jsonl"))
    return claude_transcript(f, window) if f else None


def claude_transcript(f: Path, window: int | None) -> dict:
    """State and context fill from one Claude transcript file."""
    mtime = f.stat().st_mtime
    entries = [d for d in json_tail(f) if d.get("type") in ("assistant", "user")]
    ctx = model = None
    for d in reversed(entries):
        u = (d.get("message") or {}).get("usage")
        if d["type"] == "assistant" and isinstance(u, dict):
            ctx = (
                (u.get("input_tokens") or 0)
                + (u.get("cache_read_input_tokens") or 0)
                + (u.get("cache_creation_input_tokens") or 0)
            )
            model = (d.get("message") or {}).get("model")
            break
    win = claude_window(model, window)
    state = "unknown"
    if entries:
        last = entries[-1]
        msg = last.get("message") or {}
        content = msg.get("content")
        blocks = content if isinstance(content, list) else []
        age = now_ts() - (parse_iso(last.get("timestamp")) or mtime)
        if last["type"] == "assistant":
            if any(b.get("type") == "tool_use" for b in blocks if isinstance(b, dict)):
                state = "working"
            elif msg.get("stop_reason") == "end_turn" or age > RECENT_WORKING_S:
                state = "idle"
            else:
                state = "working"
        else:  # user prompt or tool_result: the agent owes a reply
            state = "working"
    return {
        "file": str(f),
        "mtime": mtime,
        "state": state,
        "model": model,
        "ctx_tokens": ctx,
        "ctx_pct": round(100 * ctx / win, 1) if ctx is not None else None,
    }


def codex_home() -> Path:
    return Path(os.environ.get("CODEX_HOME") or Path.home() / ".codex")


def codex_sessions(since_s: float) -> dict[str, dict]:
    """Newest recent rollout per cwd: {cwd: session dict}."""
    root = codex_home() / "sessions"
    cutoff = now_ts() - since_s
    newest_by_cwd: dict[str, tuple[float, Path]] = {}
    for p in root.rglob("rollout-*.jsonl") if root.is_dir() else ():
        try:
            m = p.stat().st_mtime
            if m < cutoff:
                continue
            with p.open(encoding="utf-8") as f:
                meta = json.loads(f.readline())
        except (OSError, ValueError):
            continue
        cwd = (meta.get("payload") or {}).get("cwd") if isinstance(meta, dict) else None
        # Daemon-mode sessions record the daemon's cwd in session_meta; the
        # worker's directory is in each turn_context, so the newest one wins.
        for d in json_tail(p):
            if d.get("type") == "turn_context":
                cwd = (d.get("payload") or {}).get("cwd") or cwd
        if cwd and (cwd not in newest_by_cwd or m > newest_by_cwd[cwd][0]):
            newest_by_cwd[cwd] = (m, p)
    out = {}
    for cwd, (m, p) in newest_by_cwd.items():
        entries = json_tail(p)
        last_ts = m
        state, ctx, window, limits = "unknown", None, None, {}
        asking = False
        for d in entries:
            last_ts = parse_iso(d.get("timestamp")) or last_ts
            pl = d.get("payload") or {}
            kind = pl.get("type")
            # request_user_input_async returns at once and the turn completes
            # ("Ready" in the footer) while the question waits for Reto.
            if kind == "function_call" and str(pl.get("name", "")).startswith(
                "request_user_input"
            ):
                asking = True
            if d.get("type") != "event_msg":
                continue
            if kind == "task_started":
                state, asking = "working", False
            elif kind in ("task_complete", "turn_aborted"):
                state = "idle"
            elif kind == "token_count":
                info = pl.get("info") or {}
                use = info.get("last_token_usage") or {}
                if use:
                    ctx = use.get("input_tokens") or use.get("total_tokens")
                    window = info.get("model_context_window")
                rl = pl.get("rate_limits") or {}
                for slot in ("primary", "secondary"):
                    w = rl.get(slot)
                    if isinstance(w, dict) and w.get("used_percent") is not None:
                        limits[slot] = {
                            "used_percent": w["used_percent"],
                            "window_minutes": w.get("window_minutes"),
                        }
        if state == "unknown" and entries:
            state = "working"
        if asking:
            state = "asking"
        out[cwd] = {
            "file": str(p),
            "mtime": m,
            "last_event": last_ts,
            "state": state,
            "ctx_tokens": ctx,
            "ctx_pct": round(100 * ctx / window, 1) if ctx and window else None,
            "rate_limits": limits,
        }
    return out


def window_label(minutes) -> str:
    if not minutes:
        return "?"
    return (
        "5h"
        if minutes <= 360
        else "week"
        if minutes >= 10000
        else f"{round(minutes / 60)}h"
    )


# ---------------------------------------------------------------- assemble


def vendor_of(command: str) -> str:
    c = command.lower()
    for v in ("claude", "codex", "devin"):
        if v in c:
            return v
    return "claude" if c == "node" else c


def collect(
    repos: list[str], quiet_min: float, since_min: float, window: int | None
) -> dict:
    host = socket.gethostname().split(".")[0]
    t = now_ts()
    panes = tmux_panes()
    trees: dict[str, dict] = {}

    def add(path: str, entry: dict | None = None) -> str | None:
        key = os.path.realpath(path)
        if key not in trees:
            info = worktree_info(path, entry)
            if info is None:
                return None
            trees[info["path"]] = info
            key = info["path"]
        return key

    for repo in repos:
        for e in list_worktrees(repo):
            add(e["path"], e)
    pane_tree = {}
    for p in panes:
        if p["path"]:
            pane_tree[p["pane_id"]] = add(p["path"])
    codex = codex_sessions(since_min * 60)
    codex_by_tree = {}
    for cwd, s in codex.items():
        key = add(cwd) if os.path.isdir(cwd) else None
        if key and (
            key not in codex_by_tree or s["mtime"] > codex_by_tree[key]["mtime"]
        ):
            codex_by_tree[key] = s

    rows = []
    seen = set()

    def make_row(vendor: str, key: str, pane: dict | None) -> dict:
        wt = trees[key]
        sess = None
        if vendor == "claude":
            sess = claude_session(wt["path"], window)
        elif vendor == "codex":
            sess = codex_by_tree.get(key)
        mt = [x for x in (sess and sess["mtime"], pane and pane["activity"]) if x]
        latest = max(mt) if mt else None
        state = sess["state"] if sess else ("unknown" if pane else "idle")
        ctx_pct = sess["ctx_pct"] if sess else None
        screen = pane_screen(pane["pane_id"]) if pane else ""
        if vendor == "codex" and screen:
            f_state, f_ctx = codex_footer(screen)
            if state != "asking":
                state = f_state or state
            ctx_pct = f_ctx if f_ctx is not None else ctx_pct
        else:
            # A live footer is fresher than the last transcript usage record.
            foot = pane_ctx_pct(screen)
            if foot is not None:
                ctx_pct = foot
        if screen and any(re.search(p, screen) for p in QUESTION_PATTERNS):
            state = "asking"
        elif screen and vendor != "devin" and pane_approval(screen):
            state = "waiting"
        if wt["lock_alive"] is False or (
            pane is None
            and sess
            and sess["state"] == "working"
            and latest
            and (t - latest) / 60 > DEAD_SILENT_MIN
        ):
            state = "dead"
        row = {
            "vendor": vendor,
            "host": host,
            "project": wt["project"],
            "tree": wt["tree"],
            "path": wt["path"],
            "branch": wt["branch"],
            "dirty": wt["dirty"],
            "ahead": wt["ahead"],
            "behind": wt["behind"],
            "purpose": wt["purpose"],
            "lock_pid": wt["lock_pid"],
            "lock_alive": wt["lock_alive"],
            "state": state,
            "quiet_min": round((t - latest) / 60, 1) if latest else None,
            "last_active": iso(latest),
            "transcript_mtime": iso(sess["mtime"]) if sess else None,
            "ctx_pct": ctx_pct,
            "ctx_tokens": sess["ctx_tokens"] if sess else None,
            "model": sess.get("model") if sess else None,
            "rate_limits": sess.get("rate_limits") if sess else None,
            "tmux": None,
            "attach": None,
        }
        if pane:
            row["tmux"] = {
                k: pane[k]
                for k in (
                    "session",
                    "window",
                    "window_name",
                    "pane_id",
                    "pane_pid",
                    "command",
                )
            }
            row["attach"] = f"tmux attach -t {pane['session']}:{pane['window']}"
        return row

    for p in panes:
        if not any(a in p["command"].lower() for a in AGENT_COMMANDS):
            continue
        key = pane_tree.get(p["pane_id"])
        if not key:
            continue
        vendor = vendor_of(p["command"])
        seen.add((vendor, key))
        rows.append(make_row(vendor, key, p))
    # sessions with no pane: recent transcripts / rollouts / live locks
    for key, wt in trees.items():
        if ("claude", key) not in seen:
            c = claude_session(wt["path"], window)
            if (c and t - c["mtime"] < since_min * 60) or wt["lock_alive"]:
                rows.append(make_row("claude", key, None))
        if ("codex", key) not in seen and key in codex_by_tree:
            rows.append(make_row("codex", key, None))

    # Quota is per account: take the newest rollout anywhere, even one whose
    # cwd is not a worktree (an editor session in $HOME still reports it).
    newest_codex = max(codex.values(), key=lambda s: s["mtime"], default=None)
    annotate_done(rows)
    return {
        "host": host,
        "version": REPORT_VERSION,
        "generated": iso(t),
        "rows": rows,
        "exceptions": exceptions(rows, quiet_min),
        "quota": {"codex": newest_codex["rate_limits"]}
        if newest_codex and newest_codex["rate_limits"]
        else {},
    }


def exceptions(rows: list[dict], quiet_min: float, ctx_max: float = 60.0) -> list[dict]:
    out = []
    for r in rows:
        who = f"{r['vendor']}@{r['host']}/{r['project']}/{r['tree']}"
        reasons = []
        if r["state"] == "dead":
            reasons.append("dead")
        if r["state"] == "waiting":
            reasons.append("waiting on approval")
        if r["state"] == "asking":
            reasons.append("question for Reto")
        if (
            r["state"] == "working"
            and r["quiet_min"] is not None
            and r["quiet_min"] > quiet_min
        ):
            reasons.append(f"quiet {r['quiet_min']:.0f}m while working")
        if r["ctx_pct"] is not None and r["ctx_pct"] > ctx_max:
            reasons.append(f"context {r['ctx_pct']:.0f}%")
        for why in reasons:
            out.append({"who": who, "reason": why, "attach": r["attach"]})
    return out


#: `.claude/purpose` "<tree> at <sha>: …" marks a finished tip (mirrors
#: precis.fleet.DONE_RE; the script must stay import-free).
DONE_RE = re.compile(r"^\S+\s+at\s+([0-9a-f]{7,40})\b")
#: done@ names the BRANCH head; `scripts/ship` squash-merges it to a new
#: sha on main, so the sha itself is never an ancestor. "Landed" is
#: decided by patch-id equality of the whole branch diff
#: (`git diff <merge-base> <sha>`) against origin/main's recent commits,
#: with an exact commit-subject match as the fallback (a conflict-resolved
#: squash changes the diff but keeps the subject). The lookback bounds the
#: scan; the fleet lands far fewer commits a day.
DONE_LOOKBACK = 300
#: A done@ branch whose diff touches these never takes `scripts/ship`
#: (scripts/qgo-guard): it waits on the release owner's /go, so its row
#: reads WAITING rather than SHIPPING.
DONE_WAIT_RE = re.compile(r"migrations/|safe_fetch\.py$")
#: An explicit "did not land" note in the purpose line means WAITING even
#: when the branch objects are gone and the diff can't be checked.
DONE_WAIT_PURPOSE_RE = re.compile(r"unlanded|awaiting /go", re.I)
#: origin/main is refetched only while a done@ row exists, and at most
#: this often (FETCH_HEAD's mtime is the stamp; a failed fetch retries on
#: the next report).
DONE_FETCH_S = 120


def _git_ok(path: str, *args: str) -> bool:
    """`git -C path <args>` exits 0 (output not needed)."""
    try:
        r = subprocess.run(
            ["git", "-C", str(path), *args],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            check=False,
        )
    except (OSError, ValueError):
        return False
    return r.returncode == 0


def _patch_id(path: str, *diff_args: str) -> str | None:
    """Stable patch-id of `git -C path <diff_args>`; None on empty/failed."""
    diff = run("git", "-C", str(path), *diff_args)
    if not diff or not diff.strip():
        return None
    try:
        r = subprocess.run(
            ["git", "patch-id", "--stable"],
            input=diff,
            capture_output=True,
            text=True,
            encoding="utf-8",
            check=False,
        )
    except (OSError, ValueError):
        return None
    if r.returncode != 0 or not r.stdout.strip():
        return None
    return r.stdout.split()[0]


def _main_pids(path: str) -> set[str]:
    """patch-ids of origin/main's last DONE_LOOKBACK commits (one pipeline;
    empty commits drop out)."""
    log = run(
        "git",
        "-C",
        str(path),
        "log",
        "-p",
        f"--max-count={DONE_LOOKBACK}",
        "origin/main",
    )
    if not log:
        return set()
    try:
        r = subprocess.run(
            ["git", "patch-id", "--stable"],
            input=log,
            capture_output=True,
            text=True,
            encoding="utf-8",
            check=False,
        )
    except (OSError, ValueError):
        return set()
    return {ln.split()[0] for ln in r.stdout.splitlines() if ln.strip()}


def _main_snapshot(path: str) -> dict:
    """origin/main's recent subjects and patch-ids (empty when unfetched)."""
    subs = git(
        path, "log", f"--max-count={DONE_LOOKBACK}", "--format=%s", "origin/main"
    )
    return {"subjects": set((subs or "").splitlines()), "pids": _main_pids(path)}


def _maybe_fetch_main(path: str) -> None:
    """Best-effort `git fetch origin main`, throttled by FETCH_HEAD's mtime
    so a dead remote costs one ConnectTimeout, not a stale READY forever."""
    common = git(path, "rev-parse", "--path-format=absolute", "--git-common-dir")
    if common:
        try:
            if now_ts() - (Path(common) / "FETCH_HEAD").stat().st_mtime < DONE_FETCH_S:
                return
        except OSError:
            pass
    try:
        subprocess.run(
            ["git", "-C", str(path), "fetch", "-q", "origin", "main"],
            capture_output=True,
            text=True,
            encoding="utf-8",
            check=False,
            timeout=15,
            env={
                **os.environ,
                "GIT_TERMINAL_PROMPT": "0",
                "GIT_SSH_COMMAND": "ssh -o BatchMode=yes -o ConnectTimeout=5",
            },
        )
    except (OSError, ValueError, subprocess.TimeoutExpired):
        pass


def done_state(path: str, purpose: str, snap: dict) -> str | None:
    """'landed' | 'waiting' | 'shipping' for a done@<sha> purpose line."""
    m = DONE_RE.match(purpose or "")
    if not m:
        return None
    sha = m.group(1)
    if _git_ok(path, "merge-base", "--is-ancestor", sha, "origin/main"):
        return "landed"  # a direct merge/fast-forward kept the sha
    base = git(path, "merge-base", "origin/main", sha)
    if base is None:  # unknown sha or unfetched origin/main
        return "waiting" if DONE_WAIT_PURPOSE_RE.search(purpose) else "shipping"
    if _patch_id(path, "diff", base, sha) in snap["pids"]:
        return "landed"  # the squash carries the identical diff
    subject = git(path, "log", "-1", "--format=%s", sha)
    if subject and subject in snap["subjects"]:
        return "landed"  # resolved-squash fallback (DONE_LOOKBACK comment)
    names = git(path, "diff", "--name-only", base, sha) or ""
    if DONE_WAIT_RE.search(names) or DONE_WAIT_PURPOSE_RE.search(purpose):
        return "waiting"
    return "shipping"


def annotate_done(rows: list[dict], fetch: bool = True) -> None:
    """Set ``done_state`` on each row whose purpose carries done@<sha>.
    The origin/main snapshot is taken once per repo (common dir)."""
    snaps: dict[str, dict] = {}
    for r in rows:
        if not DONE_RE.match(r.get("purpose") or ""):
            continue
        path = r["path"]
        common = (
            git(path, "rev-parse", "--path-format=absolute", "--git-common-dir") or path
        )
        if common not in snaps:
            if fetch:
                _maybe_fetch_main(path)
            snaps[common] = _main_snapshot(path)
        r["done_state"] = done_state(path, r["purpose"], snaps[common])


def render(rep: dict) -> str:
    def who(r: dict) -> str:
        tm = r.get("tmux") or {}
        tgt = (
            f" {tm['session']}:{tm['window']}"
            if isinstance(tm, dict) and tm.get("session")
            else ""
        )
        return f"{r['vendor']}@{r['host']}/{r['project']}/{r['tree']}{tgt}"

    lines = ["EXCEPTIONS"]
    for e in rep["exceptions"]:
        tail = f"   attach: {e['attach']}" if e["attach"] else ""
        lines.append(f"  {e['who']}  {e['reason']}{tail}")
    if not rep["exceptions"]:
        lines.append("  none")
    dones = [r for r in rep["rows"] if DONE_RE.match(r["purpose"] or "")]

    def done_block(title: str, rows: list[dict]) -> None:
        lines.append(title)
        for r in rows:
            sha = DONE_RE.match(r["purpose"]).group(1)
            tail = f"   attach: {r['attach']}" if r["attach"] else ""
            lines.append(f"  {who(r)} · done@{sha}{tail}")
        if not rows:
            lines.append("  none")

    done_block("READY", [r for r in dones if r.get("done_state") == "landed"])
    done_block(
        "SHIPPING",
        [r for r in dones if r.get("done_state") not in ("landed", "waiting")],
    )
    done_block("WAITING", [r for r in dones if r.get("done_state") == "waiting"])
    lines.append("AGENTS")
    for r in rep["rows"]:
        q = f"{r['quiet_min']:.0f}m" if r["quiet_min"] is not None else "-"
        ctx = f"{r['ctx_pct']:.0f}%" if r["ctx_pct"] is not None else "-"
        purpose = r["purpose"][:40]
        m = DONE_RE.match(r["purpose"] or "")
        done = f" · done@{m.group(1)}" if m else ""
        lines.append(f"  {who(r)} · {r['state']} · {q} · {purpose} · {ctx}{done}")
    if not rep["rows"]:
        lines.append("  none")
    quota = {}
    for vendor, limits in (rep.get("quota") or {}).items():
        for w in limits.values():
            quota[(vendor, window_label(w.get("window_minutes")))] = w["used_percent"]
    for r in rep["rows"]:
        for w in (r.get("rate_limits") or {}).values():
            quota[(r["vendor"], window_label(w.get("window_minutes")))] = w[
                "used_percent"
            ]
    if quota:
        lines.append(
            "QUOTA   "
            + " · ".join(
                f"{v} {lab} {pct:.0f}%" for (v, lab), pct in sorted(quota.items())
            )
        )
    return "\n".join(lines)


# ---------------------------------------------------------------- push


# ---------------------------------------------------------------- hook mode

#: hook_event_name -> state for sessions that push their own row
#: (``--state`` wins when the harness passes one explicitly).
HOOK_STATES = {
    "sessionstart": "working",
    "userpromptsubmit": "working",
    "pretooluse": "working",
    "posttooluse": "working",
    "precompact": "working",
    "postcompaction": "working",
    "notification": "waiting",
    "permissionrequest": "waiting",
    "stop": "idle",
    "subagentstop": "idle",
    "sessionend": "dead",
}


def detect_vendor() -> str:
    """Which harness is calling, from the environment (lowest precedence)."""
    agent = (os.environ.get("AI_AGENT") or "").lower()
    for v in ("claude", "codex", "devin"):
        if agent.startswith(v):
            return v
    if os.environ.get("CLAUDECODE") or os.environ.get("CLAUDE_PROJECT_DIR"):
        return "claude"
    if os.environ.get("CODEX_HOME"):
        return "codex"
    return "agent"


def hook_stdin() -> dict:
    """The hook payload on stdin (``{}`` when absent or not JSON)."""
    if sys.stdin.isatty():
        return {}
    try:
        data = json.load(sys.stdin)
    except ValueError:
        return {}
    return data if isinstance(data, dict) else {}


def hook_report(
    vendor: str | None, state: str | None, window: int | None
) -> dict | None:
    """A one-row ``partial`` report for the calling session; None when the
    cwd is not inside a git worktree."""
    data = hook_stdin()
    cwd = str(data.get("cwd") or Path.cwd())
    wt = worktree_info(cwd)
    if wt is None:
        return None
    t = now_ts()
    pane = None
    pane_id = os.environ.get("TMUX_PANE")
    if pane_id:
        pane = next((p for p in tmux_panes() if p["pane_id"] == pane_id), None)
    if vendor is None:
        vendor = vendor_of(pane["command"]) if pane else detect_vendor()
    if state is None:
        state = HOOK_STATES.get(str(data.get("hook_event_name") or "").lower())
    state = state or "unknown"
    ctx_pct = None
    if pane and state != "dead":
        ctx_pct = pane_ctx_pct(pane_screen(pane["pane_id"]))
    if ctx_pct is None and vendor == "claude" and state != "dead":
        tp = data.get("transcript_path")
        try:
            sess = (
                claude_transcript(Path(str(tp)), window)
                if tp
                else claude_session(wt["path"], window)
            )
        except OSError:
            sess = None
        if sess:
            ctx_pct = sess["ctx_pct"]
    live = pane if pane and state != "dead" else None
    row = {
        "vendor": vendor,
        "host": socket.gethostname().split(".")[0],
        "project": wt["project"],
        "tree": wt["tree"],
        "path": wt["path"],
        "branch": wt["branch"],
        "dirty": wt["dirty"],
        "ahead": wt["ahead"],
        "behind": wt["behind"],
        "purpose": wt["purpose"],
        "state": state,
        "quiet_min": 0.0,
        "last_active": iso(t),
        "ctx_pct": ctx_pct,
        "tmux": {
            k: live[k]
            for k in (
                "session",
                "window",
                "window_name",
                "pane_id",
                "pane_pid",
                "command",
            )
        }
        if live
        else None,
        "attach": f"tmux attach -t {live['session']}:{live['window']}"
        if live
        else None,
        "via": "hook",
    }
    annotate_done([row], fetch=False)  # hooks stay off the network
    rep = {
        "host": row["host"],
        "version": REPORT_VERSION,
        "generated": iso(t),
        "rows": [row],
        # A partial report: apply_report skips dead-marking and keeps the
        # host row's quota/exceptions_hash (we measured neither).
        "partial": True,
    }
    rep["exceptions"] = exceptions([row], 30.0)
    return rep


class PushError(Exception):
    pass


def _sse_json(body: str, rpc_id: int | None) -> dict | None:
    """The JSON-RPC message in an SSE body (matching `rpc_id` when given)."""
    found = None
    for line in body.splitlines():
        if not line.startswith("data:"):
            continue
        try:
            d = json.loads(line[5:].strip())
        except ValueError:
            continue
        if isinstance(d, dict) and (rpc_id is None or d.get("id") == rpc_id):
            found = d
    return found


def mcp_post(
    url: str,
    token: str | None,
    session: str | None,
    msg: dict,
    timeout: float = 30.0,
) -> tuple[dict | None, str | None]:
    """POST one JSON-RPC message; return (response message or None, session id)."""
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json, text/event-stream",
    }
    if token:
        headers["Authorization"] = f"Bearer {token}"
    if session:
        headers["Mcp-Session-Id"] = session
    req = urllib.request.Request(
        url, data=json.dumps(msg).encode("utf-8"), headers=headers, method="POST"
    )
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            sid = resp.headers.get("Mcp-Session-Id") or session
            ctype = resp.headers.get("Content-Type", "")
            body = resp.read().decode("utf-8", "replace")
    except urllib.error.HTTPError as e:
        raise PushError(f"HTTP {e.code} from {url}") from None
    except (urllib.error.URLError, OSError) as e:
        raise PushError(f"cannot reach {url}: {e}") from None
    if "id" not in msg or not body.strip():
        return None, sid
    if "text/event-stream" in ctype:
        return _sse_json(body, msg.get("id")), sid
    try:
        d = json.loads(body)
    except ValueError:
        raise PushError(f"non-JSON reply from {url}") from None
    return (d if isinstance(d, dict) else None), sid


def push_report(rep: dict, url: str, token: str | None, timeout: float = 30.0) -> str:
    """initialize -> notifications/initialized -> tools/call put. Returns the
    tool's text reply; raises PushError on any failure."""
    init, sid = mcp_post(
        url,
        token,
        None,
        timeout=timeout,
        msg={
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-03-26",
                "capabilities": {},
                "clientInfo": {"name": "fleet-report", "version": str(REPORT_VERSION)},
            },
        },
    )
    if not init or "error" in init:
        raise PushError(f"initialize failed: {(init or {}).get('error')}")
    mcp_post(
        url,
        token,
        sid,
        {"jsonrpc": "2.0", "method": "notifications/initialized"},
        timeout=timeout,
    )
    reply, _ = mcp_post(
        url,
        token,
        sid,
        timeout=timeout,
        msg={
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/call",
            "params": {
                "name": "put",
                "arguments": {
                    "kind": "fleet",
                    "mode": "report",
                    "args": {"host": rep["host"], "report": rep},
                },
            },
        },
    )
    if not reply:
        raise PushError("no reply to tools/call")
    if "error" in reply:
        raise PushError(f"put failed: {reply['error']}")
    result = reply.get("result") or {}
    text = " ".join(
        c.get("text", "") for c in result.get("content") or [] if isinstance(c, dict)
    ).strip()
    if result.get("isError") or text.startswith("[error"):
        raise PushError(f"put refused: {text[:300]}")
    return text


#: A remote report takes a few seconds; past this the host is treated as down.
REMOTE_TIMEOUT_S = 60


def _ssh(host: str, repo: str, src: IO[str]) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        [
            "ssh",
            "-o",
            "BatchMode=yes",
            "-o",
            "ConnectTimeout=10",
            "--",
            host,
            f"python3 - --json --repo {repo}",
        ],
        stdin=src,
        capture_output=True,
        text=True,
        encoding="utf-8",
        check=False,
        timeout=REMOTE_TIMEOUT_S,
    )


def remote_report(host: str, repo: str = "~/precis-mcp") -> dict:
    """Run this script on `host` over ssh (script on stdin) and parse its JSON."""
    with Path(__file__).open(encoding="utf-8") as src:
        try:
            r = _ssh(host, repo, src)
        except subprocess.TimeoutExpired:
            raise PushError(f"ssh {host}: no report in {REMOTE_TIMEOUT_S}s") from None
    if r.returncode != 0:
        raise PushError(f"ssh {host} exited {r.returncode}: {r.stderr.strip()[:200]}")
    try:
        rep = json.loads(r.stdout)
    except ValueError:
        raise PushError(f"ssh {host}: report is not JSON") from None
    if not isinstance(rep, dict) or not rep.get("host"):
        raise PushError(f"ssh {host}: report has no host")
    return rep


def mcp_token() -> str | None:
    token = os.environ.get("PRECIS_MCP_TOKEN")
    if not token:
        token_file = os.environ.get("PRECIS_MCP_TOKEN_FILE")
        if token_file:
            try:
                token = Path(token_file).read_text(encoding="utf-8").strip()
            except OSError:
                token = None
    return token


def hook_main(args: argparse.Namespace) -> int:
    """``--hook``: report just this session. Never breaks the session: push
    failures go to stderr and the exit code stays 0. ``PRECIS_FLEET_HOOK=0``
    disables the push entirely."""
    if os.environ.get("PRECIS_FLEET_HOOK", "").lower() in ("0", "off", "false"):
        return 0
    rep = hook_report(args.vendor, args.state, args.claude_window)
    if rep is None:
        print("fleet-report --hook: not inside a git worktree", file=sys.stderr)
        return 0
    if not args.push:
        print(json.dumps(rep, indent=2) if args.json else render(rep))
        return 0
    url = os.environ.get("PRECIS_FLEET_MCP_URL") or DEFAULT_MCP_URL
    try:
        push_report(rep, url, mcp_token(), timeout=8.0)
    except PushError as e:
        print(f"fleet-report --hook: {e}", file=sys.stderr)
    return 0


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="Read-only per-host agent session report.")
    ap.add_argument(
        "--repo",
        action="append",
        default=[],
        help="repo whose worktrees to list (repeatable)",
    )
    ap.add_argument("--json", action="store_true", help="rows + exceptions as JSON")
    ap.add_argument(
        "--quiet-min",
        type=float,
        default=30.0,
        help="quiet minutes before a working session is an exception",
    )
    ap.add_argument(
        "--since-min",
        type=float,
        default=240.0,
        help="ignore pane-less transcripts older than this",
    )
    ap.add_argument(
        "--claude-window",
        type=int,
        default=None,
        metavar="TOKENS",
        help="force this context window for every Claude session "
        "(default: the transcript's model id picks from "
        "CLAUDE_MODEL_WINDOWS, 200k when unlisted)",
    )
    ap.add_argument(
        "--push",
        action="store_true",
        help="send this host's report to the precis MCP (put kind='fleet')",
    )
    ap.add_argument(
        "--remote",
        action="append",
        default=[],
        metavar="HOST",
        help="run on HOST over ssh and push its report (repeatable)",
    )
    ap.add_argument(
        "--hook",
        action="store_true",
        help="session-hook mode: report only the calling session as a "
        "partial report (reads the hook payload on stdin; combine with "
        "--push to send it)",
    )
    ap.add_argument(
        "--vendor",
        default=None,
        help="vendor for --hook (default: the pane command, else the "
        "AI_AGENT/CLAUDECODE/CODEX_HOME environment)",
    )
    ap.add_argument(
        "--state",
        default=None,
        help="state for --hook (default: mapped from hook_event_name)",
    )
    args = ap.parse_args(argv)
    if args.hook:
        return hook_main(args)
    repos = args.repo or [str(Path(__file__).resolve().parent.parent)]
    if not args.push and not args.remote:
        rep = collect(repos, args.quiet_min, args.since_min, args.claude_window)
        print(json.dumps(rep, indent=2) if args.json else render(rep))
        return 0
    url = os.environ.get("PRECIS_FLEET_MCP_URL") or DEFAULT_MCP_URL
    token = mcp_token()
    status = 0

    def push(label: str, make) -> None:
        nonlocal status
        try:
            print(f"{label}: {push_report(make(), url, token)}")
        except PushError as e:
            print(f"{label}: {e}", file=sys.stderr)
            status = 1

    if args.push:
        push(
            "local",
            lambda: collect(repos, args.quiet_min, args.since_min, args.claude_window),
        )
    for host in args.remote:
        push(host, lambda host=host: remote_report(host))
    return status


if __name__ == "__main__":
    sys.exit(main())
