fix(update): an unlistable jobs directory refuses; the rowless-agent hint names its server (Issue #72, review R2, R3)

R2: glob() hid a listing error, so an unreadable .mam/jobs (or MAM_JOBS_DIR,
or another row's jobs dir) counted as empty and the update installed over an
open job. Records are now listed with os.listdir; a directory that cannot be
listed, or stat'ed, is a blocker with its path.

R3: a discovered agent with no row was told to run stop with --session only;
stop then looked on the workspace's server, not the one the agent was found
on. The hint now carries --herdr-session <server>, quoted.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
GodopuandClaude Opus 5.5 committed 2026-10-27 08:15:23 +00:00
1 parent e3394655b3
commit ba75718ac3
3 files changed
+96 -11

No files matched your search

+36 -10
View File
@@ -31,9 +31,9 @@ from __future__ import annotations
import argparse import argparse
import fcntl import fcntl
import glob
import json import json
import os import os
import shlex
import shutil import shutil
import sqlite3 import sqlite3
import subprocess import subprocess
@@ -109,7 +109,14 @@ def server_agents(herdr: str, server: str, env: Mapping[str, str]) -> List[dict]
# --------------------------------------------------------------------------- jobs # --------------------------------------------------------------------------- jobs
def _job_records(d: str) -> Iterable[Tuple[str, Optional[dict]]]: def _job_records(d: str) -> Iterable[Tuple[str, Optional[dict]]]:
for p in sorted(glob.glob(os.path.join(d, "*.json"))): """(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: try:
with open(p, encoding="utf-8") as fh: with open(p, encoding="utf-8") as fh:
rec = json.load(fh) rec = json.load(fh)
@@ -119,6 +126,18 @@ def _job_records(d: str) -> Iterable[Tuple[str, Optional[dict]]]:
yield p, rec if isinstance(rec, dict) else None 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: def _is_open(rec: dict) -> bool:
return rec.get("status") not in TERMINAL_STATUSES and rec.get("dry_run") is not True return rec.get("status") not in TERMINAL_STATUSES and rec.get("dry_run") is not True
@@ -127,10 +146,10 @@ 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 """Every open record (not completed/error/cancelled, not dry_run) in this
workspace's .mam/jobs and MAM_JOBS_DIR; in another row's 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. control_plane.mam_jobs_dir only those addressed to a row of this registry.
A record that cannot be read is open (path only).""" A record, or a directory, that cannot be read is open (path only)."""
own, seen = [], set() own, seen = [], set()
for c in (os.path.join(workspace, ".mam", "jobs"), env.get("MAM_JOBS_DIR") or ""): for c in (os.path.join(workspace, ".mam", "jobs"), env.get("MAM_JOBS_DIR") or ""):
if c and os.path.isdir(c) and _real(c) not in seen: if c and _is_jobs_dir(c) and _real(c) not in seen:
seen.add(_real(c)) seen.add(_real(c))
own.append(c) own.append(c)
rows = state.get("herdr_sessions") or [] rows = state.get("herdr_sessions") or []
@@ -138,7 +157,7 @@ def open_jobs(workspace: str, state: dict, env: Mapping[str, str]) -> List[dict]
other = [] other = []
for s in rows: for s in rows:
c = (s.get("control_plane") or {}).get("mam_jobs_dir") or "" c = (s.get("control_plane") or {}).get("mam_jobs_dir") or ""
if c and os.path.isdir(c) and _real(c) not in seen: if c and _is_jobs_dir(c) and _real(c) not in seen:
seen.add(_real(c)) seen.add(_real(c))
other.append(c) other.append(c)
out = [] out = []
@@ -146,7 +165,7 @@ def open_jobs(workspace: str, state: dict, env: Mapping[str, str]) -> List[dict]
for d in dirs: for d in dirs:
for path, rec in _job_records(d): for path, rec in _job_records(d):
if rec is None: if rec is None:
out.append({"path": path, "unreadable": True}) out.append({"path": path, "unreadable": True, "dir": path == d})
elif _is_open(rec) and (not addressed_only or rec.get("agent_session") in names): 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 "", out.append({"path": path, "job_id": rec.get("job_id") or "",
"agent_session": rec.get("agent_session") or "", "agent_session": rec.get("agent_session") or "",
@@ -161,6 +180,9 @@ def precheck(workspace: str, state: dict, herdr: str, env: Mapping[str, str]) ->
blockers: List[str] = [] blockers: List[str] = []
stop = ("bash .agents/skills/multi-agent-mux-stop/scripts/stop_session.sh --session %s" stop = ("bash .agents/skills/multi-agent-mux-stop/scripts/stop_session.sh --session %s"
" (resume it after the update with multi-agent-mux-resume)") " (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. # 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_server, me_pane = caller_server(env), env.get("HERDR_PANE_ID") or ""
@@ -193,7 +215,7 @@ def precheck(workspace: str, state: dict, herdr: str, env: Mapping[str, str]) ->
for s in rows: for s in rows:
if s.get("status") == "running" and s is not me_row: if s.get("status") == "running" and s is not me_row:
blockers.append("session %s is running (server %s); stop it: %s" blockers.append("session %s is running (server %s); stop it: %s"
% (s.get("name"), row_server(s), stop % s.get("name"))) % (s.get("name"), row_server(s), stop % shlex.quote(s.get("name") or "")))
row_keys = {(row_server(s), sanitize_herdr_agent_name(s["name"])) 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 s in rows if s.get("status") == "running" and s.get("name")}
@@ -209,8 +231,9 @@ def precheck(workspace: str, state: dict, herdr: str, env: Mapping[str, str]) ->
continue # already listed as its running row continue # already listed as its running row
blockers.append("herdr agent %s (pane %s, server %s) runs in %s; stop it: %s" blockers.append("herdr agent %s (pane %s, server %s) runs in %s; stop it: %s"
% (name or "<unnamed>", a.get("pane_id"), server, a.get("cwd"), % (name or "<unnamed>", a.get("pane_id"), server, a.get("cwd"),
stop % name if name else stop_rowless % (shlex.quote(name), shlex.quote(server)) if name else
"herdr --session %s pane close %s" % (server, a.get("pane_id")))) "herdr --session %s pane close %s"
% (shlex.quote(server), shlex.quote(a.get("pane_id") or ""))))
jobs = open_jobs(workspace, state, env) jobs = open_jobs(workspace, state, env)
if me_row is not None: if me_row is not None:
@@ -220,7 +243,10 @@ def precheck(workspace: str, state: dict, herdr: str, env: Mapping[str, str]) ->
if len(mine) == 1: if len(mine) == 1:
jobs = [j for j in jobs if j is not mine[0]] jobs = [j for j in jobs if j is not mine[0]]
for j in jobs: for j in jobs:
if j.get("unreadable"): 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" blockers.append("job record %s cannot be read; check it, then move or delete it by hand"
% j["path"]) % j["path"])
else: else:
@@ -22,7 +22,7 @@ bash .agents/skills/multi-agent-mux-update/scripts/update.sh --yes
|---|---| |---|---|
| Where | An installed workspace only (`.mam_deploy/remove.sh` present). A git checkout updates with `git pull`. | | Where | An installed workspace only (`.mam_deploy/remove.sh` present). A git checkout updates with `git pull`. |
| Confirmation | `[y/N]` prompt; without a terminal `-y`/`--yes` is required. There is no force option. | | Confirmation | `[y/N]` prompt; without a terminal `-y`/`--yes` is required. There is no force option. |
| Refused when | a running registry row, a live herdr agent whose cwd is in the workspace (any running server), an open job (unreadable job JSON included, `dry_run` excluded), a live loop marker, `.mam/monitor.lock` held by a monitor the skills did not start, or another update. The calling agent (its herdr server + pane, row and one running job) is not counted. Each refusal lists the cleanup; nothing is changed. | | Refused when | a running registry row, a live herdr agent whose cwd is in the workspace (any running server), an open job (an unreadable job JSON or jobs directory included, `dry_run` excluded), a live loop marker, `.mam/monitor.lock` held by a monitor the skills did not start, or another update. The calling agent (its herdr server + pane, row and one running job) is not counted. Each refusal lists the cleanup; nothing is changed. |
| During the update | create, resume, `run_loop.sh`, delegate submit, stop and orc-onboard refuse (`.mam/update-in-progress`, `.mam/update-admission.lock`). | | During the update | create, resume, `run_loop.sh`, delegate submit, stop and orc-onboard refuse (`.mam/update-in-progress`, `.mam/update-admission.lock`). |
| Monitor | stopped first, started again afterwards only if it ran (`MAM_NO_SUBSCRIBER_DAEMON=1`: not started). | | Monitor | stopped first, started again afterwards only if it ran (`MAM_NO_SUBSCRIBER_DAEMON=1`: not started). |
| Install | the installer at `MAM_INSTALLER_URL` (default: Gitea `main`) runs in place; `.mam` is kept. On failure `.agents/`, `.mam_deploy/`, root files, `.venv`, the installer's `.mam` files and the registry are restored from `.mam/update-backup/`. | | Install | the installer at `MAM_INSTALLER_URL` (default: Gitea `main`) runs in place; `.mam` is kept. On failure `.agents/`, `.mam_deploy/`, root files, `.venv`, the installer's `.mam` files and the registry are restored from `.mam/update-backup/`. |
+59
View File
@@ -45,6 +45,15 @@ elif a[:2] == ["agent", "list"]:
if spec.get("fail") == "agent-list:" + server: if spec.get("fail") == "agent-list:" + server:
sys.exit(1) sys.exit(1)
print(json.dumps({"result": {"agents": spec.get("agents", {}).get(server, [])}})) print(json.dumps({"result": {"agents": spec.get("agents", {}).get(server, [])}}))
elif a[:2] == ["agent", "get"]:
hit = [x for x in spec.get("agents", {}).get(server, []) if a[2] in (x.get("name"), x.get("pane_id"))]
if not hit:
sys.exit(1)
print(json.dumps({"result": {"agent": hit[0]}}))
elif a[:2] == ["pane", "close"]:
spec.setdefault("agents", {})[server] = [x for x in spec.get("agents", {}).get(server, [])
if x.get("pane_id") != a[2]]
json.dump(spec, open(os.environ["FAKE_HERDR_SPEC"], "w"))
''' '''
STUB_MONITOR = r'''#!/usr/bin/env bash STUB_MONITOR = r'''#!/usr/bin/env bash
@@ -366,6 +375,28 @@ def test_refuses_a_rowless_agent_on_any_running_server(box):
assert '"down"' not in calls assert '"down"' not in calls
def test_the_rowless_agent_hint_stops_it_on_the_server_it_was_found_on(box):
"""R3: the printed stop command names the server; without it stop looks on
the workspace's server, where a same-named agent runs outside the workspace."""
write_rows(box.ws, [])
box.herdr(sessions={"default": True, "unrecorded-server": True},
agents={"unrecorded-server": [{"name": "stray", "pane_id": "w3:p1", "cwd": str(box.ws / "sub")}],
"default": [{"name": "stray", "pane_id": "w1:p1", "cwd": str(box.root / "elsewhere")}]})
res = box.update()
assert res.returncode == 1, res.stdout + res.stderr
line = next(ln for ln in res.stderr.splitlines() if "herdr agent stray" in ln)
cmd = line.split("stop it: ", 1)[1]
assert "--herdr-session unrecorded-server" in cmd, cmd
stop = subprocess.run(["bash", "-c", cmd], cwd=str(box.ws), env=box.env, stdin=subprocess.DEVNULL,
capture_output=True, text=True, timeout=120)
assert stop.returncode == 0, stop.stdout + stop.stderr
agents = json.loads(box.spec.read_text())["agents"]
assert agents["unrecorded-server"] == []
assert [x["pane_id"] for x in agents["default"]] == ["w1:p1"], "the same name elsewhere is left alone"
res = box.update()
assert res.returncode == 0, res.stdout + res.stderr
def test_an_agent_outside_the_workspace_does_not_refuse(box): def test_an_agent_outside_the_workspace_does_not_refuse(box):
box.herdr(agents={"default": [{"name": "other", "pane_id": "w1:p2", "cwd": str(box.root / "elsewhere")}]}) box.herdr(agents={"default": [{"name": "other", "pane_id": "w1:p2", "cwd": str(box.root / "elsewhere")}]})
res = box.update() res = box.update()
@@ -406,6 +437,34 @@ def test_open_jobs_refuse_and_terminal_or_dry_run_do_not(box, tmp_path):
del box.env["MAM_JOBS_DIR"] del box.env["MAM_JOBS_DIR"]
@pytest.mark.skipif(os.geteuid() == 0, reason="root reads a mode-000 directory")
def test_a_jobs_directory_that_cannot_be_listed_refuses(box, tmp_path):
"""R2: an unreadable .mam/jobs (or MAM_JOBS_DIR under an unreadable parent)
is not an empty registry."""
jobs = box.mam / "jobs"
write_job(jobs, "hidden1", status="running")
before = snapshot(box.ws)
jobs.chmod(0)
try:
res = box.update()
finally:
jobs.chmod(0o755)
assert_refused_unchanged(box, before, res, "jobs directory %s cannot be listed" % jobs)
(jobs / "hidden1.json").unlink()
parent = tmp_path / "locked"
write_job(parent / "jobs", "env1", status="running")
box.env["MAM_JOBS_DIR"] = str(parent / "jobs")
before = snapshot(box.ws)
parent.chmod(0)
try:
res = box.update()
finally:
parent.chmod(0o755)
assert_refused_unchanged(box, before, res, "jobs directory %s cannot be listed" % (parent / "jobs"))
del box.env["MAM_JOBS_DIR"]
def test_other_control_planes_count_only_jobs_addressed_to_this_registry(box, tmp_path): def test_other_control_planes_count_only_jobs_addressed_to_this_registry(box, tmp_path):
other = tmp_path / "other-plane" / "jobs" other = tmp_path / "other-plane" / "jobs"
write_job(other, "theirs", status="running", session="herdr:not-ours") write_job(other, "theirs", status="running", session="herdr:not-ours")