"""Decisions behind the multi-agent-mux-update skill (Issue #72). update.sh keeps the orchestration in shell (phase 1: this tree's code) and asks this module what needs parsing. Standard library only, Python 3.9. python -m lib_py.update precheck --workspace W --herdr H What refuses the update (comment 921, decisions 1 and 7): the running rows of the registry (MAM_STATE_JSON, from lib.sh `load_state_json strict`), every live herdr agent whose cwd is inside W on any server `herdr session list --json` reports running (read only: the real herdr H, never the shim, so no server is started), and the open jobs. The agent that called the update - by (server, pane): pane_lock.caller_server and HERDR_PANE_ID - is not counted, nor its row and its one running job. Prints one line per blocker with how to clear it; rc 0 nothing blocks, 1 something does (a herdr or registry that cannot be read included). python -m lib_py.update monitor --workspace W JSON {"pids": [...], "lock": "free"|"held"}: the monitor daemons daemon.find_running recognises, and whether .mam/monitor.lock is held. python -m lib_py.update target-workspaces --workspace W --agent-session S --server V --herdr H The workspaces other than W whose update admission a delegate submit to `herdr:` must take (decision 8), one per line: every installed workspace at or above the target's cwd (its row's pane.cwd; with no row the live agent's on server V) and its row's control_plane workspace_root. Nothing for a non-herdr target. rc 1: herdr cannot answer for a rowless target. python -m lib_py.update target-workspaces --workspace W --cwd C The same for an agent create or resume starts in C: every installed workspace at or above C other than W. python -m lib_py.update backup-registry --workspace W --dest D Copies the registry into D under the registry's own write lock (agent-sessions.db.lock, as atomic_yaml takes it): the DB through the SQLite backup API (commits still in -wal included), the YAML and its .bak as files. Prints the names it copied. """ from __future__ import annotations import argparse import fcntl import json import os import shlex import shutil import sqlite3 import subprocess import sys from typing import Dict, Iterable, List, Mapping, Optional, Tuple from lib_py.agents.sanitize import sanitize_herdr_agent_name from lib_py.delegation import daemon from lib_py.delegation.engine import TERMINAL_STATUSES from lib_py.pane_lock import caller_server REGISTRY_FILES = ("agent-sessions.yaml", "agent-sessions.yaml.bak", "agent-sessions.db") HERDR_TIMEOUT_S = 20 def _real(p: str) -> str: return os.path.realpath(p) def _inside(path: str, root: str) -> bool: if not path: return False p, r = _real(path), _real(root).rstrip(os.sep) return p == r or p.startswith(r + os.sep) def row_server(row: dict) -> str: return row.get("herdr_session") or row.get("herdr_server") or "default" # --------------------------------------------------------------------------- herdr class HerdrError(Exception): pass def _herdr_json(herdr: str, args: List[str], env: Mapping[str, str]) -> dict: # A pane's own scope ids would narrow the answer to the caller's workspace. cenv = {k: v for k, v in env.items() if k not in ("HERDR_WORKSPACE_ID", "HERDR_PANE_ID", "HERDR_TAB_ID")} try: res = subprocess.run([herdr, *args], env=cenv, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=HERDR_TIMEOUT_S) except (OSError, subprocess.SubprocessError) as exc: raise HerdrError("`herdr %s` failed: %s" % (" ".join(args), exc)) if res.returncode != 0: raise HerdrError("`herdr %s` exited %d: %s" % (" ".join(args), res.returncode, (res.stderr or "").strip()[:200])) try: out = res.stdout return json.loads(out[out.index("{"):]) except ValueError: raise HerdrError("`herdr %s` printed no JSON" % " ".join(args)) def running_servers(herdr: str, env: Mapping[str, str]) -> List[str]: d = _herdr_json(herdr, ["session", "list", "--json"], env) sessions = d.get("sessions") if not isinstance(sessions, list): raise HerdrError("`herdr session list --json` has no sessions list") return [s["name"] for s in sessions if isinstance(s, dict) and s.get("running") and s.get("name")] def server_agents(herdr: str, server: str, env: Mapping[str, str]) -> List[dict]: d = _herdr_json(herdr, ["--session", server, "agent", "list"], env) agents = (d.get("result") or {}).get("agents") if not isinstance(agents, list): raise HerdrError("`herdr --session %s agent list` has no agents list" % server) return [a for a in agents if isinstance(a, dict)] # --------------------------------------------------------------------------- jobs def _job_records(d: str) -> Iterable[Tuple[str, Optional[dict]]]: """(path, record) per *.json; (path, None) when it cannot be read, and (d, None) when d itself cannot be listed (not an empty registry).""" try: names = sorted(n for n in os.listdir(d) if n.endswith(".json") and not n.startswith(".")) except OSError: yield d, None return for p in (os.path.join(d, n) for n in names): try: with open(p, encoding="utf-8") as fh: rec = json.load(fh) except (OSError, ValueError): yield p, None continue yield p, rec if isinstance(rec, dict) else None def _is_jobs_dir(d: str) -> bool: """d is a directory, or cannot be looked at (an unreadable parent): only an absent d (or a file) is skipped.""" try: os.stat(d) except (FileNotFoundError, NotADirectoryError): return False except OSError: return True return os.path.isdir(d) def _is_open(rec: dict) -> bool: return rec.get("status") not in TERMINAL_STATUSES and rec.get("dry_run") is not True def open_jobs(workspace: str, state: dict, env: Mapping[str, str]) -> List[dict]: """Every open record (not completed/error/cancelled, not dry_run) in this workspace's .mam/jobs and MAM_JOBS_DIR; in another row's control_plane.mam_jobs_dir only those addressed to a row of this registry. A record, or a directory, that cannot be read is open (path only).""" own, seen = [], set() for c in (os.path.join(workspace, ".mam", "jobs"), env.get("MAM_JOBS_DIR") or ""): if c and _is_jobs_dir(c) and _real(c) not in seen: seen.add(_real(c)) own.append(c) rows = state.get("herdr_sessions") or [] names = {"herdr:%s" % s.get("name") for s in rows if s.get("name")} other = [] for s in rows: c = (s.get("control_plane") or {}).get("mam_jobs_dir") or "" if c and _is_jobs_dir(c) and _real(c) not in seen: seen.add(_real(c)) other.append(c) out = [] for dirs, addressed_only in ((own, False), (other, True)): for d in dirs: for path, rec in _job_records(d): if rec is None: out.append({"path": path, "unreadable": True, "dir": path == d}) elif _is_open(rec) and (not addressed_only or rec.get("agent_session") in names): out.append({"path": path, "job_id": rec.get("job_id") or "", "agent_session": rec.get("agent_session") or "", "status": rec.get("status"), "registry_dir": d}) return out # --------------------------------------------------------------------------- precheck def precheck(workspace: str, state: dict, herdr: str, env: Mapping[str, str]) -> List[str]: """The lines that refuse the update; [] when nothing does.""" blockers: List[str] = [] stop = ("bash .agents/skills/multi-agent-mux-stop/scripts/stop_session.sh --session %s" " (resume it after the update with multi-agent-mux-resume)") # A rowless agent: stop finds no row to take its server from (#71). stop_rowless = ("bash .agents/skills/multi-agent-mux-stop/scripts/stop_session.sh --session %s" " --herdr-session %s") # The caller (decision 7): its server and pane, both or nothing. me_server, me_pane = caller_server(env), env.get("HERDR_PANE_ID") or "" me_name: Optional[str] = None me_found = False agents_by_server: Dict[str, List[dict]] = {} try: for server in running_servers(herdr, env): agents_by_server[server] = server_agents(herdr, server, env) except HerdrError as exc: blockers.append("herdr: cannot list the live agents (%s); fix herdr and try again" % exc) if me_server and me_pane: for a in agents_by_server.get(me_server, []): if a.get("pane_id") == me_pane: me_found = True me_name = a.get("name") or None # a nameless pane is excluded by pane only break rows = state.get("herdr_sessions") or [] me_row = None if me_name: for s in rows: if (s.get("status") == "running" and row_server(s) == me_server and s.get("name") and sanitize_herdr_agent_name(s["name"]) == me_name): me_row = s break for s in rows: if s.get("status") == "running" and s is not me_row: blockers.append("session %s is running (server %s); stop it: %s" % (s.get("name"), row_server(s), stop % shlex.quote(s.get("name") or ""))) row_keys = {(row_server(s), sanitize_herdr_agent_name(s["name"])) for s in rows if s.get("status") == "running" and s.get("name")} for server, agents in agents_by_server.items(): for a in agents: if me_found and server == me_server and a.get("pane_id") == me_pane: continue if not (_inside(a.get("cwd") or "", workspace) or _inside(a.get("foreground_cwd") or "", workspace)): continue name = a.get("name") or "" if name and (server, name) in row_keys: continue # already listed as its running row blockers.append("herdr agent %s (pane %s, server %s) runs in %s; stop it: %s" % (name or "", a.get("pane_id"), server, a.get("cwd"), stop_rowless % (shlex.quote(name), shlex.quote(server)) if name else "herdr --session %s pane close %s" % (shlex.quote(server), shlex.quote(a.get("pane_id") or "")))) jobs = open_jobs(workspace, state, env) if me_row is not None: mine = [j for j in jobs if not j.get("unreadable") and j.get("agent_session") == "herdr:%s" % me_row.get("name") and j.get("status") == "running"] if len(mine) == 1: jobs = [j for j in jobs if j is not mine[0]] for j in jobs: if j.get("dir"): blockers.append("jobs directory %s cannot be listed; fix its permissions, then try again" % j["path"]) elif j.get("unreadable"): blockers.append("job record %s cannot be read; check it, then move or delete it by hand" % j["path"]) else: blockers.append("job %s (%s, %s) is open; wait for it to end or close it: python " ".agents/skills/multi-agent-mux-delegate-job/scripts/registry.py" " --registry-dir %s close --job %s --request cancelled" % (j["job_id"], j["agent_session"] or "-", j["status"], j["registry_dir"], j["job_id"])) if blockers and not me_found: blockers.append("note: no herdr agent was found at the caller's server %r and pane %r, so" " nothing was excluded as the calling agent" % (me_server, me_pane)) return blockers # --------------------------------------------------------------------------- delegate target def _installed(d: str) -> bool: """update.sh's own test of an installed workspace.""" return os.path.isfile(os.path.join(d, ".mam_deploy", "remove.sh")) or os.path.isfile(os.path.join(d, "remove.sh")) def target_workspaces(workspace: str, state: dict, agent_session: str, server: str, herdr: str, env: Mapping[str, str]) -> List[str]: """Every installed workspace at or above the target's cwd - an update of any of them counts that agent (decision 1) - and its row's control_plane.workspace_root; the caller's own workspace left out. With no row the cwd is the live agent's on (the one delivery uses), read with the real herdr. No herdr, a server that is not running, or no such agent gives nothing (delivery then fails on its own); a herdr that cannot answer raises HerdrError.""" if not agent_session.startswith("herdr:"): return [] name = agent_session[len("herdr:"):] row = next((s for s in state.get("herdr_sessions") or [] if isinstance(s, dict) and s.get("name") == name), None) cwds, out = [], [] if row is not None: cwds.append((row.get("pane") or {}).get("cwd") or "") root = (row.get("control_plane") or {}).get("workspace_root") or "" if root: out.append(_real(root)) elif herdr and server in running_servers(herdr, env): for a in server_agents(herdr, server, env): if a.get("name") in (name, sanitize_herdr_agent_name(name)): cwds += [a.get("cwd") or "", a.get("foreground_cwd") or ""] break return _others(workspace, out + installed_above(cwds)) def installed_above(cwds: Iterable[str]) -> List[str]: """Every installed workspace at or above each cwd.""" out = [] for cwd in cwds: d = _real(cwd) if cwd else "" while d: if _installed(d): out.append(d) if os.path.dirname(d) == d: break d = os.path.dirname(d) return out def _others(workspace: str, dirs: List[str]) -> List[str]: me = _real(workspace) return [d for i, d in enumerate(dirs) if d != me and d not in dirs[:i]] # --------------------------------------------------------------------------- monitor def monitor_state(workspace: str) -> dict: pids = daemon.find_running(daemon.monitor_script_path(workspace)) lock = os.path.join(workspace, ".mam", "monitor.lock") held = False if os.path.exists(lock): with open(lock, "a") as fh: try: fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB) fcntl.flock(fh, fcntl.LOCK_UN) except BlockingIOError: held = True return {"pids": pids, "lock": "held" if held else "free"} # --------------------------------------------------------------------------- registry backup def backup_registry(workspace: str, dest: str) -> List[str]: mam = os.path.join(workspace, ".mam") db = os.path.join(mam, "agent-sessions.db") os.makedirs(dest, exist_ok=True) copied = [] with open(db + ".lock", "w") as lk: fcntl.flock(lk, fcntl.LOCK_EX) try: if os.path.exists(db): src = sqlite3.connect(db, timeout=60.0) dst = sqlite3.connect(os.path.join(dest, "agent-sessions.db")) try: src.backup(dst) finally: dst.close() src.close() copied.append("agent-sessions.db") for name in REGISTRY_FILES: p = os.path.join(mam, name) if name != "agent-sessions.db" and os.path.exists(p): shutil.copy2(p, os.path.join(dest, name)) copied.append(name) finally: fcntl.flock(lk, fcntl.LOCK_UN) return copied def _state() -> dict: return json.loads(os.environ.get("MAM_STATE_JSON") or "{}") or {} def main(argv: Optional[List[str]] = None) -> int: p = argparse.ArgumentParser(prog="lib_py.update") sub = p.add_subparsers(dest="cmd", required=True) s = sub.add_parser("precheck") s.add_argument("--workspace", required=True) s.add_argument("--herdr", required=True) s = sub.add_parser("monitor") s.add_argument("--workspace", required=True) s = sub.add_parser("target-workspaces") s.add_argument("--workspace", required=True) s.add_argument("--agent-session", default="") s.add_argument("--cwd", default="") s.add_argument("--server", default="default") s.add_argument("--herdr", default="") s = sub.add_parser("backup-registry") s.add_argument("--workspace", required=True) s.add_argument("--dest", required=True) a = p.parse_args(argv) ws = _real(a.workspace) if a.cmd == "precheck": try: state = _state() except ValueError as exc: print("registry: cannot parse MAM_STATE_JSON (%s)" % exc) return 1 lines = precheck(ws, state, a.herdr, os.environ) for line in lines: print(line) return 1 if lines else 0 if a.cmd == "target-workspaces" and a.cwd: for root in _others(ws, installed_above([a.cwd])): print(root) return 0 if a.cmd == "target-workspaces": try: state = _state() except ValueError: state = {} try: roots = target_workspaces(ws, state, a.agent_session, a.server, a.herdr, os.environ) except HerdrError as exc: print("cannot find %s live: %s" % (a.agent_session, exc), file=sys.stderr) return 1 for root in roots: print(root) return 0 if a.cmd == "monitor": print(json.dumps(monitor_state(ws))) return 0 for name in backup_registry(ws, a.dest): print(name) return 0 if __name__ == "__main__": sys.exit(main())