operator-main
18db8ff80b
- Exit 0 on success (was 1 when prompts sent, confusing systemd) - [!] flag now uses word-boundary regex excluding hyphenated identities (operator-646 no longer triggers; 'operator needed' does) - State writes now hold fcntl exclusive lock with fresh reload, preventing timer check from clobbering enable/disable changes Session: sidechat/chromebox-ops
359 lines
13 KiB
Python
Executable File
359 lines
13 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""self_main_loop.py — main-chat self-monitor loop.
|
|
|
|
Failure mode it escapes: main chat is the coordination surface, but if no
|
|
agent is actively watching it, questions / job assignments / alerts posted
|
|
there go unanswered.
|
|
|
|
How it works:
|
|
1. box drives a read of each watched agent's muse.ai Main chat on a timer
|
|
(via box-chat.py thread-messages <agent> main — existing machinery).
|
|
2. On NEW messages since the per-agent watermark, the loop composes a
|
|
concise digest and posts it to that agent's prompting sidechat
|
|
(via dm.py send, opm browser as neutral sender — mirrors box notify).
|
|
3. The prompt in the sidechat triggers the operator to DM / use box
|
|
against main chat. In such, we escape the failure mode.
|
|
|
|
State: /home/super/Projects/NetVM/self-main-loop-watermark.json
|
|
{ "config": {...}, "agents": {agent: {"last_id","last_ts"}},
|
|
"last_run": iso, "last_result": {...} }
|
|
First run per agent starts at the current newest message (no backfill spam).
|
|
|
|
Contract:
|
|
- No new messages -> silent (no sidechat post), exit 0.
|
|
- Digests sent -> exit 0 (success; prompted count in JSON).
|
|
- Read/send error -> logged to stderr, exit 2 (timer stays alive).
|
|
- Overlap guard: fcntl LOCK_EX|LOCK_NB on a lockfile; a second
|
|
concurrent run prints {"ok": false, "skipped": "already running"}
|
|
and exits 0.
|
|
|
|
stdout is always a single JSON object (box-ctl.py parses it); logs go to
|
|
stderr (systemd journal).
|
|
"""
|
|
|
|
import fcntl
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timezone
|
|
|
|
BASE = "/home/super/Projects/NetVM"
|
|
BIN = os.path.join(BASE, "bin")
|
|
BOX_CHAT = os.path.join(BIN, "box-chat.py")
|
|
DM_PY = os.path.join(BIN, "dm.py")
|
|
STATE_FILE = os.path.join(BASE, "self-main-loop-watermark.json")
|
|
LOCK_FILE = os.path.join(BASE, "self-main-loop.lock")
|
|
STATE_LOCK_FILE = os.path.join(BASE, "self-main-loop-state.lock")
|
|
|
|
DEFAULT_AGENTS = ["muse", "pip", "646", "opm"]
|
|
# Prompting sidechat per agent — mirrors box-ctl.py NOTIFY_SIDECHATS.
|
|
DEFAULT_PROMPT_SIDECHAT = {
|
|
"646": "646 tasks",
|
|
"opm": "heartbeat",
|
|
"pip": "646-pip-coord",
|
|
"muse": "muse tasks",
|
|
}
|
|
DEFAULT_SENDER = "opm" # neutral sender, mirrors `box notify`
|
|
READ_LIMIT = 30
|
|
READ_TIMEOUT = 120
|
|
SEND_TIMEOUT = 180
|
|
DIGEST_MAX = 600 # well under dm.py's 1000-char non-raw truncation
|
|
PREVIEW_MAX = 120
|
|
|
|
# Word-boundary "operator" that does NOT match hyphenated identities like
|
|
# "operator-646" (the "-" counts as a boundary for \b, so we exclude it
|
|
# explicitly). "operator needed" matches; "operator-646" does not.
|
|
OPERATOR_RE = re.compile(r"(?<![\w-])operator(?![\w-])", re.IGNORECASE)
|
|
|
|
|
|
@contextmanager
|
|
def state_locked():
|
|
"""Exclusive lock for load-modify-save on the state file.
|
|
|
|
Prevents the timer's check (mid-run save) from clobbering an
|
|
enable/disable change that landed during the slow chat reads.
|
|
Held only for the brief critical section, never across network I/O.
|
|
"""
|
|
fh = open(STATE_LOCK_FILE, "w")
|
|
try:
|
|
fcntl.flock(fh, fcntl.LOCK_EX)
|
|
yield
|
|
finally:
|
|
fcntl.flock(fh, fcntl.LOCK_UN)
|
|
fh.close()
|
|
|
|
|
|
def utcnow():
|
|
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
|
|
def log(msg):
|
|
print("self-main-loop: %s" % msg, file=sys.stderr)
|
|
|
|
|
|
def load_state():
|
|
try:
|
|
with open(STATE_FILE) as f:
|
|
st = json.load(f)
|
|
if isinstance(st, dict):
|
|
return st
|
|
except (FileNotFoundError, json.JSONDecodeError, ValueError):
|
|
pass
|
|
return {}
|
|
|
|
|
|
def save_state(st):
|
|
tmp = STATE_FILE + ".tmp"
|
|
with open(tmp, "w") as f:
|
|
json.dump(st, f, indent=2)
|
|
os.replace(tmp, STATE_FILE)
|
|
|
|
|
|
def get_config(st):
|
|
cfg = st.get("config") or {}
|
|
agents = cfg.get("agents") or list(DEFAULT_AGENTS)
|
|
agents = [a for a in agents if a in DEFAULT_AGENTS]
|
|
prompt = dict(DEFAULT_PROMPT_SIDECHAT)
|
|
prompt.update(cfg.get("prompt_sidechat") or {})
|
|
enabled = dict(cfg.get("enabled") or {})
|
|
# Default any unlisted agent to enabled.
|
|
for a in agents:
|
|
enabled.setdefault(a, True)
|
|
return {
|
|
"agents": agents or list(DEFAULT_AGENTS),
|
|
"prompt_sidechat": prompt,
|
|
"sender": cfg.get("sender") or DEFAULT_SENDER,
|
|
"read_limit": int(cfg.get("read_limit") or READ_LIMIT),
|
|
"enabled": enabled,
|
|
}
|
|
|
|
|
|
def read_main_chat(agent, limit):
|
|
"""Read agent's muse.ai Main chat via box-chat.py. Returns (ok, messages|error)."""
|
|
cmd = [sys.executable, BOX_CHAT, "thread-messages", agent, "main",
|
|
"--limit", str(limit)]
|
|
try:
|
|
r = subprocess.run(cmd, capture_output=True, text=True, timeout=READ_TIMEOUT)
|
|
except subprocess.TimeoutExpired:
|
|
return False, "box-chat.py timed out after %ss" % READ_TIMEOUT
|
|
except OSError as e:
|
|
return False, "could not exec box-chat.py: %s" % e
|
|
try:
|
|
data = json.loads(r.stdout.strip())
|
|
except Exception:
|
|
return False, "box-chat.py non-JSON output: %s" % (r.stdout or r.stderr)[-200:]
|
|
if not data.get("ok"):
|
|
return False, data.get("error", "box-chat.py failed")
|
|
return True, data.get("messages") or []
|
|
|
|
|
|
def new_messages(messages, wm):
|
|
"""Messages newer than the watermark (oldest-first list)."""
|
|
if not messages:
|
|
return []
|
|
last_id = (wm or {}).get("last_id")
|
|
if not last_id:
|
|
return [] # first run: anchor at newest, no backfill
|
|
ids = [m.get("id") for m in messages]
|
|
if last_id in ids:
|
|
return messages[ids.index(last_id) + 1:]
|
|
# Watermark id fell out of the read window: fall back to ts compare.
|
|
last_ts = (wm or {}).get("last_ts") or ""
|
|
return [m for m in messages if (m.get("ts") or "") > last_ts
|
|
and m.get("id") != last_id]
|
|
|
|
|
|
def author_of(m):
|
|
frm = m.get("from") or {}
|
|
return frm.get("name") or "?"
|
|
|
|
|
|
def compose_digest(agent, new):
|
|
"""Concise, actionable digest for the prompting sidechat."""
|
|
lines = ["[main-loop] %d new in %s main chat:" % (len(new), agent)]
|
|
for m in new[:5]:
|
|
text = (m.get("text") or "").strip().replace("\n", " ")
|
|
q = " [?]" if "?" in text else ""
|
|
hot = " [!]" if (OPERATOR_RE.search(text) or "urgent" in text.lower()) else ""
|
|
lines.append("- %s: %s%s%s" % (author_of(m), text[:PREVIEW_MAX], q, hot))
|
|
if len(new) > 5:
|
|
lines.append("(+%d more)" % (len(new) - 5))
|
|
lines.append("Check main chat via box when you can.")
|
|
digest = "\n".join(lines)
|
|
return digest[:DIGEST_MAX]
|
|
|
|
|
|
def send_prompt(sender, agent, sidechat, digest):
|
|
"""Post the digest to the agent's prompting sidechat via dm.py."""
|
|
cmd = [sys.executable, DM_PY, "send",
|
|
"--agent", sender, "--to", agent, "--target", sidechat, digest]
|
|
try:
|
|
r = subprocess.run(cmd, capture_output=True, text=True, timeout=SEND_TIMEOUT)
|
|
except subprocess.TimeoutExpired:
|
|
return False, "dm.py send timed out after %ss" % SEND_TIMEOUT
|
|
except OSError as e:
|
|
return False, "could not exec dm.py: %s" % e
|
|
if r.returncode != 0:
|
|
tail = ((r.stderr or "") + (r.stdout or "")).strip()[-300:]
|
|
return False, "dm.py send failed rc=%d: %s" % (r.returncode, tail)
|
|
return True, "sent"
|
|
|
|
|
|
def do_check(only_agent=None):
|
|
st = load_state()
|
|
cfg = get_config(st)
|
|
agents = [only_agent] if only_agent else cfg["agents"]
|
|
if only_agent and only_agent not in DEFAULT_AGENTS:
|
|
return {"ok": False, "error": "unknown agent: %s" % only_agent}
|
|
|
|
agents_state = st.get("agents") or {}
|
|
enabled = cfg.get("enabled") or {}
|
|
results = {}
|
|
prompted = 0
|
|
errors = 0
|
|
|
|
for agent in agents:
|
|
if not enabled.get(agent, True):
|
|
results[agent] = {"ok": True, "new": 0, "disabled": True}
|
|
continue
|
|
wm = agents_state.get(agent) or {}
|
|
ok, payload = read_main_chat(agent, cfg["read_limit"])
|
|
if not ok:
|
|
log("%s: READ FAILED: %s" % (agent, payload))
|
|
results[agent] = {"ok": False, "error": payload}
|
|
errors += 1
|
|
continue
|
|
messages = payload
|
|
new = new_messages(messages, wm)
|
|
newest = messages[-1] if messages else None
|
|
if not new:
|
|
# Silent: advance watermark to newest seen, no sidechat post.
|
|
if newest:
|
|
agents_state[agent] = {"last_id": newest.get("id"),
|
|
"last_ts": newest.get("ts")}
|
|
results[agent] = {"ok": True, "new": 0}
|
|
continue
|
|
digest = compose_digest(agent, new)
|
|
sidechat = cfg["prompt_sidechat"].get(agent)
|
|
if not sidechat:
|
|
log("%s: no prompting sidechat configured, skipping" % agent)
|
|
results[agent] = {"ok": False, "error": "no prompting sidechat"}
|
|
errors += 1
|
|
continue
|
|
sent, detail = send_prompt(cfg["sender"], agent, sidechat, digest)
|
|
if sent:
|
|
agents_state[agent] = {"last_id": new[-1].get("id"),
|
|
"last_ts": new[-1].get("ts")}
|
|
prompted += 1
|
|
log("%s: prompted %s with %d new" % (agent, sidechat, len(new)))
|
|
results[agent] = {"ok": True, "new": len(new), "prompted": sidechat}
|
|
else:
|
|
# Keep watermark: next iteration retries the same digest.
|
|
log("%s: PROMPT SEND FAILED: %s" % (agent, detail))
|
|
results[agent] = {"ok": False, "error": detail, "new": len(new)}
|
|
errors += 1
|
|
|
|
# Save under the state lock with a fresh reload: an enable/disable may
|
|
# have landed during the slow chat reads; preserve its config changes
|
|
# and only update the keys this run owns (watermarks, last_run/result).
|
|
with state_locked():
|
|
fresh = load_state()
|
|
fresh["agents"] = agents_state
|
|
fresh["last_run"] = utcnow()
|
|
fresh["last_result"] = {"prompted": prompted, "errors": errors,
|
|
"agents": results}
|
|
# Seed defaults on first run; otherwise preserve operator-edited config.
|
|
if "config" not in fresh:
|
|
fresh["config"] = {"agents": cfg["agents"],
|
|
"prompt_sidechat": cfg["prompt_sidechat"],
|
|
"sender": cfg["sender"],
|
|
"read_limit": cfg["read_limit"],
|
|
"enabled": dict(cfg.get("enabled") or
|
|
{a: True for a in cfg["agents"]})}
|
|
save_state(fresh)
|
|
return {"ok": errors == 0, "prompted": prompted, "errors": errors,
|
|
"new_total": sum(r.get("new", 0) for r in results.values()),
|
|
"agents": results}
|
|
|
|
|
|
def _set_enabled(agent, value):
|
|
"""Enable/disable the loop for one agent (or all if agent is None)."""
|
|
with state_locked():
|
|
st = load_state()
|
|
cfg = st.get("config") or {}
|
|
agents = cfg.get("agents") or list(DEFAULT_AGENTS)
|
|
agents = [a for a in agents if a in DEFAULT_AGENTS] or list(DEFAULT_AGENTS)
|
|
if agent and agent not in DEFAULT_AGENTS:
|
|
return {"ok": False, "error": "unknown agent: %s" % agent}
|
|
enabled = dict(cfg.get("enabled") or {})
|
|
targets = [agent] if agent else agents
|
|
for a in targets:
|
|
enabled[a] = value
|
|
cfg["enabled"] = enabled
|
|
st["config"] = cfg
|
|
save_state(st)
|
|
return {"ok": True, "enabled": {a: enabled.get(a, True) for a in targets}}
|
|
|
|
|
|
def do_status():
|
|
st = load_state()
|
|
cfg = get_config(st)
|
|
agents_state = st.get("agents") or {}
|
|
return {"ok": True,
|
|
"config": cfg,
|
|
"enabled": cfg.get("enabled") or {},
|
|
"watermark": agents_state,
|
|
"last_run": st.get("last_run"),
|
|
"last_result": st.get("last_result")}
|
|
|
|
|
|
def main(argv):
|
|
action = argv[1] if len(argv) > 1 else "check"
|
|
only_agent = None
|
|
if "--agent" in argv:
|
|
i = argv.index("--agent")
|
|
if i + 1 < len(argv):
|
|
only_agent = argv[i + 1]
|
|
|
|
if action == "status":
|
|
print(json.dumps(do_status()))
|
|
return 0
|
|
|
|
if action == "enable":
|
|
print(json.dumps(_set_enabled(only_agent, True)))
|
|
return 0
|
|
|
|
if action == "disable":
|
|
print(json.dumps(_set_enabled(only_agent, False)))
|
|
return 0
|
|
|
|
if action != "check":
|
|
print(json.dumps({"ok": False, "error": "usage: check|status|enable|disable [--agent X]"}))
|
|
return 2
|
|
|
|
# Overlap guard: timer may fire while a slow run is still going.
|
|
try:
|
|
lockfh = open(LOCK_FILE, "w")
|
|
fcntl.flock(lockfh, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except (OSError, IOError):
|
|
print(json.dumps({"ok": False, "skipped": "already running"}))
|
|
return 0
|
|
|
|
try:
|
|
result = do_check(only_agent)
|
|
except Exception as e: # never crash the timer
|
|
log("UNEXPECTED ERROR: %r" % e)
|
|
print(json.dumps({"ok": False, "error": "unexpected: %s" % e}))
|
|
return 2
|
|
print(json.dumps(result))
|
|
if not result.get("ok"):
|
|
return 2
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv))
|