box/bin/muse_choice_watcher.py

#!/usr/bin/env python3
"""muse_choice_watcher.py — Per-pane daemon auto-approving Muse prompts.

Kinds (prompt -> key): native Muse approval menu -> "1", agent
interview menu -> "1", explicit-phrase request -> captured TOKEN,
lettered A/B/C choice -> "A", numbered (1)/(2) menu -> "1",
cursorless dotted 1./2./3. grill question -> "1" (rule-held for
coordinator sign-off, never expires to approve), y/n
line-end prompt -> "y", /permissions screens -> builtin-hold (never
auto-answered; the operator drives mode changes by hand).
Each prompt is answered at most once
(after stability + re-verify), with an hourly cap as backstop.

Design notes (from the agy watcher post-mortem):
  * State is namespaced by (socket, pane): pidfiles and logs embed a socket
    slug, so identical pane ids on different tmux sockets (e.g. %0 on both
    `default` and `lte`) never collide.
  * Logging never kills the loop: every log call and every poll iteration is
    exception-guarded, so auto-answers continue even when logging or tmux
    hiccups. Log rotation is best-effort.
  * Answers are once-per-prompt: a prompt signature must be stable across N
    polls, is answered at most once, and re-answering requires the prompt to
    disappear and a new signature to appear. An hourly cap bounds runaways.

Usage:
  muse_choice_watcher.py start --socket /tmp/tmux-1000/default --pane %37
  muse_choice_watcher.py start-all [--dry-run]
  muse_choice_watcher.py status
  muse_choice_watcher.py stop --socket ... --pane ...
  muse_choice_watcher.py match < pane.txt   # print JSON verdict for tuning
"""

import argparse
import fcntl
import hashlib
import json
import os
import re
import shlex
import shutil
import signal
import subprocess
import sys
import time
from datetime import datetime, timezone

STATE_DIR = "/tmp"
FILE_PREFIX = "muse-choice-watcher"
REPO_ROOT = "/home/super/Projects/NetVM"
DESIRED_STATE_FILE = os.path.join(REPO_ROOT, ".state", "muse-choices.json")
CTL_LOG = os.path.join(REPO_ROOT, "box-ctl.jsonl")
KNOWN_SOCKETS = [
    "/tmp/tmux-1000/default",
    "/tmp/tmux-1000/lte",
    "/tmp/tmux-muse.sock",
]

# Sockets whose sole purpose is self-running fleet agents. Prompts on
# fleet sockets are answered (and logged) regardless of approval-mode
# launch flags, and get a higher hourly answer budget: a headless
# builder under on-request approvals burns many answers per hour.
FLEET_SOCKETS = [
    "/tmp/tmux-muse.sock",
]

POLL_INTERVAL = 0.5
STABILITY_POLLS = 2
MAX_ANSWERS_PER_HOUR = 20
FLEET_MAX_ANSWERS_PER_HOUR = 200
ANSWERED_TTL_SECONDS = 600  # identical prompt back after 10m => stuck, allow one recovery answer
CLAIM_TTL_SECONDS = 30  # concurrent-claim window: bounds wedge if winner dies pre-send
RULES_FILE = os.path.join(REPO_ROOT, "muse-choices-rules.json")
HOLD_WINDOW_SECONDS = 120  # D3: short hold window, then expire to approve
NEGATIVE_KEYS = {"muse-approval": "2", "yn": "n"}  # D2 deny keys
QUESTION_KINDS = frozenset({"interview", "letter", "numbered", "explicit-phrase",
                            "numbered-plain", "permissions"})
HOLD_FOREVER_KINDS = frozenset({"numbered-plain"})  # coordinator-gated:
# holds renew instead of expiring to approve; resolve or manual answer only
AGENT_PANE_HINTS = ("muse-bin", "muse-code", "agy.bin")
# pane_current_command substrings: every agent harness is watched by
# default. muse-code is the launch shim: a fresh pane shows it until
# the muse-bin exec lands, so omitting it loses the launch race (no
# watcher starts and none is reported).
HARNESS_HINTS = (("muse", ("muse-bin", "muse-code")),
                 ("agy", ("agy.bin",)))  # argv[0] substrings per harness
AGY_HOLD_RULE = "agy-hold-all"  # builtin: agy sends unproven, hold all
_RULES_CACHE = {"key": None, "rules": []}
LOG_MAX_BYTES = 1_000_000
HEARTBEAT_SECONDS = 60
SUBMIT_GAP_SECONDS = 0.4  # text..Enter gap: the muse composer treats a
# back-to-back burst as paste and inserts a newline instead of submitting
COMPOSER_KINDS = frozenset({"letter", "numbered", "yn", "explicit-phrase",
                               "numbered-plain"})
VERIFY_POLL_SECONDS = 0.1
VERIFY_TEXT_TIMEOUT = 1.5  # unchanged past this: fail open to blind send
POST_ENTER_SETTLE = 0.4  # submit processing before the post-verify capture
COMPOSER_TAIL_WINDOW = 10  # trailing lines scanned for the live input row
TAIL_WINDOW = 25        # only prompts in the last N lines count (no stale scrollback)
OPTION_SPAN = 12        # A..C option lines must fit within N lines (wrapped lines ok)
MAX_OPTION_LEN = 160    # option lines longer than this are ignored (prose guard)
WRAPPED_TAIL_CAP = 4    # wrapped continuation rows skipped past the last option
CUE_BELOW_ROWS = 3      # cue may sit this far below the option block end

# A. text / B) text / C: text / C - text  (single capital letter + delimiter)
OPTION_RE = re.compile(r"^\s*([A-Z])\s*[.\)\-:]\s+\S")
# Cue that the lettered list is a choice awaiting reply (not prose).
CUE_RE = re.compile(
    r"(choose|choice|choices|select|option|options|reply|feedback|"
    r"which one|pick one|enter\s+[A-Z]\b|type\s+[A-Z]\b|press\s+[A-Z]\b|"
    r"A\s*/\s*B|A\s*,\s*B|A-C|A,B,C|\(A\))",
    re.IGNORECASE,
)
QUESTION_RE = re.compile(r"\?\s*$")

# y/n token must END the line: prompts awaiting input put the options last
# ("Proceed? (y/n)", "Overwrite? [y/N]"). Mid-line mentions are prose.
YN_RE = re.compile(r"([yY]/[nN]|\[[yY]/[nN]\])\s*[\]:)>]?\s*$")
YN_WINDOW = 6  # y/n prompt must sit in the last N content lines
# Numbered menus: (1) text / (2) text with a selection cue nearby.
NUMBERED_OPT_RE = re.compile(r"^\s*\((\d+)\)\s+\S")
NUMBERED_CUE_RE = re.compile(
    r"(Option:|Selection:|choose|choice|select an? |enter (a )?number|pick a number)",
    re.IGNORECASE,
)
# Native Muse approval dialog patterns live in MUSE_APPROVAL_*_RE below
# (single definition; the dialog shape is asserted by TestMuseApproval).# Native Muse TUI approval menu (the live auto-approve target):
#   Would you like to run the following
#     $ <command echo, often wrapped across lines>
#   > 1. Yes, proceed (y)
#     2. No, and tell Muse Code what to ...
# The cue carries no trailing "?", and the cursor may render as an ascii
# ">" or the single right-pointing angle quote (U+203A, escaped below).
MUSE_APPROVAL_CUE_RE = re.compile(r"^\s*Would you like to\b", re.IGNORECASE)
MUSE_APPROVAL_YES_RE = re.compile("^\\s*[>\\u203a]?\\s*1\\.\\s+Yes\\b")
MUSE_APPROVAL_NO_RE = re.compile("^\\s*[>\\u203a]?\\s*2\\.\\s+No\\b")
# A decided block ("approval decision accepted/denied") must never match:
# answering it would inject stray keys after the decision already landed.
MUSE_APPROVAL_DECIDED_RE = re.compile(r"approval decision", re.IGNORECASE)
MUSE_APPROVAL_SPAN = 16  # cue..options span (wrapped command echo ok)
# Collapsed approval: long commands collapse, hiding the 1.Yes/2.No pair:
#   Would you like to run the following
#     $ <first rows...>
#     ... N command rows omitted
#              ctrl+o view full command
MUSE_COLLAPSED_ROWS_RE = re.compile(r"rows?\s+omitted", re.IGNORECASE)
MUSE_COLLAPSED_KEY_RE = re.compile(r"ctrl\s*\+\s*o\b", re.IGNORECASE)

# Agent interview UI (request_user_input rendered in-pane; observed live):
#   <question...?>
#     > 1. First option (Recommended)  <desc...>
#       2. Second option               <desc...>
#       3. Both ...
#       4. None of the above ...
# The `>`/`›` cursor on an option line proves a live selectable menu
# (prose lists never carry it); the question above ends with "?".
INTERVIEW_OPT_RE = re.compile("^\\s*([>\\u203a])?\\s*(\\d+)\\.\\s+\\S")
INTERVIEW_CUE_ABOVE = 8  # question must sit within N lines above option 1

# Cursorless dotted 1./2./3. grill question (observed live on a policy
# grill): ordered dotted options, a ?-ended question above, and a strong
# pick-cue nearby. Both the question and the cue are required: bare
# prose lists must never match. Any cursor-prefixed dotted line vetoes
# (cursor menus belong to the interview matcher).
NUMBERED_PLAIN_OPT_RE = re.compile(r"^\s*(\d+)\.\s+\S")
NUMBERED_PLAIN_CURSOR_RE = re.compile("^\\s*[>\\u203a]\\s*\\d+\\.\\s+\\S")
NUMBERED_PLAIN_CUE_RE = re.compile(
    r"(?i)\bpick\s+(one|[123]|a number)\b|\breply\s+[123]\b"
    r"|\bchoose\s+(one|[123])\b|\benter\s+[123]\b"
    r"|\bselect\s+an?\s*(option|one|number)\b|\byour choice\b")

# Explicit-phrase request (model asks the user to reply a magic word;
# observed live): "Reply ACCEPT to approve this text as written ..."
# Only a single ALL-CAPS token qualifies -- lowercase/prose after Reply
# never matches, which keeps quoted instructions and chat from firing.
EXPLICIT_RE = re.compile(r"^\s*(?i:Reply)\s+([A-Z][A-Z0-9_-]{1,11})\b")
EXPLICIT_WINDOW = 8  # request must sit in the last N content lines

# /permissions mode picker (pasted verbatim from a live session):
#   Permissions
#     Selections are saved as your default for new sessions and exec runs.
#     1. Read only               Reads files only.
#     2. Ask me                  Edits workspace ...
#     3. Auto-review             Same access as Ask me; ...
#   › 4. Unrestricted (current)  No filesystem sandbox ...
# A mode change persists as the default for new sessions and exec runs,
# so the picker is builtin-hold (operator drives the TUI by hand) and
# is never auto-answered. The (current) marker names the live mode.
PERMISSIONS_HEADER_RE = re.compile(r"^\s*Permissions\s*$")
PERMISSIONS_OPT_RE = re.compile(
    "^\\s*([>\\u203a])?\\s*([1-4])\\.\\s+"
    "(Read only|Ask me|Auto-review|Unrestricted)\\b")
PERMISSIONS_CURRENT_RE = re.compile(r"\(current\)")
PERMISSIONS_SPAN = 18  # header..option 4 (wrapped descriptions ok)

# /permissions enable confirmation (pasted verbatim from live):
#   Enable unrestricted permissions?
#     Profile: Unrestricted
#     ...
#       Cancel
#     › Enable Unrestricted
# Confirming flips the session (and the saved default) out of the
# logged-approval path, so it is builtin-hold like the picker.
PERMISSIONS_CONFIRM_TITLE_RE = re.compile(
    r"^\s*Enable\s+(.+?)\s+permissions\?\s*$", re.IGNORECASE)
PERMISSIONS_CONFIRM_OPT_RE = re.compile(
    "^\\s*([>\\u203a])?\\s*(Cancel|Enable\\s+\\S.*?)\\s*$")
PERMISSIONS_CONFIRM_SPAN = 14  # title..options span

# Runtime state sensing for external agents driving panes via send-keys.
# A working pane shows a running indicator ("- running (Ns ...", "Calling
# tools (...", or the "esc to interrupt" tail, often wrapped/edge-cut);
# an idle pane sits at the prompt glyph (U+276F, escaped below).
RUNNING_RE = re.compile("([\\u2014-] running \\(|Calling tools \\(|esc to\\b)")
OPEN_PROMPT_RE = re.compile("^\\s*\\u276f")
STATE_TAIL_WINDOW = 12  # state cues must sit in the last N content lines


def slug_socket(socket_path):
    """Short filesystem-safe id for a tmux socket: basename + hash suffix.

    The hash suffix guards against two different socket paths sharing a
    basename (e.g. /tmp/a/default vs /tmp/b/default).
    """
    base = os.path.basename(socket_path.rstrip("/")) or "sock"
    safe = re.sub(r"[^A-Za-z0-9_.-]+", "_", base)[:32]
    digest = hashlib.sha1(socket_path.encode()).hexdigest()[:6]
    return "%s-%s" % (safe, digest)


def clean_pane(pane_id):
    return re.sub(r"[^A-Za-z0-9]+", "", pane_id)


def pidfile_for(socket_path, pane_id):
    return os.path.join(
        STATE_DIR, "%s-%s-%s.pid" % (FILE_PREFIX, slug_socket(socket_path), clean_pane(pane_id))
    )


def logfile_for(socket_path, pane_id):
    return os.path.join(
        STATE_DIR, "%s-%s-%s.log" % (FILE_PREFIX, slug_socket(socket_path), clean_pane(pane_id))
    )


def answered_file_for(socket_path, pane_id):
    """On-disk answered-sig store: restarts must not re-answer prompts."""
    return os.path.join(
        STATE_DIR, "%s-%s-%s.answered.json" % (FILE_PREFIX, slug_socket(socket_path), clean_pane(pane_id))
    )


class WatcherLog:
    """Never-raising JSON-lines logger with best-effort rotation."""

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

    def _rotate(self):
        try:
            if os.path.exists(self.path) and os.path.getsize(self.path) >= LOG_MAX_BYTES:
                try:
                    if os.path.exists(self.path + ".1"):
                        os.remove(self.path + ".1")
                except OSError:
                    pass
                os.rename(self.path, self.path + ".1")
        except Exception:
            pass

    def log(self, level, msg, **fields):
        try:
            self._rotate()
            rec = {
                "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
                "level": level,
                "msg": msg,
            }
            rec.update(fields)
            with open(self.path, "a") as f:
                f.write(json.dumps(rec) + "\n")
        except Exception:
            pass


def _audit_local(action, name, caller, extra):
    """Append a box-ctl.jsonl audit record without importing approvals."""
    rec = {
        "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
        "action": action,
        "type": "muse-choice",
        "name": name,
        "caller": caller,
    }
    if extra:
        rec.update(extra)
    with open(CTL_LOG, "a") as f:
        f.write(json.dumps(rec) + "\n")


def audit(action, name=None, caller="muse-choice-watcher", extra=None):
    """Feed an event to box (box-ctl.jsonl). Never raises.

    Prefers approvals.log_box_ctl so the audit schema stays unified; falls
    back to a direct append if the import fails.
    """
    try:
        try:
            import approvals
            approvals.log_box_ctl(action, name=name, caller=caller, extra=extra)
        except Exception:
            _audit_local(action, name, caller, extra or {})
    except Exception:
        pass


def get_desired():
    """Desired daemon state: {"enabled", "dry_run", ...}. Default: on.

    Policy: auto-approve is on unless explicitly disabled (`box
    muse-choices off`), the pane opted out via launch flags, or a human
    intervened. A missing/unreadable state file (or a file without the
    key) means undefined -> on.
    """
    default = {"enabled": True, "dry_run": False}
    try:
        with open(DESIRED_STATE_FILE) as f:
            data = json.load(f)
        return {"enabled": bool(data.get("enabled", True)),
                "dry_run": bool(data.get("dry_run", False)),
                "updated_at": data.get("updated_at", ""),
                "updated_by": data.get("updated_by", "")}
    except Exception:
        return dict(default)


def set_enabled(enabled, dry_run=None, by="box"):
    """Write desired state + audit the transition. Returns the new state."""
    state = get_desired()
    state["enabled"] = bool(enabled)
    if dry_run is not None:
        state["dry_run"] = bool(dry_run)
    state["updated_at"] = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    state["updated_by"] = by
    try:
        parent = os.path.dirname(DESIRED_STATE_FILE)
        if parent:
            os.makedirs(parent, exist_ok=True)
        tmp = DESIRED_STATE_FILE + ".tmp.%d" % os.getpid()
        with open(tmp, "w") as f:
            json.dump(state, f, indent=1)
        os.replace(tmp, DESIRED_STATE_FILE)
    except Exception:
        pass
    audit("muse-choice-enabled" if enabled else "muse-choice-disabled",
          caller=by, extra={"dry_run": state["dry_run"]})
    return state


def _option_markers(lines):
    """Return [(line_index, letter)] for lettered-option lines."""
    out = []
    for i, line in enumerate(lines):
        if len(line) > MAX_OPTION_LEN:
            continue
        m = OPTION_RE.match(line)
        if m:
            out.append((i, m.group(1)))
    return out


def _ordered_run(markers):
    """Find A,B[,C...] in order. Returns marker sub-list or None.

    Markers between run members (wrapped text has no markers, so any
    interleaving lettered line breaks the run) must fit OPTION_SPAN.
    """
    want = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
    best = []
    for start in range(len(markers)):
        if markers[start][1] != "A":
            continue
        run = [markers[start]]
        for idx in range(start + 1, len(markers)):
            expected = want[len(run)]
            if markers[idx][1] == expected:
                run.append(markers[idx])
            else:
                break
        if len(run) >= 2 and run[-1][0] - run[0][0] <= OPTION_SPAN:
            if len(run) > len(best):
                best = run
    return best or None


def _sig_for(kind, parts):
    src = kind + "\n" + "\n".join(parts)
    return hashlib.sha1(src.encode()).hexdigest()[:16]


def _option_block_end(window, end, marker_re):
    """Last row of an option block: skip its wrapped continuation rows.

    Captures are width-chunked, not joined, so an option's text may wrap
    onto several display rows past its marker line. Cue distance is
    measured past the block, not the marker. Stops at blank rows, at the
    next marker, and at WRAPPED_TAIL_CAP rows past the marker (bounding
    the net: a cue far below a long tail still rejects).
    """
    tail = end
    while (tail + 1 < len(window) and tail + 1 - end <= WRAPPED_TAIL_CAP
           and window[tail + 1].strip()
           and not marker_re.match(window[tail + 1])):
        tail += 1
    return tail


def _find_letter(window):
    """A/B/C lettered choice block -> {"sig", "kind", "key", ...} or None."""
    run = _ordered_run(_option_markers(window))
    if not run:
        return None
    start, end = run[0][0], run[-1][0]
    lo = max(0, start - 4)
    tail = _option_block_end(window, end, OPTION_RE)
    hi = min(len(window), tail + 1 + CUE_BELOW_ROWS)
    cue = None
    for i in range(lo, hi):
        if CUE_RE.search(window[i]) or QUESTION_RE.search(window[i]):
            cue = window[i].strip()
            break
    if cue is None:
        return None
    options = [window[i].strip()[:120] for i, _ in run]
    cue = cue[:200]
    return {"sig": _sig_for("letter", options + [cue]), "kind": "letter",
            "key": "A", "options": options, "cue": cue,
            "start": start, "end": end}


def _find_numbered(window):
    """(1)/(2) numbered menu with selection cue -> match or None."""
    markers = []
    for i, line in enumerate(window):
        if len(line) > MAX_OPTION_LEN:
            continue
        m = NUMBERED_OPT_RE.match(line)
        if m:
            markers.append((i, int(m.group(1))))
    ones = [i for i, n in markers if n == 1]
    twos = [i for i, n in markers if n == 2]
    if not ones or not twos:
        return None
    start = ones[0]
    end = next((i for i in twos if i > start), None)
    if end is None or end - start > OPTION_SPAN:
        return None
    lo = max(0, start - 2)
    tail = _option_block_end(window, end, NUMBERED_OPT_RE)
    hi = min(len(window), tail + 1 + CUE_BELOW_ROWS)
    cue = None
    for i in range(lo, hi):
        if NUMBERED_CUE_RE.search(window[i]):
            cue = window[i].strip()
            break
    if cue is None:
        return None
    options = [window[i].strip()[:120]
               for i, n in markers if start <= i <= end]
    cue = cue[:200]
    return {"sig": _sig_for("numbered", options + [cue]), "kind": "numbered",
            "key": "1", "options": options, "cue": cue,
            "start": start, "end": end}


def _find_yn(window):
    """y/n prompt at the end of a recent line -> match or None."""
    recent = window[-YN_WINDOW:]
    base = len(window) - len(recent)
    for j in range(len(recent) - 1, -1, -1):
        line = recent[j]
        if len(line) > MAX_OPTION_LEN:
            continue
        if YN_RE.search(line):
            text = line.strip()[:200]
            ctx = recent[j - 1].strip()[:120] if j > 0 else ""
            i = base + j
            return {"sig": _sig_for("yn", [text, ctx]), "kind": "yn",
                    "key": "y", "options": [text], "cue": text,
                    "start": i, "end": i}
    return None


def _find_muse_approval(window):
    """Native Muse TUI approval menu -> match or None.

    Requires the Would-you-like cue plus a 1.Yes/2.No pair below it.
    A decided block ("approval decision ...") within the span never
    matches, so an already-landed decision is never double-answered.
    """
    cue_idx = None
    for i, line in enumerate(window):
        if len(line) > MAX_OPTION_LEN:
            continue
        if MUSE_APPROVAL_CUE_RE.match(line):
            cue_idx = i
            break
    if cue_idx is None:
        return None
    yes_idx = no_idx = None
    for i in range(cue_idx + 1, min(len(window), cue_idx + 1 + MUSE_APPROVAL_SPAN)):
        line = window[i]
        if len(line) > MAX_OPTION_LEN:
            continue
        if yes_idx is None and MUSE_APPROVAL_YES_RE.match(line):
            yes_idx = i
        elif MUSE_APPROVAL_NO_RE.match(line):
            no_idx = i
            break
    if yes_idx is None or no_idx is None:
        return None
    for line in window[cue_idx:no_idx + 4]:
        if MUSE_APPROVAL_DECIDED_RE.search(line):
            return None
    cue = window[cue_idx].strip()[:200]
    options = [window[yes_idx].strip()[:120], window[no_idx].strip()[:120]]
    # Sig covers the WHOLE block (cue + $ command + options): bare
    # options+cue are byte-identical across every command approval, which
    # used to collapse all of a pane's approvals into one sig and wedge
    # every dialog after the first (observed live 2026-10-06).
    context = [ln.strip()[:120] for ln in window[cue_idx:no_idx + 1]]
    return {"sig": _sig_for("muse-approval", context),
            "kind": "muse-approval", "key": "1", "options": options,
            "cue": cue, "context": context,
            "start": cue_idx, "end": no_idx}


def _find_collapsed_approval(window):
    """Collapsed approval menu (long command, options hidden) or None.

    Requires the Would-you-like cue plus the collapsed markers ("N
    command rows omitted" + "ctrl+o view full command"). Answers with
    a bare Enter: that expands the block to the full 1/2 form (or
    accepts outright), and the expanded form answers as a fresh
    muse-approval prompt on later polls. (ctrl+o is NOT used: it opens
    a modal pager that strands the session.)
    """
    cue_idx = None
    for i, line in enumerate(window):
        if len(line) > MAX_OPTION_LEN:
            continue
        if MUSE_APPROVAL_CUE_RE.match(line):
            cue_idx = i
            break
    if cue_idx is None:
        return None
    rows_idx = key_idx = None
    for i in range(cue_idx + 1, min(len(window), cue_idx + 1 + MUSE_APPROVAL_SPAN)):
        line = window[i]
        if len(line) > MAX_OPTION_LEN:
            continue
        if rows_idx is None and MUSE_COLLAPSED_ROWS_RE.search(line):
            rows_idx = i
        if MUSE_COLLAPSED_KEY_RE.search(line):
            key_idx = i
    if rows_idx is None or key_idx is None:
        return None
    end = max(rows_idx, key_idx)
    for line in window[cue_idx:end + 4]:
        if MUSE_APPROVAL_DECIDED_RE.search(line):
            return None
    cue = window[cue_idx].strip()[:200]
    options = [window[rows_idx].strip()[:120],
               window[key_idx].strip()[:120]]
    context = [ln.strip()[:120] for ln in window[cue_idx:end + 1]]
    return {"sig": _sig_for("muse-approval-collapsed", context),
            "kind": "muse-approval-collapsed", "key": "Enter",
            "enter": False, "options": options, "cue": cue,
            "context": context, "start": cue_idx, "end": end}


def _find_interview(window):
    """Agent interview menu (cursor + ordered 1./2./.. + question) or None.

    Requires an ordered run starting at 1 with at least options 1 and 2,
    a selection cursor on one of the run's lines, and a "?"-ended
    question above. The cursor requirement is what separates a live
    interview from a prose numbered list.
    """
    markers = []
    for i, line in enumerate(window):
        if len(line) > MAX_OPTION_LEN:
            continue
        m = INTERVIEW_OPT_RE.match(line)
        if m:
            markers.append((i, int(m.group(2)), bool(m.group(1))))
    ones = [i for i, n, c in markers if n == 1]
    if not ones:
        return None
    start = ones[0]
    run = [start]
    for i, n, c in markers:
        if i > start and n == len(run) + 1 and i - start <= OPTION_SPAN:
            run.append(i)
    if len(run) < 2:
        return None
    if not any(c for i, n, c in markers if run[0] <= i <= run[-1]):
        return None
    question = None
    for j in range(max(0, start - INTERVIEW_CUE_ABOVE), start):
        if QUESTION_RE.search(window[j]):
            question = window[j].strip()
    if question is None:
        return None
    options = [window[i].strip()[:120] for i in run]
    question = question[:200]
    return {"sig": _sig_for("interview", options + [question]),
            "kind": "interview", "key": "1", "options": options,
            "cue": question, "start": start, "end": run[-1]}


def _ordered_int_run(markers):
    """Complete 1,2[,3...] runs; returns the NEWEST (last) one or None.

    Unlike _ordered_run (first on ties), grill windows stack an answered
    block above the live one: the live prompt is the lower run.
    """
    runs = []
    for start in range(len(markers)):
        if markers[start][1] != 1:
            continue
        run = [markers[start]]
        for idx in range(start + 1, len(markers)):
            if markers[idx][1] == len(run) + 1:
                run.append(markers[idx])
            else:
                break
        if len(run) >= 2 and run[-1][0] - run[0][0] <= OPTION_SPAN:
            runs.append(run)
    return runs[-1] if runs else None


def _find_numbered_plain(window):
    """Cursorless dotted 1./2./3. grill question -> match or None.

    Ordered dotted options starting at 1, a ?-ended question above, and
    a strong pick-cue nearby. Both the question and the cue are
    required: bare prose lists must never match. Any cursor-prefixed
    dotted line vetoes (cursor menus belong to _find_interview).
    """
    for line in window:
        if NUMBERED_PLAIN_CURSOR_RE.match(line):
            return None
    markers = []
    for i, line in enumerate(window):
        if len(line) > MAX_OPTION_LEN:
            continue
        m = NUMBERED_PLAIN_OPT_RE.match(line)
        if m:
            markers.append((i, int(m.group(1))))
    run = _ordered_int_run(markers)
    if not run:
        return None
    start, end = run[0][0], run[-1][0]
    question = None
    for qi in range(max(0, start - INTERVIEW_CUE_ABOVE), start):
        if QUESTION_RE.search(window[qi]):
            question = window[qi].strip()
    if question is None:
        return None
    lo = max(0, start - 4)
    tail = _option_block_end(window, end, NUMBERED_PLAIN_OPT_RE)
    hi = min(len(window), tail + 1 + CUE_BELOW_ROWS)
    cue = None
    for i in range(lo, hi):
        if NUMBERED_PLAIN_CUE_RE.search(window[i]):
            cue = window[i].strip()
            break
    if cue is None:
        return None
    options = [window[i].strip()[:120] for i, _ in run]
    question = question[:200]
    return {"sig": _sig_for("numbered-plain", options + [question]),
            "kind": "numbered-plain", "key": "1", "options": options,
            "cue": question, "start": start, "end": end}


def _find_explicit(window):
    """Explicit-phrase request ("Reply ACCEPT to ...") or None.

    Scans the last EXPLICIT_WINDOW content lines bottom-up so the
    freshest request wins. The key is the captured token itself.
    """
    recent = window[-EXPLICIT_WINDOW:]
    base = len(window) - len(recent)
    for j in range(len(recent) - 1, -1, -1):
        line = recent[j]
        if len(line) > MAX_OPTION_LEN:
            continue
        m = EXPLICIT_RE.match(line)
        if m:
            token = m.group(1)
            text = line.strip()[:200]
            i = base + j
            return {"sig": _sig_for("explicit-phrase", [text]),
                    "kind": "explicit-phrase", "key": token,
                    "options": [text], "cue": text,
                    "start": i, "end": i}
    return None


def _find_permissions_picker(window):
    """/permissions mode picker -> match or None.

    Requires the exact "Permissions" header, all four mode options in
    1..4 order, a (current) marker naming the live mode, and a cursor
    on an option line (proves a live selectable menu). The sig covers
    labels + current mode but not cursor position, so operator
    navigation doesn't churn holds while an actual mode change does.
    Key is None: the picker is builtin-hold, never auto-answered.
    """
    header = None
    for i, line in enumerate(window):
        if len(line) > MAX_OPTION_LEN:
            continue
        if PERMISSIONS_HEADER_RE.match(line):
            header = i
            break
    if header is None:
        return None
    found = {}
    cursor = False
    end = header
    for i in range(header + 1, min(len(window), header + 1 + PERMISSIONS_SPAN)):
        line = window[i]
        if len(line) > MAX_OPTION_LEN:
            continue
        m = PERMISSIONS_OPT_RE.match(line)
        if not m:
            continue
        if m.group(1):
            cursor = True
        found[int(m.group(2))] = (i, m.group(3),
                                  bool(PERMISSIONS_CURRENT_RE.search(line)))
        end = i
    if sorted(found) != [1, 2, 3, 4] or not cursor:
        return None
    current = next((label for _, label, is_cur in found.values() if is_cur),
                   None)
    if current is None:
        return None
    options = ["%d. %s" % (n, found[n][1]) for n in (1, 2, 3, 4)]
    cue = "Permissions (current: %s)" % current
    return {"sig": _sig_for("permissions", options + [current]),
            "kind": "permissions", "key": None, "options": options,
            "cue": cue, "current": current,
            "start": header, "end": end}


def _find_permissions_confirm(window):
    """/permissions enable confirmation -> match or None.

    Requires an "Enable <X> permissions?" title plus Cancel and
    Enable-... option lines with a cursor on one. Sig excludes cursor
    position (navigation churn, see picker). Key None: builtin-hold.
    """
    title = None
    profile = None
    for i, line in enumerate(window):
        if len(line) > MAX_OPTION_LEN:
            continue
        m = PERMISSIONS_CONFIRM_TITLE_RE.match(line)
        if m:
            title = i
            profile = m.group(1).strip()
            break
    if title is None:
        return None
    seen_cancel = seen_enable = cursor = False
    end = title
    for i in range(title + 1, min(len(window), title + 1 + PERMISSIONS_CONFIRM_SPAN)):
        line = window[i]
        if len(line) > MAX_OPTION_LEN:
            continue
        m = PERMISSIONS_CONFIRM_OPT_RE.match(line)
        if not m:
            continue
        if m.group(1):
            cursor = True
        opt = m.group(2)
        if opt == "Cancel":
            seen_cancel = True
        elif opt.startswith("Enable"):
            seen_enable = True
        end = i
    if not (seen_cancel and seen_enable and cursor):
        return None
    cue = window[title].strip()[:200]
    options = ["Cancel", "Enable %s" % profile] if profile else ["Cancel"]
    return {"sig": _sig_for("permissions", [cue]),
            "kind": "permissions", "key": None, "options": options,
            "cue": cue, "current": None,
            "start": title, "end": end}


def find_choice_prompt(text, tail_window=TAIL_WINDOW):
    """Detect a Muse prompt awaiting reply.

    Kinds (priority order): permissions (/permissions picker or
    enable confirmation -> builtin-hold, never auto-answered),
    muse-approval (native Would-you-like menu -> key "1"),
    muse-approval-collapsed (collapsed long command -> bare Enter,
    expand-or-accept), interview (cursor + ordered 1./2. menu +
    question -> key "1"), numbered-plain (cursorless dotted 1./2./3.
    grill question + pick-cue -> key "1", coordinator-held),
    explicit-phrase ("Reply TOKEN to ..." ->
    key TOKEN), letter (A/B/C -> key "A"), numbered ((1)/(2) menu ->
    key "1"), yn (y/n line-end -> key "y").
    Returns {"sig", "kind", "key", "options", "cue", "start", "end"}
    or None. Only blocks in the last `tail_window` content lines are
    eligible, so answered/stale prompts in scrollback never re-trigger.
    """
    if not text:
        return None
    lines = text.split("\n")
    # Trailing blank lines (tall panes, fresh prompts) must not push a live
    # prompt out of the window; staleness is judged on content lines.
    while lines and not lines[-1].strip():
        lines.pop()
    window = lines[-tail_window:]
    if not window:
        return None
    return (_find_permissions_picker(window) or _find_permissions_confirm(window)
            or _find_muse_approval(window) or _find_collapsed_approval(window)
            or _find_interview(window) or _find_numbered_plain(window)
            or _find_explicit(window)
            or _find_letter(window) or _find_numbered(window)
            or _find_yn(window))


def runtime_state(text):
    """Classify a Muse pane capture for agentic send-keys drivers.

    Returns {"state", "match"} where state is one of approval-pending
    (a choice prompt awaits reply), working (running indicator),
    open-prompt (idle at the input prompt), or unknown. Priority is
    approval-pending > working > open-prompt: a working pane still
    renders its prompt footer, and an approval can arrive mid-run.
    """
    match = find_choice_prompt(text)
    if match is not None:
        return {"state": "approval-pending", "match": match}
    lines = (text or "").split("\n")
    while lines and not lines[-1].strip():
        lines.pop()
    tail = lines[-STATE_TAIL_WINDOW:]
    for line in tail:
        if RUNNING_RE.search(line):
            return {"state": "working", "match": None}
    for line in tail:
        if len(line) > MAX_OPTION_LEN:
            continue
        if OPEN_PROMPT_RE.match(line):
            return {"state": "open-prompt", "match": None}
    return {"state": "unknown", "match": None}


def _cmdline(pid):
    """Argv of a pid as a list (empty when unreadable). Never raises."""
    try:
        with open("/proc/%d/cmdline" % pid, "rb") as f:
            raw = f.read().split(b"\0")
        return [p.decode(errors="replace") for p in raw if p]
    except Exception:
        return []


def _child_pids(pid):
    """Direct children of pid via a bounded /proc scan. Never raises."""
    out = []
    try:
        names = os.listdir("/proc")
    except Exception:
        return out
    for name in names:
        if not name.isdigit():
            continue
        try:
            with open("/proc/%s/stat" % name) as f:
                parts = f.read().rsplit(")", 1)[1].split()
            if int(parts[1]) == pid:
                out.append(int(name))
        except Exception:
            continue
    return out


def muse_approval_flags(cmd_argv):
    """Approval posture of a muse argv.

    Returns {"auto_approve", "flags", "profile", "mode", "bypass"}.
    profile is the --permission-profile id (built-ins :read-only,
    :standard, :unrestricted) or None; mode is "yolo" for --yolo, the
    profile id when one was passed, else "default"; bypass is True
    when tool-approval prompts are disabled at launch (yolo,
    --disable-approval, or --approval-mode=never), in which case the
    pane shows no approval dialogs (questions may still render, so the
    watcher stays active).
    """
    flags = []
    argv = cmd_argv or []
    if "--yolo" in argv:
        flags.append("yolo")
    if "--disable-approval" in argv:
        flags.append("disable-approval")
    if "--disable-sandbox" in argv:
        flags.append("disable-sandbox")
    if "--trust-workspace" in argv:
        flags.append("trust-workspace")
    profile = None
    for i, arg in enumerate(argv):
        if arg == "--approval-mode" and i + 1 < len(argv):
            flags.append("approval-mode=%s" % argv[i + 1])
        elif arg.startswith("--approval-mode="):
            flags.append("approval-mode=%s" % arg.split("=", 1)[1])
        elif arg == "--permission-profile" and i + 1 < len(argv):
            profile = argv[i + 1]
            flags.append("permission-profile=%s" % profile)
        elif arg.startswith("--permission-profile="):
            profile = arg.split("=", 1)[1]
            flags.append("permission-profile=%s" % profile)
    auto = ("yolo" in flags or "disable-approval" in flags
            or "approval-mode=never" in flags)
    if "yolo" in flags:
        mode = "yolo"
    elif profile is not None:
        mode = profile
    else:
        mode = "default"
    return {"auto_approve": auto, "flags": flags, "profile": profile,
            "mode": mode, "bypass": auto}


def launch_opt_out(cmd_argv):
    """True when a muse argv explicitly opts out of watcher auto-approve.

    Bare argv (no approval flags) means undefined -> the watcher
    answers. An explicit --approval-mode other than never (on-request,
    untrusted, ...) declares human-in-the-loop intent -> hold.
    """
    argv = cmd_argv or []
    for i, arg in enumerate(argv):
        mode = None
        if arg == "--approval-mode" and i + 1 < len(argv):
            mode = argv[i + 1]
        elif arg.startswith("--approval-mode="):
            mode = arg.split("=", 1)[1]
        if mode is not None and mode != "never":
            return True
    return False


def pane_holds_for_optout(socket_path, cmd_argv):
    """True when the watcher must hold prompts for launch-flag opt-out.

    Fleet sockets host self-running agents: prompts there are answered
    (and logged) regardless of approval-mode flags, since nobody is
    watching to answer by hand. Elsewhere an explicit --approval-mode
    other than never declares human-in-the-loop intent -> hold.
    """
    if socket_path in FLEET_SOCKETS:
        return False
    return launch_opt_out(cmd_argv)


def hourly_cap_for(socket_path):
    """Hourly answer budget for a pane loop on socket_path."""
    if socket_path in FLEET_SOCKETS:
        return FLEET_MAX_ANSWERS_PER_HOUR
    return MAX_ANSWERS_PER_HOUR


def approval_determined(cmd_argv):
    """True when argv already determines approval behavior.

    yolo / disable-approval / approval-mode / permission-profile all
    settle it; sandbox and workspace-trust flags do not (they ride
    along with whatever approval posture applies).
    """
    posture = muse_approval_flags(cmd_argv)
    if posture["profile"] is not None:
        return True
    flags = posture["flags"]
    if "yolo" in flags or "disable-approval" in flags:
        return True
    return any(f.startswith("approval-mode=") for f in flags)


def muse_launch_cmdline(muse_args):
    """Build a fleet launch command line.

    Strips a leading "--" separator; bare launches (no
    approval-determining flags) get on-request approvals injected so
    prompts render for the watcher trail. Returns (cmdline, injected).
    Shared by `box runtime launch` and the runtime reconciler so both
    spawn identical sessions.
    """
    args = list(muse_args or [])
    if args[:1] == ["--"]:
        args = args[1:]
    injected = ([] if approval_determined(args)
                else ["--approval-mode", "on-request"])
    launcher = (shutil.which("muse-code")
                or "/home/super/.local/bin/muse-code")
    return (shlex.join([launcher] + injected + args), injected)


def saved_default_profile(settings_path=None):
    """Saved default permission profile id, or None.

    Reads permissions.default_profile from the muse settings file
    (e.g. ":unrestricted" after driving the /permissions picker). The
    app applies this to sessions launched without explicit approval
    flags; launch flags override it ("Launch overrides" footer).
    Never raises.
    """
    path = settings_path or os.path.expanduser(
        "~/.config/muse/settings.json")
    try:
        with open(path) as f:
            data = json.load(f)
    except Exception:
        return None
    try:
        return data.get("permissions", {}).get("default_profile")
    except Exception:
        return None


def resolve_socket(sock):
    """Resolve a user-supplied tmux socket alias to a server path.

    Tables show basenames (tmux-muse.sock), so callers copy them back
    as --socket values. As-given wins when it exists; otherwise match
    by basename against KNOWN_SOCKETS, then /tmp/<name>. Unknown
    values pass through unchanged so downstream errors still name
    what was given. Pure apart from existence probes.
    """
    if not sock or os.path.exists(sock):
        return sock
    base = os.path.basename(sock)
    for known in KNOWN_SOCKETS:
        if os.path.basename(known) == base:
            return known
    cand = os.path.join("/tmp", base)
    if os.path.exists(cand):
        return cand
    return sock


def effective_posture(posture, saved_default=None):
    """Effective approval posture: explicit flags win, else saved default.

    Returns {"effective_mode", "effective_bypass"}. effective_bypass is
    True when the pane shows no approval dialogs (explicit yolo /
    disable-approval / never, an unrestricted profile, or an
    unrestricted saved default with no explicit flags). Modes are
    display-ready (leading ":" stripped from profile ids). Pure.
    """
    flags = posture.get("flags") or []
    profile = posture.get("profile")
    approval_mode = None
    for f in flags:
        if f.startswith("approval-mode="):
            approval_mode = f.split("=", 1)[1]
    if "yolo" in flags:
        return {"effective_mode": "yolo", "effective_bypass": True}
    if "disable-approval" in flags or approval_mode == "never":
        if profile is not None:
            mode = profile.lstrip(":")
        elif approval_mode is not None:
            mode = approval_mode
        else:
            mode = "default"
        return {"effective_mode": mode, "effective_bypass": True}
    if approval_mode is not None:
        return {"effective_mode": approval_mode,
                "effective_bypass": False}
    if profile is not None:
        return {"effective_mode": profile.lstrip(":"),
                "effective_bypass": profile == ":unrestricted"}
    if saved_default:
        return {"effective_mode": str(saved_default).lstrip(":"),
                "effective_bypass": saved_default == ":unrestricted"}
    return {"effective_mode": "default", "effective_bypass": False}


def _pane_tree_argvs(socket_path, pane_id):
    """Argvs of a pane's root pid plus its direct children (may be empty)."""
    try:
        r = _tmux(socket_path, "list-panes", "-a", "-F",
                  "#{pane_id} #{pane_pid}", timeout=5)
        if r.returncode != 0:
            return []
    except Exception:
        return []
    pane_pid = None
    for line in r.stdout.split("\n"):
        parts = line.split()
        if len(parts) == 2 and parts[0] == pane_id:
            try:
                pane_pid = int(parts[1])
            except ValueError:
                return []
    if pane_pid is None:
        return []
    return [argv for argv in (_cmdline(c)
                             for c in [pane_pid] + _child_pids(pane_pid))
            if argv]


def pane_muse_argv(socket_path, pane_id):
    """Argv of the muse process in a pane (empty when not found)."""
    for argv in _pane_tree_argvs(socket_path, pane_id):
        if "muse-bin" in argv[0] or "muse-code" in argv[0]:
            return argv
    return []


def pane_harness(socket_path, pane_id):
    """Agent harness owning a pane: "muse", "agy", or None.

    Matches the harness binary among the pane's processes (stable:
    independent of which child is in the foreground).
    """
    for argv in _pane_tree_argvs(socket_path, pane_id):
        for harness, hints in HARNESS_HINTS:
            if any(h in argv[0] for h in hints):
                return harness
    return None


# Minimum pane geometry for reliable approval rendering. Empirically
# derived: a 35x7 tile drops approval text the matcher needs, while
# 35x35/36x35/71x27 panes answer cleanly (width 35 works when tall
# enough; capture uses -J so wrapping is width-independent). Below
# either bound the pane is flagged squeezed (see `box runtime layout` /
# `spread`).
MIN_APPROVAL_WIDTH = 35
MIN_APPROVAL_HEIGHT = 12

# Session naming convention (see NODES.md): <node>--<role>--<id>
# separates node runtimes on the shared stable server, e.g.
# pip--worker--01. Ad-hoc sessions carry no node and show "-".
NODE_NAMES = ("muse", "pip", "646", "opm", "def", "dev")
NODE_SESSION_RE = re.compile(r"^([A-Za-z0-9]+)--([A-Za-z0-9-]+)--([A-Za-z0-9]+)$")


def node_from_session(session_name):
    """Node owning a tmux session per the naming convention.

    Returns the node name, or None for ad-hoc sessions and unknown
    node prefixes. Pure function for supervision display.
    """
    if not session_name:
        return None
    m = NODE_SESSION_RE.match(session_name)
    if not m:
        return None
    node = m.group(1).lower()
    return node if node in NODE_NAMES else None


def runtime_rows(socket_path, errors=None):
    """One row per pane: identity, approval posture, live state, watcher.

    Rows are JSON-serializable dicts for `box runtime list` and external
    agents. Panes that vanish mid-scan are skipped, never fatal. When
    the socket itself is unreachable and errors is a dict, the tmux
    failure is recorded as errors[socket_path] (one line) instead of
    silently yielding zero rows.
    """
    rows = []
    try:
        r = _tmux(socket_path, "list-panes", "-a", "-F",
                  "#{session_name}\t#{window_index}\t#{pane_id}\t"
                  "#{pane_current_command}\t#{pane_pid}\t"
                  "#{pane_width}\t#{pane_height}", timeout=10)
        if r.returncode != 0:
            if errors is not None:
                errors[socket_path] = _tmux_err(r)
            return rows
    except Exception as e:
        if errors is not None:
            errors[socket_path] = str(e)[:160] or "tmux error"
        return rows
    saved_default = saved_default_profile()
    for line in r.stdout.split("\n"):
        parts = line.split("\t")
        if len(parts) != 7:
            continue
        session, window, pane_id, cmd, pid, width, height = parts
        try:
            pane_pid = int(pid)
        except ValueError:
            pane_pid = None
        try:
            pane_width, pane_height = int(width), int(height)
        except ValueError:
            pane_width = pane_height = None
        squeezed = (pane_width is not None and pane_height is not None
                    and (pane_width < MIN_APPROVAL_WIDTH
                         or pane_height < MIN_APPROVAL_HEIGHT))
        node = node_from_session(session)
        is_muse = "muse-bin" in cmd or "muse-code" in cmd
        posture = {"auto_approve": None, "flags": [], "profile": None,
                   "mode": None, "bypass": None}
        if is_muse and pane_pid:
            candidates = [pane_pid] + _child_pids(pane_pid)
            for cand in candidates:
                argv = _cmdline(cand)
                if argv and ("muse-bin" in argv[0]
                             or "muse-code" in argv[0]):
                    posture = muse_approval_flags(argv)
                    break
        text = capture_pane(socket_path, pane_id)
        if text is None:
            continue
        st = runtime_state(text)
        match = st["match"] or {}
        watcher_pid = watcher_alive(socket_path, pane_id)
        if posture["mode"] is None:
            effective = {"effective_mode": None, "effective_bypass": None}
        else:
            effective = effective_posture(posture, saved_default)
        rows.append({
            "socket": socket_path, "session": session,
            "window": window, "pane": pane_id, "cmd": cmd,
            "pid": pane_pid, "is_muse": is_muse,
            "node": node,
            "width": pane_width, "height": pane_height,
            "squeezed": squeezed,
            "auto_approve": posture["auto_approve"],
            "approval_flags": posture["flags"],
            "permission_mode": posture["mode"],
            "permission_profile": posture["profile"],
            "permission_bypass": posture["bypass"],
            "effective_mode": effective["effective_mode"],
            "effective_bypass": effective["effective_bypass"],
            "state": st["state"],
            "prompt_kind": match.get("kind"),
            "prompt_key": match.get("key"),
            "watcher_alive": watcher_pid is not None,
            "watcher_pid": watcher_pid,
        })
    return rows


def spread_targets(rows):
    """Rows that `box runtime spread` would break into own windows.

    A row is a target when it runs a coding runtime (muse today; other
    harnesses follow the registry) and its geometry is squeezed below
    the reliable-approval minimums. Pure function of rows for testing.
    """
    return [r for r in rows
            if r.get("is_muse") and r.get("squeezed")]


def _tmux_err(r):
    """First line of a failed tmux result, bounded."""
    text = ((getattr(r, "stderr", "") or "") + "\n"
            + (getattr(r, "stdout", "") or "")).strip().split("\n")
    line = next((ln.strip() for ln in text if ln.strip()), "")
    return (line or "tmux error")[:160]


def all_runtime_rows(sockets=None, errors=None):
    """runtime_rows across every known socket that exists.

    Sockets given explicitly but missing from the filesystem are
    reported via errors (when a dict) instead of silently skipped:
    callers that name a socket deserve to know it is absent.
    """
    rows = []
    explicit = sockets is not None
    for sock in sockets or KNOWN_SOCKETS:
        if not os.path.exists(sock):
            if errors is not None and explicit:
                errors[sock] = "no such socket"
            continue
        rows.extend(runtime_rows(sock, errors=errors))
    return rows


def pane_state(socket_path, pane_id):
    """Single-pane runtime row, or {"error": ...} when not found.

    Distinguishes a dead socket ("socket_unreachable", with the tmux
    failure in "detail") from a live socket without that pane
    ("no_such_pane").
    """
    errors = {}
    for row in runtime_rows(socket_path, errors=errors):
        if row["pane"] == pane_id:
            return row
    if errors.get(socket_path):
        return {"error": "socket_unreachable", "socket": socket_path,
                "pane": pane_id, "detail": errors[socket_path]}
    return {"error": "no_such_pane", "socket": socket_path,
            "pane": pane_id}


class WatcherState:
    """Tracks prompt stability and once-per-prompt answering.

    When persist_path is set, answered sigs survive restarts (atomic
    JSON store): a daemon that restarts while an answered prompt is
    still visible must not answer it again (observed live: same sig
    re-answered minutes later, stray "1" landing in the input box).
    """

    def __init__(self, persist_path=None, hourly_cap=None):
        self.pending_sig = None
        self.stable_count = 0
        self.answered_sigs = {}
        self.answer_times = []
        self.last_capped_sig = None
        self.hourly_cap = hourly_cap or MAX_ANSWERS_PER_HOUR
        self._persist_path = persist_path
        if persist_path:
            self._load_answered()

    def _load_answered(self):
        try:
            with open(self._persist_path) as f:
                data = json.load(f)
        except Exception:
            return
        if not isinstance(data, dict):
            return
        now = time.time()
        for sig, ts in data.items():
            if (isinstance(sig, str) and isinstance(ts, (int, float))
                    and now - ts < ANSWERED_TTL_SECONDS):
                self.answered_sigs[sig] = ts

    def _save_answered(self):
        if not self._persist_path:
            return
        try:
            tmp = "%s.tmp.%d" % (self._persist_path, os.getpid())
            with open(tmp, "w") as f:
                json.dump(self.answered_sigs, f)
            os.replace(tmp, self._persist_path)
        except Exception:
            pass

    def refresh_answered(self):
        """Merge on-disk answered sigs into memory. Never raises.

        Cross-process once-per-prompt: a peer watcher that answered
        after our startup persisted its sig; re-reading before we type
        suppresses the duplicate ('11' in the input box).
        """
        if not self._persist_path:
            return
        try:
            with open(self._persist_path) as f:
                data = json.load(f)
        except Exception:
            return
        if not isinstance(data, dict):
            return
        now = time.time()
        for sig, ts in data.items():
            if (isinstance(sig, str) and isinstance(ts, (int, float))
                    and now - ts < ANSWERED_TTL_SECONDS):
                if sig not in self.answered_sigs:
                    self.answered_sigs[sig] = ts

    def try_claim(self, sig, now=None):
        """Atomically claim a sig for this process. True iff we won.

        O_CREAT|O_EXCL makes the first claimer win even when two
        watchers reach the same stable prompt in the same poll window;
        the loser suppresses instead of double-typing. Claims expire
        after CLAIM_TTL_SECONDS (concurrent window only; stuck-dialog
        recovery is governed by the answered store's longer TTL), so a
        winner that dies between claim and send wedges at most briefly.
        Memory-only states (no persist path) always win. Never raises.
        """
        if not self._persist_path:
            return True
        now = time.time() if now is None else now
        base = os.path.basename(self._persist_path)
        directory = os.path.dirname(self._persist_path) or STATE_DIR
        # Prune expired claims for this pane (best-effort).
        try:
            for name in os.listdir(directory):
                if not name.startswith(base + ".") or not name.endswith(".claim"):
                    continue
                p = os.path.join(directory, name)
                try:
                    with open(p) as f:
                        rec = json.load(f)
                    ts = float(rec.get("ts", 0))
                except Exception:
                    ts = 0
                try:
                    if now - ts >= CLAIM_TTL_SECONDS:
                        os.remove(p)
                except OSError:
                    pass
        except Exception:
            pass
        path = "%s.%s.claim" % (self._persist_path, sig)
        try:
            fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
        except FileExistsError:
            return False
        except OSError:
            return True  # claim store unavailable: fail open, send once
        try:
            os.write(fd, json.dumps({"pid": os.getpid(),
                                     "ts": now}).encode())
        except Exception:
            pass
        try:
            os.close(fd)
        except Exception:
            pass
        return True

    def _prune_times(self, now):
        cutoff = now - 3600
        self.answer_times = [t for t in self.answer_times if t >= cutoff]

    def cap_reached(self, now):
        self._prune_times(now)
        return len(self.answer_times) >= self.hourly_cap

    def observe(self, match, now):
        """Feed one poll's match (or None). Returns "answer" when the prompt
        is stable, unanswered, and under the hourly cap."""
        if match is None:
            self.pending_sig = None
            self.stable_count = 0
            return "none"
        sig = match["sig"]
        if sig in self.answered_sigs:
            if now - self.answered_sigs[sig] < ANSWERED_TTL_SECONDS:
                return "none"
            # Expired: an identical prompt still (or again) present long
            # after its answer is a stuck dialog, not a duplicate -- allow
            # one recovery answer instead of wedging forever.
            del self.answered_sigs[sig]
        if sig == self.pending_sig:
            self.stable_count += 1
        else:
            self.pending_sig = sig
            self.stable_count = 1
        if self.stable_count < STABILITY_POLLS:
            return "wait"
        if self.cap_reached(now):
            return "capped"
        return "answer"

    def record_answer(self, sig, now):
        self.answered_sigs[sig] = now
        # Bound memory: drop expired entries, then oldest-first past the cap.
        cutoff = now - ANSWERED_TTL_SECONDS
        for s in [s for s, t in self.answered_sigs.items() if t < cutoff]:
            del self.answered_sigs[s]
        if len(self.answered_sigs) > 200:
            for s in sorted(self.answered_sigs,
                            key=lambda s: self.answered_sigs[s])[:-100]:
                del self.answered_sigs[s]
        self.answer_times.append(now)
        self.pending_sig = None
        self.stable_count = 0
        self._save_answered()


def _tmux(socket_path, *args, timeout=5):
    cmd = ["tmux", "-S", socket_path] + list(args)
    return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)


def pane_exists(socket_path, pane_id):
    try:
        r = _tmux(socket_path, "list-panes", "-a", "-F", "#{pane_id}", timeout=5)
        if r.returncode != 0:
            return False
        return pane_id in r.stdout.split()
    except Exception:
        return False


def capture_pane(socket_path, pane_id, history=80):
    """Capture pane text with wrapped rows joined (-J).

    -J makes matching width-independent: narrow panes wrap the same
    dialog onto more physical rows, which otherwise pushes cue/option
    spans apart and breaks the matcher.
    """
    try:
        r = _tmux(socket_path, "capture-pane", "-p", "-J", "-t", pane_id,
                  "-S", "-%d" % history, timeout=5)
        if r.returncode != 0:
            return None
        return r.stdout
    except Exception:
        return None


def _tail_composer_line(text, window=COMPOSER_TAIL_WINDOW):
    """Last input-prompt-led line in the trailing window, or None.

    Transcript history may hold older input echoes of submitted messages;
    the LAST prompt-led row is the live composer input. None when no such
    row shows (shell prompts, menus with no input row).
    """
    lines = (text or "").split("\n")
    while lines and not lines[-1].strip():
        lines.pop()
    for line in reversed(lines[-window:]):
        if OPEN_PROMPT_RE.match(line):
            return line
    return None


def _composer_shows_letter(text, letter):
    """True when the live composer input is exactly our letter.

    Exact match only: stale or concurrent input ("ABC", placeholder hints)
    must never read as ours, so a retry can only re-submit the known
    newline-bug shape (our letter sitting unsubmitted).
    """
    if not letter:
        return False
    line = _tail_composer_line(text)
    if line is None:
        return False
    return re.match(r"^\s*\u276f\s*%s\s*$" % re.escape(letter),
                     line) is not None


def _await_render(socket_path, pane_id, before):
    """True once the screen differs from `before`. Never raises.

    A changed screen proves the app consumed the typed text, so whatever
    we send next is a separate input event. Unchanged past the timeout
    (slow pane, dead pane, tmux hiccup) fails open: the caller still
    sends Enter on the blind timing.
    """
    if before is None:
        return False
    try:
        deadline = time.time() + VERIFY_TEXT_TIMEOUT
        while time.time() < deadline:
            now = capture_pane(socket_path, pane_id)
            if now is not None and now != before:
                return True
            time.sleep(VERIFY_POLL_SECONDS)
    except Exception:
        pass
    return False


def send_answer(socket_path, pane_id, letter="A", enter=True, kind=None,
                sig=None):
    """Type the choice key (+ Enter unless enter=False).

    Returns (ok, detail): ok is False only when tmux refused a send or a
    composer answer is still sitting unsubmitted after one retry; detail
    is {"verified": True/False/None, "retried": bool} for logs/audit.

    Composer kinds (letter/numbered/yn/explicit-phrase: the answer goes to
    a text input) are capture-verified: the text must render on screen
    before Enter goes out (proving the app consumed it as its own input
    event — a back-to-back burst arrives as paste and lands a newline
    instead of submitting), and after Enter the prompt must clear or the
    composer must empty, else Enter goes once more. Timeouts fail open to
    the blind send so a slow pane still gets its answer.
    Menu kinds (muse-approval/interview/collapsed: single-key widgets that
    never render text) keep the blind gap send, which is proven there.
    """
    blind = {"verified": None, "retried": False}
    try:
        if not enter:
            r = _tmux(socket_path, "send-keys", "-t", pane_id, letter,
                       timeout=5)
            return (r.returncode == 0), dict(blind)
        verify = kind in COMPOSER_KINDS
        before = capture_pane(socket_path, pane_id) if verify else None
        r1 = _tmux(socket_path, "send-keys", "-l", "-t", pane_id, letter,
                   timeout=5)
        if r1.returncode != 0:
            return False, dict(blind)
        verified = _await_render(socket_path, pane_id, before) \
            if verify else None
        time.sleep(SUBMIT_GAP_SECONDS)
        r2 = _tmux(socket_path, "send-keys", "-t", pane_id, "Enter",
                   timeout=5)
        if r2.returncode != 0:
            return False, {"verified": verified, "retried": False}
        if not verify or sig is None:
            return True, {"verified": verified, "retried": False}
        # Post-verify: same sig + our letter still in the composer means
        # the Enter landed as a newline; one more Enter submits it. Same
        # sig with a clear composer is a lingering transcript, not a miss.
        time.sleep(POST_ENTER_SETTLE)
        fresh = capture_pane(socket_path, pane_id)
        if fresh is None:
            return True, {"verified": verified, "retried": False}
        m = find_choice_prompt(fresh)
        if m is None or m["sig"] != sig:
            return True, {"verified": verified, "retried": False}
        if not _composer_shows_letter(fresh, letter):
            return True, {"verified": verified, "retried": False}
        r3 = _tmux(socket_path, "send-keys", "-t", pane_id, "Enter",
                   timeout=5)
        if r3.returncode != 0:
            return False, {"verified": verified, "retried": True}
        time.sleep(POST_ENTER_SETTLE)
        again = capture_pane(socket_path, pane_id)
        stuck = (again is not None
                 and _composer_shows_letter(again, letter))
        return (not stuck), {"verified": verified, "retried": True}
    except Exception:
        return False, {"verified": None, "retried": False}


def _load_rules(log=None):
    """Load the D0 rules dictionary, cached by (mtime, size). Never raises:
    a missing or broken file means no rules (fail open per D4)."""
    global _RULES_CACHE
    try:
        st = os.stat(RULES_FILE)
        key = (st.st_mtime, st.st_size)
    except OSError:
        key = "missing"
    if _RULES_CACHE["key"] == key:
        return _RULES_CACHE["rules"]
    rules = []
    if key != "missing":
        try:
            with open(RULES_FILE) as f:
                data = json.load(f)
            rules = [r for r in data.get("rules", [])
                     if isinstance(r, dict) and r.get("id")]
        except Exception as e:
            if log is not None:
                try:
                    log.log("warn", "rules file unreadable, failing open",
                            error="%r" % (e,))
                except Exception:
                    pass
            rules = []
    _RULES_CACHE = {"key": key, "rules": rules}
    return rules


def _as_list(v):
    return v if isinstance(v, list) else [v]


def _rule_matches(rule, match):
    kind = rule.get("kind")
    if kind is not None and match["kind"] not in _as_list(kind):
        return False
    token = rule.get("token")
    if token is not None:
        if match["kind"] != "explicit-phrase":
            return False
        if match["key"] not in _as_list(token):
            return False
    cmd = rule.get("command")
    if cmd is not None:
        ctx = match.get("context") or match.get("options", [])
        try:
            if not re.search(cmd, "\n".join(ctx)):
                return False
        except re.error:
            return False
    text = rule.get("text")
    if text is not None:
        hay = "\n".join(match.get("options", []) + [match.get("cue") or ""])
        try:
            if not re.search(text, hay):
                return False
        except re.error:
            return False
    return True


def evaluate_rules(match, log=None):
    """First matching rule wins -> (decision, rule). No match -> approve.

    D2: deny rules are skipped for question kinds (questions always
    resolve top-choice). Collapsed approvals have no visible No option,
    so deny downgrades to hold there rather than auto-approving
    something a rule flagged risky.
    """
    for rule in _load_rules(log=log):
        if not _rule_matches(rule, match):
            continue
        d = rule.get("decision", "approve")
        if d not in ("approve", "deny", "hold"):
            d = "approve"
        if d == "deny":
            if match["kind"] in NEGATIVE_KEYS:
                return "deny", rule
            if match["kind"] in QUESTION_KINDS:
                continue
            downgraded = dict(rule)
            downgraded["downgraded_from"] = "deny"
            return "hold", downgraded
        return d, rule
    return "approve", None


def holdfile_for(socket_path, pane_id):
    return os.path.join(
        STATE_DIR, "%s-%s-%s.held.json" % (FILE_PREFIX, slug_socket(socket_path),
                                          clean_pane(pane_id)))


def write_hold(socket_path, pane_id, rec):
    try:
        with open(holdfile_for(socket_path, pane_id), "w") as f:
            json.dump(rec, f)
        return True
    except Exception:
        return False


def read_hold(socket_path, pane_id):
    try:
        with open(holdfile_for(socket_path, pane_id)) as f:
            rec = json.load(f)
        return rec if isinstance(rec, dict) and rec.get("sig") else None
    except Exception:
        return None


def clear_hold(socket_path, pane_id):
    try:
        os.remove(holdfile_for(socket_path, pane_id))
    except OSError:
        pass


def list_holds():
    """All holdfiles, pruning ones long expired with no watcher to own them."""
    out = []
    try:
        names = os.listdir(STATE_DIR)
    except OSError:
        return out
    now = time.time()
    for name in names:
        if not name.startswith(FILE_PREFIX + "-") or not name.endswith(".held.json"):
            continue
        path = os.path.join(STATE_DIR, name)
        try:
            with open(path) as f:
                rec = json.load(f)
        except Exception:
            continue
        if not isinstance(rec, dict) or not rec.get("sig"):
            continue
        try:
            expired_ago = now - float(rec.get("held_until", 0))
        except (TypeError, ValueError):
            expired_ago = 0
        if expired_ago > 300:
            try:
                os.remove(path)
            except OSError:
                pass
            continue
        rec["_file"] = name
        out.append(rec)
    return out


def _peer_suppressed(state, sig, now, log):
    """True when a peer watcher already owns sig; suppress our send.

    Checks the shared on-disk answered store first (peer answered
    earlier and persisted), then attempts an atomic claim (peer racing
    us in the same window). On suppression the pending prompt is
    reset so later polls re-observe cleanly; answered-store hits are
    already merged into memory, claim-loss is not recorded (peer's
    send owns it, and its claim expires quickly if it dies). Never
    raises; memory-only states never suppress.
    """
    try:
        state.refresh_answered()
    except Exception:
        pass
    try:
        ts = state.answered_sigs.get(sig)
        if ts is not None and now - ts < ANSWERED_TTL_SECONDS:
            try:
                log.log("info", "duplicate suppressed (peer answered)",
                        sig=sig)
            except Exception:
                pass
            state.pending_sig = None
            state.stable_count = 0
            return True
    except Exception:
        pass
    try:
        if not state.try_claim(sig, now):
            try:
                log.log("info", "duplicate suppressed (peer claimed)",
                        sig=sig)
            except Exception:
                pass
            state.pending_sig = None
            state.stable_count = 0
            return True
    except Exception:
        pass
    return False


def hold_renews_forever(hold, kind):
    """True when a hold renews instead of expiring to approve.

    Coordinator-gated kinds (grill interviews) and every agy hold
    (unproven sends): release only by operator directive.
    """
    return (kind in HOLD_FOREVER_KINDS
            or (hold or {}).get("harness") == "agy")


def _check_hold(socket_path, pane_id, state, match, now, log):
    """Rule-hold gate. Returns None (fresh: evaluate rules), "released"
    (hold over: approve WITHOUT re-evaluating, else expiry would
    re-hold forever), "held", "denied", or "duplicate-suppressed".

    The holdfile is the single source of truth (crash-safe,
    box-visible): present + fresh => suppress; directive deny => deny
    now; expired or operator-cleared => release back to approve.
    """
    sig = match["sig"]
    hf = read_hold(socket_path, pane_id)
    if hf is None:
        return None
    if hf.get("sig") != sig:
        clear_hold(socket_path, pane_id)  # stale hold for a gone prompt
        return None
    if hf.get("directive") == "approve":
        clear_hold(socket_path, pane_id)
        log.log("info", "hold released by operator, answering", sig=sig)
        return "released"
    if hf.get("directive") == "deny":
        neg = NEGATIVE_KEYS.get(match["kind"])
        if neg is None:
            hf["directive"] = None
            write_hold(socket_path, pane_id, hf)
            log.log("warn", "resolve-deny refused: D2 never denies questions",
                    sig=sig, kind=match["kind"])
            return "held"
        if _peer_suppressed(state, sig, now, log):
            return "duplicate-suppressed"
        ok, detail = send_answer(socket_path, pane_id, neg, enter=True,
                                    kind=match["kind"], sig=sig)
        clear_hold(socket_path, pane_id)
        state.record_answer(sig, now)
        log.log("info" if ok else "error", "denied %s (operator resolve)" % neg,
                sig=sig, ok=ok, kind=match["kind"], key=neg,
                verified=detail["verified"], retried=detail["retried"])
        audit("muse-choice-denied",
              name="%s:%s" % (os.path.basename(socket_path), pane_id),
              extra={"socket": socket_path, "pane": pane_id, "sig": sig,
                     "ok": ok, "kind": match["kind"], "key": neg,
                     "via": "resolve", "verified": detail["verified"],
                     "retried": detail["retried"]})
        return "denied"
    try:
        expired = now >= float(hf.get("held_until", 0))
    except (TypeError, ValueError):
        expired = True
    if expired and hold_renews_forever(hf, match["kind"]):
        # Coordinator-gated kinds, and every agy hold, never expire
        # to approve: renew the window while the prompt persists.
        # Release only by operator directive, or when the prompt
        # scrolls away (sig mismatch) or the operator answers by
        # hand. Renewal failures keep holding; they never fail open
        # to approve.
        try:
            hf["held_until"] = now + HOLD_WINDOW_SECONDS
            hf["renewals"] = int(hf.get("renewals", 0) or 0) + 1
            write_hold(socket_path, pane_id, hf)
        except Exception:
            pass
        else:
            why = ("agy hold-all" if hf.get("harness") == "agy"
                   else "coordinator gate")
            log.log("info", "hold renewed (%s, never auto)" % why,
                    sig=sig, kind=match["kind"],
                    renewals=hf["renewals"])
        return "held"
    if expired:
        clear_hold(socket_path, pane_id)
        log.log("info", "hold expired, releasing to approve", sig=sig)
        audit("muse-choice-hold-expired",
              name="%s:%s" % (os.path.basename(socket_path), pane_id),
              extra={"socket": socket_path, "pane": pane_id, "sig": sig,
                     "kind": match["kind"], "rule": hf.get("rule")})
        return "released"
    return "held"


def _poll_once(socket_path, pane_id, state, log, dry_run=False):
    """Run one poll iteration. Returns an outcome string ("gone" tells the
    caller to exit). Logs only state transitions, never per-poll spam."""
    if not pane_exists(socket_path, pane_id):
        return "gone"
    text = capture_pane(socket_path, pane_id)
    if text is None:
        log.log("warn", "capture failed, continuing")
        return "capture-failed"
    match = find_choice_prompt(text)
    now = time.time()
    gate = _check_hold(socket_path, pane_id, state, match, now, log) \
        if match is not None else None
    if gate in ("held", "denied", "duplicate-suppressed"):
        if gate in ("denied", "duplicate-suppressed"):
            state.pending_sig = None
            state.stable_count = 0
        return gate
    skip_eval = (gate == "released")
    verdict = state.observe(match, now)
    if verdict == "wait" and state.stable_count == 1:
        log.log("info", "prompt seen", kind=match["kind"], key=match["key"],
                sig=match["sig"], text=(match["cue"] or "")[:200])
        return "seen"
    if verdict == "answer":
        log.log("info", "prompt stable, answering", kind=match["kind"],
                key=match["key"], sig=match["sig"])
        # Re-verify the prompt is still present before typing.
        confirm = capture_pane(socket_path, pane_id)
        cmatch = find_choice_prompt(confirm) if confirm else None
        if cmatch is None or cmatch["sig"] != match["sig"]:
            log.log("info", "prompt vanished before answer, skipping",
                    sig=match["sig"])
            state.pending_sig = None
            state.stable_count = 0
            return "vanished"
        if pane_holds_for_optout(socket_path,
                                 pane_muse_argv(socket_path, pane_id)):
            log.log("info", "held: pane opted out via launch flags",
                    sig=match["sig"], kind=match["kind"])
            state.record_answer(match["sig"], now)
            return "held"
        if pane_harness(socket_path, pane_id) == "agy" and not skip_eval:
            # Hold-all: agy auto-sends are unproven (its native menus
            # may ignore digit keys), so every agy prompt holds for
            # coordinator resolve instead of auto-answering. The hold
            # renews forever; a released hold (resolve-approve) falls
            # through and sends the kind's key with a human in the
            # loop -- hence the skip_eval guard (cf. permissions).
            until = now + HOLD_WINDOW_SECONDS
            write_hold(socket_path, pane_id, {
                "sig": match["sig"], "kind": match["kind"],
                "key": match["key"],
                "text": (match["cue"] or "")[:200],
                "rule": AGY_HOLD_RULE,
                "reason": ("agy harness: sends unproven; "
                           "coordinator resolve required"),
                "downgraded_from": None, "harness": "agy",
                "held_until": until, "directive": None,
                "socket": socket_path, "pane": pane_id})
            log.log("info", "held: agy harness (hold-all)",
                    sig=match["sig"], kind=match["kind"])
            if not dry_run:
                audit("muse-choice-held",
                      name="%s:%s" % (os.path.basename(socket_path), pane_id),
                      extra={"socket": socket_path, "pane": pane_id,
                             "sig": match["sig"], "kind": match["kind"],
                             "rule": AGY_HOLD_RULE, "harness": "agy",
                             "held_until": until})
            return "held"
        if match["kind"] == "permissions" and not skip_eval:
            # Builtin hold: a mode change persists as the default for
            # new sessions and exec runs, so no rule may auto-answer
            # it -- the operator drives the TUI by hand. Rules are
            # not consulted for this kind.
            until = now + HOLD_WINDOW_SECONDS
            write_hold(socket_path, pane_id, {
                "sig": match["sig"], "kind": "permissions", "key": None,
                "text": (match["cue"] or "")[:200],
                "rule": "builtin-permissions",
                "reason": ("permission mode screen; operator drives "
                           "the TUI by hand"),
                "downgraded_from": None,
                "held_until": until, "directive": None,
                "socket": socket_path, "pane": pane_id})
            log.log("info", "held: /permissions screen (builtin)",
                    sig=match["sig"], kind="permissions")
            if not dry_run:
                audit("muse-choice-held",
                      name="%s:%s" % (os.path.basename(socket_path), pane_id),
                      extra={"socket": socket_path, "pane": pane_id,
                             "sig": match["sig"], "kind": "permissions",
                             "rule": "builtin-permissions",
                             "held_until": until})
            return "held"
        if skip_eval:
            decision, rule = "approve", None
            via = "hold-released"
        else:
            decision, rule = evaluate_rules(match, log)
            via = "direct"
        key = match["key"]
        if decision == "deny":
            neg = NEGATIVE_KEYS[match["kind"]]
            if dry_run:
                log.log("info", "dry-run would deny %s" % neg,
                        sig=match["sig"], kind=match["kind"], key=neg,
                        rule=rule["id"] if rule else None)
                state.record_answer(match["sig"], now)
                return "dry-denied"
            if _peer_suppressed(state, match["sig"], now, log):
                return "duplicate-suppressed"
            ok, detail = send_answer(socket_path, pane_id, neg, enter=True,
                                        kind=match["kind"],
                                        sig=match["sig"])
            state.record_answer(match["sig"], now)
            log.log("info" if ok else "error", "denied %s" % neg,
                    sig=match["sig"], ok=ok, kind=match["kind"], key=neg,
                    rule=rule["id"] if rule else None,
                    verified=detail["verified"], retried=detail["retried"])
            audit("muse-choice-denied",
                  name="%s:%s" % (os.path.basename(socket_path), pane_id),
                  extra={"socket": socket_path, "pane": pane_id,
                         "sig": match["sig"], "ok": ok, "kind": match["kind"],
                         "key": neg, "via": "rule",
                         "rule": rule["id"] if rule else None,
                         "verified": detail["verified"],
                         "retried": detail["retried"]})
            return "denied"
        if decision == "hold":
            until = now + HOLD_WINDOW_SECONDS
            write_hold(socket_path, pane_id, {
                "sig": match["sig"], "kind": match["kind"], "key": key,
                "text": (match["cue"] or "")[:200],
                "rule": rule["id"] if rule else None,
                "reason": (rule.get("reason") if rule else None) or "",
                "downgraded_from": (rule.get("downgraded_from")
                                    if rule else None),
                "held_until": until, "directive": None,
                "socket": socket_path, "pane": pane_id})
            log.log("info", "held by rule %s" % (rule["id"] if rule else "?"),
                    sig=match["sig"], kind=match["kind"],
                    rule=rule["id"] if rule else None,
                    reason=(rule.get("reason") if rule else None) or "")
            if not dry_run:
                audit("muse-choice-held",
                      name="%s:%s" % (os.path.basename(socket_path), pane_id),
                      extra={"socket": socket_path, "pane": pane_id,
                             "sig": match["sig"], "kind": match["kind"],
                             "rule": rule["id"] if rule else None,
                             "held_until": until})
            return "held"
        if dry_run:
            log.log("info", "dry-run would answer %s" % key,
                    sig=match["sig"], kind=match["kind"], key=key,
                    options=match["options"], cue=match["cue"])
            state.record_answer(match["sig"], now)
            return "dry-answered"
        if match["kind"] == "permissions":
            # Released (operator resolve or hold expiry): still send
            # nothing -- the operator drives mode changes by hand.
            # Record so this screen stops holding.
            log.log("info", "permissions released to operator (no keys sent)",
                    sig=match["sig"], cue=match["cue"])
            state.record_answer(match["sig"], now)
            audit("muse-choice-answered",
                  name="%s:%s" % (os.path.basename(socket_path), pane_id),
                  extra={"socket": socket_path, "pane": pane_id,
                         "sig": match["sig"], "ok": True,
                         "kind": "permissions", "key": None,
                         "options": match["options"], "cue": match["cue"],
                         "via": via,
                         "rule": rule["id"] if rule else None})
            return "answered"
        if _peer_suppressed(state, match["sig"], now, log):
            return "duplicate-suppressed"
        ok, detail = send_answer(socket_path, pane_id, key,
                                    enter=match.get("enter", True),
                                    kind=match["kind"], sig=match["sig"])
        state.record_answer(match["sig"], now)
        log.log("info" if ok else "error", "answered %s" % key,
                sig=match["sig"], ok=ok, kind=match["kind"], key=key,
                options=match["options"], cue=match["cue"],
                verified=detail["verified"], retried=detail["retried"])
        audit("muse-choice-answered",
              name="%s:%s" % (os.path.basename(socket_path), pane_id),
              extra={"socket": socket_path, "pane": pane_id,
                     "sig": match["sig"], "ok": ok, "kind": match["kind"],
                     "key": key, "options": match["options"],
                     "cue": match["cue"], "via": via,
                     "rule": rule["id"] if rule else None,
                     "verified": detail["verified"],
                     "retried": detail["retried"]})
        return "answered"
    if verdict == "capped":
        if state.last_capped_sig != match["sig"]:
            state.last_capped_sig = match["sig"]
            log.log("warn", "hourly answer cap reached, holding",
                    sig=match["sig"], kind=match["kind"])
        return "capped"
    if verdict == "none":
        state.last_capped_sig = None
        return "none"
    return "waiting"


def _log_posture(socket_path, pane_id, log, saved_default=None):
    """Log the pane's permission posture once at watcher start.

    The watcher answers with per-choice logging in every mode (that
    trail is the default path and informs policy); an effective bypass
    posture (yolo / approval disabled / unrestricted saved default
    with no explicit flags) additionally gets a box audit record,
    since the session then makes choices outside the trail. Never
    raises.
    """
    try:
        posture = muse_approval_flags(pane_muse_argv(socket_path, pane_id))
    except Exception:
        return
    try:
        effective = effective_posture(posture, saved_default)
    except Exception:
        effective = {"effective_mode": posture["mode"],
                     "effective_bypass": posture["bypass"]}
    try:
        log.log("info", "pane posture", mode=posture["mode"],
                bypass=posture["bypass"], flags=posture["flags"],
                effective_mode=effective["effective_mode"],
                effective_bypass=effective["effective_bypass"])
    except Exception:
        pass
    try:
        if (effective["effective_bypass"]
                or effective["effective_mode"] not in ("default", None)):
            audit("muse-choice-posture",
                  name="%s:%s" % (os.path.basename(socket_path), pane_id),
                  extra={"socket": socket_path, "pane": pane_id,
                         "mode": posture["mode"],
                         "profile": posture["profile"],
                         "bypass": posture["bypass"],
                         "flags": posture["flags"],
                         "effective_mode": effective["effective_mode"],
                         "effective_bypass":
                             effective["effective_bypass"]})
    except Exception:
        pass


def watch_loop(socket_path, pane_id, dry_run=False):
    """Main daemon loop. Returns only when the pane is gone or signalled."""
    log = WatcherLog(logfile_for(socket_path, pane_id))
    state = WatcherState(
        persist_path=answered_file_for(socket_path, pane_id),
        hourly_cap=hourly_cap_for(socket_path))
    log.log("info", "watcher started", socket=socket_path, pane=pane_id,
            dry_run=dry_run, pid=os.getpid())
    _log_posture(socket_path, pane_id, log,
                 saved_default=saved_default_profile())
    polls = 0
    answers = 0
    last_heartbeat = time.time()
    while True:
        try:
            outcome = _poll_once(socket_path, pane_id, state, log,
                                 dry_run=dry_run)
            if outcome == "gone":
                log.log("info", "pane gone, exiting", polls=polls,
                        answers=answers)
                return 0
            if outcome in ("answered", "dry-answered"):
                answers += 1
            polls += 1
            if time.time() - last_heartbeat >= HEARTBEAT_SECONDS:
                last_heartbeat = time.time()
                log.log("info", "heartbeat", polls=polls, answers=answers,
                        pending=state.pending_sig)
        except Exception as e:
            try:
                log.log("error", "poll iteration failed, continuing",
                        error="%r" % (e,))
            except Exception:
                pass
        time.sleep(POLL_INTERVAL)


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


def _pid_is_watcher(pid):
    """True if pid looks like our watcher (guards against pid reuse)."""
    try:
        with open("/proc/%d/cmdline" % pid, "rb") as f:
            cmd = f.read().decode(errors="replace")
        return "muse_choice_watcher" in cmd
    except Exception:
        return False


def _watch_procs(proc_root="/proc"):
    """Live processes running this module's foreground `watch` argv.

    Returns [{"pid", "socket", "pane"}]. Exact argv match only -- it can
    never misidentify another program, so callers may act on the result.
    """
    out = []
    try:
        pids = os.listdir(proc_root)
    except OSError:
        return out
    me = os.getpid()
    for pid in pids:
        if not pid.isdigit() or int(pid) == me:
            continue
        try:
            with open(os.path.join(proc_root, pid, "cmdline"), "rb") as f:
                parts = f.read().split(b"\0")
        except (OSError, IOError):
            continue
        try:
            args = [p.decode(errors="replace") for p in parts if p]
        except Exception:
            continue
        if len(args) < 6:
            continue
        if not args[1].endswith("muse_choice_watcher.py") or args[2] != "watch":
            continue
        try:
            sock = args[args.index("--socket") + 1]
            pane = args[args.index("--pane") + 1]
        except (ValueError, IndexError):
            continue
        out.append({"pid": int(pid), "socket": sock, "pane": pane})
    return out


def is_running(socket_path, pane_id):
    pidfile = pidfile_for(socket_path, pane_id)
    try:
        with open(pidfile) as f:
            pid = int(f.read().strip())
    except Exception:
        return None
    if _pid_alive(pid) and _pid_is_watcher(pid):
        return pid
    try:
        os.remove(pidfile)
    except OSError:
        pass
    return None


def watcher_alive(socket_path, pane_id):
    """Pid of the live watcher for a pane, pidfile or orphan.

    is_running covers the normal pidfile case; _watch_procs catches an
    identical watcher alive with a lost pidfile (/tmp cleaned under
    it). Starters must consult this (not is_running alone) or they
    spawn a second daemon onto the same pane (double answers). Never
    raises; None means no live watcher.
    """
    try:
        pid = is_running(socket_path, pane_id)
        if pid:
            return pid
    except Exception:
        pass
    try:
        for proc in _watch_procs():
            if proc["socket"] == socket_path and proc["pane"] == pane_id:
                return proc["pid"]
    except Exception:
        pass
    return None


def _daemonize():
    """Double-fork away from the controlling terminal (survives shell exit)."""
    if os.fork() != 0:
        os._exit(0)
    os.setsid()
    if os.fork() != 0:
        os._exit(0)
    devnull = os.open(os.devnull, os.O_RDWR)
    os.dup2(devnull, 0)
    os.dup2(devnull, 1)
    os.dup2(devnull, 2)
    if devnull > 2:
        os.close(devnull)


_PIDFILE_LOCK_FH = None


def _claim_pidfile(pidfile):
    """Claim a pidfile for this process. Returns False if another
    starter holds it. Kernel-enforced: the winner holds an exclusive
    flock for its whole lifetime, so concurrent box + timer starts can
    never pile two daemons onto one pane (double answers, '11' in the
    input box). Call _release_pidfile() on exit."""
    global _PIDFILE_LOCK_FH
    me = os.getpid()
    try:
        fh = open(pidfile, "a+")
    except OSError:
        return False
    try:
        fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except (OSError, IOError):
        fh.close()
        return False
    try:
        fh.seek(0)
        content = fh.read().strip()
        other = int(content) if content else None
    except Exception:
        other = None
    if other and other != me and _pid_alive(other) and _pid_is_watcher(other):
        # Live foreign owner (e.g. a daemon from before locking existed,
        # or a pid recycled into a watcher): yield to it.
        try:
            fcntl.flock(fh, fcntl.LOCK_UN)
        except Exception:
            pass
        fh.close()
        return False
    try:
        fh.seek(0)
        fh.truncate()
        fh.write(str(me))
        fh.flush()
    except Exception:
        pass
    if _PIDFILE_LOCK_FH is not None:
        try:
            _PIDFILE_LOCK_FH.close()
        except Exception:
            pass
    _PIDFILE_LOCK_FH = fh
    return True


def _release_pidfile(pidfile):
    """Drop the claim taken by _claim_pidfile: remove the pidfile only
    if we still own it, then release the lock. Never raises."""
    global _PIDFILE_LOCK_FH
    try:
        with open(pidfile) as f:
            owner = f.read().strip()
    except Exception:
        owner = ""
    if owner == str(os.getpid()):
        try:
            os.remove(pidfile)
        except OSError:
            pass
    if _PIDFILE_LOCK_FH is not None:
        try:
            fcntl.flock(_PIDFILE_LOCK_FH, fcntl.LOCK_UN)
        except Exception:
            pass
        try:
            _PIDFILE_LOCK_FH.close()
        except Exception:
            pass
        _PIDFILE_LOCK_FH = None


def start_watcher(socket_path, pane_id, dry_run=False):
    pid = watcher_alive(socket_path, pane_id)
    if pid:
        return {"ok": False, "status": "already_running", "pid": pid}
    if not pane_exists(socket_path, pane_id):
        return {"ok": False, "status": "no_such_pane"}
    _daemonize()
    # Child continues here with a clean identity: re-exec ourselves as a
    # foreground `watch` process so /proc cmdline always names this module
    # (liveness checks and ps output stay correct no matter who spawned us --
    # box, timer, or direct CLI).
    args = [sys.executable, os.path.abspath(__file__), "watch",
            "--socket", socket_path, "--pane", pane_id]
    if dry_run:
        args.append("--dry-run")
    os.execvp(sys.executable, args)
    os._exit(1)  # unreachable; exec replaces the image


def stop_watcher(socket_path, pane_id, timeout=5):
    pid = watcher_alive(socket_path, pane_id)
    if not pid:
        return {"ok": True, "status": "not_running"}
    try:
        os.kill(pid, signal.SIGTERM)
    except ProcessLookupError:
        pass
    deadline = time.time() + timeout
    while time.time() < deadline and _pid_alive(pid):
        time.sleep(0.1)
    if _pid_alive(pid):
        try:
            os.kill(pid, signal.SIGKILL)
        except ProcessLookupError:
            pass
    try:
        os.remove(pidfile_for(socket_path, pane_id))
    except OSError:
        pass
    return {"ok": True, "status": "stopped", "pid": pid}


def agent_panes(socket_path):
    """Pane ids on a socket whose current command looks like an agent.

    Every agent harness is watched by default; extend
    AGENT_PANE_HINTS as new harnesses appear.
    """
    try:
        r = _tmux(socket_path, "list-panes", "-a", "-F",
                  "#{pane_id} #{pane_current_command}", timeout=10)
        if r.returncode != 0:
            return []
    except Exception:
        return []
    out = []
    for line in r.stdout.split("\n"):
        parts = line.strip().split(None, 1)
        if len(parts) == 2 and any(h in parts[1]
                                  for h in AGENT_PANE_HINTS):
            out.append(parts[0])
    return out


muse_panes = agent_panes  # backward-compatible alias


def wait_for_session_pane(socket_path, session, timeout=10, interval=0.5):
    """True once the session has a pane running an agent command.

    Launch-time boot race: `new-session` returns before the agent
    binary is visible as pane_current_command, so a reconcile fired
    immediately would find no agent panes and start no watcher.
    Bounded and never fatal: on timeout just proceed to reconcile
    (the timer heals the rest). Never raises.
    """
    try:
        deadline = time.time() + timeout
    except Exception:
        return False
    while True:
        try:
            r = _tmux(socket_path, "list-panes", "-a", "-F",
                      "#{session_name} #{pane_id} #{pane_current_command}",
                      timeout=10)
        except Exception:
            return False
        if r.returncode == 0:
            for line in r.stdout.split("\n"):
                parts = line.strip().split(None, 2)
                if len(parts) == 3 and parts[0] == session and any(
                        h in parts[2] for h in AGENT_PANE_HINTS):
                    return True
        try:
            if time.time() >= deadline:
                return False
            time.sleep(interval)
        except Exception:
            return False


def _start_detached(socket_path, pane_id, dry_run=False):
    """Fork off a watcher via start_watcher (which daemonizes further).
    Returns True if a watcher is running for the pane afterwards."""
    pid = os.fork()
    if pid == 0:
        start_watcher(socket_path, pane_id, dry_run=dry_run)
        os._exit(0)
    os.waitpid(pid, 0)
    time.sleep(0.2)
    return watcher_alive(socket_path, pane_id) is not None


def start_all(dry_run=False, sockets=None):
    results = []
    for sock in sockets or KNOWN_SOCKETS:
        if not os.path.exists(sock):
            results.append({"socket": sock, "status": "no_socket"})
            continue
        panes = agent_panes(sock)
        if not panes:
            results.append({"socket": sock, "status": "no_muse_panes"})
        for pane in panes:
            if watcher_alive(sock, pane):
                results.append({"socket": sock, "pane": pane,
                                "status": "already_running"})
                continue
            ok = _start_detached(sock, pane, dry_run=dry_run)
            results.append({"socket": sock, "pane": pane,
                            "status": "started" if ok else "failed"})
    return results


def stop_all():
    results = []
    stopped_pids = set()
    prefix = FILE_PREFIX + "-"
    try:
        names = os.listdir(STATE_DIR)
    except OSError:
        names = []
    for name in names:
        if not name.startswith(prefix) or not name.endswith(".pid"):
            continue
        try:
            with open(os.path.join(STATE_DIR, name)) as f:
                pid = int(f.read().strip())
        except Exception:
            continue
        if _pid_alive(pid) and _pid_is_watcher(pid):
            try:
                os.kill(pid, signal.SIGTERM)
            except ProcessLookupError:
                pass
            stopped_pids.add(pid)
            results.append({"pidfile": name, "pid": pid, "status": "stopped"})
        try:
            os.remove(os.path.join(STATE_DIR, name))
        except OSError:
            pass
    # "off means off": also stop exact-argv watchers running pidfile-less.
    for proc in _watch_procs():
        if proc["pid"] in stopped_pids:
            continue
        try:
            os.kill(proc["pid"], signal.SIGTERM)
        except ProcessLookupError:
            pass
        results.append({"pidfile": None, "pid": proc["pid"],
                        "status": "stopped-orphan", "socket": proc["socket"],
                        "pane": proc["pane"]})
    return results


def reconcile(sockets=None):
    """Enforce desired state: start missing watchers when enabled, stop all
    when disabled, prune dead pidfiles. Audits only when it changed
    something. Returns a summary dict."""
    desired = get_desired()
    started, already, pruned, failed = [], [], [], []
    stopped = []
    if desired["enabled"]:
        for sock in sockets or KNOWN_SOCKETS:
            if not os.path.exists(sock):
                continue
            for pane in agent_panes(sock):
                if watcher_alive(sock, pane):
                    already.append("%s:%s" % (sock, pane))
                    continue
                if _start_detached(sock, pane, dry_run=desired["dry_run"]):
                    started.append("%s:%s" % (sock, pane))
                else:
                    failed.append("%s:%s" % (sock, pane))
        for row in status_all():
            if not row["alive"]:
                try:
                    os.remove(os.path.join(STATE_DIR, row["pidfile"]))
                    pruned.append(row["pidfile"])
                except OSError:
                    pass
    else:
        stopped = ["%s:%s" % (r.get("pidfile"), r.get("pid")) for r in stop_all()]
    if started or stopped or pruned or failed:
        audit("muse-choice-reconciled",
              extra={"started": started, "stopped": stopped, "pruned": pruned,
                     "failed": failed})
    return {"enabled": desired["enabled"], "dry_run": desired["dry_run"],
            "started": started, "already": already,
            "stopped": stopped, "pruned": pruned, "failed": failed}


def recent_answers(limit=10, scan_lines=3000):
    """Most recent muse-choice-answered audit records (newest last)."""
    try:
        with open(CTL_LOG) as f:
            lines = f.readlines()[-scan_lines:]
    except OSError:
        return []
    out = []
    for line in lines:
        try:
            rec = json.loads(line)
        except Exception:
            continue
        if rec.get("action") == "muse-choice-answered":
            out.append(rec)
    return out[-limit:]


def status_all():
    rows = []
    prefix = FILE_PREFIX + "-"
    try:
        names = sorted(os.listdir(STATE_DIR))
    except OSError:
        return rows
    for name in names:
        if not name.startswith(prefix) or not name.endswith(".pid"):
            continue
        pid = None
        try:
            with open(os.path.join(STATE_DIR, name)) as f:
                pid = int(f.read().strip())
        except Exception:
            pass
        alive = bool(pid) and _pid_alive(pid) and _pid_is_watcher(pid)
        logname = name[:-4] + ".log"
        logsize = None
        try:
            logsize = os.path.getsize(os.path.join(STATE_DIR, logname))
        except OSError:
            pass
        rows.append({"pidfile": name, "pid": pid, "alive": alive,
                     "log": logname, "log_bytes": logsize})
    owned = set(r["pid"] for r in rows if r["alive"] and r["pid"])
    for proc in _watch_procs():
        if proc["pid"] in owned:
            continue
        logname = os.path.basename(
            logfile_for(proc["socket"], proc["pane"]))
        try:
            logsize = os.path.getsize(os.path.join(STATE_DIR, logname))
        except OSError:
            logsize = None
        rows.append({"pidfile": None, "pid": proc["pid"], "alive": True,
                     "orphan": True, "socket": proc["socket"],
                     "pane": proc["pane"], "log": logname,
                     "log_bytes": logsize})
    return rows


def main(argv=None):
    ap = argparse.ArgumentParser(description="Muse A/B/C choice watcher")
    sub = ap.add_subparsers(dest="cmd", required=True)

    p = sub.add_parser("start", help="start watching one pane (daemonizes)")
    p.add_argument("--socket", required=True)
    p.add_argument("--pane", required=True)
    p.add_argument("--dry-run", action="store_true",
                   help="log answers instead of sending keys")

    p = sub.add_parser("stop", help="stop one pane watcher")
    p.add_argument("--socket", required=True)
    p.add_argument("--pane", required=True)

    p = sub.add_parser("start-all", help="watch every Muse pane on known sockets")
    p.add_argument("--dry-run", action="store_true")
    p.add_argument("--socket", action="append", dest="sockets", default=None)

    sub.add_parser("stop-all", help="stop all watchers")
    sub.add_parser("status", help="list watcher state files + liveness")

    p = sub.add_parser("on", help="enable auto-answers (desired state) + reconcile now")
    p.add_argument("--dry-run", action="store_true",
                   help="log answers instead of sending keys")
    p.add_argument("--by", default="cli", help="caller identity for audit")

    p = sub.add_parser("off", help="disable auto-answers (desired state) + stop all now")
    p.add_argument("--by", default="cli", help="caller identity for audit")

    p = sub.add_parser("reconcile", help="enforce desired state (for timer)")
    p.add_argument("--socket", action="append", dest="sockets", default=None)

    p = sub.add_parser("watch", help="run one pane loop in foreground (daemon target)")
    p.add_argument("--socket", required=True)
    p.add_argument("--pane", required=True)
    p.add_argument("--dry-run", action="store_true")

    p = sub.add_parser("match", help="print JSON verdict for text on stdin")
    p.add_argument("--tail-window", type=int, default=TAIL_WINDOW)

    p = sub.add_parser("logs", help="tail one pane watcher log")
    p.add_argument("--socket", required=True)
    p.add_argument("--pane", required=True)
    p.add_argument("-n", type=int, default=20)

    p = sub.add_parser("state", help="print JSON runtime state for one pane")
    p.add_argument("--socket", required=True)
    p.add_argument("--pane", required=True)

    args = ap.parse_args(argv)
    if args.cmd == "start":
        existing = watcher_alive(args.socket, args.pane)
        if existing:
            print(json.dumps({"ok": True, "status": "already_running",
                              "pid": existing,
                              "log": logfile_for(args.socket, args.pane)}))
            return 0
        # start_watcher never returns in the daemon child; the parent fork
        # dance happens inside, so wrap: fork here, child calls start.
        pid = os.fork()
        if pid == 0:
            start_watcher(args.socket, args.pane, dry_run=args.dry_run)
            os._exit(0)
        _, status = os.waitpid(pid, 0)
        time.sleep(0.3)
        running = watcher_alive(args.socket, args.pane)
        print(json.dumps({"ok": running is not None, "pid": running,
                          "log": logfile_for(args.socket, args.pane)}))
        return 0 if running else 1
    if args.cmd == "stop":
        print(json.dumps(stop_watcher(args.socket, args.pane)))
        return 0
    if args.cmd == "start-all":
        print(json.dumps(start_all(dry_run=args.dry_run, sockets=args.sockets), indent=1))
        return 0
    if args.cmd == "stop-all":
        print(json.dumps(stop_all(), indent=1))
        return 0
    if args.cmd == "status":
        print(json.dumps({"desired": get_desired(), "watchers": status_all()},
                         indent=1))
        return 0
    if args.cmd == "on":
        state = set_enabled(True, dry_run=args.dry_run, by=args.by)
        res = reconcile()
        print(json.dumps({"desired": state, "reconcile": res}, indent=1))
        return 0
    if args.cmd == "off":
        state = set_enabled(False, by=args.by)
        stopped = stop_all()
        print(json.dumps({"desired": state, "stopped": stopped}, indent=1))
        return 0
    if args.cmd == "reconcile":
        print(json.dumps(reconcile(sockets=args.sockets), indent=1))
        return 0
    if args.cmd == "watch":
        # Refuse to pile on: an identical watcher may be alive with a lost
        # pidfile (e.g. /tmp cleaned under it). Exact argv match, no guessing.
        for proc in _watch_procs():
            if proc["socket"] == args.socket and proc["pane"] == args.pane:
                return 3
        pidfile = pidfile_for(args.socket, args.pane)
        if not _claim_pidfile(pidfile):
            return 3
        try:
            return watch_loop(args.socket, args.pane, dry_run=args.dry_run)
        finally:
            _release_pidfile(pidfile)
    if args.cmd == "match":
        text = sys.stdin.read()
        print(json.dumps(find_choice_prompt(text, tail_window=args.tail_window), indent=1))
        return 0
    if args.cmd == "logs":
        path = logfile_for(args.socket, args.pane)
        try:
            with open(path) as f:
                lines = f.readlines()
        except OSError as e:
            print("no log: %s (%s)" % (path, e))
            return 1
        for line in lines[-args.n:]:
            print(line.rstrip("\n"))
        return 0
    if args.cmd == "state":
        print(json.dumps(pane_state(args.socket, args.pane), indent=1))
        return 0
    return 1


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