#!/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 --invocation-id ID --session-started-at TIME
           --invocation-budget-seconds N (--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 errno
import hashlib
import json
import os
import re
import signal
import subprocess
import sys
import tempfile
import threading
import time
import uuid
from contextlib import contextmanager
from datetime import datetime, timedelta, timezone
from urllib.parse import quote, urlsplit

IS_WINDOWS = sys.platform == "win32"

if IS_WINDOWS:
    import msvcrt
else:
    import fcntl

# 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"
CURRENCY_CLAIMED = "claimed"
CURRENCY_CONFIRMED = "confirmed"
CURRENCY_OUTCOME_MUTATION_OBSERVED = "mutation-observed"
CURRENCY_OUTCOME_PROVEN_NO_MUTATION = "proven-no-mutation"
CURRENCY_OUTCOME_AMBIGUOUS = "ambiguous"
CURRENCY_ATTENTION_CLAIM = "claim"
CURRENCY_ATTENTION_INSPECT = "inspect"
CURRENCY_ATTENTION_RECONCILE = "reconcile"

MANAGER_CONFIRMED = "confirmed"
MANAGER_ABSENT = "absent"
MANAGER_PROBE_ERROR = "probe-error"
RELATIONSHIP_DEPENDENT = "dependent"
RELATIONSHIP_INDEPENDENT = "independent"
RELATIONSHIP_PROBE_ERROR = "probe-error"
BASE_REF_CURRENT = "current"
BASE_REF_RACE = "race"
BASE_REF_MERGEABILITY_PENDING = "mergeability-pending"
BASE_REF_PROBE_ERROR = "probe-error"
BASE_REF_LEGACY_STALE = "stale"

# First judgment point for an incomplete review lifecycle. The agent may extend once to 1800s when
# concrete prior-round timing supports it; the skill contract owns that semantic decision.
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

# One invocation is a bounded monitoring shift, not the lifetime of the PR. Eight hours covers a
# maximum-length GitHub-hosted Actions job (six hours) plus reaction and settle time. Callers may
# choose another fixed value on the first snapshot; later commands must present the same value.
DEFAULT_INVOCATION_BUDGET_SECONDS = 8 * 60 * 60
# The 8h budget is spent in *active watch-capability time*, not raw wall-clock: a span where the
# whole watch process was suspended (laptop asleep) is excluded. Detection is coarse — an activity
# gap wider than this threshold (well above the 150s poll interval) is charged to dead time, minus
# the threshold itself so ordinary polls/ticks never register. The 3-day backstop stays wall-clock.
DEAD_TIME_THRESHOLD_SECONDS = 15 * 60
DEFAULT_INVOCATION_BACKSTOP_SECONDS = 3 * 24 * 60 * 60
CURRENCY_RETRY_BACKOFF_SECONDS = 30


# --- cross-platform state primitives ------------------------------------------
#
# The watcher needs three POSIX facilities that native Windows Python does not
# have: an advisory file lock (fcntl.flock), a process-identity probe for the
# PID-reuse guard (`ps -o lstart=`), and an unconditional rename. Each Windows
# analog differs in a way that matters here, so each is named rather than
# translated literally. See
# docs/solutions/architecture-patterns/posix-process-supervision-on-native-windows.md.

# Poll spacing while waiting on a Windows lock or a blocked rename.
_WIN_RETRY_SECONDS = 0.05
# Bound only the rename retry. A lock wait is unbounded, matching flock.
_WIN_REPLACE_TIMEOUT_SECONDS = 10.0

if IS_WINDOWS:
    import ctypes
    from ctypes import wintypes

    _kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
    _PROCESS_QUERY_LIMITED_INFORMATION = 0x1000

    # Declare argtypes/restype explicitly: without them ctypes truncates HANDLEs
    # to 32-bit ints on Win64, so a valid handle silently becomes a bad one.
    _kernel32.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD]
    _kernel32.OpenProcess.restype = wintypes.HANDLE
    _kernel32.CloseHandle.argtypes = [wintypes.HANDLE]
    _kernel32.CloseHandle.restype = wintypes.BOOL
    _kernel32.GetProcessTimes.argtypes = [
        wintypes.HANDLE, ctypes.POINTER(wintypes.FILETIME),
        ctypes.POINTER(wintypes.FILETIME), ctypes.POINTER(wintypes.FILETIME),
        ctypes.POINTER(wintypes.FILETIME)]
    _kernel32.GetProcessTimes.restype = wintypes.BOOL
    _kernel32.QueryFullProcessImageNameW.argtypes = [
        wintypes.HANDLE, wintypes.DWORD, wintypes.LPWSTR,
        ctypes.POINTER(wintypes.DWORD)]
    _kernel32.QueryFullProcessImageNameW.restype = wintypes.BOOL

    def _win_process_identity(pid):
        """Creation time plus image path — the Windows analog of `ps -o lstart= -o command=`.

        Creation time is the part that makes this a PID-reuse guard: Windows recycles PIDs
        aggressively, and a recycled PID always carries a later creation time than the watcher
        we recorded. Returns None when the process is gone or unopenable, which the caller
        treats as "identity not proven" and declines to act on."""
        handle = _kernel32.OpenProcess(_PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
        if not handle:
            return None
        try:
            created, exited, kernel, user = (wintypes.FILETIME() for _ in range(4))
            if not _kernel32.GetProcessTimes(
                    handle, ctypes.byref(created), ctypes.byref(exited),
                    ctypes.byref(kernel), ctypes.byref(user)):
                return None
            started = (created.dwHighDateTime << 32) | created.dwLowDateTime
            if not started:
                return None
            size = wintypes.DWORD(32768)
            buf = ctypes.create_unicode_buffer(size.value)
            image = (buf.value if _kernel32.QueryFullProcessImageNameW(
                handle, 0, buf, ctypes.byref(size)) else "")
        finally:
            _kernel32.CloseHandle(handle)
        return "{} {}".format(started, image)


def _lock_acquire(fd, exclusive):
    """Block until this process owns the lock file.

    msvcrt.locking is the Windows analog of flock, with three differences. It locks a byte range
    from the current offset rather than the whole file, so the offset is pinned to 0. It has no
    shared mode, so a reader takes the same exclusive lock a writer does — free here, because every
    critical section is a small local read/write with the network fetch already completed outside
    the lock. And its blocking mode (LK_LOCK) gives up after roughly ten seconds instead of waiting,
    so retrying the non-blocking mode is what actually reproduces flock's wait-for-the-holder
    behavior. As on POSIX, the OS releases the lock if a holder dies, so a crash cannot wedge this."""
    if not IS_WINDOWS:
        fcntl.flock(fd, fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH)
        return
    os.lseek(fd, 0, os.SEEK_SET)
    while True:
        try:
            msvcrt.locking(fd, msvcrt.LK_NBLCK, 1)
            return
        except OSError as e:
            # Retry only genuine contention. msvcrt reports a held range as EACCES; anything
            # else (EBADF from a closed descriptor, EINVAL from a bad range) is a defect that
            # retrying cannot clear, and swallowing it here would spin this loop forever with
            # no output — a silent hang, which for a watcher is indistinguishable from the
            # "monitoring is quietly not running" failure this port exists to fix.
            if e.errno != errno.EACCES:
                raise
            time.sleep(_WIN_RETRY_SECONDS)


def _lock_release(fd):
    if not IS_WINDOWS:
        fcntl.flock(fd, fcntl.LOCK_UN)
        return
    os.lseek(fd, 0, os.SEEK_SET)
    try:
        msvcrt.locking(fd, msvcrt.LK_UNLCK, 1)
    except OSError:
        pass


@contextmanager
def _state_lock(state_dir, exclusive=True):
    """Serialize access to the state dir for the duration of the body."""
    os.makedirs(state_dir, exist_ok=True)
    # O_RDWR|O_CREAT rather than a truncating open("w"): on Windows, truncating a file another
    # process holds a byte-range lock on fails, which would make lock acquisition itself the race.
    fd = os.open(os.path.join(state_dir, "lock"), os.O_RDWR | os.O_CREAT, 0o600)
    try:
        _lock_acquire(fd, exclusive)
        try:
            yield
        finally:
            _lock_release(fd)
    finally:
        os.close(fd)


def _replace_atomic(src, dst):
    """Atomically move `src` onto `dst`.

    POSIX rename always succeeds over an open destination. Windows refuses while any other handle
    to `dst` is open, so an unlocked best-effort reader — or a virus scanner that opened the file
    behind us — would otherwise turn a routine concurrent read into a crashed writer. Retry briefly
    rather than treating that transient sharing violation as a failure."""
    if not IS_WINDOWS:
        os.replace(src, dst)
        return
    deadline = time.monotonic() + _WIN_REPLACE_TIMEOUT_SECONDS
    while True:
        try:
            os.replace(src, dst)
            return
        except PermissionError:
            if time.monotonic() >= deadline:
                raise
            time.sleep(_WIN_RETRY_SECONDS)


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


class _InvocationSuperseded(Exception):
    """Internal control flow: this watch belongs to an invocation that was replaced."""

    def __init__(self, current_invocation_id):
        self.current_invocation_id = current_invocation_id


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, encoding="utf-8")


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 _valid_oid(value):
    return isinstance(value, str) and bool(
        re.fullmatch(r"(?:[0-9a-fA-F]{40}|[0-9a-fA-F]{64})", value))


def fetch_pr_merge_identity(pr, owner, name, host=None):
    """Read mergeability and the identities it was computed from in one GraphQL observation."""
    query = """
query($owner:String!,$repo:String!,$pr:Int!){
  repository(owner:$owner,name:$repo){ pullRequest(number:$pr){
    mergeable mergeStateStatus headRefOid baseRefOid baseRefName viewerCanUpdateBranch
    baseRef { target { oid } }
    potentialMergeCommit { oid parents(first:2) { nodes { oid } } }
  } }
}"""
    r = _run(["gh", "api", "graphql", *_host_args(host), "-f", f"owner={owner}",
              "-f", f"repo={name}", "-F", f"pr={pr}", "-f", f"query={query}"])
    if r.returncode != 0:
        return None
    try:
        value = json.loads(r.stdout)["data"]["repository"]["pullRequest"]
    except (KeyError, TypeError, ValueError, json.JSONDecodeError):
        return None
    return value if isinstance(value, dict) else None


def fetch_base_ref(owner, name, ref, merge_identity, host=None):
    """Bind GitHub's mergeability observation to an independently probed current base ref.

    `baseRefOid` is retained only as historical PR metadata. Current identity comes from both the
    PR's `baseRef.target.oid` and an exact Git-ref read; a usable generated test merge must then
    name that base and the observed PR head as its two parents.
    """
    historical_oid = (merge_identity or {}).get("baseRefOid")
    head_oid = (merge_identity or {}).get("headRefOid")
    mergeable = (merge_identity or {}).get("mergeable")
    merge_state_status = (merge_identity or {}).get("mergeStateStatus")
    graphql_oid = (((merge_identity or {}).get("baseRef") or {}).get("target") or {}).get("oid")
    potential_merge = (merge_identity or {}).get("potentialMergeCommit")
    merge_commit_oid = potential_merge.get("oid") if isinstance(potential_merge, dict) else None
    merge_parent_oids = (((potential_merge or {}).get("parents") or {}).get("nodes")
                         if isinstance(potential_merge, dict) else None)
    if isinstance(merge_parent_oids, list):
        merge_parent_oids = [node.get("oid") if isinstance(node, dict) else None
                             for node in merge_parent_oids]
    else:
        merge_parent_oids = []
    base = {
        "host": host or "github.com",
        "repository": f"{owner}/{name}",
        "ref": ref,
        "oid": None,
        "graphql_oid": graphql_oid,
        "historical_oid": historical_oid,
        "merge_commit_oid": merge_commit_oid,
        "merge_parent_oids": merge_parent_oids,
        "identity": BASE_REF_PROBE_ERROR,
    }
    if not ref:
        return base
    encoded_ref = quote(ref, safe="/")
    r = _run([
        "gh", "api", *_host_args(host),
        f"repos/{owner}/{name}/git/ref/heads/{encoded_ref}",
        "--jq", ".object.sha",
    ])
    current_oid = (r.stdout or "").strip()
    if r.returncode != 0 or not _valid_oid(current_oid):
        return base
    base["oid"] = current_oid
    if not (_valid_oid(graphql_oid) and _valid_oid(head_oid)):
        return base
    if current_oid.lower() != graphql_oid.lower():
        base["identity"] = BASE_REF_RACE
        return base

    if mergeable == "MERGEABLE" and merge_state_status not in (None, "UNKNOWN", "DIRTY"):
        if potential_merge is None:
            base["identity"] = BASE_REF_MERGEABILITY_PENDING
            return base
        if (not _valid_oid(merge_commit_oid) or len(merge_parent_oids) != 2
                or not all(_valid_oid(oid) for oid in merge_parent_oids)):
            return base
        if (merge_parent_oids[0].lower() != current_oid.lower()
                or merge_parent_oids[1].lower() != head_oid.lower()):
            base["identity"] = BASE_REF_RACE
            return base
    elif not (mergeable == "CONFLICTING" and merge_state_status == "DIRTY"):
        base["identity"] = BASE_REF_MERGEABILITY_PENDING
        return base

    base["identity"] = BASE_REF_CURRENT
    return base


def _base_ref_blocker(base):
    identity = (base or {}).get("identity")
    if identity is None:
        # Compatibility for pre-change --fetch-file fixtures and persisted snapshots.
        identity = (base or {}).get("freshness")
    if identity == BASE_REF_CURRENT:
        return None
    if identity in (BASE_REF_RACE, BASE_REF_MERGEABILITY_PENDING):
        return identity
    if identity == BASE_REF_LEGACY_STALE:
        return BASE_REF_LEGACY_STALE
    return BASE_REF_PROBE_ERROR


def _eyes_reaction_identities(reaction_pages):
    """Stable identities for current PR-body eyes reactors from paginated REST results."""
    if not isinstance(reaction_pages, list):
        return []
    reactions = [reaction for page in reaction_pages for reaction in page] \
        if reaction_pages and all(isinstance(page, list) for page in reaction_pages) else reaction_pages
    identities = []
    for reaction in reactions:
        if not isinstance(reaction, dict) or reaction.get("content") not in (None, "eyes"):
            continue
        user = reaction.get("user") or {}
        identity = user.get("node_id") or user.get("id") or user.get("login")
        # Deleted-user reactions have no user identity. The reaction identity still lets an
        # add/remove or same-count substitution move the lifecycle instead of becoming invisible.
        identity = identity or reaction.get("node_id") or reaction.get("id")
        if identity is not None:
            identities.append(str(identity))
    return sorted(set(identities))


def fetch_eyes_reactors(pr, owner, name, host=None):
    """Fetch every current PR-body eyes reactor; `gh pr view` exposes only their count."""
    r = _run_checked([
        "gh", "api", *_host_args(host), "--paginate", "--slurp",
        f"repos/{owner}/{name}/issues/{pr}/reactions?content=eyes&per_page=100",
    ], "gh api reactions")
    return _eyes_reaction_identities(json.loads(r.stdout))


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 _prior_thread_activity_identity(t):
    """The last external-review baseline already accepted for this parked thread."""
    acted_identity = t.get("acted_identity")
    if (t.get("disposition") in (DISPOSITION_DISPATCHED, DISPOSITION_NEEDS_HUMAN)
            and isinstance(acted_identity, (list, tuple))
            and len(acted_identity) == 2):
        return tuple(acted_identity)
    return _thread_identity(t)


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,
        "default_branch": 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:
        chain["default_branch"] = default_branch
        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,
        "default_branch": default_branch,
        "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,baseRefOid,baseRefName,headRefName,url,number,isDraft,"
                         "statusCheckRollup,author,comments,reviews"],
                        "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"),
            })

    owner, name, host = _resolve_repo_ref(repo, v.get("url"))
    merge_identity = fetch_pr_merge_identity(v.get("number") or pr, owner, name, host)
    merge_observation = merge_identity or v
    mergeable = merge_observation.get("mergeable")
    merge_state_status = merge_observation.get("mergeStateStatus")
    head = v.get("headRefOid")
    base = fetch_base_ref(owner, name, v.get("baseRefName"), merge_observation, host)
    if merge_identity is not None:
        identity_head = merge_identity.get("headRefOid")
        identity_base_ref = merge_identity.get("baseRefName")
        if (not isinstance(identity_base_ref, str)
                or not (_valid_oid(identity_head) and _valid_oid(head))):
            base["identity"] = BASE_REF_PROBE_ERROR
        elif (identity_head.lower() != head.lower()
                or identity_base_ref != v.get("baseRefName")):
            # The checks and review payload came from `v`; do not combine them with mergeability
            # observed for a different head or base branch.
            base["identity"] = BASE_REF_RACE
    pr_chain = fetch_pr_chain(v.get("number") or pr, repo, v.get("url"),
                              v.get("baseRefName"), v.get("headRefName"), owner, name, host)
    currency_probe = {
        "mergeable": mergeable,
        "merge_state_status": merge_state_status,
        "base": base,
        "pr_chain": pr_chain,
    }
    host_branch_update_capability = "unknown"
    if (merge_state_status == "BEHIND"
            and _branch_currency_observation(currency_probe, head) is not None):
        capability = merge_observation.get("viewerCanUpdateBranch")
        if isinstance(capability, bool):
            host_branch_update_capability = capability

    # 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_signal_identities = fetch_eyes_reactors(v.get("number") or pr, owner, name, host)
    review_signal_count = len(review_signal_identities)
    review_in_progress = review_signal_count > 0

    return {
        "pr_state": v.get("state"),
        "pr_is_draft": v.get("isDraft"),
        "mergeable": mergeable,
        "merge_state_status": merge_state_status,
        "review_decision": v.get("reviewDecision") or None,
        "head_sha": head,
        "base": base,
        "host_branch_update_capability": host_branch_update_capability,
        "url": v.get("url"),
        "checks": checks,
        "review_in_progress": review_in_progress,
        "review_signal_count": review_signal_count,
        "review_signal_identities": review_signal_identities,
        "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": pr_chain,
    }


def _body_hash(body):
    """Edit identity for a top-level comment / review body: `gh pr view --json` exposes no
    updatedAt, so hash the body. Feeds the external_review_moved activity signal (an edit wakes
    the watch and extends the settle clock) and `needs-human` reactivation (a human may answer a
    parked question by editing the same comment), but deliberately NOT `dispatched` reactivation:
    status bots rewrite their bodies on every push, so edit-keyed reactivation of handled items
    re-actionized bot comments forever (#1309)."""
    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: an API/permission failure returns None so the tick can preserve the
    last proven gate state rather than treating an unknown result as a proven clear."""
    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 None
    try:
        return int((r.stdout or "").strip() or 0)
    except ValueError:
        return None


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)
    created_at = _iso(now)
    return {
        "pr": {"owner": owner, "repo": name, "number": pr, "url": url},
        "head_sha": None, "tick": 0, "state_created_at": created_at,
        "started_at": created_at, "invocation_id": None, "invocation_budget_seconds": None,
        "last_activity_at": created_at, "dead_time_seconds": 0.0,
        "invocation_backstop_seconds": DEFAULT_INVOCATION_BACKSTOP_SECONDS,
        "checks": {}, "threads": {}, "ci_dispatched": {},
        "review_decision": None, "mergeable": None, "merge_state_status": None,
        "review_in_progress": False, "review_signal_count": 0, "review_signal_identities": [],
        "review_signal_seen_on_head": False,
        "review_signal_first_seen_at": None, "review_signal_last_changed_at": None,
        "blocked_external_head_sha": None,
        "blocked_external_first_seen_at": None,
        "blocked_external_review_last_activity_at": None,
        "pr_chain": _default_pr_chain(),
        "base": None,
        "branch_currency_state": _empty_branch_currency_state(),
        "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_branch_currency_state():
    return {"current_key": None, "items": {}, "semantic_parks": {}, "head_sha": None}


def _load_branch_currency_state(state):
    """Conservatively migrate durable state written before branch currency existed."""
    currency = state.get("branch_currency_state")
    if not isinstance(currency, dict):
        currency = {}
    defaults = _empty_branch_currency_state()
    for key, value in defaults.items():
        currency.setdefault(key, value)
    if not isinstance(currency.get("items"), dict):
        currency["items"] = {}
    if not isinstance(currency.get("semantic_parks"), dict):
        currency["semantic_parks"] = {}
    state["branch_currency_state"] = currency
    return currency


def _mergeability_certain(mergeable, merge_state_status, base):
    """GitHub mergeability is usable only when its cached base is proven current."""
    if _base_ref_blocker(base) is not None:
        return False
    if mergeable not in ("MERGEABLE", "CONFLICTING"):
        return False
    if not merge_state_status or merge_state_status == "UNKNOWN":
        return False
    if merge_state_status == "DIRTY":
        return mergeable == "CONFLICTING"
    if merge_state_status == "BEHIND":
        return mergeable == "MERGEABLE"
    return True


def _normal_base_route(cur):
    """Return the only route U1 may classify: an unmanaged PR rooted on its normal base.

    Ordinary dependents are intentionally harmless. An open or unknown parent means this target is
    itself in a manual dependency chain and therefore outside this unit's authority.
    """
    chain = cur.get("pr_chain") or _default_pr_chain()
    base = cur.get("base") or {}
    if chain.get("manager_status") != MANAGER_ABSENT:
        return None
    if chain.get("relationship_status") == RELATIONSHIP_PROBE_ERROR:
        return None
    if not chain.get("default_branch") or base.get("ref") != chain.get("default_branch"):
        return None
    parents = chain.get("parent_prs") or []
    if any(not isinstance(parent, dict) or parent.get("state") not in ("CLOSED", "MERGED")
           for parent in parents):
        return None
    return "normal-base"


def _branch_currency_observation(cur, head):
    status = cur.get("merge_state_status")
    certain = _mergeability_certain(cur.get("mergeable"), status, cur.get("base"))
    if not certain or status not in ("BEHIND", "DIRTY"):
        return None
    route = _normal_base_route(cur)
    base = cur.get("base") or {}
    if (route is None or not head
            or not all(base.get(key) for key in ("host", "repository", "ref", "oid"))):
        return None
    identity = {
        "host": base["host"],
        "base_repository": base["repository"],
        "base_ref": base["ref"],
        "base_oid": base["oid"],
        "head_sha": head,
        "status": status,
        "route": route,
    }
    encoded = json.dumps(identity, sort_keys=True, separators=(",", ":")).encode("utf-8")
    return {"key": f"currency:{hashlib.sha256(encoded).hexdigest()}", **identity}


def _claimed_currency_items(currency):
    """Return unresolved mutation claims with the current item first.

    Base or expected head movement may be the result of the claimed mutation. It is evidence to
    reconcile, not permission to discard the claim and start another attempt.
    """
    items = currency.get("items") or {}
    current_key = currency.get("current_key")
    claimed = []
    current = items.get(current_key)
    if isinstance(current, dict) and current.get("disposition") == CURRENCY_CLAIMED:
        claimed.append((current_key, current))
    claimed.extend(
        (key, item) for key, item in items.items()
        if key != current_key and isinstance(item, dict)
        and item.get("disposition") == CURRENCY_CLAIMED
    )
    return claimed


def _prepare_claimed_currency_item(state, item, cur=None, head=None):
    invocation_id = state.get("invocation_id")
    already_handled_here = (
        item.get("claimed_invocation_id") == invocation_id
        or item.get("reconciled_invocation_id") == invocation_id
    )
    base = (cur or {}).get("base") or {}
    async_evidence_moved = (
        item.get("recovery_state") in (
            CURRENCY_OUTCOME_MUTATION_OBSERVED, CURRENCY_OUTCOME_AMBIGUOUS)
        and ((head and head != item.get("head_sha"))
             or (base.get("oid") and base.get("oid") != item.get("base_oid")))
    )
    item["attention"] = (CURRENCY_ATTENTION_RECONCILE
                         if async_evidence_moved or not already_handled_here else None)
    item["reconciliation_only"] = True
    return item


def _apply_branch_currency(state, cur, head, head_changed, now):
    currency = _load_branch_currency_state(state)
    carried_claims = _claimed_currency_items(currency)
    if head_changed or (currency.get("head_sha") not in (None, head)):
        currency = _empty_branch_currency_state()
        currency["items"] = dict(carried_claims)
        state["branch_currency_state"] = currency
    currency["head_sha"] = head

    # An unresolved claim owns the branch-currency lane until it is explicitly reconciled. A base
    # move or expected head move may be the mutation's result, so a new observation must not replace
    # the old exact key or make the PR look clear in the meantime.
    claimed_items = _claimed_currency_items(currency)
    if claimed_items:
        key, item = claimed_items[0]
        currency["current_key"] = key
        return _prepare_claimed_currency_item(state, item, cur, head)

    observation = _branch_currency_observation(cur, head)
    if observation is None:
        currency["current_key"] = None
        return None

    key = observation["key"]
    items = currency["items"]
    item = items.get(key)
    if not isinstance(item, dict):
        item = {**observation, "disposition": DISPOSITION_OPEN}
        prior_host_branch_update_capability = None
    else:
        prior_host_branch_update_capability = item.get(
            "host_branch_update_capability", "unknown")
        # Refresh non-identity facts without changing an invocation-fenced disposition.
        item.update(observation)
    item["host_branch_update_capability"] = cur.get(
        "host_branch_update_capability", "unknown")
    parks = currency.get("semantic_parks") or {}
    item["parked_semantic_fingerprints"] = sorted(parks)
    item.setdefault("retry_count", 0)
    item.setdefault("mutation_consumed", False)
    disposition = item.get("disposition", DISPOSITION_OPEN)
    # A transient capability park is not a durable decision: the unchanged observation becomes
    # claimable as soon as the host confirms that the branch-update operation is authorized.
    if (disposition == DISPOSITION_NEEDS_HUMAN
            and item.get("status") == "BEHIND"
            and item.get("recovery_state") is None
            and prior_host_branch_update_capability in (False, "unknown")
            and item.get("host_branch_update_capability") is True):
        item["disposition"] = DISPOSITION_OPEN
        disposition = DISPOSITION_OPEN
    if disposition == DISPOSITION_OPEN:
        inspection_required = (item.get("status") == "DIRTY"
                               and bool(parks)
                               and item.get("inspection_result") != "changed")
        item["inspection_required"] = inspection_required
        retry_wait_seconds = 0
        retry_not_before = item.get("retry_not_before")
        if retry_not_before:
            try:
                retry_wait_seconds = max(
                    0, int((_parse_iso8601(retry_not_before) - now).total_seconds()))
            except (ValueError, TypeError):
                retry_wait_seconds = 0
        item["retry_wait_seconds"] = retry_wait_seconds
        if inspection_required:
            item["attention"] = CURRENCY_ATTENTION_INSPECT
        elif retry_wait_seconds > 0:
            item["attention"] = None
        else:
            item["attention"] = CURRENCY_ATTENTION_CLAIM
        item["reconciliation_only"] = False
    elif disposition == CURRENCY_CLAIMED:
        _prepare_claimed_currency_item(state, item, cur, head)
    else:
        item["attention"] = None
        item["reconciliation_only"] = False
    items[key] = item
    currency["current_key"] = key

    # Old observations are useful only when they carry an unresolved claim or semantic park. Claims
    # are never pruned; the small bounded tail applies only to parked base generations.
    retained = {item_key: value for item_key, value in items.items()
                if item_key == key
                or value.get("disposition") in (CURRENCY_CLAIMED, DISPOSITION_NEEDS_HUMAN)}
    if len(retained) > 8:
        parked_keys = [item_key for item_key, value in retained.items()
                       if item_key != key
                       and value.get("disposition") == DISPOSITION_NEEDS_HUMAN]
        for item_key in parked_keys[:-7]:
            retained.pop(item_key, None)
    currency["items"] = retained
    return item


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,
                        reactivate_dispositions=(DISPOSITION_DISPATCHED, DISPOSITION_NEEDS_HUMAN)):
    """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, an item whose disposition is in `reactivate_dispositions` 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 on the thread, or an edit of the
    parked 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.

    The feedback stream excludes DISPATCHED from `reactivate_dispositions` deliberately: status
    bots rewrite their comment bodies on every push, so edit-keyed reactivation re-actionized
    handled bot comments forever and merge-ready could never fire (#1309). A `needs-human`
    feedback item still reactivates on edit — a human answering by editing their own comment must
    wake the loop, mirroring the thread semantics. (A bot comment parked `needs-human` can churn
    open on bot self-edits; that costs a redundant resolve pass, not a stuck loop — merge-ready is
    blocked by open_needs_human either way.) A brand-new comment is just a new id, always
    actionable."""
    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 reactivate_dispositions 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_feedback_activity = {
        fid: item.get("edit_id") for fid, item in prior_feedback.items()
    }
    prior_review_decision = state.get("review_decision")
    prior_change_sig = _change_sig(state)
    prior_review_in_progress = bool(state.get("review_in_progress", False))
    prior_review_signal_count = state.get("review_signal_count")
    if prior_review_signal_count is None:
        prior_review_signal_count = 1 if prior_review_in_progress else 0
    review_in_progress = bool(cur.get("review_in_progress", False))
    review_signal_count = cur.get("review_signal_count")
    if review_signal_count is None:
        review_signal_count = 1 if review_in_progress else 0
    prior_review_signal_identities = state.get("review_signal_identities")
    if not isinstance(prior_review_signal_identities, list):
        prior_review_signal_identities = None
    else:
        prior_review_signal_identities = sorted(set(map(str, prior_review_signal_identities)))
    review_signal_identities = cur.get("review_signal_identities")
    if not isinstance(review_signal_identities, list):
        review_signal_identities = None
    else:
        review_signal_identities = sorted(set(map(str, review_signal_identities)))
        review_signal_count = len(review_signal_identities)
    review_in_progress = review_signal_count > 0
    signal_changed = (review_signal_identities != prior_review_signal_identities
                      if review_signal_identities is not None else
                      review_signal_count != prior_review_signal_count)

    if head_changed:
        # SHA-scoped state is meaningless on a new head.
        state["ci_dispatched"] = {}
        state["review_signal_seen_on_head"] = review_in_progress
        state["review_signal_first_seen_at"] = _iso(now) if review_in_progress else None
        state["review_signal_last_changed_at"] = _iso(now) if review_in_progress else None
    else:
        # Preserve a signal seen on this head after it disappears; the agent owns completion judgment,
        # while this structural fact distinguishes an incomplete lifecycle from "no signal ever."
        if (prior_review_in_progress or review_in_progress) and not state.get("review_signal_seen_on_head", False):
            state["review_signal_seen_on_head"] = True
            legacy_seen_at = state.get("last_change_at") if prior_review_in_progress else None
            state["review_signal_first_seen_at"] = legacy_seen_at or _iso(now)
        if signal_changed:
            state["review_signal_last_changed_at"] = _iso(now)

    # --- 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. Its disappearance is likewise the resolver confirming the dispatched work,
    #     not fresh review activity; removals of open or `needs-human` threads remain external.
    #     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.
    #     Reactivation on a body edit is per-disposition: a `dispatched` item is NEVER
    #     auto-reactivated — status bots (changeset-bot, CodeRabbit, Codecov) rewrite their own
    #     comment bodies on every push, so edit-keyed reactivation re-actionized handled bot
    #     comments forever and the merge-ready gate could never fire (#1309) — while a
    #     `needs-human` item still reactivates on edit, because a human may answer the parked
    #     question by editing the same comment/review body. New requests otherwise arrive as
    #     review threads or brand-new comments (new ids), both still actionable; edit_id also
    #     feeds external_review_moved below, so a dispatched item's edit wakes/extends the settle
    #     clock without re-actionizing it.
    # 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")],
        reactivate_dispositions=(DISPOSITION_NEEDS_HUMAN,))
    current_thread_activity = {
        item.get("thread_id"): _thread_identity(item)
        for item in cur["threads"] if item.get("thread_id")
    }
    prior_thread_activity = {
        tid: _prior_thread_activity_identity(item) for tid, item in prior_threads.items()
        if not (item.get("disposition") == DISPOSITION_DISPATCHED
                and tid not in current_thread_activity)
    }
    current_feedback_activity = {
        item.get("id"): item.get("edit_id")
        for item in (cur.get("feedback") or []) if item.get("id")
    }
    external_review_moved = (
        head_changed
        or current_thread_activity != prior_thread_activity
        or current_feedback_activity != prior_feedback_activity
        or cur.get("review_decision") != prior_review_decision
        or signal_changed
    )
    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"] = review_in_progress
    state["review_signal_count"] = review_signal_count
    state["review_signal_identities"] = review_signal_identities
    observed_awaiting_approval = cur.get("awaiting_approval")
    approval_probe_succeeded = observed_awaiting_approval is not None
    awaiting_approval = (
        observed_awaiting_approval
        if approval_probe_succeeded
        else state.get("awaiting_approval", 0)
    )
    state["awaiting_approval"] = awaiting_approval
    if awaiting_approval > 0:
        gate_is_new_for_head = (
            state.get("blocked_external_head_sha") != head
            or not state.get("blocked_external_review_last_activity_at")
        )
        if gate_is_new_for_head:
            state["blocked_external_head_sha"] = head
            state["blocked_external_first_seen_at"] = _iso(now)
            state["blocked_external_review_last_activity_at"] = _iso(now)
        elif external_review_moved:
            state["blocked_external_review_last_activity_at"] = _iso(now)
    elif approval_probe_succeeded:
        state["blocked_external_head_sha"] = None
        state["blocked_external_first_seen_at"] = None
        state["blocked_external_review_last_activity_at"] = None
    state["pr_chain"] = cur.get("pr_chain") or _default_pr_chain()
    state["base"] = cur.get("base")
    branch_currency = _apply_branch_currency(state, cur, head, head_changed, now)

    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"]
    blocked_external_review_quiet_seconds = (
        _elapsed(state.get("blocked_external_review_last_activity_at"), now)
        if awaiting_approval > 0 else 0
    )
    # "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)
    invocation_elapsed = _active_elapsed(state, now)

    return {
        "pr_state": cur["pr_state"],
        "pr_is_draft": cur.get("pr_is_draft"),
        "mergeable": cur["mergeable"],
        "merge_state_status": cur["merge_state_status"],
        "review_decision": cur["review_decision"],
        "head_sha": head,
        "head_changed": head_changed,
        "base": state.get("base"),
        "mergeability_certain": _mergeability_certain(
            cur.get("mergeable"), cur.get("merge_state_status"), cur.get("base")),
        "base_ref_blocker": _base_ref_blocker(cur.get("base")),
        "host_branch_update_capability": cur.get("host_branch_update_capability", "unknown"),
        "branch_currency": branch_currency,
        "branch_currency_blocker": ({
            "key": branch_currency.get("key"),
            "disposition": branch_currency.get("disposition"),
            "recovery_state": branch_currency.get("recovery_state"),
        } if branch_currency else None),
        "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": review_in_progress,
        "review_signal_count": review_signal_count,
        "review_signal_identities": review_signal_identities,
        "review_signal_seen_on_head": bool(state.get("review_signal_seen_on_head", False)),
        "review_signal_first_seen_at": state.get("review_signal_first_seen_at"),
        "review_signal_last_changed_at": state.get("review_signal_last_changed_at"),
        "checks_awaiting_approval": awaiting_approval,
        "blocked_external": blocked_external,
        "blocked_external_first_seen_at": state.get("blocked_external_first_seen_at"),
        "blocked_external_review_last_activity_at": state.get(
            "blocked_external_review_last_activity_at"),
        "blocked_external_review_quiet_seconds": blocked_external_review_quiet_seconds,
        "blocked_external_review_moved_this_tick": (
            awaiting_approval > 0 and external_review_moved),
        "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,
        "invocation_id": state.get("invocation_id"),
        "invocation_started_at": state.get("started_at"),
        "invocation_elapsed_seconds": invocation_elapsed,
        "invocation_budget_seconds": state.get("invocation_budget_seconds"),
        "invocation_remaining_seconds": max(
            0,
            (state.get("invocation_budget_seconds") or 0) - invocation_elapsed,
        ),
        "invocation_wall_elapsed_seconds": _elapsed(state.get("started_at"), now),
        "invocation_dead_time_seconds": int(state.get("dead_time_seconds") or 0),
        "invocation_backstop_seconds": state.get("invocation_backstop_seconds"),
        "persisted_state_created_at": state.get("state_created_at"),
        "persisted_state_age_seconds": _elapsed(state.get("state_created_at"), now),
        # Compatibility aliases for callers migrating from the original single-clock contract.
        "session_started_at": state.get("started_at"),
        "session_seconds": invocation_elapsed,
        "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()}
    # edit_id is part of the signal so a body edit of a silenced item still resets the settle
    # clock (review activity happened) even though it no longer reopens the item (#1309).
    feedback = {fid: (v.get("disposition"), v.get("edit_id"))
                for fid, v in state.get("feedback", {}).items()}
    currency = _load_branch_currency_state(state)
    current_currency = currency.get("items", {}).get(currency.get("current_key")) or {}
    return (checks, threads, feedback, state.get("review_decision"), state.get("mergeable"),
            state.get("merge_state_status"), state.get("review_in_progress"),
            state.get("review_signal_count"),
            tuple(state.get("review_signal_identities") or []),
            json.dumps(state.get("pr_chain") or _default_pr_chain(), sort_keys=True),
            json.dumps(state.get("base"), sort_keys=True),
            (current_currency.get("host_branch_update_capability")
             if current_currency.get("status") == "BEHIND" else None),
            currency.get("current_key"), current_currency.get("disposition"),
            current_currency.get("attention"), current_currency.get("recovery_state"),
            current_currency.get("retry_count"), current_currency.get("mutation_consumed"),
            # 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 _parse_iso8601(value):
    if isinstance(value, str) and value.endswith("Z"):
        value = value[:-1] + "+00:00"
    return datetime.fromisoformat(value)


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


def _advance_activity(state, now, threshold=DEAD_TIME_THRESHOLD_SECONDS, accumulate=True):
    """Advance the activity heartbeat, and (for the in-session watch) accumulate suspended time.

    Every watch poll and every agent snapshot/mark marks activity. A gap wider than `threshold`
    means the whole process was not running — a suspended machine — so, when accumulating, the
    excess beyond the threshold is charged to dead time and later excluded from active elapsed.
    Agent snapshots/marks bump the heartbeat with `accumulate=False`, so active agent work keeps the
    heartbeat fresh and is not refunded as long as the agent touches state (snapshots or marks) within
    the threshold; a single silent tick longer than the threshold is the coarse-discriminator limit
    the plan defers, bounded by the wall-clock backstop. Dead time only ever grows in the watch loop.
    Clock-backward safe: `_elapsed` floors negative deltas at 0 and the accumulator never decreases.
    """
    last = state.get("last_activity_at") or state.get("started_at")
    if accumulate:
        gap = _elapsed(last, now)
        if gap > threshold:
            state["dead_time_seconds"] = (state.get("dead_time_seconds") or 0) + (gap - threshold)
    state["last_activity_at"] = _iso(now)


def _active_elapsed(state, now):
    """Wall-clock elapsed since the invocation anchor, minus accumulated suspended (dead) time."""
    wall = _elapsed(state.get("started_at"), now)
    return max(0, wall - int(state.get("dead_time_seconds") or 0))


def _session_started_at(value):
    """Parse one invocation-wide, timezone-aware budget anchor for reuse across state dirs."""
    try:
        parsed = _parse_iso8601(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):
    state_path = os.path.join(state_dir, "state.json")
    with _state_lock(state_dir):
        if os.path.exists(state_path):
            with open(state_path) as f:
                state = json.load(f)
        else:
            state = _empty_state(pr, repo, None, now)
        # Older state used `started_at` for both durable-state age and the active budget. Preserve
        # that original value as the state birth time before a new invocation replaces the clock.
        state.setdefault("state_created_at", state.get("started_at") or _iso(now))
        # Active-time accounting fields migrate onto legacy state on first observation. Seed the
        # heartbeat to *now*, not the old anchor: there is no recorded activity for the pre-migration
        # period, so charging that whole span as one supra-threshold gap would refund the entire
        # historical invocation (a 9h-old 8h run would read as ~15 min active and never max-runtime).
        # Seeding to load time makes the first poll see a ~0 gap and keeps elapsed on real wall-clock.
        state.setdefault("last_activity_at", _iso(now))
        state.setdefault("dead_time_seconds", 0.0)
        state.setdefault("invocation_backstop_seconds", DEFAULT_INVOCATION_BACKSTOP_SECONDS)
        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()
        _replace_atomic(tmp.name, state_path)


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_invocation(box, args, now):
    """Start explicitly, or reuse the one fixed clock named by an explicit invocation token."""
    state = box["state"]
    start_requested = bool(getattr(args, "start_invocation", False)
                           or getattr(args, "reset_session", False))
    continue_requested = bool(getattr(args, "continue_invocation", False))
    invocation_id = getattr(args, "invocation_id", None)
    requested_started_at = getattr(args, "session_started_at", None)
    requested_budget = getattr(args, "invocation_budget_seconds", None)

    if start_requested:
        invocation_id = uuid.uuid4().hex
        requested_started_at = _iso(now)
        requested_budget = requested_budget or DEFAULT_INVOCATION_BUDGET_SECONDS
        state["invocation_id"] = invocation_id
        state["started_at"] = requested_started_at
        state["invocation_budget_seconds"] = requested_budget
        # A fresh invocation resets the active-time clock: heartbeat at now, no dead time yet.
        state["last_activity_at"] = _iso(now)
        state["dead_time_seconds"] = 0.0
        state.setdefault("invocation_backstop_seconds", DEFAULT_INVOCATION_BACKSTOP_SECONDS)
        args.invocation_id = invocation_id
        args.session_started_at = requested_started_at
        args.invocation_budget_seconds = requested_budget
        args._started_new_invocation = True
        return

    if not invocation_id:
        raise SystemExit("pass --start-invocation on the first snapshot or --invocation-id to resume it")

    if continue_requested:
        if not requested_started_at or not requested_budget:
            raise SystemExit("--continue-invocation requires its fixed start and budget")
        if state.get("invocation_id") == invocation_id:
            try:
                same_anchor = (_parse_iso8601(state.get("started_at"))
                               == _parse_iso8601(requested_started_at))
            except (ValueError, TypeError):
                same_anchor = False
            if not same_anchor or state.get("invocation_budget_seconds") != requested_budget:
                raise SystemExit("continuation cannot renew or extend an existing invocation")
            # Honor a carry arg on a same-invocation re-continue too, as a monotonic floor: raise the
            # dead-time to the carried value without clobbering time this dir already accumulated
            # (dead time only ever grows). This makes the carry order-independent — it lands whether
            # it arrives on the first adopt or a later re-continue.
            continue_dead_time = getattr(args, "continue_dead_time_seconds", None)
            if continue_dead_time is not None:
                state["dead_time_seconds"] = max(
                    state.get("dead_time_seconds") or 0, float(continue_dead_time))
            return
        state["invocation_id"] = invocation_id
        state["started_at"] = requested_started_at
        state["invocation_budget_seconds"] = requested_budget
        # Adopting an invocation into this state dir establishes a fresh active-time clock. Seed the
        # heartbeat to now (not the shared anchor, which may be hours old) so the first poll sees a
        # ~0 gap and does not refund the span between the anchor and adoption as dead time.
        state["last_activity_at"] = _iso(now)
        # Dead time is per-state-dir, but a managed-stack continuation shares one active-time budget
        # across layers, so it explicitly carries the prior layer's accumulated dead time here. Absent
        # that arg (an ordinary adopt of an unrelated dir), reset to 0 rather than inherit stale state.
        continue_dead_time = getattr(args, "continue_dead_time_seconds", None)
        state["dead_time_seconds"] = float(continue_dead_time) if continue_dead_time is not None else 0.0
        state.setdefault("invocation_backstop_seconds", DEFAULT_INVOCATION_BACKSTOP_SECONDS)
        return

    persisted_id = state.get("invocation_id")
    persisted_started_at = state.get("started_at")
    persisted_budget = state.get("invocation_budget_seconds")
    if persisted_id != invocation_id:
        raise SystemExit("invocation token does not match persisted state; start or continue explicitly")
    try:
        anchors_match = (_parse_iso8601(persisted_started_at)
                         == _parse_iso8601(requested_started_at))
    except (ValueError, TypeError):
        anchors_match = False
    if not anchors_match:
        raise SystemExit("invocation token does not match the persisted budget anchor")
    if persisted_budget != requested_budget:
        raise SystemExit("invocation token does not match the persisted fixed budget")


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"]
    _apply_invocation(box, args, 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)
    # Re-capture the clock *after* the (blocking) fetch so activity accounting reflects post-fetch
    # time: a suspend that lands during the fetch is then credited as dead time by this poll's
    # _advance_activity, instead of leaving the next top-of-loop budget check to fire max-runtime on
    # a gap it hasn't yet excluded. It also keeps a new invocation's slow first fetch off its budget.
    now = _now()
    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()
        if (watch_generation is not None
                and box["state"].get("invocation_id") != getattr(args, "invocation_id", None)):
            raise _InvocationSuperseded(box["state"].get("invocation_id"))
        # Only the self-sustaining in-session watch (watch_generation set) accumulates dead time;
        # an agent-driven snapshot bumps the heartbeat without accumulating, so checkpoint/durable
        # runs — which have no continuous poll cadence — keep pure wall-clock accounting.
        if box["state"].get("started_at"):
            _advance_activity(box["state"], now, accumulate=watch_generation is not None)
        actionable, box["state"] = _apply_snapshot(box, args, cur, now, advance_trajectory)
    return actionable


def cmd_snapshot(args):
    result = _run_snapshot(args, _now())
    if (getattr(args, "_started_new_invocation", False)
            and _elapsed(result.get("invocation_started_at"), _now()) > 60):
        raise SystemExit("new invocation unexpectedly inherited more than 60 seconds of elapsed time")
    print(json.dumps(result, 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"
    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"
    base_ref_blocker = (a.get("base_ref_blocker")
                        if "base_ref_blocker" in a else _base_ref_blocker(a.get("base")))
    if base_ref_blocker:
        return "base-ref-blocked"
    branch_currency = a.get("branch_currency") or {}
    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"
    # Branch maintenance is a third attention stream. It runs only after new review/CI work and
    # standing red blockers are clear, but a passive in-progress check or review signal does not
    # delay a BEHIND update that will invalidate that work anyway. Claimed observations wake only
    # for reconciliation; confirmed and parked residuals remain quiet.
    if branch_currency.get("attention") in (CURRENCY_ATTENTION_CLAIM,
                                             CURRENCY_ATTENTION_INSPECT,
                                             CURRENCY_ATTENTION_RECONCILE):
        return "branch-currency"
    if a.get("blocked_external"):
        return "blocked-external"
    if (a.get("open_needs_human", 0)
            or branch_currency.get("disposition") == DISPOSITION_NEEDS_HUMAN):
        return "needs-human"         # parked items block ready; surface them
    # merge-ready candidate: green + settled with no fresh review movement. A live 👀 blocks the
    # ordinary 300s settle, but wakes the agent after 900 quiet seconds for the semantic stale-review
    # judgment. The agent may extend once to 1800s by re-arming with that settle value; the detector
    # never decides whether the reviewer is genuinely slow or stuck.
    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.)
    mergeability_certain = a.get("mergeability_certain")
    if mergeability_certain is None:
        mergeability_certain = _mergeability_certain(
            a.get("mergeable"), a.get("merge_state_status"), a.get("base"))
    if (mergeability_certain and 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 a.get("branch_currency_blocker") is None
            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__")
        sig.add(("approval-review-state", bool(a.get("review_signal_seen_on_head")),
                 tuple(a.get("review_signal_identities") or []), a.get("review_decision")))
    if a.get("stack_blocker"):
        sig.add(f"__stack__:{a['stack_blocker']}")
    if a.get("base_ref_blocker"):
        sig.add(f"__base_ref__:{a['base_ref_blocker']}")
    branch_currency = a.get("branch_currency") or {}
    currency_disposition = branch_currency.get("disposition")
    if (currency_disposition in (CURRENCY_CONFIRMED, DISPOSITION_NEEDS_HUMAN)
            or (currency_disposition == CURRENCY_CLAIMED
                and not branch_currency.get("attention"))):
        sig.add(("currency", branch_currency.get("key"), currency_disposition,
                 branch_currency.get("recovery_state")))
    # 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."""
    if IS_WINDOWS:
        return _win_process_identity(pid)
    try:
        r = subprocess.run(["ps", "-p", str(pid), "-o", "lstart=", "-o", "command="],
                           capture_output=True, text=True, encoding="utf-8")
    except OSError:
        return None   # no ps on PATH: unproven identity, same as a failed lookup
    if r.returncode != 0:
        return None
    return (r.stdout or "").strip() or None


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 _state_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()
        _replace_atomic(tmp.name, _watch_candidate_path(args))
    return previous


def _clear_watch_candidate(args, generation):
    with _state_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
        # The arming poll is a real in-session watch poll: accumulate any suspended span since the
        # last activity (e.g. the watch armed right after the machine resumed) before reading elapsed.
        if state.get("started_at"):
            _advance_activity(state, now, accumulate=True)
        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 _state_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. Invocation replacement also supersedes every ordinary
    wake from the old budget, even when the replacement kept the same watch generation."""
    state_path = os.path.join(args.state_dir, "state.json")
    with _state_lock(args.state_dir, exclusive=False):
        if not os.path.exists(state_path):
            return False
        with open(state_path) as f:
            state = json.load(f)
        if state.get("watch_generation") != generation:
            return False
        current_invocation_id = state.get("invocation_id")
        watcher_invocation_id = getattr(args, "invocation_id", None)
        if current_invocation_id != watcher_invocation_id and reason != "invocation-superseded":
            _emit_wake(
                "invocation-superseded", watch_generation=generation,
                superseded_invocation_id=watcher_invocation_id,
                current_invocation_id=current_invocation_id,
            )
            return True
        _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.

    On Windows os.kill maps to TerminateProcess, so the replaced watcher dies abruptly instead of
    unwinding a SIGTERM handler: it cannot clear its own watch candidate, nor finish terminating a
    predecessor it was mid-handoff with. Neither leaks. The replacement has already written its own
    candidate before calling this, and any watcher left running reads a foreign watch_generation on
    its next poll and exits on its own — so the abrupt path costs at most one poll interval."""
    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 OSError:
        pass   # exited between the identity check and here (Windows reports this as a plain OSError)


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()

    # Cooperative supersession. Windows accepts this handler but never delivers SIGTERM across
    # processes, so there it is inert and the loop's own watch_generation checks are what retire a
    # replaced watcher — the handler is the fast path, not the correctness guarantee.
    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
        _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
            now = _now()
            # Active elapsed = raw wall-clock minus accumulated suspended time (dead_time was
            # updated by the previous poll's _run_snapshot, including any gap from a resumed
            # suspension). The 8h budget spends active time; the 3-day backstop is raw wall-clock.
            wall_elapsed = _elapsed(actionable.get("invocation_started_at"), now)
            dead_time = int(actionable.get("invocation_dead_time_seconds") or 0)
            invocation_elapsed = max(0, wall_elapsed - dead_time)
            budget = actionable.get("invocation_budget_seconds") or 0
            backstop = actionable.get("invocation_backstop_seconds") or 0
            actionable["invocation_elapsed_seconds"] = invocation_elapsed
            actionable["invocation_remaining_seconds"] = max(0, budget - invocation_elapsed)
            actionable["invocation_wall_elapsed_seconds"] = wall_elapsed
            actionable["persisted_state_age_seconds"] = _elapsed(
                actionable.get("persisted_state_created_at"), now)
            reason = _wake_reason(actionable, args.settle_seconds)
            if reason in ("terminal", "merge-ready"):
                _emit_wake_if_current(args, generation, reason, url=actionable.get("url"),
                                      pr_state=actionable.get("pr_state"), counts=actionable.get("counts"))
                return
            active_cap_hit = bool(budget) and invocation_elapsed >= budget
            backstop_hit = bool(backstop) and wall_elapsed >= backstop
            if active_cap_hit or backstop_hit:
                _emit_wake_if_current(
                    args, generation, "max-runtime", url=actionable.get("url"),
                    invocation_id=actionable.get("invocation_id"),
                    invocation_started_at=actionable.get("invocation_started_at"),
                    invocation_elapsed_seconds=actionable.get("invocation_elapsed_seconds"),
                    invocation_budget_seconds=budget,
                    invocation_wall_elapsed_seconds=wall_elapsed,
                    invocation_backstop_seconds=backstop or None,
                    max_runtime_ceiling=("backstop" if backstop_hit and not active_cap_hit
                                         else "active-budget"),
                    persisted_state_age_seconds=actionable.get("persisted_state_age_seconds"),
                )
                return
            drain_seconds = getattr(args, "blocked_external_drain_seconds", None)
            drain_quiet = actionable.get("blocked_external_review_quiet_seconds", 0)
            if (drain_seconds is not None and actionable.get("blocked_external")
                    and drain_quiet >= drain_seconds):
                _emit_wake_if_current(
                    args, generation, "blocked-external-drained", url=actionable.get("url"),
                    pr_state=actionable.get("pr_state"), counts=actionable.get("counts"),
                    blocked_external_review_quiet_seconds=drain_quiet,
                    blocked_external_drain_seconds=drain_seconds,
                )
                return
            approval_review_moved = (
                reason == "blocked-external"
                and actionable.get("blocked_external_review_moved_this_tick"))
            if (reason in ("needs-human", "blocked-failing", "blocked-external",
                           "stack-blocked", "base-ref-blocked")
                    and not approval_review_moved 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
            remaining = max(0, budget - invocation_elapsed) if budget else args.interval
            wait_seconds = min(args.interval, remaining)
            if stop_requested.wait(wait_seconds):
                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
    except _InvocationSuperseded as exc:
        superseded = True
        _emit_wake_if_current(
            args, generation, "invocation-superseded",
            superseded_invocation_id=getattr(args, "invocation_id", None),
            current_invocation_id=exc.current_invocation_id,
        )
        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 _mark_branch_currency(args, state, now):
    currency = _load_branch_currency_state(state)
    key = args.currency_key
    if not key or currency.get("current_key") != key:
        raise SystemExit("currency mark requires the exact current observation key")
    item = currency.get("items", {}).get(key)
    if not isinstance(item, dict):
        raise SystemExit("currency mark requires a current observed item")
    prior = item.get("disposition", DISPOSITION_OPEN)
    outcome = args.currency_outcome
    inspected_fingerprint = args.currency_inspected_fingerprint

    if outcome:
        if prior != CURRENCY_CLAIMED:
            raise SystemExit("currency outcomes require a claimed observation")
        item["reconciled_invocation_id"] = state.get("invocation_id")
        item["reconciled_at"] = _iso(now)
        if outcome == CURRENCY_OUTCOME_MUTATION_OBSERVED:
            item["mutation_consumed"] = True
            item["mutation_observed_at"] = _iso(now)
            item["recovery_state"] = CURRENCY_OUTCOME_MUTATION_OBSERVED
        elif outcome == CURRENCY_OUTCOME_AMBIGUOUS:
            item["recovery_state"] = CURRENCY_OUTCOME_AMBIGUOUS
        elif outcome == CURRENCY_OUTCOME_PROVEN_NO_MUTATION:
            if item.get("mutation_consumed"):
                raise SystemExit("cannot record no mutation after mutation start was observed")
            retries = int(item.get("retry_count", 0))
            if retries < 1:
                item["retry_count"] = retries + 1
                item["disposition"] = DISPOSITION_OPEN
                item["recovery_state"] = "retry-authorized"
                item["retry_not_before"] = _iso(
                    now + timedelta(seconds=CURRENCY_RETRY_BACKOFF_SECONDS))
                item.pop("claimed_invocation_id", None)
                item.pop("reconciled_invocation_id", None)
            else:
                item["disposition"] = DISPOSITION_NEEDS_HUMAN
                item["recovery_state"] = "retry-exhausted"
        state["last_action"] = f"{outcome} currency {key}"
        return

    if inspected_fingerprint:
        if prior != DISPOSITION_OPEN:
            raise SystemExit("currency inspection requires an open observation")
        parks = currency.get("semantic_parks") or {}
        if not parks:
            raise SystemExit("currency inspection requires carried semantic conflict evidence")
        item["inspected_semantic_conflict_fingerprint"] = inspected_fingerprint
        item["inspected_at"] = _iso(now)
        if inspected_fingerprint in parks:
            item["disposition"] = DISPOSITION_NEEDS_HUMAN
            item["semantic_conflict_fingerprint"] = inspected_fingerprint
            item["recovery_state"] = "semantic-unchanged"
            item["inspection_result"] = "unchanged"
        else:
            item["inspection_required"] = False
            item["inspection_result"] = "changed"
            item["recovery_state"] = "semantic-changed"
            # The preview describes the current conflict set for this head. Once it differs, the
            # carried fingerprints are no longer standing residuals and must not tax every later
            # base generation with inspection of stale evidence.
            currency["semantic_parks"] = {}
            item["parked_semantic_fingerprints"] = []
        state["last_action"] = f"inspected currency {key}"
        return

    disposition = args.currency_disposition
    allowed = {
        DISPOSITION_OPEN: {DISPOSITION_OPEN, CURRENCY_CLAIMED, DISPOSITION_NEEDS_HUMAN},
        CURRENCY_CLAIMED: {CURRENCY_CLAIMED, CURRENCY_CONFIRMED, DISPOSITION_NEEDS_HUMAN},
        CURRENCY_CONFIRMED: {CURRENCY_CONFIRMED, DISPOSITION_OPEN},
        DISPOSITION_NEEDS_HUMAN: {DISPOSITION_NEEDS_HUMAN, DISPOSITION_OPEN},
    }
    if disposition not in allowed.get(prior, set()):
        raise SystemExit(f"invalid currency transition: {prior} -> {disposition}")
    if (prior == DISPOSITION_OPEN and disposition == CURRENCY_CLAIMED
            and item.get("inspection_required")):
        raise SystemExit("currency claim requires semantic-fingerprint inspection first")
    if prior == DISPOSITION_OPEN and disposition == CURRENCY_CLAIMED:
        retry_not_before = item.get("retry_not_before")
        if retry_not_before:
            try:
                if _parse_iso8601(retry_not_before) > now:
                    raise SystemExit("currency retry backoff has not elapsed")
            except (ValueError, TypeError):
                raise SystemExit("currency retry backoff is invalid")
        budget = state.get("invocation_budget_seconds")
        backstop = state.get("invocation_backstop_seconds")
        if ((budget is not None and _active_elapsed(state, now) >= budget)
                or (backstop is not None and _elapsed(state.get("started_at"), now) >= backstop)):
            raise SystemExit("currency claim cannot start after max-runtime")
    item["disposition"] = disposition
    item["transitioned_at"] = _iso(now)
    if disposition == CURRENCY_CLAIMED:
        # A repeated claim is deliberately idempotent. It cannot transfer or renew an existing
        # claim; a later invocation must reconcile it and record an explicit outcome.
        if prior == DISPOSITION_OPEN:
            item["claimed_invocation_id"] = state.get("invocation_id")
            item["claimed_at"] = _iso(now)
            item["attempt_number"] = int(item.get("retry_count", 0)) + 1
            item["mutation_consumed"] = False
            item["recovery_state"] = "claimed"
            item.pop("reconciled_invocation_id", None)
    elif disposition == CURRENCY_CONFIRMED:
        item["confirmed_invocation_id"] = state.get("invocation_id")
    elif disposition == DISPOSITION_NEEDS_HUMAN:
        fingerprint = args.semantic_conflict_fingerprint
        if fingerprint:
            item["semantic_conflict_fingerprint"] = fingerprint
            currency.setdefault("semantic_parks", {})[fingerprint] = {
                "head_sha": item.get("head_sha"),
                "status": item.get("status"),
                "route": item.get("route"),
                "observation_key": key,
            }
    elif disposition == DISPOSITION_OPEN:
        old_fingerprint = item.pop("semantic_conflict_fingerprint", None)
        if old_fingerprint:
            currency.setdefault("semantic_parks", {}).pop(old_fingerprint, None)
        item.pop("claimed_invocation_id", None)
        item.pop("confirmed_invocation_id", None)
        item.pop("reconciled_invocation_id", None)
        item.pop("inspected_semantic_conflict_fingerprint", None)
        item.pop("inspection_result", None)
        item["retry_count"] = 0
        item["mutation_consumed"] = False
        item.pop("retry_not_before", None)
        item.pop("recovery_state", None)
    item["parked_semantic_fingerprints"] = sorted(currency.get("semantic_parks") or {})
    state["last_action"] = f"{disposition} currency {key}"


def cmd_mark(args):
    now = _now()
    with locked_state(args.state_dir, args.pr, args.repo, now) as box:
        _apply_invocation(box, args, now)
        state = box["state"]
        # An agent-driven mark is activity: bump the heartbeat (no accumulation) so a long tick
        # that only marks — never snapshots — still never has its active time refunded as dead time.
        if state.get("started_at"):
            _advance_activity(state, now, accumulate=False)
        if (args.currency_key or args.currency_disposition or args.currency_outcome
                or args.currency_inspected_fingerprint):
            actions = sum(bool(value) for value in (
                args.currency_disposition, args.currency_outcome,
                args.currency_inspected_fingerprint))
            if not args.currency_key or actions != 1:
                raise SystemExit("currency marks require --currency-key and exactly one currency action")
            _mark_branch_currency(args, state, now)
        elif 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). Only a `needs-human` mark ever reads it (dispatched comments no
                    # longer reactivate on edit, #1309), but storing it unconditionally is
                    # harmless and closes the answered-by-edit race for parked comments.
                    entry["acted_identity"] = [args.acted_edit_id]
            state["last_action"] = f"{args.disposition} {label} {item_id}"
    print(json.dumps({"marked": args.currency_key or args.check or args.thread or args.comment}))


WATCH_BOOTSTRAP_FLAGS = ("--state-dir", "--invocation-id", "--session-started-at",
                         "--invocation-budget-seconds")
WATCH_BOOTSTRAP_HINT = """\
watch cannot start a babysit run: it only arms the change detector for an
invocation that was already bootstrapped by a snapshot --start-invocation run
(the ce-babysit-pr Step 2 bootstrap).
To recover, invoke the ce-babysit-pr skill through your harness's callable
skill mechanism with this PR: its instructions own the bootstrap, arming,
marks, and stop protocol. Do not drive pr-snapshot directly outside that skill.
Never mint the bootstrap values yourself; they come from the skill's
bootstrap snapshot."""


class _WatchHintingParser(argparse.ArgumentParser):
    # A bare `watch --pr N` is the observed illegal start path (an agent arming the
    # detector without the Step 2 bootstrap). Keep the fail-closed exit 2, but make
    # the refusal carry its own recovery path instead of a raw usage dump.
    def error(self, message):
        if (self.prog.endswith(" watch") and "required" in message
                and any(flag in message for flag in WATCH_BOOTSTRAP_FLAGS)):
            self.print_usage(sys.stderr)
            self.exit(2, f"{self.prog}: error: {message}\n\n{WATCH_BOOTSTRAP_HINT}\n")
        super().error(message)


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

    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("--start-invocation", action="store_true",
                    help="mint one new fixed budget (first snapshot only)")
    sg.add_argument("--reset-session", action="store_true",
                    help=argparse.SUPPRESS)  # deprecated alias for --start-invocation
    sg.add_argument("--continue-invocation", action="store_true",
                    help="carry the current fixed budget into a managed-stack layer state dir")
    s.add_argument("--invocation-id", default=None,
                   help="token emitted by the first snapshot; required on every resume")
    s.add_argument("--session-started-at", type=_session_started_at,
                   help="fixed anchor emitted by the first snapshot; required for a new state dir")
    s.add_argument("--invocation-budget-seconds", type=float,
                   help=f"fixed total budget; first snapshot default {DEFAULT_INVOCATION_BUDGET_SECONDS}")
    s.add_argument("--continue-dead-time-seconds", type=float, default=None,
                   help="carry the prior layer's accumulated dead time into a managed-stack layer "
                        "(with --continue-invocation), so the shared active-time budget stays correct")
    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("--invocation-id", required=True,
                   help="token emitted by the invocation's first snapshot")
    m.add_argument("--session-started-at", type=_session_started_at, required=True,
                   help="fixed anchor emitted by the invocation's first snapshot")
    m.add_argument("--invocation-budget-seconds", type=float, required=True,
                   help="fixed budget emitted by the invocation's first snapshot")
    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; read only by needs-human comment reactivation (#1309)
    m.add_argument("--currency-key", default=None,
                   help="exact branch-currency observation key emitted by snapshot")
    m.add_argument("--currency-disposition",
                   choices=[DISPOSITION_OPEN, CURRENCY_CLAIMED, CURRENCY_CONFIRMED,
                            DISPOSITION_NEEDS_HUMAN], default=None)
    m.add_argument("--semantic-conflict-fingerprint", default=None)
    m.add_argument("--currency-outcome",
                   choices=[CURRENCY_OUTCOME_MUTATION_OBSERVED,
                            CURRENCY_OUTCOME_PROVEN_NO_MUTATION,
                            CURRENCY_OUTCOME_AMBIGUOUS], default=None)
    m.add_argument("--currency-inspected-fingerprint", default=None)
    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("--settle-seconds", type=float, default=300.0, help="quiet window before a merge-ready wake")
    w.add_argument("--blocked-external-drain-seconds", type=float, default=None,
                   help="head-scoped review-quiet bound before a gated-CI handback")
    w.add_argument("--stop-file", default=None, help="path whose existence stops the watch")
    w.add_argument("--fetch-file", default=None)
    w.add_argument("--invocation-id", required=True,
                   help="token emitted by the invocation's first snapshot")
    w.add_argument("--session-started-at", type=_session_started_at, required=True,
                   help="fixed anchor emitted by the first snapshot")
    w.add_argument("--invocation-budget-seconds", type=float, required=True,
                   help="fixed budget emitted by the invocation's first snapshot")
    w.set_defaults(func=cmd_watch)

    args = p.parse_args()
    if (args.cmd in ("snapshot", "watch")
            and args.invocation_budget_seconds is not None
            and args.invocation_budget_seconds <= 0):
        p.error("--invocation-budget-seconds must be positive")
    if (args.cmd == "watch" and args.blocked_external_drain_seconds is not None
            and args.blocked_external_drain_seconds <= 0):
        p.error("--blocked-external-drain-seconds must be positive")
    if args.cmd == "snapshot":
        starting = args.start_invocation or args.reset_session
        continuing = args.continue_invocation
        if starting and (args.invocation_id or args.session_started_at):
            p.error("the first snapshot cannot combine --start-invocation with resume fields")
        if continuing and not (args.invocation_id and args.session_started_at
                               and args.invocation_budget_seconds):
            p.error("--continue-invocation requires --invocation-id, --session-started-at, and --invocation-budget-seconds")
        if not starting and not continuing and not (args.invocation_id and args.session_started_at
                                                     and args.invocation_budget_seconds):
            p.error("snapshot requires --start-invocation or --invocation-id")
    args.func(args)


if __name__ == "__main__":
    main()
