#!/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 re
import signal
import subprocess
import tempfile
import threading
import uuid
from contextlib import contextmanager
from datetime import datetime, timezone
from urllib.parse import urlsplit

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

MANAGER_CONFIRMED = "confirmed"
MANAGER_ABSENT = "absent"
MANAGER_PROBE_ERROR = "probe-error"
RELATIONSHIP_DEPENDENT = "dependent"
RELATIONSHIP_INDEPENDENT = "independent"
RELATIONSHIP_PROBE_ERROR = "probe-error"

# 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

# 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


class _WatchSuperseded(Exception):
    """Internal control flow: this watch lost ownership and must unwind immediately."""


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 _stack_schema_unavailable(stderr):
    """Recognize only GraphQL schema errors for the private-preview PullRequest stack fields."""
    unavailable_markers = ("doesn't exist on type", "does not exist on type",
                           "cannot query field", "unknown field")
    for line in (stderr or "").splitlines():
        lowered = line.lower()
        if ("pullrequest" in lowered
                and re.search(r"\bstack(?:entry)?\b", lowered)
                and any(marker in lowered for marker in unavailable_markers)):
            return True
    return False


def _fetch_default_branch(owner, name, host=None):
    """Read the default branch without relying on private-preview GraphQL fields."""
    result = _run(["gh", "api", *_host_args(host), f"repos/{owner}/{name}",
                   "--jq", ".default_branch"])
    if result.returncode != 0:
        return None
    branch = result.stdout.strip()
    return branch if branch and branch != "null" else None


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 _default_pr_chain():
    """Backward-compatible neutral shape for injected/older snapshots with no chain facts."""
    return {
        "manager_status": MANAGER_ABSENT,
        "manager_source": None,
        "relationship_status": RELATIONSHIP_INDEPENDENT,
        "trunk": None,
        "current_branch": None,
        "target_position": None,
        "target_needs_rebase": None,
        "upstack_needs_rebase": [],
        "entries": [],
        "parent_prs": [],
        "dependent_prs": [],
    }


def _pr_summary(pr):
    if not pr:
        return None
    return {k: pr.get(k) for k in ("number", "url", "state", "isDraft", "baseRefName", "headRefName")
            if pr.get(k) is not None}


def _pr_url_identity(url):
    """Normalize a GitHub PR URL to a repository-scoped identity tuple."""
    if not isinstance(url, str):
        return None
    try:
        parsed = urlsplit(url.strip())
        parts = [part for part in parsed.path.split("/") if part]
        if (parsed.scheme.lower() not in ("http", "https") or not parsed.netloc
                or len(parts) < 4 or parts[-2].lower() != "pull"):
            return None
        number = int(parts[-1])
    except (TypeError, ValueError):
        return None
    return (parsed.netloc.lower(), parts[-4].lower(), parts[-3].lower(), number)


def _chain_from_entries(entries, pr, source, trunk=None, current_branch=None,
                        manager_id=None, manager_number=None, target_url=None):
    """Normalize an ordered manager-owned chain and locate the requested PR within it."""
    target_identity = _pr_url_identity(target_url) if target_url is not None else None
    target_index = next((i for i, e in enumerate(entries)
                         if (e.get("number") == pr or str(e.get("number")) == str(pr))
                         and (target_url is None
                              or (target_identity is not None
                                  and _pr_url_identity(e.get("url")) == target_identity))), None)
    if target_index is None:
        return None
    target = entries[target_index]
    upstack_stale = [
        {k: e.get(k) for k in ("number", "position", "name", "url") if e.get(k) is not None}
        for e in entries[target_index + 1:] if e.get("needs_rebase") is True
    ]
    chain = _default_pr_chain()
    chain.update({
        "manager_status": MANAGER_CONFIRMED,
        "manager_source": source,
        "manager_id": manager_id,
        "manager_number": manager_number,
        "relationship_status": RELATIONSHIP_DEPENDENT if len(entries) > 1 else RELATIONSHIP_INDEPENDENT,
        "trunk": trunk,
        "current_branch": current_branch,
        "target_position": target.get("position") or target_index + 1,
        "target_needs_rebase": target.get("needs_rebase"),
        "upstack_needs_rebase": upstack_stale,
        "entries": entries,
        "parent_prs": [_pr_summary(entries[target_index - 1])] if target_index > 0 else [],
        "dependent_prs": [_pr_summary(e) for e in entries[target_index + 1:]],
    })
    return chain


def _chain_from_stack_view(raw, pr, target_url):
    entries = []
    for index, branch in enumerate((raw or {}).get("branches") or []):
        remote_pr = branch.get("pr") or {}
        entries.append({
            "position": index + 1,
            "name": branch.get("name"),
            "head": branch.get("head"),
            "base": branch.get("base"),
            "is_current": bool(branch.get("isCurrent")),
            "is_merged": bool(branch.get("isMerged")),
            "is_queued": bool(branch.get("isQueued")),
            "needs_rebase": branch.get("needsRebase"),
            "number": remote_pr.get("number"),
            "url": remote_pr.get("url"),
            "state": remote_pr.get("state"),
            "isDraft": remote_pr.get("isDraft"),
        })
    return _chain_from_entries(entries, pr, "gh-stack", (raw or {}).get("trunk"),
                               (raw or {}).get("currentBranch"), target_url=target_url)


def _fetch_graphql_stack(pr, owner, name, host=None):
    """Read-only remote membership fallback. A successful null stack is distinct from failure."""
    query = """
query($owner:String!,$repo:String!,$pr:Int!){
  repository(owner:$owner,name:$repo){
    defaultBranchRef{ name }
    pullRequest(number:$pr){
    stackEntry{ position }
    stack{ id number size baseRefName
      entries(first:100){ nodes{ position pullRequest{
        number url state isDraft baseRefName headRefName headRefOid
      } } }
    }
  } }
}"""
    args = ["gh", "api", "graphql", *_host_args(host), "-f", f"owner={owner}", "-f", f"repo={name}",
            "-F", f"pr={pr}", "-f", f"query={query}"]
    r = _run(args)
    if r.returncode != 0:
        if _stack_schema_unavailable(r.stderr):
            default_branch = _fetch_default_branch(owner, name, host)
            if default_branch:
                return MANAGER_ABSENT, None, default_branch
        return MANAGER_PROBE_ERROR, None, None
    try:
        repository = json.loads(r.stdout)["data"]["repository"]
        default_branch = (repository.get("defaultBranchRef") or {}).get("name")
        node = repository["pullRequest"]
        stack = node.get("stack")
    except (KeyError, TypeError, ValueError, json.JSONDecodeError):
        return MANAGER_PROBE_ERROR, None, None
    if stack is None:
        return MANAGER_ABSENT, None, default_branch
    entries = []
    for raw in (stack.get("entries") or {}).get("nodes") or []:
        remote_pr = raw.get("pullRequest") or {}
        entries.append({
            "position": raw.get("position"),
            "name": remote_pr.get("headRefName"),
            "head": remote_pr.get("headRefOid"),
            "base_ref_name": remote_pr.get("baseRefName"),
            "needs_rebase": None,
            **(_pr_summary(remote_pr) or {}),
        })
    chain = _chain_from_entries(entries, pr, "graphql", stack.get("baseRefName"),
                                manager_id=stack.get("id"), manager_number=stack.get("number"))
    return ((MANAGER_CONFIRMED, chain, default_branch) if chain
            else (MANAGER_PROBE_ERROR, None, default_branch))


def _manual_relationships(pr, repo, base_ref_name, head_ref_name, owner, name,
                          host=None, default_branch=None):
    """Find ordinary parent/children with narrow read-only branch filters."""
    repo_ref = repo or f"{owner}/{name}"
    if host and repo_ref.count("/") == 1:
        repo_ref = f"{host}/{repo_ref}"
    fields = "number,url,state,isDraft,baseRefName,headRefName"
    parent_result = None
    if base_ref_name and base_ref_name != default_branch:
        parent_result = _run(["gh", "pr", "list", "--repo", repo_ref, "--state", "all",
                              "--head", base_ref_name, "--limit", "20", "--json", fields])
    dependent_result = _run(["gh", "pr", "list", "--repo", repo_ref, "--state", "open",
                             "--base", head_ref_name or "", "--limit", "100", "--json", fields])
    if ((parent_result is not None and parent_result.returncode != 0)
            or dependent_result.returncode != 0):
        return RELATIONSHIP_PROBE_ERROR, [], []
    try:
        parent_candidates = json.loads(parent_result.stdout) or [] if parent_result else []
        dependent_candidates = json.loads(dependent_result.stdout) or []
    except (TypeError, ValueError, json.JSONDecodeError):
        return RELATIONSHIP_PROBE_ERROR, [], []
    parents = [_pr_summary(p) for p in parent_candidates
               if p.get("number") != pr and p.get("headRefName") == base_ref_name]
    dependents = [_pr_summary(p) for p in dependent_candidates
                  if p.get("number") != pr and p.get("state") == "OPEN"
                  and p.get("baseRefName") == head_ref_name]
    status = RELATIONSHIP_DEPENDENT if parents or dependents else RELATIONSHIP_INDEPENDENT
    return status, parents, dependents


def fetch_pr_chain(pr, repo, url, base_ref_name, head_ref_name, owner, name, host=None):
    """Classify manager membership and ordinary dependency relationships without mutation.

    The local manager is the fast/rich path, but its output is accepted only when it contains the
    requested PR. `gh stack view` has no target argument and may describe a different current stack.
    """
    local = _run(["gh", "stack", "view", "--json"])
    if local.returncode == 0:
        try:
            chain = _chain_from_stack_view(json.loads(local.stdout), pr, url)
        except (TypeError, ValueError, json.JSONDecodeError):
            chain = None
        if chain:
            return chain

    manager_status, chain, default_branch = _fetch_graphql_stack(pr, owner, name, host)
    if manager_status == MANAGER_CONFIRMED:
        return chain

    relationship, parents, dependents = _manual_relationships(
        pr, repo, base_ref_name, head_ref_name, owner, name, host, default_branch)
    result = _default_pr_chain()
    result.update({
        "manager_status": manager_status,
        "relationship_status": relationship,
        "parent_prs": parents,
        "dependent_prs": dependents,
    })
    return result


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,baseRefName,headRefName,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 entirely the
        # resolver's judgment; this deterministic layer keeps every non-empty non-author body.
        "feedback": _extract_feedback(v),
        "awaiting_approval": fetch_awaiting_approval(owner, name, head, host),
        "pr_chain": fetch_pr_chain(v.get("number") or pr, repo, v.get("url"),
                                   v.get("baseRefName"), v.get("headRefName"), owner, name, host),
    }


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):
    """Every non-empty top-level PR comment and review body not known to be from the PR author.
    `gh pr view --json comments,reviews` returns flat arrays (not GraphQL {nodes}). Content,
    identity, and surface are evidence for the resolver to judge, not deterministic exclusions."""
    author = (v.get("author") or {}).get("login")
    out = []
    for c in v.get("comments") or []:
        a = (c.get("author") or {}).get("login")
        if (not author or a != author) 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 (not author or a != author) 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,
        "pr_chain": _default_pr_chain(),
        "last_change_at": None, "last_action": None, "stop_reason": None,
        "watch_generation": None, "watch_pid": None, "watch_process_identity": 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)
    state["pr_chain"] = cur.get("pr_chain") or _default_pr_chain()

    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,
        "pr_chain": state["pr_chain"],
        "stack_blocker": _stack_blocker(state["pr_chain"]),
        "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_started_at": state.get("started_at"),
        "session_seconds": _elapsed(state.get("started_at"), now),
        "watch_generation": state.get("watch_generation"),
        "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"),
            json.dumps(state.get("pr_chain") or _default_pr_chain(), sort_keys=True),
            # 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 _stack_blocker(chain):
    """Return the manager-currency residual that blocks target readiness, if any."""
    status = (chain or {}).get("manager_status")
    if status == MANAGER_PROBE_ERROR:
        return "manager-probe-error"
    if (chain or {}).get("relationship_status") == RELATIONSHIP_PROBE_ERROR:
        return "relationship-probe-error"
    if status == MANAGER_CONFIRMED:
        freshness = chain.get("target_needs_rebase")
        if freshness is True:
            return "target-needs-rebase"
        if freshness is not False:
            return "managed-freshness-unknown"
    return None


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


def _session_started_at(value):
    """Parse one invocation-wide, timezone-aware budget anchor for reuse across state dirs."""
    try:
        parsed = datetime.fromisoformat(value)
    except (ValueError, TypeError):
        raise argparse.ArgumentTypeError("must be an ISO-8601 timestamp")
    if parsed.tzinfo is None:
        raise argparse.ArgumentTypeError("must include a timezone")
    return _iso(parsed.astimezone(timezone.utc))


@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 _fetch_snapshot(args):
    """Fetch current PR state without mutating the persisted babysit state."""
    if args.fetch_file:
        with open(args.fetch_file) as f:
            return json.load(f)
    return fetch(args.pr, args.repo)


def _apply_snapshot(box, args, cur, now, advance_trajectory):
    """Apply one fetched snapshot to a caller-owned locked state box."""
    if box["state"].get("pr", {}).get("url") is None and cur.get("url"):
        box["state"]["pr"]["url"] = cur["url"]
    if getattr(args, "session_started_at", None):
        # Carry one invocation-wide clock through watch re-arms and managed-stack layer state dirs.
        box["state"]["started_at"] = args.session_started_at
    elif getattr(args, "reset_session", False):
        # A fresh babysit invocation 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 before doing any work.
        box["state"]["started_at"] = _iso(now)
    return diff(box["state"], cur, now, advance_trajectory=advance_trajectory)


def _run_snapshot(args, now, advance_trajectory=True, watch_generation=None):
    """One fetch -> diff -> persist. Returns the actionable/state dict."""
    cur = _fetch_snapshot(args)
    with locked_state(args.state_dir, args.pr, args.repo, now) as box:
        if watch_generation is not None and box["state"].get("watch_generation") != watch_generation:
            raise _WatchSuperseded()
        actionable, box["state"] = _apply_snapshot(box, args, cur, now, 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("ci", 0):
        return "actionable"
    if c.get("comments", 0):
        # Non-thread bodies are feedback candidates, not a deterministic conclusion that work is
        # required. The resolver may legitimately silent-drop status noise or review wrappers.
        return "feedback-candidate"
    if a.get("stack_blocker"):
        return "stack-blocked"
    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 _persisted_watch_generation(args):
    """Best-effort read of current ownership without creating or mutating watch state."""
    try:
        with open(os.path.join(args.state_dir, "state.json")) as f:
            state = json.load(f)
    except (OSError, json.JSONDecodeError):
        return None
    generation = state.get("watch_generation") if isinstance(state, dict) else None
    return generation if isinstance(generation, str) and generation else None


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__")
    if a.get("stack_blocker"):
        sig.add(f"__stack__:{a['stack_blocker']}")
    # 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 _process_identity(pid):
    """Best-effort PID-reuse guard for replacing a prior watcher. If process identity cannot be
    proven, generation invalidation still suppresses its wake and we leave OS cleanup alone."""
    r = subprocess.run(["ps", "-p", str(pid), "-o", "lstart=", "-o", "command="],
                       capture_output=True, text=True)
    if r.returncode != 0:
        return None
    return (r.stdout or "").strip() or None


@contextmanager
def _watch_lock(state_dir, exclusive):
    import fcntl
    os.makedirs(state_dir, exist_ok=True)
    with open(os.path.join(state_dir, "lock"), "w") as lf:
        fcntl.flock(lf, fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH)
        try:
            yield
        finally:
            fcntl.flock(lf, fcntl.LOCK_UN)


def _watch_candidate_path(args):
    return os.path.join(args.state_dir, "watch-candidate.json")


def _read_watch_candidate(args):
    try:
        with open(_watch_candidate_path(args)) as f:
            candidate = json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        return {}
    return candidate if isinstance(candidate, dict) else {}


def _reserve_watch_candidate(args, generation):
    """Make this invocation the newest candidate without displacing the active watcher."""
    candidate = {
        "generation": generation,
        "pid": os.getpid(),
        "process_identity": _process_identity(os.getpid()),
    }
    with _watch_lock(args.state_dir, exclusive=True):
        previous = _read_watch_candidate(args)
        tmp = tempfile.NamedTemporaryFile("w", dir=args.state_dir, delete=False)
        json.dump(candidate, tmp)
        tmp.flush()
        os.fsync(tmp.fileno())
        tmp.close()
        os.replace(tmp.name, _watch_candidate_path(args))
    return previous


def _clear_watch_candidate(args, generation):
    with _watch_lock(args.state_dir, exclusive=True):
        if _read_watch_candidate(args).get("generation") != generation:
            return
        try:
            os.unlink(_watch_candidate_path(args))
        except FileNotFoundError:
            pass


def _activate_watch(args, generation, now, cur):
    """Atomically activate this watcher and persist its successfully fetched preflight."""
    identity = _process_identity(os.getpid())
    with locked_state(args.state_dir, args.pr, args.repo, now) as box:
        if _read_watch_candidate(args).get("generation") != generation:
            return None, None
        state = box["state"]
        previous = {
            "pid": state.get("watch_pid"),
            "process_identity": state.get("watch_process_identity"),
        }
        state["watch_generation"] = generation
        state["watch_pid"] = os.getpid()
        state["watch_process_identity"] = identity
        actionable, box["state"] = _apply_snapshot(box, args, cur, now, advance_trajectory=False)
        try:
            os.unlink(_watch_candidate_path(args))
        except FileNotFoundError:
            pass
    return previous, actionable


def _watch_is_current(args, generation):
    """Read the ownership generation under the state lock without rewriting state.json."""
    state_path = os.path.join(args.state_dir, "state.json")
    with _watch_lock(args.state_dir, exclusive=False):
        if not os.path.exists(state_path):
            return False
        with open(state_path) as f:
            return json.load(f).get("watch_generation") == generation


def _emit_wake_if_current(args, generation, reason, **fields):
    """Serialize the final ownership check with takeover so an old generation cannot emit after
    the new generation has become current. A wake emitted just before takeover is valid at emission
    and will be recognized as stale against the next fresh snapshot."""
    state_path = os.path.join(args.state_dir, "state.json")
    with _watch_lock(args.state_dir, exclusive=False):
        if not os.path.exists(state_path):
            return False
        with open(state_path) as f:
            if json.load(f).get("watch_generation") != generation:
                return False
        _emit_wake(reason, watch_generation=generation, **fields)
        return True


def _terminate_replaced_watch(previous):
    """Promptly stop the replaced process, but never signal a PID whose identity no longer matches."""
    pid = previous.get("pid")
    identity = previous.get("process_identity")
    if not isinstance(pid, int) or pid == os.getpid() or not identity:
        return
    if _process_identity(pid) != identity:
        return
    try:
        os.kill(pid, signal.SIGTERM)
    except ProcessLookupError:
        pass


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
    (work to inspect 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."""
    stop_requested = threading.Event()
    interrupt_immediately = True
    superseded = False
    generation = uuid.uuid4().hex

    # An already-requested stop is authoritative before this invocation becomes a candidate. Use
    # current ownership for the wake when one exists, but do not fetch, reserve, mutate watch state,
    # or disturb that incumbent merely to report the stop condition.
    if args.stop_file and os.path.exists(args.stop_file):
        _emit_wake("stop-signal", watch_generation=_persisted_watch_generation(args) or generation)
        return

    def request_stop(_signum, _frame):
        nonlocal superseded
        superseded = True
        stop_requested.set()
        if interrupt_immediately:
            raise _WatchSuperseded()

    prior_sigterm = signal.getsignal(signal.SIGTERM)
    signal.signal(signal.SIGTERM, request_stop)
    try:
        previous_candidate = _reserve_watch_candidate(args, generation)
        _terminate_replaced_watch(previous_candidate)

        # Preflight before takeover: an invalid fetch/auth/config must not displace a healthy watcher.
        cur = _fetch_snapshot(args)
        if stop_requested.is_set():
            return

        # Finish the handoff even if an even newer watcher arrives: otherwise that watcher could
        # stop us between activation and terminating our predecessor, orphaning the oldest process.
        interrupt_immediately = False
        previous, actionable = _activate_watch(args, generation, _now(), cur)
        if previous is None:
            superseded = True
            return
        args.reset_session = False
        _terminate_replaced_watch(previous)
        interrupt_immediately = True
        if stop_requested.is_set():
            return

        armed = _blocker_sig(actionable)  # blockers already surfaced when this generation armed
        while True:
            if stop_requested.is_set():
                return
            if not _watch_is_current(args, generation):
                superseded = True
                return
            if args.stop_file and os.path.exists(args.stop_file):
                _emit_wake_if_current(args, generation, "stop-signal")
                return
            reason = _wake_reason(actionable, args.settle_seconds)
            if reason in ("needs-human", "blocked-failing", "blocked-external", "stack-blocked") and _blocker_sig(actionable) <= armed:
                reason = None   # already-surfaced residual — keep watching, do not re-wake or terminate
            if reason:
                _emit_wake_if_current(args, generation, reason, url=actionable.get("url"),
                                      pr_state=actionable.get("pr_state"), counts=actionable.get("counts"))
                return
            if args.max_runtime and actionable.get("session_seconds", 0) >= args.max_runtime:
                _emit_wake_if_current(args, generation, "max-runtime", url=actionable.get("url"))
                return
            if stop_requested.wait(args.interval):
                return
            if not _watch_is_current(args, generation):
                superseded = True
                return
            actionable = _run_snapshot(args, _now(), advance_trajectory=False,
                                       watch_generation=generation)
    except _WatchSuperseded:
        superseded = True
        return
    finally:
        interrupt_immediately = False
        _clear_watch_candidate(args, generation)
        # A newer owner can still signal our recorded PID after observing us stale but before this
        # process has exited. Keep that late takeover signal harmless; ordinary wake/timeout/stop
        # returns restore the embedding caller's handler as before.
        signal.signal(signal.SIGTERM, signal.SIG_IGN if superseded else prior_sigterm)


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)
    sg = s.add_mutually_exclusive_group()
    sg.add_argument("--reset-session", action="store_true")  # start the budget clock (a fresh invocation)
    sg.add_argument("--session-started-at", type=_session_started_at,
                    help="carry a prior invocation's ISO-8601 budget anchor")
    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)
    wg = w.add_mutually_exclusive_group()
    wg.add_argument("--reset-session", action="store_true")
    wg.add_argument("--session-started-at", type=_session_started_at,
                    help="carry the invocation-wide ISO-8601 budget anchor")
    w.set_defaults(func=cmd_watch)

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


if __name__ == "__main__":
    main()
