operator
2067 lines
75 KiB
Python
Executable File
2067 lines
75 KiB
Python
Executable File
#!/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", y/n
|
||
line-end prompt -> "y".
|
||
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 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",
|
||
]
|
||
|
||
POLL_INTERVAL = 0.5
|
||
STABILITY_POLLS = 2
|
||
MAX_ANSWERS_PER_HOUR = 20
|
||
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"})
|
||
_RULES_CACHE = {"key": None, "rules": []}
|
||
LOG_MAX_BYTES = 1_000_000
|
||
HEARTBEAT_SECONDS = 60
|
||
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)
|
||
|
||
# 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
|
||
|
||
# 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
|
||
|
||
# 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 _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)
|
||
hi = min(len(window), end + 5)
|
||
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)
|
||
hi = min(len(window), end + 4)
|
||
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 _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_choice_prompt(text, tail_window=TAIL_WINDOW):
|
||
"""Detect a Muse prompt awaiting reply.
|
||
|
||
Kinds (priority order): 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"), 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_muse_approval(window) or _find_collapsed_approval(window)
|
||
or _find_interview(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_muse_argv(socket_path, pane_id):
|
||
"""Argv of the muse process in a pane (empty when not found)."""
|
||
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 []
|
||
for cand in [pane_pid] + _child_pids(pane_pid):
|
||
argv = _cmdline(cand)
|
||
if argv and ("muse-bin" in argv[0] or "muse-code" in argv[0]):
|
||
return argv
|
||
return []
|
||
|
||
|
||
# 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):
|
||
"""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.
|
||
"""
|
||
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:
|
||
return rows
|
||
except Exception:
|
||
return rows
|
||
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)
|
||
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"],
|
||
"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 all_runtime_rows(sockets=None):
|
||
"""runtime_rows across every known socket that exists."""
|
||
rows = []
|
||
for sock in sockets or KNOWN_SOCKETS:
|
||
if not os.path.exists(sock):
|
||
continue
|
||
rows.extend(runtime_rows(sock))
|
||
return rows
|
||
|
||
|
||
def pane_state(socket_path, pane_id):
|
||
"""Single-pane runtime row, or {"error": ...} when not found."""
|
||
for row in runtime_rows(socket_path):
|
||
if row["pane"] == pane_id:
|
||
return row
|
||
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):
|
||
self.pending_sig = None
|
||
self.stable_count = 0
|
||
self.answered_sigs = {}
|
||
self.answer_times = []
|
||
self.last_capped_sig = None
|
||
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) >= MAX_ANSWERS_PER_HOUR
|
||
|
||
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 send_answer(socket_path, pane_id, letter="A", enter=True):
|
||
"""Type the choice key (+ Enter unless enter=False). True on success."""
|
||
try:
|
||
r1 = _tmux(socket_path, "send-keys", "-t", pane_id, letter, timeout=5)
|
||
if r1.returncode != 0:
|
||
return False
|
||
if not enter:
|
||
return True
|
||
r2 = _tmux(socket_path, "send-keys", "-t", pane_id, "Enter", timeout=5)
|
||
return r2.returncode == 0
|
||
except Exception:
|
||
return 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 _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 = send_answer(socket_path, pane_id, neg, enter=True)
|
||
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)
|
||
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"})
|
||
return "denied"
|
||
try:
|
||
expired = now >= float(hf.get("held_until", 0))
|
||
except (TypeError, ValueError):
|
||
expired = True
|
||
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 launch_opt_out(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 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 = send_answer(socket_path, pane_id, neg, enter=True)
|
||
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)
|
||
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})
|
||
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 _peer_suppressed(state, match["sig"], now, log):
|
||
return "duplicate-suppressed"
|
||
ok = send_answer(socket_path, pane_id, key,
|
||
enter=match.get("enter", True))
|
||
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"])
|
||
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})
|
||
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):
|
||
"""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); a bypass posture
|
||
(yolo / approval disabled) 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:
|
||
log.log("info", "pane posture", mode=posture["mode"],
|
||
bypass=posture["bypass"], flags=posture["flags"])
|
||
except Exception:
|
||
pass
|
||
try:
|
||
if posture["bypass"] or posture["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"]})
|
||
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))
|
||
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)
|
||
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 muse_panes(socket_path):
|
||
"""Pane ids on a socket whose current command looks like Muse."""
|
||
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 "muse-bin" in parts[1]:
|
||
out.append(parts[0])
|
||
return out
|
||
|
||
|
||
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 = muse_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 muse_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())
|