#!/usr/bin/env python3
"""Deterministic snapshot + state helper for ce-babysit-pr.

The agent (SKILL.md) owns judgment and mutations; this script owns the parts
prose cannot do reliably: a combined fetch of both event streams, atomic state
read/write under a file lock, and dedup keyed on remote truth.

The dedup model is claim -> act -> confirm. `snapshot` never marks an item
handled just because it observed it; an item stays actionable until either the
agent confirms it acted (`mark`) OR remote truth removes it (a resolved thread
drops out of the unresolved fetch). So if a resolve/debug pass crashes or fails
before the agent marks it, the item is still actionable on the next tick.

Subcommands:
  snapshot --pr N [--repo O/R] --state-dir DIR [--fetch-file F]
           Fetch (or load F), diff against on-disk state, persist the
           observed state atomically, emit the actionable set as JSON.
  mark     --state-dir DIR (--thread ID --disposition needs-human|dispatched
           | --comment ID --disposition needs-human|dispatched | --check KEY)
           Record that the agent acted on an item. A `dispatched` thread is
           re-emitted when a later reviewer comment moves its last-comment
           identity past the one we acted on (our own reply does not
           re-trigger); a `needs-human` thread stays parked until an explicit
           `--disposition open`. A non-thread feedback item never drops out of
           the fetch on its own, so `--comment` is the only way to silence a
           handled one. A new head SHA clears dispatched CI checks.

--fetch-file injects a pre-captured combined snapshot instead of calling gh,
so the diff logic is testable without a live PR.
"""
import argparse
import hashlib
import json
import os
import subprocess
import tempfile
import time
from contextlib import contextmanager
from datetime import datetime, timezone

# Conclusions that mean "this check needs attention" (failing states).
FAILING = {"FAILURE", "TIMED_OUT", "CANCELLED", "ACTION_REQUIRED", "STARTUP_FAILURE", "STALE"}

DISPOSITION_OPEN = "open"
DISPOSITION_NEEDS_HUMAN = "needs-human"
DISPOSITION_DISPATCHED = "dispatched"

# A present in-progress review signal (👀) blocks a merge-ready read — but never terminally: after
# this much quiet a stale/never-cleared signal is ignored so the PR can still be called ready.
REVIEW_INPROGRESS_MAX_WAIT = 900

# Pure CI/status bots whose top-level comments are never review feedback (coverage deltas,
# deploy previews). Code-review bots (Codex, CodeRabbit, Greptile, etc.) are deliberately NOT
# here — their comments ARE feedback and must reach ce-resolve-pr-feedback. Matched case-folded
# against both the bare login and its "[bot]" suffix form.
CI_BOT_LOGINS = {
    "github-actions", "codecov", "codecov-commenter", "netlify", "vercel",
}

# Trajectory check-history states (persisted, compared by ==).
CHECK_UNKNOWN = "unknown"
CHECK_CLEAR = "clear"
CHECK_FAILING = "failing"
# Trajectory single-stream activity labels.
STREAM_CI = "ci"
STREAM_REVIEW = "review"
# Keep a check's recurrence memory across a transient absence (a one-tick gap from a
# workflow-registration lag or a paths-filtered run), but bound growth: evict entries
# unseen for this many ticks.
CHECK_HISTORY_TTL = 30


def _now():
    return datetime.now(timezone.utc)


def _iso(dt):
    return dt.isoformat()


def _run(cmd):
    return subprocess.run(cmd, capture_output=True, text=True)


def _run_checked(cmd, label):
    r = _run(cmd)
    if r.returncode != 0:
        raise SystemExit(f"{label} failed: {r.stderr.strip()}")
    return r


def _split_repo(repo):
    """Parse "[HOST/]OWNER/NAME" into (owner, name), or (None, None). A host-qualified ref (gh's
    documented `[HOST/]OWNER/REPO` selector) drops the host — the last two segments are what the
    GraphQL lookup needs; treating the host as the owner would query a nonexistent repo on GHE."""
    if not repo:
        return None, None
    parts = repo.strip("/").split("/")
    if len(parts) >= 2:
        return parts[-2], parts[-1]
    return None, None


def _resolve_repo_ref(repo, url):
    """Resolve (owner, name, host) from --repo + the PR url, else one `gh repo view` call.
    The host is parsed from the url and threaded into every `gh api` call so a GitHub Enterprise
    PR queries the right host — without it, `gh api` defaults to github.com and a GHE babysitter
    fetches the PR via `gh pr view` but then reads review threads / workflow runs from github.com.
    Parsing the url (already fetched by `fetch`) also avoids a redundant `gh repo view` per tick."""
    owner, name = _split_repo(repo)
    host = None
    if url:
        # https://<host>/OWNER/NAME/pull/N
        parts = url.rstrip("/").split("/")
        if len(parts) >= 5 and parts[0].startswith("http"):
            host = parts[2]
            if not owner:
                owner, name = parts[-4], parts[-3]
    if not owner:
        r = _run(["gh", "repo", "view", "--json", "owner,name"])
        if r.returncode == 0:
            info = json.loads(r.stdout)
            owner, name = info.get("owner", {}).get("login"), info.get("name")
    if not owner or not name:
        raise SystemExit("could not resolve owner/repo; pass --repo OWNER/REPO")
    return owner, name, host


def _host_args(host):
    """`gh api --hostname` selector so GHE calls hit the PR's host, not the default github.com."""
    return ["--hostname", host] if host else []


def _thread_identity(t):
    """The remote-truth identity of a thread's latest state."""
    return (t.get("last_comment_id"), t.get("last_comment_at"))


def fetch(pr, repo):
    """Fetch both event streams via gh into one combined snapshot dict."""
    repo_args = ["--repo", repo] if repo else []
    view = _run_checked(["gh", "pr", "view", str(pr), *repo_args, "--json",
                         "state,mergeable,mergeStateStatus,reviewDecision,headRefOid,url,number,"
                         "statusCheckRollup,author,comments,reviews,reactionGroups"],
                        "gh pr view")
    v = json.loads(view.stdout)

    checks = []
    for c in v.get("statusCheckRollup") or []:
        # CheckRun entries carry name/status/conclusion/workflowName/detailsUrl.
        # StatusContext (legacy commit statuses) carry context/state/targetUrl.
        if c.get("__typename") == "StatusContext":
            name = c.get("context") or "status"
            state = (c.get("state") or "").upper()
            checks.append({
                "key": name, "name": name,
                "status": "COMPLETED" if state in ("SUCCESS", "FAILURE", "ERROR") else "IN_PROGRESS",
                "conclusion": {"ERROR": "FAILURE"}.get(state, state) or None,
                "details_url": c.get("targetUrl"),
            })
        else:
            wf = c.get("workflowName")
            name = c.get("name") or "check"
            checks.append({
                "key": f"{wf}/{name}" if wf else name, "name": name,
                "status": (c.get("status") or "").upper() or "IN_PROGRESS",
                "conclusion": (c.get("conclusion") or None) and c["conclusion"].upper(),
                "details_url": c.get("detailsUrl"),
            })

    # In-progress review signal: an 👀 (EYES) reaction on the PR body is how several review bots
    # (Codex among them) announce a review is *underway* — a present in-progress signal means the PR
    # is NOT settled, regardless of quiet time. A present signal is meaningful; absence tells us
    # nothing (many reviewers give no signal), so this only ever *delays* a merge-ready read.
    review_in_progress = any(
        g.get("content") == "EYES" and (g.get("users") or {}).get("totalCount", 0) > 0
        for g in v.get("reactionGroups") or [])

    owner, name, host = _resolve_repo_ref(repo, v.get("url"))
    head = v.get("headRefOid")
    return {
        "pr_state": v.get("state"),
        "mergeable": v.get("mergeable"),
        "merge_state_status": v.get("mergeStateStatus"),
        "review_decision": v.get("reviewDecision") or None,
        "head_sha": head,
        "url": v.get("url"),
        "checks": checks,
        "review_in_progress": review_in_progress,
        "threads": fetch_threads(pr, owner, name, host),
        # Non-thread feedback: top-level PR comments + review submission bodies. ce-resolve-pr-feedback
        # handles these too, so a Changes-Requested review body or an actionable top-level comment with
        # NO inline thread must not be invisible to the loop. Content-actionability is the resolver's
        # judgment; here we just surface non-empty, non-author, non-CI-bot items so the loop invokes it.
        "feedback": _extract_feedback(v),
        "awaiting_approval": fetch_awaiting_approval(owner, name, head, host),
    }


def _is_ci_bot(login):
    """A pure CI/status bot login, tolerant of the "[bot]" suffix gh reports."""
    return login.lower().removesuffix("[bot]") in CI_BOT_LOGINS


def _body_hash(body):
    """Edit identity for a top-level comment / review body: `gh pr view --json` exposes no
    updatedAt, so hash the body. When a dispatched wrapper is later edited to add an actionable
    request, this changes and the item reactivates (unlike a thread, our reply is a separate
    top-level comment and never edits the original, so there is no self-reply retrigger)."""
    return hashlib.sha1((body or "").encode("utf-8")).hexdigest()[:16]


def _extract_feedback(v):
    """Top-level PR comments + non-empty review bodies, excluding the PR author and CI/status
    bots. `gh pr view --json comments,reviews` returns flat arrays (not GraphQL {nodes})."""
    author = (v.get("author") or {}).get("login")
    out = []
    for c in v.get("comments") or []:
        a = (c.get("author") or {}).get("login")
        if a and a != author and not _is_ci_bot(a) and (c.get("body") or "").strip():
            out.append({"id": c.get("id"), "kind": "comment", "author": a, "edit_id": _body_hash(c.get("body"))})
    for r in v.get("reviews") or []:
        a = (r.get("author") or {}).get("login")
        if a and a != author and not _is_ci_bot(a) and (r.get("body") or "").strip():
            out.append({"id": r.get("id"), "kind": "review", "author": a, "state": r.get("state"),
                        "edit_id": _body_hash(r.get("body"))})
    return out


def fetch_awaiting_approval(owner, name, head, host=None):
    """Count Actions workflow runs on this head that are awaiting maintainer approval —
    the fork-PR security gate. Such a run has created NO check-run yet, so it is invisible
    to statusCheckRollup; without this, a fork PR blocked on approval reads as 'all checks ok'.
    Best-effort: any API/permission failure returns 0 rather than failing the tick."""
    if not head:
        return 0
    r = _run(["gh", "api", *_host_args(host),
              f"repos/{owner}/{name}/actions/runs?head_sha={head}&per_page=50",
              "--jq", '[.workflow_runs[] | select(.status==\"action_required\" or .status==\"waiting\" '
                      'or .conclusion==\"action_required\")] | length'])
    if r.returncode != 0:
        return 0
    try:
        return int((r.stdout or "").strip() or 0)
    except ValueError:
        return 0


def fetch_threads(pr, owner, name, host=None):
    """Unresolved review threads with their last-comment identity."""
    query = """
query($owner:String!,$repo:String!,$pr:Int!,$cursor:String){
  repository(owner:$owner,name:$repo){ pullRequest(number:$pr){
    reviewThreads(first:100,after:$cursor){
      nodes{ id isResolved path line
        comments(last:100){ nodes{ id createdAt lastEditedAt } } }
      pageInfo{ hasNextPage endCursor } } } } }"""
    out = []
    cursor = None
    while True:
        args = ["gh", "api", "graphql", *_host_args(host), "-f", f"owner={owner}", "-f", f"repo={name}",
                "-F", f"pr={pr}", "-f", f"query={query}"]
        if cursor:
            args += ["-f", f"cursor={cursor}"]
        r = _run_checked(args, "gh api graphql")
        data = json.loads(r.stdout)["data"]["repository"]["pullRequest"]["reviewThreads"]
        for n in data["nodes"]:
            if n.get("isResolved"):
                continue
            cs = n.get("comments", {}).get("nodes") or []
            last = cs[-1] if cs else {}
            # `last_comment_at` is the reactivation signal — the MAX edit/create time across every
            # comment in the thread, not just the last one, so a reviewer editing an *earlier* comment
            # (their original request) after the agent's reply still moves the identity and re-opens it.
            # Bounded to the last 100 comments (see the query): review threads are ~never that long; an
            # edit to a comment outside that window would be missed (acceptable vs paginating per thread).
            edit_at = max((c.get("lastEditedAt") or c.get("createdAt") or "" for c in cs), default="")
            out.append({
                "thread_id": n["id"],
                "last_comment_id": last.get("id"),
                "last_comment_at": edit_at or last.get("lastEditedAt") or last.get("createdAt"),
                "path": n.get("path"),
                "line": n.get("line"),
            })
        if not data["pageInfo"]["hasNextPage"]:
            break
        cursor = data["pageInfo"]["endCursor"]
    return out


def _empty_state(pr, repo, url, now):
    owner, name = _split_repo(repo)
    return {
        "pr": {"owner": owner, "repo": name, "number": pr, "url": url},
        "head_sha": None, "tick": 0, "started_at": _iso(now),
        "checks": {}, "threads": {}, "ci_dispatched": {},
        "review_decision": None, "mergeable": None, "merge_state_status": None,
        "last_change_at": None, "last_action": None, "stop_reason": None,
        "trajectory": _empty_trajectory(),
    }


def _empty_trajectory():
    """Deterministic cross-tick facts babysit hands the leaves (facts, not judgment):
    babysit ships the trajectory, the leaf decides whether it means non-convergence."""
    return {
        "check_history": {},           # check key -> {state, last_head, recur, seen_tick}
        "seen_threads": {},            # unresolved thread id -> first-seen tick
        "unresolved_series": [],       # unresolved-thread count per tick (last 6)
        "stream_series": [],           # single-stream activity per tick (last 8)
        "problem_keys": [],            # last tick's failing checks + non-parked threads (progress detection)
        "min_open_problems": None,     # lowest total open-problem count seen
        "heads_since_progress": 0,     # head changes since progress (a new low OR something cleared)
        "last_head": None,             # head as of the last AGENT tick — hsp counts moves between ticks,
                                       # NOT poll-observed head moves (state["head_sha"] advances on polls)
    }


def _load_trajectory(state):
    """Load the trajectory, tolerating a partial or non-dict value from an older on-disk
    state.json (persisted in /tmp across script versions) — backfill missing keys so a new
    field never KeyErrors an old state, and a null/garbage value never crashes."""
    tj = state.get("trajectory")
    if not isinstance(tj, dict):
        tj = {}
    for key, value in _empty_trajectory().items():
        tj.setdefault(key, value)
    state["trajectory"] = tj
    return tj


def _push_bounded(lst, item, cap):
    """Append to a sliding window that keeps only the last `cap` items."""
    lst.append(item)
    del lst[:-cap]


def _stream_alternations(series):
    """Count flips between ci-active and review-active ticks — the cross-stream churn signal."""
    flips = 0
    prev = None
    for s in series:
        if prev is not None and s != prev:
            flips += 1
        prev = s
    return flips


def _trend(series):
    if len(series) < 3:
        return "flat"
    if series[-1] > series[0]:
        return "rising"
    if series[-1] < series[0]:
        return "falling"
    return "flat"


def _record_check_history(state, head, new_checks):
    """Record CI fail->clear->fail recurrence transitions. Runs on EVERY snapshot — agent ticks AND
    watch polls — so a CLEAR (or FAIL) observed only between agent ticks is not lost, which would
    otherwise make the ping-pong recurrence trigger under-fire under the self-sustaining watch.
    Idempotent per transition: once a check is FAILING, re-observing FAIL does not re-increment."""
    tj = _load_trajectory(state)
    tick = state.get("tick", 0)
    hist = tj["check_history"]
    for key, c in new_checks.items():
        h = hist.setdefault(key, {"state": CHECK_UNKNOWN, "last_head": None, "recur": 0, "seen_tick": tick})
        h["seen_tick"] = tick
        if c["conclusion"] in FAILING:
            if h["state"] == CHECK_CLEAR and h["last_head"] != head:  # fail after a clear on a new head
                h["recur"] += 1
            h["state"] = CHECK_FAILING
            h["last_head"] = head
        elif c["status"] == "COMPLETED":  # observed non-failing terminal — a genuine clear
            h["state"] = CHECK_CLEAR
            h["last_head"] = head
        # IN_PROGRESS/QUEUED: leave prior state untouched (not yet a clear)
    # Evict entries unseen for TTL agent-ticks (polls don't advance the tick) — bounds growth without
    # erasing a check that was briefly absent (a one-tick gap must not lose real recurrence history).
    tj["check_history"] = {k: v for k, v in hist.items() if tick - v.get("seen_tick", tick) <= CHECK_HISTORY_TTL}


def _update_trajectory(state, head, new_checks, new_threads, new_feedback, actionable):
    """Maintain and emit the deterministic trajectory. Coarse by design: check-name-level
    recurrence, backlog trend, cross-stream alternation, no-progress heads. Fine, invariant-
    level judgment (log signatures, nit root-clustering) is the leaf's job, not this script's.
    `actionable` is the `{ci, threads, comments}` set diff() also returns. Non-thread feedback
    (top-level comments + review bodies) counts as review-stream activity and as an open problem
    for the stall signal, but the thread-named backlog fields stay scoped to inline threads."""
    tj = _load_trajectory(state)

    # --- CI recurrence (fail->clear->fail on a *different* head): recorded on EVERY observation
    # (agent ticks AND watch polls, via _record_check_history in diff) so a clear seen only between
    # agent ticks is not lost. Here we just READ the accumulated history for the trigger. recur_max
    # reflects only checks present this tick, so a stale key can't keep it elevated. ---
    hist = tj["check_history"]
    recur_max = max((hist[k]["recur"] for k in new_checks if k in hist), default=0)
    recurring = [{"key": k, "recur": hist[k]["recur"]} for k in new_checks if k in hist and hist[k]["recur"] > 0]

    # --- Review-thread backlog: trend of unresolved-thread count + genuinely new threads this
    # tick. Scoped to inline threads (non-thread feedback is captured in the total-problem stall
    # signal below), so these thread-named fields stay accurate. ---
    seen = tj.get("seen_threads", {})
    new_arrivals = [tid for tid in new_threads if tid not in seen]
    tj["seen_threads"] = {tid: seen.get(tid, state.get("tick", 0)) for tid in new_threads}
    _push_bounded(tj["unresolved_series"], len(new_threads), 6)

    # --- Cross-stream churn: alternation between ci-only and review-only active ticks.
    # Review is active when either threads OR non-thread feedback is actionable. ---
    review_active = bool(actionable["threads"] or actionable.get("comments"))
    if actionable["ci"] and not review_active:
        active = STREAM_CI
    elif review_active and not actionable["ci"]:
        active = STREAM_REVIEW
    else:
        active = None  # both or neither — not a single-stream tick, don't record
    if active:
        _push_bounded(tj["stream_series"], active, 8)

    # --- No-progress heads: measured from TOTAL open problems (failing checks + non-parked
    # unresolved threads + non-parked non-thread feedback), NOT the post-claim `actionable` set —
    # marking items dispatched shrinks
    # actionable and would fake progress. Reset the counter when the head moves and either the
    # total set a new low OR a previously-failing item cleared: progressive migration (A cleared
    # while B appears) is progress, not a stall, so it must not accrue heads_since_progress. ---
    # Only genuinely-OPEN items are unresolved work: a `dispatched` item is handled (a top-level
    # comment never drops out of the fetch, so counting it would keep heads_since_progress climbing
    # forever and falsely trip non-convergence on unrelated later work), and `needs-human` is parked.
    problem_keys = {f"c:{k}" for k, c in new_checks.items() if c["conclusion"] in FAILING}
    problem_keys |= {f"t:{tid}" for tid, t in new_threads.items() if t.get("disposition") == DISPOSITION_OPEN}
    problem_keys |= {f"m:{fid}" for fid, f in new_feedback.items() if f.get("disposition") == DISPOSITION_OPEN}
    cleared_something = bool(set(tj.get("problem_keys", [])) - problem_keys)
    open_problems = len(problem_keys)
    minp = tj.get("min_open_problems")
    new_low = minp is None or open_problems < minp
    if new_low:
        tj["min_open_problems"] = open_problems
    # heads_since_progress counts head moves BETWEEN AGENT TICKS (tj["last_head"]), not poll-observed
    # moves — a watch poll advances state["head_sha"], so a plain head_changed would read False at the
    # agent's tick and starve this stall signal under the default self-sustaining watch.
    traj_head_moved = tj.get("last_head") is not None and head != tj.get("last_head")
    if new_low or cleared_something:
        tj["heads_since_progress"] = 0
    elif traj_head_moved:
        tj["heads_since_progress"] = tj.get("heads_since_progress", 0) + 1
    tj["problem_keys"] = sorted(problem_keys)
    tj["last_head"] = head

    return {
        "recurring_checks": recurring,
        "check_recur_max": recur_max,
        "unresolved_threads": len(new_threads),
        "unresolved_series": list(tj["unresolved_series"]),
        "unresolved_trend": _trend(tj["unresolved_series"]),
        "new_threads_this_tick": len(new_arrivals),
        "stream_alternations": _stream_alternations(tj["stream_series"]),
        "heads_since_progress": tj["heads_since_progress"],
    }


def _apply_dispositions(items, id_key, prior, identity_fn=None):
    """Claim->act->confirm dedup for a review stream: an item stays actionable until `mark`
    records a non-open disposition (persisted in prior[id]['disposition']). Returns
    (persisted_by_id, actionable_list, open_needs_human_count).

    When identity_fn is given, a `dispatched` **or** `needs-human` item is *reactivated* — set back
    to open and re-actionized — once its last-comment / edit identity moves past the one we acted on.
    That acted-on identity is captured lazily on the first tick we observe the item parked, which is
    *after* our own reply (a fix acknowledgment, or a needs-human `decision_context` reply) has
    already landed in the fetch — so our reply becomes the baseline and does not re-trigger, while a
    genuine later comment does. For `needs-human` this is exactly how a **human answering the parked
    question** (a reviewer reply, or an edit of the top-level comment) re-opens it and wakes the loop,
    instead of it sitting parked forever; an explicit `mark --disposition open` still forces it too.
    This is also what stops a dispatched-but-unresolved thread with fresh reviewer activity from being
    silenced out of counts.threads and letting the merge-ready gate call the PR ready. An item with no
    identity_fn never auto-reactivates (a brand-new top-level comment is just a new id)."""
    persisted, actionable, needs_human = {}, [], 0
    for it in items:
        iid = it.get(id_key)
        if not iid:
            continue
        pri = prior.get(iid, {})
        disposition = pri.get("disposition", DISPOSITION_OPEN)
        acted_identity = pri.get("acted_identity")
        if disposition in (DISPOSITION_DISPATCHED, DISPOSITION_NEEDS_HUMAN) and identity_fn is not None:
            current_identity = identity_fn(it)
            if acted_identity is None:
                acted_identity = current_identity   # first post-action observation: adopt as baseline
            elif current_identity != acted_identity:
                disposition = DISPOSITION_OPEN       # a later human/reviewer reply past our baseline -> reactivate
                acted_identity = None
        if disposition == DISPOSITION_OPEN:
            actionable.append(it)
        elif disposition == DISPOSITION_NEEDS_HUMAN:
            needs_human += 1
        rec = {**it, "disposition": disposition}
        if acted_identity is not None:
            rec["acted_identity"] = acted_identity
        persisted[iid] = rec
    return persisted, actionable, needs_human


def diff(state, cur, now=None, advance_trajectory=True):
    """Pure: given prior state + current snapshot, compute the actionable set
    and the persisted observed state. `now` is injectable for tests."""
    now = now or _now()
    # A transient null/empty head (a gh hiccup) falls back to the last known head,
    # so a momentary null does not look like a new head and wipe ci_dispatched.
    head = cur["head_sha"] or state.get("head_sha")
    head_changed = state.get("head_sha") is not None and head != state["head_sha"]

    prior_threads = state.get("threads", {})
    prior_feedback = state.get("feedback", {})
    prior_change_sig = _change_sig(state)

    if head_changed:
        # SHA-scoped state is meaningless on a new head.
        state["ci_dispatched"] = {}

    # --- CI: a failing check on the current head is actionable until the agent
    # marks it dispatched (recorded in ci_dispatched[head]) or the head moves.
    # `checks_terminal` = every check has finished (none IN_PROGRESS/QUEUED).
    # Duplicate check keys (same workflow/name) are disambiguated with a #n suffix
    # so one never shadows another and silently drops a failing check. ---
    dispatched = set(state.get("ci_dispatched", {}).get(head, []))
    new_checks = {}
    actionable_ci = []
    has_failing = False
    checks_terminal = True
    seen_keys = {}
    for c in cur["checks"]:
        key = c["key"]
        if key in seen_keys:
            seen_keys[key] += 1
            key = f"{key}#{seen_keys[key]}"
        else:
            seen_keys[key] = 0
        new_checks[key] = {"name": c["name"], "status": c["status"],
                           "conclusion": c["conclusion"], "head_sha": head}
        if c["status"] != "COMPLETED":
            checks_terminal = False
        if c["conclusion"] in FAILING:
            has_failing = True
            if key not in dispatched:
                actionable_ci.append({"key": key, "name": c["name"],
                                      "conclusion": c["conclusion"], "details_url": c["details_url"]})

    # Both review streams share claim->act->confirm dedup (see _apply_dispositions), differing
    # in how an item leaves the fetch and whether new activity re-opens it:
    #   - Threads: a resolved thread drops out of the unresolved fetch. A `dispatched` thread that
    #     is still unresolved is reactivated once its last-comment identity moves past the one we
    #     acted on (acted_identity) — so a reviewer re-engaging is not silently dropped, yet the
    #     acting loop's own reply (captured as the baseline on first observation) does not
    #     re-trigger. A `needs-human` thread stays parked (open_needs_human keeps merge-ready from
    #     firing) until a *human* answers it — a later reply/edit past that same baseline reopens and
    #     wakes it — or an explicit `mark --disposition open` forces it.
    #   - Feedback (top-level comments + review bodies): no remote "resolve" exists, so an item
    #     never drops out on its own — `mark --comment <id>` is the only thing that silences it. A
    #     dispatched item reactivates only when its own body is *edited* (edit_id changes) to add a
    #     new request; a brand-new comment is just a new id. Our reply is a separate top-level
    #     comment, never an edit of the original, so it never retriggers.
    # Either stream's open needs-human items feed open_needs_human so the merge-ready gate
    # refuses to call the PR ready while a human decision is still pending.
    new_threads, actionable_threads, human_threads = _apply_dispositions(
        cur["threads"], "thread_id", prior_threads,
        identity_fn=lambda t: [t.get("last_comment_id"), t.get("last_comment_at")])
    new_feedback, actionable_feedback, human_feedback = _apply_dispositions(
        cur.get("feedback") or [], "id", prior_feedback, identity_fn=lambda c: [c.get("edit_id")])
    open_needs_human = human_threads + human_feedback
    # Identities of the currently-parked needs-human items, so the watch can tell an already-
    # surfaced residual (do not re-wake) from a newly-arrived one (wake) — a parked human decision
    # must not busy-wake the loop or terminate it.
    needs_human_ids = sorted(
        [k for k, r in new_threads.items() if r.get("disposition") == DISPOSITION_NEEDS_HUMAN]
        + [k for k, r in new_feedback.items() if r.get("disposition") == DISPOSITION_NEEDS_HUMAN])

    state["head_sha"] = head or state.get("head_sha")
    state["checks"] = new_checks
    state["threads"] = new_threads
    state["feedback"] = new_feedback
    state["review_decision"] = cur["review_decision"]
    state["mergeable"] = cur["mergeable"]
    state["merge_state_status"] = cur["merge_state_status"]
    state["review_in_progress"] = cur.get("review_in_progress", False)
    state["awaiting_approval"] = cur.get("awaiting_approval", 0)

    actionable = {"ci": actionable_ci, "threads": actionable_threads, "comments": actionable_feedback}
    if advance_trajectory:
        state["tick"] = state.get("tick", 0) + 1
    # Record CI recurrence transitions on EVERY observation (polls too) so a fail->clear->fail seen
    # only between agent ticks is not lost. head_sha for the check-level last_head is the observed
    # head; the trajectory-level last_head (for heads_since_progress) is agent-tick-only, inside
    # _update_trajectory.
    _record_check_history(state, head, new_checks)
    if advance_trajectory:
        trajectory = _update_trajectory(state, head, new_checks, new_threads, new_feedback, actionable)
    else:
        # A watch poll detects change and advances the settle clock, but must NOT roll the rest of the
        # trajectory (tick counter, seen_threads, unresolved_series, heads_since_progress). Advancing
        # it would consume new_threads_this_tick — the waking poll marks the just-arrived thread
        # "seen", so the agent's real tick reads 0 new arrivals and the review-bot-treadmill
        # non-convergence trigger never fires. Only the agent's tick (advance_trajectory=True) rolls it.
        trajectory = {}

    # --- Settle window: any observable movement resets the quiet clock. ---
    changed_this_tick = head_changed or _change_sig(state) != prior_change_sig or state.get("last_change_at") is None
    if changed_this_tick:
        state["last_change_at"] = _iso(now)
    quiet_seconds = _elapsed(state.get("last_change_at"), now)

    # Workflow runs awaiting maintainer approval (fork-PR gate) create no check-run, so they are
    # invisible to the rollup above — surface them, and never call CI "ok" while the real CI is
    # gated. blocked_external = the loop cannot progress (no failing check to fix, but CI can't run)
    # and no one in this loop can unblock it — it is up to a maintainer of the base repo.
    awaiting_approval = state["awaiting_approval"]
    # "OK" requires every check terminal, none failing, AND none gated on approval. A still-
    # IN_PROGRESS or awaiting-approval check is neither ok nor failing — do not exit green.
    all_checks_ok = checks_terminal and not has_failing and bool(cur["checks"]) and awaiting_approval == 0
    blocked_external = (awaiting_approval > 0 and not has_failing and checks_terminal
                        and not actionable_threads and not actionable_feedback)

    return {
        "pr_state": cur["pr_state"],
        "mergeable": cur["mergeable"],
        "merge_state_status": cur["merge_state_status"],
        "review_decision": cur["review_decision"],
        "head_sha": head,
        "head_changed": head_changed,
        "url": cur["url"],
        "has_failing_checks": has_failing,
        "checks_terminal": checks_terminal,
        "checks_present": bool(cur["checks"]),
        "all_checks_ok": all_checks_ok,
        "review_in_progress": cur.get("review_in_progress", False),
        "checks_awaiting_approval": awaiting_approval,
        "blocked_external": blocked_external,
        "open_needs_human": open_needs_human,
        "needs_human_ids": needs_human_ids,
        "actionable": {"threads": actionable_threads, "ci": actionable_ci, "comments": actionable_feedback},
        "counts": {"threads": len(actionable_threads), "ci": len(actionable_ci),
                   "comments": len(actionable_feedback), "needs_human": open_needs_human},
        "changed_this_tick": changed_this_tick,
        "quiet_seconds": quiet_seconds,
        "session_seconds": _elapsed(state.get("started_at"), now),
        "tick": state["tick"],
        "trajectory": trajectory,
    }, state


def _change_sig(state):
    """Everything whose movement should reset the settle clock."""
    checks = {k: (v.get("status"), v.get("conclusion")) for k, v in state.get("checks", {}).items()}
    threads = {tid: _thread_identity(v) for tid, v in state.get("threads", {}).items()}
    feedback = {fid: v.get("disposition") for fid, v in state.get("feedback", {}).items()}
    return (checks, threads, feedback, state.get("review_decision"), state.get("mergeable"),
            state.get("merge_state_status"), state.get("review_in_progress"),
            # awaiting-approval clearing is movement: a fork gate lifting must reset the settle clock
            # so merge-ready waits for the now-imminent check-runs instead of firing on an empty rollup.
            bool(state.get("awaiting_approval")))


def _elapsed(iso_str, now):
    try:
        return int((now - datetime.fromisoformat(iso_str)).total_seconds())
    except (ValueError, TypeError):
        return 0


@contextmanager
def locked_state(state_dir, pr, repo, now):
    import fcntl
    os.makedirs(state_dir, exist_ok=True)
    lock_path = os.path.join(state_dir, "lock")
    state_path = os.path.join(state_dir, "state.json")
    with open(lock_path, "w") as lf:
        fcntl.flock(lf, fcntl.LOCK_EX)
        if os.path.exists(state_path):
            with open(state_path) as f:
                state = json.load(f)
        else:
            state = _empty_state(pr, repo, None, now)
        box = {"state": state}
        yield box
        tmp = tempfile.NamedTemporaryFile("w", dir=state_dir, delete=False)
        json.dump(box["state"], tmp, indent=2)
        tmp.flush()
        os.fsync(tmp.fileno())
        tmp.close()
        os.replace(tmp.name, state_path)
        fcntl.flock(lf, fcntl.LOCK_UN)


def _run_snapshot(args, now, advance_trajectory=True):
    """One fetch -> diff -> persist. Returns the actionable/state dict."""
    if args.fetch_file:
        with open(args.fetch_file) as f:
            cur = json.load(f)
    else:
        cur = fetch(args.pr, args.repo)
    with locked_state(args.state_dir, args.pr, args.repo, now) as box:
        if box["state"].get("pr", {}).get("url") is None and cur.get("url"):
            box["state"]["pr"]["url"] = cur["url"]
        if getattr(args, "reset_session", False):
            # A new babysit session (fresh invocation / watch arm) starts the budget clock now. Without
            # this, `started_at` persists from state creation, so resuming against day-old state would
            # read session_seconds as huge and hand back as "budget exhausted" before doing any work.
            box["state"]["started_at"] = _iso(now)
        actionable, box["state"] = diff(box["state"], cur, now, advance_trajectory=advance_trajectory)
    return actionable


def cmd_snapshot(args):
    print(json.dumps(_run_snapshot(args, _now()), indent=2))


def _wake_reason(a, settle_seconds):
    """Why the in-session agent should wake and run a tick, or None to keep watching.
    Ordered by precedence — a terminal/blocked/needs-human state outranks a merge-ready read."""
    if a.get("pr_state") in ("MERGED", "CLOSED"):
        return "terminal"
    if a.get("blocked_external"):
        return "blocked-external"
    c = a.get("counts") or {}
    if c.get("threads", 0) or c.get("comments", 0) or c.get("ci", 0):
        return "actionable"
    if a.get("open_needs_human", 0):
        return "needs-human"         # parked items block ready; surface them
    if a.get("has_failing_checks") and a.get("checks_terminal"):
        # a dispatched check left terminally red (counts.ci is 0 — nothing new to dispatch) is a
        # blocker to hand back, not a reason to idle to max-runtime.
        return "blocked-failing"
    # merge-ready candidate: green + settled AND no review in flight. An 👀-style in-progress signal
    # means a review is underway, so the PR is NOT ready no matter how long it has been quiet — time
    # is one component of "settled", not the gate. Honor the signal only up to a generous bound so a
    # stale reaction that never clears cannot block ready forever (absence never blocks; presence
    # only delays). Only wake once quiet has actually elapsed, so a stable green PR is not roused
    # every poll while it cools off.
    review_blocking = a.get("review_in_progress") and a.get("quiet_seconds", 0) < REVIEW_INPROGRESS_MAX_WAIT
    # Interactive merge-ready does NOT require `all_checks_ok`'s "at least one observed check": a repo
    # with no configured checks has a CLEAN/MERGEABLE PR that should be callable ready. (That guard
    # stays in pipeline success, where a not-yet-created rollup must not read as a pass.)
    if (a.get("mergeable") == "MERGEABLE" and a.get("merge_state_status") == "CLEAN"
            and a.get("checks_terminal") and not a.get("has_failing_checks")
            and a.get("checks_awaiting_approval", 0) == 0
            and not review_blocking
            and a.get("quiet_seconds", 0) >= settle_seconds):
        return "merge-ready"
    return None


def _emit_wake(reason, **fields):
    print(json.dumps({"event": "BABYSIT_WAKE", "reason": reason, **fields}), flush=True)


def _blocker_sig(a):
    """Identity of blockers the agent surfaces once but cannot self-clear — parked needs-human
    items, a dispatched terminally-red check, and a fork workflow awaiting maintainer approval. The
    watch captures this at arm time and does not re-wake on a blocker already in that baseline, so a
    parked residual keeps the watch alive (or the blocked-external bounded watch keeps polling for
    the gate to clear) instead of busy-waking or terminating on the same condition."""
    sig = set(a.get("needs_human_ids") or [])
    if a.get("has_failing_checks") and a.get("checks_terminal") and not (a.get("counts") or {}).get("ci"):
        sig.add("__terminal_red__")
    if a.get("blocked_external"):
        sig.add("__blocked_external__")
    # A new head is "context materially changed": a human may have pushed a commit that answers or
    # supersedes a parked residual, so the head is part of the baseline — when it moves, the residual
    # is no longer "already-surfaced against this state" and the watch wakes to give the agent a tick
    # to reopen/reprocess it, instead of parking forever while it still blocks merge-ready.
    if sig:
        sig.add(("head", a.get("head_sha")))
    return frozenset(sig)


def cmd_watch(args):
    """Deterministic background change-detector (no agent tokens between changes): poll on an
    interval, print one wake sentinel line and exit when there is something for the agent to do
    (an actionable change or a stop condition), or exit on the stop-signal file / max-runtime.
    A residual already present at arm time (a parked needs-human, or a dispatched terminal-red the
    agent already handed back) does NOT re-wake the loop — it keeps watching the other streams;
    only a *new* blocker (signature grown past the baseline) wakes."""
    deadline = (time.monotonic() + args.max_runtime) if args.max_runtime else None
    armed = None  # baseline blocker signature captured on the first poll (already-surfaced residuals)
    args.reset_session = True   # arming a watch is a fresh session -> start the budget clock now
    while True:
        if args.stop_file and os.path.exists(args.stop_file):
            _emit_wake("stop-signal")
            return
        actionable = _run_snapshot(args, _now(), advance_trajectory=False)  # a poll must not roll the trajectory
        args.reset_session = False   # ...but later polls are ticks of that same session, not new ones
        if armed is None:
            armed = _blocker_sig(actionable)
        reason = _wake_reason(actionable, args.settle_seconds)
        if reason in ("needs-human", "blocked-failing", "blocked-external") and _blocker_sig(actionable) <= armed:
            reason = None   # already-surfaced residual — keep watching, do not re-wake or terminate
        if reason:
            _emit_wake(reason, url=actionable.get("url"),
                       pr_state=actionable.get("pr_state"), counts=actionable.get("counts"))
            return
        if deadline is not None and time.monotonic() >= deadline:
            _emit_wake("max-runtime", url=actionable.get("url"))
            return
        time.sleep(args.interval)


def _mark_thread_baseline(item_id, args, state):
    """Best-effort current last-comment identity of a thread, captured at mark time so our just-posted
    reply becomes the acted baseline directly. Without this the baseline is adopted lazily on the next
    snapshot, so a reviewer reply that races in between the reply and that snapshot would be adopted as
    the baseline and silently swallowed instead of reactivating the thread. A fetch hiccup falls back
    to the lazy baseline (acted_identity stays unset)."""
    try:
        if getattr(args, "fetch_file", None):
            threads = json.load(open(args.fetch_file)).get("threads", [])
        else:
            owner, name, host = _resolve_repo_ref(args.repo, (state.get("pr") or {}).get("url"))
            threads = fetch_threads(args.pr, owner, name, host)
        for t in threads:
            if t.get("thread_id") == item_id:
                return [t.get("last_comment_id"), t.get("last_comment_at")]
    except (SystemExit, Exception):  # SystemExit (from _run_checked) is not an Exception subclass
        pass
    return None


def cmd_mark(args):
    now = _now()
    with locked_state(args.state_dir, args.pr, args.repo, now) as box:
        state = box["state"]
        if args.check:
            head = state.get("head_sha")
            if not head:
                raise SystemExit("mark --check requires a prior snapshot (state has no head_sha)")
            state.setdefault("ci_dispatched", {}).setdefault(head, [])
            if args.check not in state["ci_dispatched"][head]:
                state["ci_dispatched"][head].append(args.check)
            state["last_action"] = f"dispatched check {args.check}"
        elif args.thread or args.comment:
            item_id = args.thread or args.comment
            collection, id_field, label = (
                ("threads", "thread_id", "thread") if args.thread else ("feedback", "id", "comment"))
            entry = state.setdefault(collection, {}).setdefault(item_id, {id_field: item_id})
            entry["disposition"] = args.disposition
            if args.disposition == DISPOSITION_OPEN:
                entry.pop("acted_identity", None)   # reopened -> next dispatch/park re-baselines
            elif args.disposition in (DISPOSITION_DISPATCHED, DISPOSITION_NEEDS_HUMAN):
                if args.thread:
                    # our reply moved the thread's last comment, so re-read it now as the baseline.
                    ident = _mark_thread_baseline(item_id, args, state)
                    if ident is not None:
                        entry["acted_identity"] = ident
                elif args.comment and args.acted_edit_id:
                    # our reply is a separate top-level comment and never edits THIS one, so the
                    # snapshot-time edit_id the agent passes is already the correct baseline (no fetch).
                    entry["acted_identity"] = [args.acted_edit_id]
            state["last_action"] = f"{args.disposition} {label} {item_id}"
    print(json.dumps({"marked": args.check or args.thread or args.comment}))


def main():
    p = argparse.ArgumentParser(prog="pr-snapshot")
    sub = p.add_subparsers(dest="cmd", required=True)

    s = sub.add_parser("snapshot")
    s.add_argument("--pr", type=int, required=True)
    s.add_argument("--repo", default=None)
    s.add_argument("--state-dir", required=True)
    s.add_argument("--fetch-file", default=None)
    s.add_argument("--reset-session", action="store_true")  # start the budget clock (a fresh invocation)
    s.set_defaults(func=cmd_snapshot)

    m = sub.add_parser("mark")
    m.add_argument("--pr", type=int, default=0)
    m.add_argument("--repo", default=None)
    m.add_argument("--state-dir", required=True)
    m.add_argument("--thread", default=None)
    # `open` re-actionizes a parked thread — the explicit re-open the SKILL prose relies on when a
    # parked stream's context materially changes (a human pushed, the check universe moved).
    m.add_argument("--disposition", choices=[DISPOSITION_NEEDS_HUMAN, DISPOSITION_DISPATCHED, DISPOSITION_OPEN], default=DISPOSITION_DISPATCHED)
    m.add_argument("--check", default=None)
    m.add_argument("--comment", default=None)
    m.add_argument("--fetch-file", default=None)  # reuse the tick's fetch for the at-mark baseline
    m.add_argument("--acted-edit-id", default=None)  # snapshot-time edit_id baseline for a --comment
    m.set_defaults(func=cmd_mark)

    w = sub.add_parser("watch")
    w.add_argument("--pr", type=int, required=True)
    w.add_argument("--repo", default=None)
    w.add_argument("--state-dir", required=True)
    w.add_argument("--interval", type=float, default=150.0, help="poll cadence seconds")
    w.add_argument("--max-runtime", type=float, default=14400.0, help="hard stop seconds (0 = unbounded)")
    w.add_argument("--settle-seconds", type=float, default=300.0, help="quiet window before a merge-ready wake")
    w.add_argument("--stop-file", default=None, help="path whose existence stops the watch")
    w.add_argument("--fetch-file", default=None)
    w.add_argument("--reset-session", action="store_true")  # cmd_watch also forces this on arm
    w.set_defaults(func=cmd_watch)

    args = p.parse_args()
    args.func(args)


if __name__ == "__main__":
    main()
