Compare commits

...
6 Commits
16 changed files with 3019 additions and 9 deletions

No files matched your search

+4 -1
View File
@@ -197,11 +197,14 @@ class AgyAgentAdapter(BaseAgentAdapter):
# the DB already gone, and those orphans must still be reachable.
return any(os.path.exists(p) for p in self._artifact_paths(uuid, ctx))
def in_use_paths(self, uuid: str, ctx: DiscoveryContext) -> list:
return self._artifact_paths(uuid, ctx)
def conversation_in_use(self, uuid: str, ctx: DiscoveryContext) -> bool:
# The holder may keep the presence lock (or a sidecar) open rather
# than the DB, and may name the uuid nowhere; the lsof leg of the
# check must see every path of the conversation.
return conversation_in_use(uuid, self._artifact_paths(uuid, ctx))
return conversation_in_use(uuid, self.in_use_paths(uuid, ctx))
def purge_artifacts(self, uuid: str, ctx: DiscoveryContext) -> list:
# Issue #10: a purge must leave nothing resumable behind. Besides the
+23 -2
View File
@@ -1,6 +1,6 @@
import os, json, glob, sqlite3, subprocess
import os, json, glob, shutil, sqlite3, subprocess
from typing import Optional, Any, List
from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext, _ro_connect, scan_names, db_rows_naming
from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext, _plausible_id, _ro_connect, scan_names, db_rows_naming
from lib_py.verify_session import workspace_key
class CodexAgentAdapter(BaseAgentAdapter):
@@ -131,6 +131,22 @@ class CodexAgentAdapter(BaseAgentAdapter):
except Exception:
return False
def _purge_tui_thread_capability(self, uuid: str, ctx: DiscoveryContext, purged: List[str]) -> None:
if not _plausible_id(uuid):
return
cap = os.path.join(self._codex_home(ctx), "tui-thread-reference-capabilities", uuid)
if not os.path.lexists(cap):
return
try:
if os.path.isdir(cap) and not os.path.islink(cap):
shutil.rmtree(cap)
else:
os.unlink(cap)
if not os.path.lexists(cap):
purged.append(cap)
except Exception:
pass
def purge_artifacts(self, uuid: str, ctx: DiscoveryContext) -> List[str]:
purged = []
path = self._find_rollout(uuid, ctx)
@@ -140,12 +156,16 @@ class CodexAgentAdapter(BaseAgentAdapter):
if res.returncode == 0:
if path:
purged.append(path)
self._purge_tui_thread_capability(uuid, ctx, purged)
return purged
except Exception:
pass
if path and os.path.exists(path):
try:
os.remove(path)
purged.append(path)
except Exception:
pass
db = self._state_db(ctx)
if db and os.path.exists(db):
try:
@@ -156,6 +176,7 @@ class CodexAgentAdapter(BaseAgentAdapter):
purged.append(f"state db row for thread: {uuid}")
except Exception:
pass
self._purge_tui_thread_capability(uuid, ctx, purged)
return purged
def residual_artifacts(self, uuid: str, ctx: DiscoveryContext) -> List[str]:
@@ -13,6 +13,11 @@ _LEGACY_TABLES = (("message", "session_id"), ("messages", "session_id"), ("part"
# (binary, HOME, XDG_DATA_HOME) -> the db `debug paths` printed for them.
_DB_PATHS = {}
def _is_shared_service(cmd: str) -> bool:
"""The per-user opencode background service (`serve --service`)."""
return "serve --service" in (cmd or "")
class OpenCodeAgentAdapter(BaseAgentAdapter):
@property
def name(self) -> str:
@@ -250,6 +255,13 @@ class OpenCodeAgentAdapter(BaseAgentAdapter):
found += rows if i == uuid else [f"{r} [child session {i}]" for r in rows]
return found
def in_use_paths(self, uuid: str, ctx: DiscoveryContext) -> list:
db = self._resolve_db(ctx)
return [db] if db else []
def _holder_counts(self, cmd: str) -> bool:
return not _is_shared_service(cmd)
def conversation_in_use(self, uuid: str, ctx: DiscoveryContext) -> bool:
"""A process naming the session, or an opencode process holding the
shared database. The per-user background service (`serve --service`)
@@ -270,13 +282,14 @@ class OpenCodeAgentAdapter(BaseAgentAdapter):
timeout=10).stdout
except Exception:
return True
if cmd.strip() and "serve --service" not in cmd:
if cmd.strip() and not _is_shared_service(cmd):
return True
return False
def in_use_hints(self, uuid: str, ctx: DiscoveryContext) -> list:
db = self._resolve_db(ctx)
return super().in_use_hints(uuid, ctx) + [
f"another opencode session holding {self._resolve_db(ctx)}: lsof {self._resolve_db(ctx)}"]
f"another opencode session holding {db}: lsof {db}"]
# opencode 2.x: the default agent is build (`--agent` is gone from the
# TUI), and --standalone keeps the server a child of the pane's TUI
+42 -3
View File
@@ -1,7 +1,7 @@
# base.py — BaseAgentAdapter abstract base class, SpawnSpec, and DiscoveryContext
import os, json, sqlite3, shutil, subprocess, tempfile
from typing import Optional, Dict, Any, List
from typing import Optional, Dict, Any, List, Tuple
from urllib.parse import quote
from lib_py.paths import resolve_home
from lib_py.verify_session import verify_session_uuid, workspace_key
@@ -86,6 +86,34 @@ def conversation_in_use(uuid: str, paths=()) -> bool:
return False
def file_holders(paths=()) -> List[Tuple[int, str]]:
"""(pid, command_line) for each process holding one of `paths` open via lsof (#78).
Informational only: returns [] when lsof is absent or fails."""
existing = [p for p in paths if p and os.path.exists(p)]
if not existing or not shutil.which("lsof"):
return []
try:
r = subprocess.run(["lsof", "-t"] + existing, capture_output=True, text=True, timeout=10)
except Exception:
return []
pids: List[int] = []
for tok in (r.stdout or "").split():
if tok.isdigit():
pid = int(tok)
if pid not in pids:
pids.append(pid)
out: List[Tuple[int, str]] = []
for pid in pids:
try:
cmd = subprocess.run(["ps", "-o", "command=", "-p", str(pid)],
capture_output=True, text=True, timeout=10).stdout.strip()
except Exception:
continue
if cmd:
out.append((pid, cmd))
return out
def _plausible_id(uuid: str) -> bool:
"""An id safe to build a path from and to search for: not empty, not
made of dots alone ("." names the parent dir itself), no "/", ".." or "*"."""
@@ -354,14 +382,25 @@ class BaseAgentAdapter:
path = self.artifact_path(uuid, ctx)
return uuid in path and os.path.exists(path)
def in_use_paths(self, uuid: str, ctx: DiscoveryContext) -> list:
"""On-disk paths checked with lsof for conversation_in_use and in_use_hints (#78)."""
return [self.artifact_path(uuid, ctx)]
def _holder_counts(self, cmd: str) -> bool:
return True
def conversation_in_use(self, uuid: str, ctx: DiscoveryContext) -> bool:
"""True while a live process still references the conversation."""
return conversation_in_use(uuid, [self.artifact_path(uuid, ctx)])
return conversation_in_use(uuid, self.in_use_paths(uuid, ctx))
def in_use_hints(self, uuid: str, ctx: DiscoveryContext) -> list:
"""Where to look for what holds the conversation, shown when a purge
is refused because it is in use."""
return [f"any process with it on its command line (an editor, a grep, ...): ps -axww | grep {uuid}"]
hints = [f"any process with it on its command line (an editor, a grep, ...): ps -axww | grep {uuid}"]
for pid, cmd in file_holders(self.in_use_paths(uuid, ctx)):
if self._holder_counts(cmd):
hints.append(f"held open by pid {pid}: {cmd} (not stopped by MAM; end it yourself if it is safe)")
return hints
def spawn_spec(self, binary: str, session_uuid: str = "", use_wrapper: bool = False) -> str:
raise NotImplementedError
+812
View File
@@ -0,0 +1,812 @@
"""Decisions behind the multi-agent-mux-reset skill (Issue #78).
reset.sh keeps the shell sequence (arg parsing, terminal check, update
admission, loop lock, sequential stop/create invocations, exit codes) and
delegates target selection, prechecks, atomic spec-file persistence, 5-state
progress classification, and result-table formatting to this module.
Standard library only, Python 3.9 compatible.
"""
from __future__ import annotations
import argparse
import datetime
import json
import os
import shlex
import shutil
import sqlite3
import subprocess
import sys
import tempfile
from typing import Dict, List, Mapping, Optional, Tuple
from lib_py.agents.base import AGENT_SESSION_ENV
from lib_py.agents.registry import agent_of_row, get_adapter, own_key
from lib_py.update import open_jobs, row_server
SPEC_VERSION = 1
SPEC_REL = os.path.join(".mam", "reset-spec.json")
class SpecError(Exception):
pass
def _real(p: str) -> str:
return os.path.realpath(p) if p else ""
def spec_path(workspace: str) -> str:
return os.path.join(_real(workspace), SPEC_REL)
def _state(workspace: str = "") -> dict:
raw = os.environ.get("MAM_STATE_JSON")
if raw is not None:
return json.loads(raw or "{}") or {}
yaml_path = os.environ.get("AGENT_SESSIONS_YAML") or (
os.path.join(_real(workspace), ".mam", "agent-sessions.yaml") if workspace else ""
)
if not yaml_path:
return {}
db_path = os.path.splitext(yaml_path)[0] + ".db"
if os.path.exists(db_path):
tmp_dir = tempfile.mkdtemp(prefix="mam-reset-ro.")
try:
tmp_db = os.path.join(tmp_dir, "agent-sessions.db")
shutil.copyfile(db_path, tmp_db)
for suf in ("-wal", "-shm"):
if os.path.exists(db_path + suf):
shutil.copyfile(db_path + suf, tmp_db + suf)
conn = sqlite3.connect(tmp_db, timeout=60.0)
try:
d: dict = {}
row = conn.execute("SELECT data FROM state WHERE id=1").fetchone()
if row:
d = json.loads(row[0])
cursor = conn.execute("SELECT data FROM sessions")
d["herdr_sessions"] = [json.loads(r[0]) for r in cursor.fetchall()]
return d
finally:
conn.close()
except Exception as exc:
raise ValueError("the registry %s cannot be read (%s); fix it first" % (yaml_path, exc))
finally:
shutil.rmtree(tmp_dir, ignore_errors=True)
if os.path.exists(yaml_path):
try:
import yaml
with open(yaml_path, encoding="utf-8") as fh:
data = yaml.safe_load(fh)
if data is not None and not isinstance(data, dict):
raise ValueError("top-level YAML is not a mapping")
return data or {}
except Exception as exc:
raise ValueError("the registry %s cannot be read (%s); fix it first" % (yaml_path, exc))
return {}
# --------------------------------------------------------------------------- D2: target selection
def select_targets(workspace: str, state: dict) -> dict:
"""Select running control-root sessions and classify non-targets.
Returns {"targets": [...], "skipped": [{"name", "reason"}],
"orchestrators": [...], "errors": [...]}.
"""
ws_real = _real(workspace)
targets: List[dict] = []
skipped: List[dict] = []
errors: List[str] = []
for row in state.get("herdr_sessions") or []:
if not isinstance(row, dict):
continue
name = str(row.get("name") or "")
status = str(row.get("status") or "")
if status != "running":
skipped.append({"name": name, "reason": "status %s" % (status or "unknown")})
continue
cwd = str((row.get("pane") or {}).get("cwd") or "")
cwd_real = _real(cwd) if cwd else ""
if cwd_real != ws_real:
skipped.append({
"name": name,
"reason": "pane.cwd %s is not the control root (a worktree creator?)" % (cwd or "<empty>"),
})
continue
agent = agent_of_row(row)
role = str(row.get("role") or "").strip()
if not agent:
errors.append(
"running session %s has no recognisable agent type (agent=%r, pane.cmd=%r); "
"fix its registry row first"
% (name, row.get("agent"), (row.get("pane") or {}).get("cmd"))
)
continue
if not role:
errors.append(
"running session %s (%s) has an empty role; fix its registry row first"
% (name, agent)
)
continue
key = own_key(agent)
raw_id = row.get(key)
old_id = str(raw_id).strip() if isinstance(raw_id, str) and raw_id.strip() else None
targets.append({
"name": name,
"agent": agent,
"role": role,
"herdr_server": str(row_server(row)),
"herdr_workspace": str(row.get("herdr_workspace") or ""),
"old_id": old_id,
"old_created_at": str(row.get("herdr_session_created_at") or ""),
})
orchestrators = [str(u) for u in (state.get("orchestrator_uuids") or []) if u]
return {
"targets": targets,
"skipped": skipped,
"orchestrators": orchestrators,
"errors": errors,
}
# --------------------------------------------------------------------------- D5: spec file
def write_spec(workspace: str, agents: List[dict]) -> str:
"""Atomically write .mam/reset-spec.json (mkstemp -> fsync -> replace -> dir fsync)."""
ws_real = _real(workspace)
mam_dir = os.path.join(ws_real, ".mam")
os.makedirs(mam_dir, exist_ok=True)
path = os.path.join(mam_dir, "reset-spec.json")
now_iso = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
doc = {
"version": SPEC_VERSION,
"workspace": ws_real,
"written_at": now_iso,
"agents": agents,
}
payload = (json.dumps(doc, indent=2) + "\n").encode("utf-8")
fd, tmp_path = tempfile.mkstemp(prefix=".reset-spec.", suffix=".tmp", dir=mam_dir)
try:
with os.fdopen(fd, "wb") as fh:
fh.write(payload)
fh.flush()
os.fsync(fh.fileno())
os.replace(tmp_path, path)
except BaseException:
try:
os.unlink(tmp_path)
except OSError:
pass
raise
try:
dir_fd = os.open(mam_dir, os.O_RDONLY)
try:
os.fsync(dir_fd)
finally:
os.close(dir_fd)
except OSError:
pass
return path
def read_spec(workspace: str) -> dict:
"""Read and validate .mam/reset-spec.json; raise SpecError if invalid."""
ws_real = _real(workspace)
path = spec_path(ws_real)
try:
with open(path, encoding="utf-8") as fh:
doc = json.load(fh)
except (OSError, ValueError) as exc:
raise SpecError("spec file %s cannot be read (%s); check it, then remove it by hand" % (path, exc))
if not isinstance(doc, dict) or doc.get("version") != SPEC_VERSION:
raise SpecError(
"spec file %s has unsupported version %r; check it, then remove it by hand"
% (path, doc.get("version") if isinstance(doc, dict) else None)
)
spec_ws = _real(str(doc.get("workspace") or ""))
if spec_ws != ws_real:
raise SpecError(
"spec file %s is for workspace %r, not %r; check it, then remove it by hand"
% (path, doc.get("workspace"), ws_real)
)
agents = doc.get("agents")
if not isinstance(agents, list):
raise SpecError("spec file %s has no agents list; check it, then remove it by hand" % path)
for item in agents:
if not isinstance(item, dict):
raise SpecError("spec file %s has a non-object agent entry; check it, then remove it by hand" % path)
name = item.get("name")
agent = item.get("agent")
role = item.get("role")
if not (isinstance(name, str) and name and isinstance(agent, str) and agent and isinstance(role, str) and role):
raise SpecError("spec file %s has an incomplete agent entry; check it, then remove it by hand" % path)
adapter = get_adapter(agent)
if adapter is None or adapter.name != agent or not own_key(agent):
raise SpecError(
"spec file %s has unknown agent %r; check it, then remove it by hand" % (path, agent)
)
return doc
def record_old_id(workspace: str, state: dict, session_name: str) -> Optional[str]:
"""Persist a newly captured conversation id onto an entry with old_id=None.
Called after step 1 (normal stop) captures a POST_DISCOVERED session's id
onto the original instance's registry row, so a subsequent resume + rerun
cannot mistake the original conversation for a new one (review F1).
Reads the id from `state` (never from command-line args).
"""
ws_real = _real(workspace)
spec = read_spec(ws_real)
rows_by_name = {
str(s.get("name")): s
for s in (state.get("herdr_sessions") or [])
if isinstance(s, dict) and s.get("name")
}
row = rows_by_name.get(session_name)
if row is None:
return None
changed = False
recorded: Optional[str] = None
for entry in spec["agents"]:
if str(entry.get("name") or "") != session_name:
continue
raw_old = entry.get("old_id")
old_id = str(raw_old).strip() if isinstance(raw_old, str) and raw_old.strip() else None
if old_id:
return old_id
cur_created_at = str(row.get("herdr_session_created_at") or "")
old_created_at = str(entry.get("old_created_at") or "")
if cur_created_at != old_created_at:
return None
key = own_key(entry["agent"])
raw_cur = row.get(key)
cur_id = str(raw_cur).strip() if isinstance(raw_cur, str) and raw_cur.strip() else None
if cur_id:
entry["old_id"] = cur_id
recorded = cur_id
changed = True
if changed:
write_spec(ws_real, spec["agents"])
return recorded
def new_running_rows(workspace: str, state: dict, spec_agents: List[dict]) -> List[str]:
"""Names of running control-root rows not present in spec_agents."""
known = {str(a.get("name") or "") for a in spec_agents}
sel = select_targets(workspace, state)
return [t["name"] for t in sel["targets"] if t["name"] not in known]
# --------------------------------------------------------------------------- D6: progress classification
def classify_entry(entry: dict, row: Optional[dict]) -> Tuple[str, Optional[str], bool]:
"""Classify a spec entry against its current registry row (matched by name only).
Returns (r_state, new_id, same_instance):
1. row is None -> ("purged", None, False)
2. r.herdr_session_created_at == old_created_at -> ("pending", None, True)
3. not same_instance and r[id] == old_id (non-empty) -> ("pending", cur_id, False)
4. not same_instance and r[id] is empty -> ("created", None, False)
5. not same_instance and r[id] is non-empty and != old_id -> ("done", cur_id, False)
"""
if row is None:
return "purged", None, False
key = own_key(entry["agent"])
cur_created_at = str(row.get("herdr_session_created_at") or "")
old_created_at = str(entry.get("old_created_at") or "")
raw_cur = row.get(key)
cur_id = str(raw_cur).strip() if isinstance(raw_cur, str) and raw_cur.strip() else None
raw_old = entry.get("old_id")
old_id = str(raw_old).strip() if isinstance(raw_old, str) and raw_old.strip() else None
if cur_created_at == old_created_at:
return "pending", None, True
if cur_id and old_id and cur_id == old_id:
return "pending", cur_id, False
if not cur_id:
return "created", None, False
return "done", cur_id, False
# --------------------------------------------------------------------------- D4: precheck
def _ancestor_pids(start_pid: Optional[int] = None) -> set:
pid = os.getpid() if start_pid is None else start_pid
try:
res = subprocess.run(
["ps", "-eo", "pid=,ppid="],
stdin=subprocess.DEVNULL,
capture_output=True,
text=True,
timeout=5,
)
except (OSError, subprocess.SubprocessError):
return set()
if res.returncode != 0:
return set()
ppid_of: Dict[int, int] = {}
for line in (res.stdout or "").splitlines():
parts = line.split()
if len(parts) == 2 and parts[0].isdigit() and parts[1].isdigit():
ppid_of[int(parts[0])] = int(parts[1])
ancestors = set()
cur = pid
while cur and cur > 1 and cur not in ancestors:
ancestors.add(cur)
cur = ppid_of.get(cur, 0)
return ancestors
def _caller_blockers(
targets: List[dict],
rows_by_name: Mapping[str, dict],
env: Mapping[str, str],
ancestors: Optional[set] = None,
) -> List[str]:
env_ids = {
str(env.get(k) or "").strip()
for k in AGENT_SESSION_ENV
if str(env.get(k) or "").strip()
}
anc = _ancestor_pids() if ancestors is None else ancestors
out: List[str] = []
for t in targets:
name = t["name"]
row = rows_by_name.get(name)
r_state, _, _ = classify_entry(t, row)
if r_state != "pending":
continue
old_id = t.get("old_id")
key = own_key(t["agent"])
raw_cur = (row or {}).get(key)
cur_id = str(raw_cur).strip() if isinstance(raw_cur, str) and raw_cur.strip() else None
matched = False
if env_ids and ((old_id and old_id in env_ids) or (cur_id and cur_id in env_ids)):
matched = True
elif row is not None and anc:
pane_pid = (row.get("pane") or {}).get("pid")
child_pid = row.get("child_pid")
for p in (pane_pid, child_pid):
if isinstance(p, int) and p > 1 and p in anc:
matched = True
break
if matched:
out.append("you are %s, one of the targets; run the reset from the orchestrator" % name)
return out
def precheck(
workspace: str,
state: dict,
env: Mapping[str, str],
ancestors: Optional[set] = None,
) -> List[str]:
"""Return refusal lines for open jobs, unreadable spec, un-recreatable targets, or caller==target."""
ws_real = _real(workspace)
blockers: List[str] = []
spath = spec_path(ws_real)
targets: List[dict] = []
if os.path.lexists(spath):
try:
spec = read_spec(ws_real)
targets = spec["agents"]
except SpecError as exc:
blockers.append(str(exc))
else:
sel = select_targets(ws_real, state)
blockers.extend(sel["errors"])
targets = sel["targets"]
for j in open_jobs(ws_real, state, env):
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"],
)
)
rows_by_name = {
str(s.get("name")): s
for s in (state.get("herdr_sessions") or [])
if isinstance(s, dict) and s.get("name")
}
blockers.extend(_caller_blockers(targets, rows_by_name, env, ancestors=ancestors))
return blockers
# --------------------------------------------------------------------------- D1: entries shell lines
def format_entry_line(entry: dict, row: Optional[dict]) -> str:
r_state, _, _ = classify_entry(entry, row)
key = own_key(entry["agent"])
raw_cur = (row or {}).get(key)
row_id = str(raw_cur).strip() if isinstance(raw_cur, str) and raw_cur.strip() else ""
pairs = [
("R_STATE", r_state),
("R_NAME", str(entry.get("name") or "")),
("R_AGENT", str(entry.get("agent") or "")),
("R_ROLE", str(entry.get("role") or "")),
("R_SERVER", str(entry.get("herdr_server") or "default")),
("R_LABEL", str(entry.get("herdr_workspace") or "")),
("R_OLD_ID", str(entry.get("old_id") or "")),
("R_ROW_ID", row_id),
]
return " ".join("%s=%s" % (k, shlex.quote(v)) for k, v in pairs)
# --------------------------------------------------------------------------- D8: result table
def _parse_skipped(items: Optional[List[str]]) -> Dict[str, str]:
out: Dict[str, str] = {}
for raw in items or []:
for token in raw.split(","):
token = token.strip()
if not token:
continue
if ":" in token:
name, kind = token.split(":", 1)
out[name] = kind
else:
out[token] = "purge_refused"
return out
def _parse_failed(raw: Optional[str]) -> Optional[Tuple[str, str, str]]:
if not raw:
return None
parts = raw.split(":", 2)
if len(parts) == 3:
return parts[0], parts[1], parts[2]
if len(parts) == 2:
return parts[0], parts[1], "1"
return parts[0], "stop", "1"
def build_table(
agents: List[dict],
state: dict,
skipped_items: Optional[List[str]] = None,
failed_spec: Optional[str] = None,
) -> Tuple[str, int]:
"""Format the D8 result table and compute the D9 exit code (0, 1, 4, or 5)."""
rows_by_name = {
str(s.get("name")): s
for s in (state.get("herdr_sessions") or [])
if isinstance(s, dict) and s.get("name")
}
skipped_map = _parse_skipped(skipped_items)
failed = _parse_failed(failed_spec)
failed_name = failed[0] if failed else None
after_failed = False
headers = ("NAME", "AGENT", "OLD_ID", "NEW_ID", "RECORDED", "NEW≠OLD", "RESULT")
table_rows: List[Tuple[str, str, str, str, str, str, str]] = []
has_skip_or_incomplete = bool(skipped_map)
has_created = False
for entry in agents:
name = str(entry.get("name") or "")
agent = str(entry.get("agent") or "")
raw_old = entry.get("old_id")
old_id = str(raw_old).strip() if isinstance(raw_old, str) and raw_old.strip() else None
old_col = old_id if old_id else "-"
row = rows_by_name.get(name)
r_state, new_id, _ = classify_entry(entry, row)
new_col = new_id if new_id else "-"
recorded_col = "yes" if new_id else "no"
if not new_id:
diff_col = "-"
elif not old_id or new_id != old_id:
diff_col = "yes"
else:
diff_col = "no"
if failed and name == failed_name:
result_col = "failed at %s (rc %s)" % (failed[1], failed[2])
after_failed = True
has_skip_or_incomplete = True
elif after_failed and r_state in ("pending", "purged"):
result_col = "not reached"
has_skip_or_incomplete = True
elif name in skipped_map:
kind = skipped_map[name]
if kind == "no_id":
result_col = "skipped: no conversation id recorded"
else:
result_col = "skipped: purge refused"
has_skip_or_incomplete = True
elif r_state == "done":
result_col = "done"
elif r_state == "created":
result_col = "created; id not recorded yet"
has_created = True
elif r_state == "purged":
result_col = "purged"
has_skip_or_incomplete = True
else:
result_col = "pending"
has_skip_or_incomplete = True
table_rows.append((name, agent, old_col, new_col, recorded_col, diff_col, result_col))
widths = [len(h) for h in headers]
for r in table_rows:
for i in range(len(headers) - 1):
if len(r[i]) > widths[i]:
widths[i] = len(r[i])
def _fmt(cols: Tuple[str, ...]) -> str:
parts = [cols[i].ljust(widths[i]) for i in range(len(cols) - 1)]
parts.append(cols[-1])
return " ".join(parts)
lines = [_fmt(headers)]
for r in table_rows:
lines.append(_fmt(r))
if failed:
rc = 1
elif has_skip_or_incomplete:
rc = 4
elif has_created:
rc = 5
else:
rc = 0
return "\n".join(lines), rc
# --------------------------------------------------------------------------- D10: dry-run
def format_dry_run(
workspace: str,
state: dict,
env: Mapping[str, str],
extra_refusals: Optional[List[str]] = None,
) -> Tuple[str, int]:
ws_real = _real(workspace)
lines: List[str] = []
refusals: List[str] = list(extra_refusals or [])
refusals.extend(precheck(ws_real, state, env))
rows_by_name = {
str(s.get("name")): s
for s in (state.get("herdr_sessions") or [])
if isinstance(s, dict) and s.get("name")
}
sel = select_targets(ws_real, state)
spath = spec_path(ws_real)
action_map = {
"pending": "stop + purge + create",
"purged": "create",
"created": "none (waiting for id)",
"done": "none (done)",
}
if os.path.lexists(spath):
try:
spec = read_spec(ws_real)
except SpecError:
spec = None
if spec is not None:
lines.append("spec file: %s" % spath)
for e in spec["agents"]:
r_state, _, _ = classify_entry(e, rows_by_name.get(e["name"]))
lines.append(
"target: %s agent=%s role=%s herdr_server=%s herdr_workspace=%s "
"old_id=%s old_created_at=%s state=%s action=%s"
% (
e["name"],
e["agent"],
e["role"],
e.get("herdr_server") or "default",
e.get("herdr_workspace") or "",
e.get("old_id") or "-",
e.get("old_created_at") or "-",
r_state,
action_map.get(r_state, r_state),
)
)
extra_running = new_running_rows(ws_real, state, spec["agents"])
if extra_running:
lines.append(
"NOTE: running control-root session(s) not in reset-spec.json "
"(not targeted by this rerun): %s" % ", ".join(extra_running)
)
else:
if sel["targets"]:
for t in sel["targets"]:
lines.append(
"target: %s agent=%s role=%s herdr_server=%s herdr_workspace=%s "
"old_id=%s old_created_at=%s"
% (
t["name"],
t["agent"],
t["role"],
t["herdr_server"],
t["herdr_workspace"],
t["old_id"] or "-",
t["old_created_at"] or "-",
)
)
else:
lines.append("targets: (none)")
for sk in sel["skipped"]:
lines.append("skip: %s (%s)" % (sk["name"], sk["reason"]))
for u in sel["orchestrators"]:
lines.append("orchestrator %s (not a row; never reset)" % u)
if refusals:
lines.append("would refuse:")
for r in refusals:
lines.append(" - %s" % r)
return "\n".join(lines), 1
return "\n".join(lines), 0
# --------------------------------------------------------------------------- CLI
def main(argv: Optional[List[str]] = None) -> int:
p = argparse.ArgumentParser(prog="lib_py.reset")
sub = p.add_subparsers(dest="cmd", required=True)
s = sub.add_parser("plan")
s.add_argument("--workspace", required=True)
s = sub.add_parser("precheck")
s.add_argument("--workspace", required=True)
s = sub.add_parser("write-spec")
s.add_argument("--workspace", required=True)
s = sub.add_parser("entries")
s.add_argument("--workspace", required=True)
s.add_argument("--session", default="")
s.add_argument("--warn-new", action="store_true")
s = sub.add_parser("record-id")
s.add_argument("--workspace", required=True)
s.add_argument("--session", required=True)
s = sub.add_parser("table")
s.add_argument("--workspace", required=True)
s.add_argument("--skipped", action="append", default=[])
s.add_argument("--failed", default="")
s = sub.add_parser("dry-run")
s.add_argument("--workspace", required=True)
s.add_argument("--refusal", action="append", default=[])
a = p.parse_args(argv)
ws = _real(a.workspace)
if a.cmd == "write-spec":
try:
raw = json.load(sys.stdin)
except ValueError as exc:
print("ERROR: invalid targets JSON on stdin (%s)" % exc, file=sys.stderr)
return 1
agents = raw.get("targets") if isinstance(raw, dict) else raw
if not isinstance(agents, list):
print("ERROR: targets JSON must be a list", file=sys.stderr)
return 1
try:
write_spec(ws, agents)
except OSError as exc:
print("ERROR: cannot write %s (%s)" % (spec_path(ws), exc), file=sys.stderr)
return 1
return 0
try:
state = _state(ws)
except ValueError as exc:
if a.cmd == "dry-run":
refs = list(a.refusal or []) + [str(exc)]
out, rc = format_dry_run(ws, {}, os.environ, extra_refusals=refs)
print(out)
return rc
if a.cmd == "precheck":
print("registry: cannot parse MAM_STATE_JSON (%s)" % exc)
return 1
print("ERROR: cannot parse MAM_STATE_JSON (%s)" % exc, file=sys.stderr)
return 1
if a.cmd == "plan":
sel = select_targets(ws, state)
print(json.dumps(sel))
return 1 if sel["errors"] else 0
if a.cmd == "precheck":
lines = precheck(ws, state, os.environ)
for line in lines:
print(line)
return 1 if lines else 0
if a.cmd == "entries":
try:
spec = read_spec(ws)
except SpecError as exc:
print("ERROR: %s" % exc, file=sys.stderr)
return 1
if a.warn_new:
extra = new_running_rows(ws, state, spec["agents"])
if extra:
print(
"NOTE: running control-root session(s) not in %s (not targeted by this rerun): %s"
% (spec_path(ws), ", ".join(extra)),
file=sys.stderr,
)
rows_by_name = {
str(s.get("name")): s
for s in (state.get("herdr_sessions") or [])
if isinstance(s, dict) and s.get("name")
}
for e in spec["agents"]:
if a.session and e.get("name") != a.session:
continue
print(format_entry_line(e, rows_by_name.get(e["name"])))
return 0
if a.cmd == "record-id":
try:
record_old_id(ws, state, a.session)
except (SpecError, OSError) as exc:
print("ERROR: %s" % exc, file=sys.stderr)
return 1
return 0
if a.cmd == "table":
try:
spec = read_spec(ws)
except SpecError as exc:
print("ERROR: %s" % exc, file=sys.stderr)
return 1
out, rc = build_table(
spec["agents"],
state,
skipped_items=a.skipped,
failed_spec=a.failed or None,
)
print(out)
return rc
if a.cmd == "dry-run":
out, rc = format_dry_run(ws, state, os.environ, extra_refusals=a.refusal)
print(out)
return rc
return 2
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,48 @@
---
name: multi-agent-mux-reset
description: "Purge the control root's running resident agents and recreate each with the same name, agent, role, herdr server and workspace label, one at a time, with onboarding. Persists .mam/reset-spec.json before the first purge so an interrupted or skipped run resumes where it left off."
version: 4.13.0
author: godopu
license: MIT
platforms: [linux, macos]
environments: [terminal, herdr]
metadata:
hermes:
tags: [reset, purge, recreate, lifecycle, onboarding]
related_skills: [multi-agent-mux-stop, multi-agent-mux-create, multi-agent-mux-status]
---
# Multi-Agent Reset
```bash
bash .agents/skills/multi-agent-mux-reset/scripts/reset.sh --yes [--dry-run]
```
| Item | Behaviour |
|---|---|
| Targets | Every registry row with `status: running` and `realpath(pane.cwd) == realpath(WS)`. Worktree rows, `stopped`/`terminated` rows, and `orchestrator_uuids` are skipped and listed with their reason. |
| Confirmation | `[y/N]` prompt; without a terminal `-y`/`--yes` is required (exits `3` otherwise). `--dry-run` inspects without `--yes` and changes nothing. |
| Prechecks | Refuses before any change when an open job exists (prints the `registry.py ... close --request cancelled` command), when `.mam/loop-guard-active` or `.mam/update-in-progress` is held, when a target row lacks a valid agent or role, or when the caller itself is one of the targets. |
| Lock & concurrency | Takes update admission (`mam_update_admit`), acquires `.mam/loop-guard-active` (`mam_acquire_loop_lock`), then releases update admission so `run_loop.sh`, `update.sh`, and concurrent resets refuse while the reset runs. |
| Spec file & rerun | Writes `.mam/reset-spec.json` atomically before the first stop/purge and removes it only when all targets reach `done`. Rerunning resumes from the spec file without re-purging already recreated agents (`created` or `done`). |
| Sequence per target | Sequential (never parallel): ① normal `stop_session.sh --reason reset` (captures verified ID onto the row while herdr is live and sets `status: stopped`), ② checks that the row has a recorded conversation ID, ③ `stop_session.sh --reason reset --purge-conversation --yes` (using the row's recorded ID, never a CLI UUID), then `create_session.sh` with the original name, agent, role, `--herdr-session`, and `--herdr-workspace` (onboarding enabled by default). |
| When skipped (exit 4) | If step ② finds no recorded ID or step ③ exits `4` (in-use gate), the target is skipped and the reset continues to the next agent: the skipped agent is stopped but resumable (its row keeps its id) and its conversation files are untouched; rerun once its holders end, or resume it. |
| Exit | Meaning |
|---|---|
| 0 | All targeted agents reached `done` and `.mam/reset-spec.json` was removed; or nothing to reset, cancelled at prompt, or `--dry-run` with no blockers |
| 1 | Refused by precheck (nothing changed), corrupt/mismatched `.mam/reset-spec.json`, or a `stop`/`create` step failed (`.mam/reset-spec.json` is kept so rerunning resumes) |
| 2 | Invalid arguments |
| 3 | Not a terminal and `-y`/`--yes` was not passed; nothing was changed |
| 4 | Completed the pass, but one or more agents were skipped (purge refused with stop exit 4, or no conversation ID recorded); `.mam/reset-spec.json` is kept for rerun |
| 5 | No agent was skipped or failed, but one or more newly created post-discovered agents do not have their new conversation ID recorded yet; `.mam/reset-spec.json` is kept so rerunning re-checks without re-purging |
| other | Under `set -e`, an unexpected sub-step failure passes its nonzero exit code through |
## Safety Notes
- **Irreversible**: purged conversations cannot be recovered. Confirm with `--dry-run` first.
- **No external process termination**: processes holding conversation files open (reported via `lsof` in `in_use_hints`) are never signalled by MAM; end them yourself when safe, then rerun.
- **Shared stores preserved**: Grok's shared `memory-v2.db` is left intact during purge; only session rows and per-session files are removed.
- **Orchestrator hook during reset**: while `.mam/loop-guard-active` is held by the reset, the AGY orchestrator's direct file-edit hook (`loop_delegation_guard.sh`) blocks direct edits until the reset finishes.
- **Post-precheck job submissions**: `multi-agent-mux-delegate-job submit` does not check `.mam/loop-guard-active`; do not submit new jobs to resident agents while a reset is running.
- **Same-type post-discovered ordering**: when resetting multiple post-discovered agents of the same type, the second `create_session.sh` waits up to `MAM_ID_ORDER_WAIT_S` for the first agent's ID to be recorded; if it times out, rerun the reset once the ID is recorded.
+286
View File
@@ -0,0 +1,286 @@
#!/usr/bin/env bash
# reset.sh — multi-agent-mux-reset (Issue #78): purge the control root's
# running resident agents and recreate each with the same name, agent, role,
# herdr server and workspace label, one at a time, with onboarding.
# Usage:
# bash .agents/skills/multi-agent-mux-reset/scripts/reset.sh [-y|--yes] [--dry-run]
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
SKILL_DIR="$(cd "$SCRIPT_DIR/../.." && pwd)"
WS="$(cd -P "$SCRIPT_DIR/../../../.." && pwd -P)"
export WORKSPACE_ROOT="$WS"
export AGENT_SESSIONS_YAML="${AGENT_SESSIONS_YAML:-$WS/.mam/agent-sessions.yaml}"
export PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}"
usage() {
cat <<'USAGE'
Usage: bash .agents/skills/multi-agent-mux-reset/scripts/reset.sh [-y|--yes] [--dry-run]
Options:
-y, --yes Confirm purging and recreating the control root's running agents
--dry-run Inspect targets, skipped rows, spec progress and blockers without changing anything
-h, --help Show this help
USAGE
}
YES=0
DRY_RUN=0
while [ $# -gt 0 ]; do
case "$1" in
-y|--yes) YES=1; shift ;;
--dry-run) DRY_RUN=1; shift ;;
-h|--help) usage; exit 0 ;;
*) echo "ERROR: unknown arg: $1" >&2; usage >&2; exit 2 ;;
esac
done
if [ "$DRY_RUN" != "1" ] && [ "$YES" != "1" ] && [ ! -t 0 ]; then
echo "ERROR: not a terminal; run with -y/--yes (or --dry-run to inspect)" >&2
exit 3
fi
MAM="$WS/.mam"
MARKER="${MAM_LOOP_MARKER:-$MAM/loop-guard-active}"
SPEC_FILE="$MAM/reset-spec.json"
# shellcheck disable=SC1091
source "$SKILL_DIR/multi-agent-mux-loop/scripts/loop_lock.sh"
if [ "$DRY_RUN" = "1" ]; then
if [ -x "$WS/.venv/bin/python" ]; then
PY="$WS/.venv/bin/python"
elif command -v python3 >/dev/null 2>&1; then
PY="python3"
else
PY="python"
fi
DRY_ARGS=()
if [ -f "$MAM/update-in-progress" ]; then
upd_st="$(_mam_lock_holder_state "$MAM/update-in-progress")"
case "$upd_st" in
stale) echo "NOTE: stale update marker $MAM/update-in-progress (its update is gone); not a reason to refuse" ;;
*) DRY_ARGS+=(--refusal "an update of $WS is in progress (pid $(_mam_lock_field "$MAM/update-in-progress" pid), holder $upd_st); wait for it to finish") ;;
esac
fi
if [ -f "$MARKER" ]; then
loop_st="$(_mam_lock_holder_state "$MARKER")"
case "$loop_st" in
stale) echo "NOTE: stale loop marker $MARKER (its loop is gone); not a reason to refuse" ;;
*) DRY_ARGS+=(--refusal "a review loop or another reset is running (pid $(_mam_lock_field "$MARKER" pid), holder $loop_st); wait for it to end") ;;
esac
fi
dry_rc=0
"$PY" -m lib_py.reset dry-run --workspace "$WS" ${DRY_ARGS[@]+"${DRY_ARGS[@]}"} || dry_rc=$?
if [ "$dry_rc" -eq 0 ]; then
exit 0
fi
exit 1
fi
# shellcheck disable=SC1091
source "$WS/.agents/skills/lib.sh"
PY="$(_delegate_py_bin)"
mam_update_admit "$WS" || exit 1
_acquire_rc=0
mam_acquire_loop_lock "$MARKER" || _acquire_rc=$?
case "$_acquire_rc" in
0) ;;
3)
mam_update_admit_release
_holder_pid="$(_mam_lock_field "$MARKER" "pid" 2>/dev/null || echo "unknown")"
echo "ERROR: reset refused; nothing was changed:" >&2
echo " - a review loop or another reset is running (pid $_holder_pid); wait for it to end" >&2
exit 1
;;
*)
mam_update_admit_release
echo "ERROR: reset refused; nothing was changed:" >&2
echo " - cannot determine whether a review loop or another reset is running (ps unavailable?)" >&2
exit 1
;;
esac
_release_reset_lock() {
mam_release_loop_lock "$MARKER" || true
}
_on_reset_signal() {
local sig_rc="$1"
_release_reset_lock
exit "$sig_rc"
}
trap _release_reset_lock EXIT
trap '_on_reset_signal 130' INT
trap '_on_reset_signal 143' TERM
trap '_on_reset_signal 129' HUP
mam_update_admit_release
STATE_JSON=""
if ! STATE_JSON="$(load_state_json strict)"; then
echo "ERROR: reset refused; nothing was changed:" >&2
echo " - the registry $AGENT_SESSIONS_YAML cannot be read; fix it first" >&2
exit 1
fi
pre_out=""
pre_rc=0
pre_out="$(MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset precheck --workspace "$WS")" || pre_rc=$?
if [ "$pre_rc" -ne 0 ]; then
echo "ERROR: reset refused; nothing was changed:" >&2
while IFS= read -r line; do
[ -n "$line" ] && echo " - $line" >&2
done <<< "${pre_out:-lib_py.reset precheck failed (rc $pre_rc)}"
exit 1
fi
if [ -f "$SPEC_FILE" ]; then
MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset dry-run --workspace "$WS" || exit 1
if [ "$YES" != "1" ]; then
ans=""
read -r -p "Resume reset from $SPEC_FILE? [y/N]: " ans || ans=""
case "$ans" in
y|Y|yes|Yes|YES) ;;
*) echo "Reset cancelled; nothing was changed"; exit 0 ;;
esac
fi
else
PLAN_JSON="$(MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset plan --workspace "$WS")" || exit 1
TARGET_COUNT="$(printf '%s' "$PLAN_JSON" | "$PY" -c 'import json,sys; print(len(json.load(sys.stdin).get("targets", [])))')"
MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset dry-run --workspace "$WS" || exit 1
if [ "$TARGET_COUNT" = "0" ]; then
echo "nothing to reset"
exit 0
fi
if [ "$YES" != "1" ]; then
ans=""
read -r -p "Purge and recreate $TARGET_COUNT agents? [y/N]: " ans || ans=""
case "$ans" in
y|Y|yes|Yes|YES) ;;
*) echo "Reset cancelled; nothing was changed"; exit 0 ;;
esac
fi
printf '%s' "$PLAN_JSON" | "$PY" -m lib_py.reset write-spec --workspace "$WS" || exit 1
fi
ENTRIES_OUT="$(MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset entries --workspace "$WS" --warn-new)" || exit 1
ENTRY_NAMES=()
while IFS= read -r line; do
[ -n "$line" ] || continue
eval "$line"
ENTRY_NAMES+=("$R_NAME")
done <<< "$ENTRIES_OUT"
SKIPPED_ARGS=()
FAILED_SPEC=""
for target_name in ${ENTRY_NAMES[@]+"${ENTRY_NAMES[@]}"}; do
if ! STATE_JSON="$(load_state_json strict)"; then
FAILED_SPEC="$target_name:stop:1"
break
fi
entry_line="$(MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset entries --workspace "$WS" --session "$target_name")" || {
FAILED_SPEC="$target_name:stop:1"
break
}
eval "$entry_line"
case "$R_STATE" in
done|created)
continue
;;
pending)
STOP_CMD=(bash "$SKILL_DIR/multi-agent-mux-stop/scripts/stop_session.sh" --session "$R_NAME" --agent "$R_AGENT")
if [ -n "$R_SERVER" ]; then
STOP_CMD+=(--herdr-session "$R_SERVER")
fi
STOP_CMD+=(--reason reset)
echo "reset [$R_NAME]: ${STOP_CMD[*]}"
stop_rc=0
"${STOP_CMD[@]}" || stop_rc=$?
if [ "$stop_rc" -ne 0 ]; then
FAILED_SPEC="$R_NAME:stop:$stop_rc"
break
fi
if ! STATE_JSON="$(load_state_json strict)"; then
FAILED_SPEC="$R_NAME:stop:1"
break
fi
if [ -z "$R_OLD_ID" ]; then
MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset record-id --workspace "$WS" --session "$R_NAME" || {
FAILED_SPEC="$R_NAME:stop:1"
break
}
fi
entry_line="$(MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset entries --workspace "$WS" --session "$R_NAME")" || {
FAILED_SPEC="$R_NAME:stop:1"
break
}
eval "$entry_line"
if [ -z "$R_ROW_ID" ]; then
echo "skipped $R_NAME: no conversation id is recorded, so a purge would drop the row and leave the conversation on disk; resume it, or purge it by hand (stop_session.sh --session $R_NAME --purge-conversation --yes) and rerun the reset to recreate it"
SKIPPED_ARGS+=(--skipped "$R_NAME:no_id")
continue
fi
echo "reset [$R_NAME]: ${STOP_CMD[*]} --purge-conversation --yes"
purge_rc=0
"${STOP_CMD[@]}" --purge-conversation --yes || purge_rc=$?
if [ "$purge_rc" -eq 4 ]; then
echo "skipped $R_NAME: its purge was refused (stop exit 4); the row (now stopped, id recorded) and the conversation are kept; the processes above hold it - they are not killed; rerun the reset once they end"
SKIPPED_ARGS+=(--skipped "$R_NAME:purge_refused")
continue
fi
if [ "$purge_rc" -ne 0 ]; then
FAILED_SPEC="$R_NAME:stop:$purge_rc"
break
fi
R_STATE="purged"
;;
esac
if [ "$R_STATE" = "purged" ]; then
CREATE_CMD=(bash "$SKILL_DIR/multi-agent-mux-create/scripts/create_session.sh" --workspace "$WS" --agent "$R_AGENT" --role "$R_ROLE" --session "$R_NAME")
if [ -n "$R_SERVER" ]; then
CREATE_CMD+=(--herdr-session "$R_SERVER")
fi
if [ -n "$R_LABEL" ]; then
CREATE_CMD+=(--herdr-workspace "$R_LABEL")
fi
echo "reset [$R_NAME]: ${CREATE_CMD[*]}"
create_rc=0
"${CREATE_CMD[@]}" || create_rc=$?
if [ "$create_rc" -ne 0 ]; then
FAILED_SPEC="$R_NAME:create:$create_rc"
break
fi
fi
done
STATE_JSON="$(load_state_json strict)" || exit 1
TABLE_ARGS=(--workspace "$WS")
if [ ${#SKIPPED_ARGS[@]} -gt 0 ]; then
TABLE_ARGS+=("${SKIPPED_ARGS[@]}")
fi
if [ -n "$FAILED_SPEC" ]; then
TABLE_ARGS+=(--failed "$FAILED_SPEC")
fi
table_rc=0
MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset table "${TABLE_ARGS[@]}" || table_rc=$?
case "$table_rc" in
0)
rm -f "$SPEC_FILE"
exit 0
;;
4)
exit 4
;;
5)
exit 5
;;
*)
exit 1
;;
esac
+1
View File
@@ -57,6 +57,7 @@ bash deploy/install.sh
| **`multi-agent-mux-loop`** | 계획·구현·리뷰: 선택적인 planner 와 creator, 그다음 선택한 리뷰어(`--reviewer A,B`, 또는 `--all-reviewer` 로 creator 를 뺀 실행 중인 리뷰어 전원)가 모두 `PASS` 를 낼 때까지(또는 `--max-loop` 까지). 둘 다 없으면 creator 가 스스로 리뷰합니다. 루프 밖에서 끝난 작업은 `--review-only` 로 리뷰합니다. 통과한 리포트는 `.agents/reports/<session>/` 으로 옮겨 보관합니다. |
| **`multi-agent-mux-orc-onboard`** | 오케스트레이터 자신의 세션 ID 를 등록해, 세션 탐색이 그것을 하위 에이전트로 착각하지 않게 합니다. |
| **`multi-agent-mux-update`** | 설치된 작업 공간을 제자리에서 업데이트합니다. 실행 중인 세션·열린 잡·루프가 있으면 거부하고, 모니터를 멈췄다가 원래대로 되돌리며, 설치가 실패하면 되돌리고, 새 코드로 레지스트리를 변환합니다. |
| **`multi-agent-mux-reset`** | 제어 루트의 실행 중인 상주 에이전트를 하나씩 정리(purge)하고 같은 이름·에이전트·역할·herdr 서버·작업 공간 라벨로 다시 만들며 온보딩합니다(`.mam/reset-spec.json` 으로 중단된 지점부터 재개). |
---
+1
View File
@@ -57,6 +57,7 @@ Every skill lives under `.agents/skills/` with a `SKILL.md` that documents its o
| **`multi-agent-mux-loop`** | Plan, implement, review: an optional planner and a creator, then the selected reviewers (`--reviewer A,B`, or `--all-reviewer` for every running reviewer except the creator) until all of them `PASS` or `--max-loop` is reached. With neither option the creator reviews its own work. `--review-only` reviews work finished outside the loop. Passing reports are promoted to `.agents/reports/<session>/`. |
| **`multi-agent-mux-orc-onboard`** | Registers the orchestrator's own session id, so session discovery never mistakes it for a sub-agent. |
| **`multi-agent-mux-update`** | Updates an installed workspace in place: refuses while sessions, open jobs or a loop are live; stops and restores the monitor; rolls back on a failed install; converts the registry with the new code. |
| **`multi-agent-mux-reset`** | Purges the control root's running resident agents and recreates each with the same name, agent, role, herdr server and workspace label, one at a time, with onboarding (`.mam/reset-spec.json` for resumable reruns). |
---
+1
View File
@@ -42,6 +42,7 @@ jobs:
shellcheck deploy/install_mam.sh
shellcheck deploy/generate-env.sh
shellcheck .agents/skills/multi-agent-mux-update/scripts/*.sh
shellcheck .agents/skills/multi-agent-mux-reset/scripts/*.sh
shellcheck deploy/remove.sh
echo "✅ ShellCheck completed successfully."
+3
View File
@@ -668,5 +668,8 @@ cat <<EOF
3. Update installed skills in the future:
$ bash .agents/skills/multi-agent-mux-update/scripts/update.sh
4. Reset running control-root agents with fresh conversations:
$ bash .agents/skills/multi-agent-mux-reset/scripts/reset.sh --dry-run
--------------------------------------------------------------------------------
EOF
+1
View File
@@ -93,6 +93,7 @@ else
".agents/skills/multi-agent-mux-status"
".agents/skills/multi-agent-mux-stop"
".agents/skills/multi-agent-mux-update"
".agents/skills/multi-agent-mux-reset"
".agents/hooks"
".agents/hooks.json"
".agents/MULTI_AGENT_RULES.md"
+27
View File
@@ -477,6 +477,20 @@
],
"mismatch": null,
"note": "entry point call (Issue #72)"
},
{
"skill": "multi-agent-mux-reset",
"start": 17,
"end": 19,
"lang": "bash",
"sha": "3e9ea3c85bb0",
"class": "c",
"kind": "invocation",
"evidence": [
"multi-agent-mux-reset/scripts/reset.sh:6"
],
"mismatch": null,
"note": "entry point call (Issue #78)"
}
],
"removed": [
@@ -635,6 +649,19 @@
"passthrough": {
"4": "multi-agent-mux-create/scripts/create_session.sh:554"
}
},
"multi-agent-mux-reset/scripts/reset.sh": {
"skill": "multi-agent-mux-reset",
"exit_codes": [
"0",
"1",
"2",
"3",
"4",
"5",
"other"
],
"passthrough": {}
}
},
"prose_mismatches": [
+1 -1
View File
@@ -116,7 +116,7 @@ def test_inventory_matches_skill_md(skill):
def test_inventory_covers_nine_skills():
assert len(SKILL_NAMES) == 9 # Issue #72: + multi-agent-mux-update
assert len(SKILL_NAMES) == 10 # Issue #78: + multi-agent-mux-reset
assert {b["skill"] for b in DATA["blocks"]} == set(SKILL_NAMES)
+338
View File
@@ -0,0 +1,338 @@
"""Issue #78: in_use_hints lists lsof holders, and codex purge removes
tui-thread-reference-capabilities/<uuid>.
"""
from __future__ import annotations
import json
import os
import sqlite3
import subprocess
import sys
import uuid as uuidlib
from pathlib import Path
import pytest
REPO_ROOT = Path(__file__).resolve().parent.parent
SKILLS = REPO_ROOT / ".agents" / "skills"
if str(SKILLS) not in sys.path:
sys.path.insert(0, str(SKILLS))
from lib_py.agents.base import DiscoveryContext # noqa: E402
from lib_py.agents.registry import get_adapter # noqa: E402
pytestmark = pytest.mark.usefixtures("own_workspace_root")
def _clean_env(extra: dict | None = None) -> dict:
env = {
k: v
for k, v in os.environ.items()
if not k.startswith("XDG_")
and not k.startswith("HERDR_")
and k
not in (
"CLAUDE_CODE_SESSION_ID",
"ANTIGRAVITY_CONVERSATION_ID",
"HERMES_SESSION_ID",
"GROK_SESSION_ID",
"OPENCODE_SESSION_ID",
"CLAUDE_CONFIG_DIR",
"CLAUDE_PROJECT_DIR",
"CODEX_HOME",
)
}
env["PYTHONPATH"] = str(SKILLS)
if extra:
env.update(extra)
return env
def _write_fake_lsof(bindir: Path, mapping: dict[str, list[int]]) -> Path:
"""Fake lsof that prints mapped PIDs when any argument matches a key substring."""
bindir.mkdir(parents=True, exist_ok=True)
cases = []
for needle, pids in mapping.items():
echoes = "\n".join(f" echo {p}" for p in pids)
cases.append(f" *{needle}*)\n{echoes}\n found=1 ;;")
body = "\n".join(cases)
exe = bindir / "lsof"
exe.write_text(
"#!/bin/sh\n"
"found=0\n"
'for a in "$@"; do\n'
' case "$a" in\n'
f"{body}\n"
" esac\n"
"done\n"
'[ "$found" = "1" ] && exit 0\n'
"exit 1\n",
encoding="utf-8",
)
exe.chmod(0o755)
return exe
def _materialize_artifact(agent: str, conv: str, ctx: DiscoveryContext) -> str:
adapter = get_adapter(agent)
if agent == "agy":
# Hold the presence lock so agy's in_use_paths (_artifact_paths) is exercised too.
paths = adapter._artifact_paths(conv, ctx)
target = paths[-1]
else:
target = adapter.artifact_path(conv, ctx)
Path(target).parent.mkdir(parents=True, exist_ok=True)
Path(target).write_text("{}\n", encoding="utf-8")
return target
@pytest.mark.parametrize("agent", ["claude", "agy", "codex", "grok", "hermes"])
def test_in_use_hints_list_lsof_holders(tmp_path, monkeypatch, agent):
home = tmp_path / "home"
ws = tmp_path / "ws"
bindir = tmp_path / "bin"
no_lsof_bin = tmp_path / "no-lsof-bin"
for d in (home, ws, bindir, no_lsof_bin):
d.mkdir(parents=True, exist_ok=True)
# Populate no_lsof_bin with symlinks to basic tools (ps) but not lsof.
for tool in ("ps", "sh", "bash"):
for sysdir in ("/bin", "/usr/bin"):
cand = Path(sysdir) / tool
if cand.exists() and not (no_lsof_bin / tool).exists():
(no_lsof_bin / tool).symlink_to(cand)
conv = str(uuidlib.uuid4())
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("HOME_DIR", str(home))
monkeypatch.setenv("CODEX_HOME", str(home / ".codex"))
ctx = DiscoveryContext(workspace=str(ws), agent_name=agent, home_dir=str(home))
target = _materialize_artifact(agent, conv, ctx)
adapter = get_adapter(agent)
holder = subprocess.Popen(["sleep", "300"])
try:
_write_fake_lsof(bindir, {conv: [holder.pid]})
env = _clean_env({
"HOME": str(home),
"HOME_DIR": str(home),
"CODEX_HOME": str(home / ".codex"),
"PATH": f"{bindir}:{os.environ.get('PATH', '')}",
})
res = subprocess.run(
[sys.executable, "-m", "lib_py.agents", "in-use", agent, conv, str(ws)],
capture_output=True,
text=True,
env=env,
timeout=30,
)
assert res.returncode == 4, res.stdout + res.stderr
assert json.loads(res.stdout) == {"agent": agent, "in_use": True}
assert f"ps -axww | grep {conv}" in res.stderr
assert f"held open by pid {holder.pid}:" in res.stderr
assert "sleep 300" in res.stderr
assert "not stopped by MAM" in res.stderr
monkeypatch.setenv("PATH", env["PATH"])
hints = adapter.in_use_hints(conv, ctx)
if agent == "claude":
assert hints[0].startswith("a claude process whose session file under ")
assert f"ps -axww | grep {conv}" in hints[1]
else:
assert f"ps -axww | grep {conv}" in hints[0]
assert any(f"held open by pid {holder.pid}:" in h and "sleep 300" in h for h in hints)
assert target in adapter.in_use_paths(conv, ctx)
finally:
holder.kill()
holder.wait()
def test_in_use_hints_no_lsof_unchanged(tmp_path, monkeypatch):
"""§5.1 regression control: without lsof on PATH, hints match pre-#78 lines."""
home = tmp_path / "home"
ws = tmp_path / "ws"
no_lsof_bin = tmp_path / "no-lsof-bin"
for d in (home, ws, no_lsof_bin):
d.mkdir(parents=True, exist_ok=True)
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("HOME_DIR", str(home))
monkeypatch.setenv("CODEX_HOME", str(home / ".codex"))
monkeypatch.setenv("PATH", str(no_lsof_bin))
for agent in ("claude", "agy", "codex", "grok", "hermes"):
conv = str(uuidlib.uuid4())
ctx = DiscoveryContext(workspace=str(ws), agent_name=agent, home_dir=str(home))
_materialize_artifact(agent, conv, ctx)
adapter = get_adapter(agent)
hints = adapter.in_use_hints(conv, ctx)
assert not any("held open by pid" in h for h in hints)
assert any(f"ps -axww | grep {conv}" in h for h in hints)
def test_in_use_hints_opencode_skip_service(tmp_path, monkeypatch):
home = tmp_path / "home"
ws = tmp_path / "ws"
bindir = tmp_path / "bin"
for d in (home, ws, bindir):
d.mkdir(parents=True, exist_ok=True)
db = home / ".local" / "share" / "opencode" / "opencode.db"
db.parent.mkdir(parents=True, exist_ok=True)
db.write_text("", encoding="utf-8")
adapter = get_adapter("opencode")
monkeypatch.setattr(adapter, "_bin", lambda ctx: None)
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("HOME_DIR", str(home))
ctx = DiscoveryContext(workspace=str(ws), agent_name="opencode", home_dir=str(home))
service_proc = subprocess.Popen(
[sys.executable, "-c", "import time; time.sleep(300)", "serve", "--service"]
)
holder_proc = subprocess.Popen(["sleep", "300"])
try:
_write_fake_lsof(bindir, {"opencode.db": [service_proc.pid, holder_proc.pid]})
monkeypatch.setenv("PATH", f"{bindir}:{os.environ.get('PATH', '')}")
hints = adapter.in_use_hints("ses_test78", ctx)
assert hints[0] == "any process with it on its command line (an editor, a grep, ...): ps -axww | grep ses_test78"
assert any(f"held open by pid {holder_proc.pid}:" in h and "sleep 300" in h for h in hints)
assert not any(f"pid {service_proc.pid}" in h or "serve --service" in h for h in hints)
assert hints[-1] == f"another opencode session holding {db}: lsof {db}"
finally:
for p in (service_proc, holder_proc):
p.kill()
p.wait()
def test_stop_purge_refusal_prints_holders(mam_sandbox, mock_herdr, monkeypatch):
conv = str(uuidlib.uuid4())
name = "issue78-reviewer-claude"
ws = mam_sandbox
ctx = DiscoveryContext(workspace=str(ws), agent_name="claude", home_dir=str(mam_sandbox))
transcript = Path(get_adapter("claude").artifact_path(conv, ctx))
transcript.parent.mkdir(parents=True, exist_ok=True)
transcript.write_text(json.dumps({"sessionId": conv}) + "\n", encoding="utf-8")
lib = mam_sandbox / ".agents" / "skills" / "lib.sh"
yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
mutation = (
f"d['herdr_sessions'] = [{{'name': {name!r}, 'status': 'running', 'agent': 'claude', "
f"'role': 'reviewer', 'claude_session_id_own': {conv!r}, "
f"'pane': {{'cwd': {os.path.realpath(str(ws))!r}}}}}]\n"
)
res_init = subprocess.run(
["bash", "-c", f'source "{lib}" && atomic_dump_yaml "{yaml_path}"'],
input=mutation,
capture_output=True,
text=True,
env=_clean_env({
"HOME": str(mam_sandbox),
"HOME_DIR": str(mam_sandbox),
"WORKSPACE_ROOT": str(mam_sandbox),
"AGENT_SESSIONS_YAML": str(yaml_path),
}),
)
assert res_init.returncode == 0, res_init.stdout + res_init.stderr
holder = subprocess.Popen(["sleep", "300"])
try:
bindir = mam_sandbox / ".local" / "bin"
_write_fake_lsof(bindir, {conv: [holder.pid]})
stop_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh"
env = _clean_env({
"HOME": str(mam_sandbox),
"HOME_DIR": str(mam_sandbox),
"CLAUDE_PROJECT_DIR": str(mam_sandbox / ".claude" / "projects"),
"WORKSPACE_ROOT": str(mam_sandbox),
"AGENT_SESSIONS_YAML": str(yaml_path),
"PATH": f"{bindir}:{os.environ.get('PATH', '')}",
})
res = subprocess.run(
["bash", str(stop_sh), "--session", name, "--agent", "claude", "--purge-conversation", "--yes"],
capture_output=True,
text=True,
cwd=str(mam_sandbox),
env=env,
timeout=60,
)
assert res.returncode == 4, res.stdout + res.stderr
assert f"held open by pid {holder.pid}:" in res.stderr
assert "sleep 300" in res.stderr
assert transcript.exists()
assert name in yaml_path.read_text(encoding="utf-8")
finally:
holder.kill()
holder.wait()
@pytest.mark.parametrize("delete_ok", [True, False], ids=["delete_ok", "delete_fails"])
def test_codex_purge_removes_tui_thread_capability(tmp_path, monkeypatch, delete_ok):
home = tmp_path / "home"
codex_home = tmp_path / "codex-home"
ws = tmp_path / "ws"
bindir = tmp_path / "bin"
for d in (home, codex_home, ws, bindir):
d.mkdir(parents=True, exist_ok=True)
conv = str(uuidlib.uuid4())
other = str(uuidlib.uuid4())
rollout = codex_home / "sessions" / "2026" / "10" / "28" / f"rollout-2026-10-28T00-00-00-{conv}.jsonl"
rollout.parent.mkdir(parents=True, exist_ok=True)
rollout.write_text(json.dumps({"payload": {"id": conv, "cwd": str(ws)}}) + "\n", encoding="utf-8")
cap_dir = codex_home / "tui-thread-reference-capabilities"
cap_dir.mkdir(parents=True, exist_ok=True)
cap_mine = cap_dir / conv
cap_other = cap_dir / other
cap_mine.write_text("cap-mine", encoding="utf-8")
cap_other.write_text("cap-other", encoding="utf-8")
state_db = codex_home / "state_5.sqlite"
conn = sqlite3.connect(str(state_db))
conn.execute("CREATE TABLE threads (id TEXT PRIMARY KEY, rollout_path TEXT, cwd TEXT)")
conn.execute("INSERT INTO threads VALUES (?, ?, ?)", (conv, str(rollout), str(ws)))
conn.execute("INSERT INTO threads VALUES (?, ?, ?)", (other, "", str(ws)))
conn.commit()
conn.close()
fake_codex = bindir / "codex"
if delete_ok:
# Simulate `codex delete <uuid> --force` removing rollout + DB row, leaving tui-thread-reference-capabilities.
fake_codex.write_text(
f"#!{sys.executable}\n"
"import os, sqlite3, sys\n"
"if len(sys.argv) >= 3 and sys.argv[1] == 'delete':\n"
f" p = {str(rollout)!r}\n"
" if os.path.exists(p):\n"
" os.remove(p)\n"
f" conn = sqlite3.connect({str(state_db)!r})\n"
" conn.execute('DELETE FROM threads WHERE id=?', (sys.argv[2],))\n"
" conn.commit()\n"
" conn.close()\n"
" sys.exit(0)\n"
"sys.exit(1)\n",
encoding="utf-8",
)
else:
fake_codex.write_text("#!/bin/sh\nexit 1\n", encoding="utf-8")
fake_codex.chmod(0o755)
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("HOME_DIR", str(home))
monkeypatch.setenv("CODEX_HOME", str(codex_home))
monkeypatch.setenv("PATH", f"{bindir}:{os.environ.get('PATH', '')}")
adapter = get_adapter("codex")
ctx = DiscoveryContext(workspace=str(ws), agent_name="codex", home_dir=str(home))
assert str(cap_mine) in adapter.residual_artifacts(conv, ctx)
purged = adapter.purge_artifacts(conv, ctx)
assert str(cap_mine) in purged
assert not cap_mine.exists()
assert cap_other.exists()
assert not rollout.exists()
assert adapter.residual_artifacts(conv, ctx) == []
assert str(cap_other) in adapter.residual_artifacts(other, ctx)
File diff suppressed because it is too large. Load diff