fix(reset): never take another live reset's fence, and fence absent row registries (#78)
Review round 4 (codex): - R1: fence_registries reads an existing marker under the registry lock and refuses the reset, changing nothing, when another live (or undecidable) reset holds it - registries may be shared between workspaces; only a stale marker is replaced, and the reset's exit removes only its own fences. - R2: the registries to fence come from update.jobs_dirs, the enumeration open_jobs now uses too, whether or not they exist; an absent row registry is created and fenced (a registration would create it), and one that cannot be created refuses the reset. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
568385c77a
commit
b49314c3d0
6 files changed
+213
-74
No files matched your search
@@ -94,32 +94,45 @@ def _lstart(pid: str) -> str:
|
||||
return " ".join(out.split())
|
||||
|
||||
|
||||
def reset_refusal(registry_dir: str, agent_session: str) -> Optional[str]:
|
||||
"""Why registering a job for ``agent_session`` in ``registry_dir`` is
|
||||
refused, or None: a live reset fenced this registry (for that session) and
|
||||
the caller is not its own create. Holder states as loop_lock.sh decides
|
||||
them: a dead or reused pid is stale; ps failing on this process is unknown
|
||||
(refused)."""
|
||||
def reset_marker(registry_dir: str) -> Tuple[str, Dict[str, str]]:
|
||||
"""(holder state, fields) of the registry's RESET_MARKER: "absent", or as
|
||||
loop_lock.sh decides - "stale" (dead or reused pid), "live", "unknown" (ps
|
||||
fails on this process; a marker that cannot be read is unknown too)."""
|
||||
marker = os.path.join(registry_dir, RESET_MARKER)
|
||||
try:
|
||||
with open(marker, encoding="utf-8") as fh:
|
||||
fields = dict(line.split("=", 1) for line in fh.read().splitlines() if "=" in line)
|
||||
except FileNotFoundError:
|
||||
return None
|
||||
return "absent", {}
|
||||
except (OSError, ValueError) as exc:
|
||||
return "the reset marker %s cannot be read (%s)" % (marker, exc)
|
||||
if "sessions" in fields and agent_session not in fields["sessions"].split(","):
|
||||
return None
|
||||
return "unknown", {"error": str(exc)}
|
||||
pid = fields.get("pid", "")
|
||||
if not pid.isdigit() or os.environ.get(RESET_OWNER_ENV) == pid:
|
||||
return None
|
||||
if not pid.isdigit():
|
||||
return "stale", fields
|
||||
if not _lstart(str(os.getpid())):
|
||||
return "cannot tell whether the reset in %s (pid %s) is live (ps unavailable?)" % (marker, pid)
|
||||
return "unknown", fields
|
||||
cur = _lstart(pid)
|
||||
if not cur or (fields.get("lstart") and cur != fields["lstart"]):
|
||||
return "stale", fields
|
||||
return "live", fields
|
||||
|
||||
|
||||
def reset_refusal(registry_dir: str, agent_session: str) -> Optional[str]:
|
||||
"""Why registering a job for ``agent_session`` in ``registry_dir`` is
|
||||
refused, or None: a live (or undecidable) reset fenced this registry, for
|
||||
that session, and the caller is not its own create."""
|
||||
state, fields = reset_marker(registry_dir)
|
||||
if state in ("absent", "stale"):
|
||||
return None
|
||||
if "sessions" in fields and agent_session not in fields["sessions"].split(","):
|
||||
return None
|
||||
if fields.get("pid") and os.environ.get(RESET_OWNER_ENV) == fields["pid"]:
|
||||
return None
|
||||
if state == "unknown":
|
||||
return "cannot tell whether the reset fencing %s is live (%s)" % (
|
||||
os.path.join(registry_dir, RESET_MARKER), fields.get("error") or "ps unavailable?")
|
||||
return ("a reset of %s is in progress (pid %s); its agents are being purged and recreated;"
|
||||
" register the job when it has finished" % (fields.get("workspace", "?"), pid))
|
||||
" register the job when it has finished" % (fields.get("workspace", "?"), fields["pid"]))
|
||||
|
||||
|
||||
def register_job(
|
||||
|
||||
@@ -23,8 +23,8 @@ from typing import Dict, Iterator, List, Mapping, Optional, Tuple
|
||||
from lib_py.agents.base import AGENT_SESSION_ENV, POST_DISCOVERED
|
||||
from lib_py.agents.registry import agent_of_row, get_adapter, own_key
|
||||
from lib_py.delegation.mqtt_common import registry_lock
|
||||
from lib_py.delegation.registry import RESET_MARKER
|
||||
from lib_py.update import open_jobs, row_server
|
||||
from lib_py.delegation.registry import RESET_MARKER, reset_marker
|
||||
from lib_py.update import jobs_dirs, open_jobs, row_server
|
||||
from lib_py.verify_session import verify_session_uuid
|
||||
|
||||
SPEC_VERSION = 1
|
||||
@@ -485,35 +485,36 @@ def target_open_jobs(workspace: str, state: dict, env: Mapping[str, str], sessio
|
||||
return out
|
||||
|
||||
|
||||
class FenceBusy(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def fence_registries(workspace: str, state: dict, env: Mapping[str, str]) -> Iterator[str]:
|
||||
"""Write RESET_MARKER into every registry open_jobs reads, each under its
|
||||
"""Write RESET_MARKER into every registry open_jobs reads (jobs_dirs,
|
||||
created when absent - a registration would create it too), each under its
|
||||
registry lock (a registration that read no marker has finished writing),
|
||||
and yield its real path as it is written. This workspace's .mam/jobs and
|
||||
MAM_JOBS_DIR (created when absent) refuse every registration; another
|
||||
control plane's registry named by a row refuses only those addressed to
|
||||
this registry's rows. The marker copies .mam/reset-in-progress."""
|
||||
and yield its real path as it is written. FenceBusy when one is already
|
||||
fenced by another live (or undecidable) reset - registries may be shared
|
||||
between workspaces; its marker is left as it is. This workspace's own
|
||||
registries refuse every registration; another control plane's only those
|
||||
addressed to this registry's rows. The marker copies .mam/reset-in-progress."""
|
||||
with open(os.path.join(workspace, ".mam", "reset-in-progress"), encoding="utf-8") as fh:
|
||||
base = "".join(l + "\n" for l in fh.read().splitlines() if l.split("=", 1)[0] in ("pid", "lstart", "started"))
|
||||
me = dict(l.split("=", 1) for l in base.splitlines()).get("pid", "")
|
||||
base += "workspace=%s\n" % workspace
|
||||
rows = state.get("herdr_sessions") or []
|
||||
sessions = ",".join("herdr:%s" % s["name"] for s in rows if s.get("name"))
|
||||
plan, seen = [], set()
|
||||
for d in (os.path.join(workspace, ".mam", "jobs"), env.get("MAM_JOBS_DIR") or ""):
|
||||
if d:
|
||||
os.makedirs(d, exist_ok=True)
|
||||
plan.append((d, base))
|
||||
for s in rows:
|
||||
d = (s.get("control_plane") or {}).get("mam_jobs_dir") or ""
|
||||
if d and os.path.isdir(d):
|
||||
plan.append((d, base + "sessions=%s\n" % sessions))
|
||||
for d, text in plan:
|
||||
if _real(d) in seen:
|
||||
continue
|
||||
seen.add(_real(d))
|
||||
for d, addressed_only in jobs_dirs(workspace, state, env):
|
||||
os.makedirs(d, exist_ok=True)
|
||||
with registry_lock(d):
|
||||
held, fields = reset_marker(d)
|
||||
if held in ("live", "unknown") and fields.get("pid") != me:
|
||||
raise FenceBusy(
|
||||
"the job registry %s is fenced by a reset of %s (pid %s, holder %s); wait for it to end"
|
||||
% (d, fields.get("workspace", "?"), fields.get("pid", "?"), held))
|
||||
fd, tmp = tempfile.mkstemp(dir=d, prefix=RESET_MARKER + ".")
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as fh:
|
||||
fh.write(text)
|
||||
fh.write(base + ("sessions=%s\n" % sessions if addressed_only else ""))
|
||||
os.replace(tmp, os.path.join(d, RESET_MARKER))
|
||||
yield _real(d)
|
||||
|
||||
@@ -886,9 +887,12 @@ def main(argv: Optional[List[str]] = None) -> int:
|
||||
if a.cmd == "fence":
|
||||
try:
|
||||
for d in fence_registries(ws, state, os.environ):
|
||||
print(d, flush=True)
|
||||
print("fenced %s" % d, flush=True)
|
||||
except FenceBusy as exc:
|
||||
print("refused %s" % exc)
|
||||
return 1
|
||||
except OSError as exc:
|
||||
print("ERROR: cannot fence the job registries (%s)" % exc, file=sys.stderr)
|
||||
print("refused the job registries cannot be fenced against new registrations (%s)" % exc)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
@@ -147,34 +147,40 @@ 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:
|
||||
def jobs_dirs(workspace: str, state: dict, env: Mapping[str, str]) -> List[Tuple[str, bool]]:
|
||||
"""(dir, addressed_only) per registry an open-job check reads, once per
|
||||
real path, whether or not it exists yet: this workspace's .mam/jobs and
|
||||
MAM_JOBS_DIR, then another row's control_plane.mam_jobs_dir, where only
|
||||
records addressed to a row of this registry count (addressed_only)."""
|
||||
rows = state.get("herdr_sessions") or []
|
||||
cands = [(c, False) for c in (os.path.join(workspace, ".mam", "jobs"), env.get("MAM_JOBS_DIR") or "")]
|
||||
cands += [((s.get("control_plane") or {}).get("mam_jobs_dir") or "", True) for s in rows]
|
||||
out, seen = [], set()
|
||||
for c, addressed_only in cands:
|
||||
if c and _real(c) not in seen:
|
||||
seen.add(_real(c))
|
||||
own.append(c)
|
||||
out.append((c, addressed_only))
|
||||
return out
|
||||
|
||||
|
||||
def open_jobs(workspace: str, state: dict, env: Mapping[str, str]) -> List[dict]:
|
||||
"""Every open record (not completed/error/cancelled, not dry_run) in the
|
||||
registries jobs_dirs names; in another row's control-plane registry only
|
||||
those addressed to a row of this registry. A record, or a directory, that
|
||||
cannot be read is open (path only)."""
|
||||
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})
|
||||
for d, addressed_only in jobs_dirs(workspace, state, env):
|
||||
if not _is_jobs_dir(d):
|
||||
continue
|
||||
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
|
||||
|
||||
|
||||
|
||||
@@ -23,7 +23,7 @@ bash .agents/skills/multi-agent-mux-reset/scripts/reset.sh --yes [--dry-run]
|
||||
| Targets | Every registry row with `status: running` and `realpath(pane.cwd) == realpath(WS)`. Rows with any other status, worktree 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, when a running `claude`/`grok` target has no recorded conversation id (the monitor had not pinned one; record it, or stop/purge it by hand first), when `.mam/reset-spec.json` is invalid, or when the caller itself is one of the targets. |
|
||||
| Lock & concurrency | Takes the update admission lock exclusively, acquires `.mam/loop-guard-active` and publishes `.mam/reset-in-progress`, then releases the admission lock and runs the prechecks. `run_loop.sh`, `update.sh` and another reset refuse on the loop marker. The reset then writes `.reset-in-progress` into every job registry its open-job check reads (`.mam/jobs`, `MAM_JOBS_DIR`, and a row's other `control_plane.mam_jobs_dir`), each under that registry's lock, before the check. A registration (`multi-agent-mux-delegate-job submit`, `registry.py register`, the registry API) into a fenced registry - by any path to that directory - refuses; in another control plane's registry only for this workspace's sessions. The onboarding jobs of the reset's own `create_session.sh` are let through. |
|
||||
| Lock & concurrency | Takes the update admission lock exclusively, acquires `.mam/loop-guard-active` and publishes `.mam/reset-in-progress`, then releases the admission lock and runs the prechecks. `run_loop.sh`, `update.sh` and another reset refuse on the loop marker. The reset then writes `.reset-in-progress` into every job registry its open-job check reads (`.mam/jobs`, `MAM_JOBS_DIR`, and a row's other `control_plane.mam_jobs_dir`; created when absent), each under that registry's lock, before the check. It refuses, changing nothing, when a registry shared with another workspace is already fenced by that workspace's live reset; it never replaces or removes another reset's marker. A registration (`multi-agent-mux-delegate-job submit`, `registry.py register`, the registry API) into a fenced registry - by any path to that directory - refuses; in another control plane's registry only for this workspace's sessions. The onboarding jobs of the reset's own `create_session.sh` are let through. |
|
||||
| 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): ⓪ re-checks that its row still has the spec's agent and herdr server and that no job addressed to it is open, ① normal `stop_session.sh --reason reset` (a post-discovered agent's id is captured onto the row while herdr is live; `status: stopped`), ② checks that the row's id is the one recorded before the stop and, for a post-discovered agent, that it still verifies as a conversation of the row's folder, ③ `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) | The target is skipped and the reset continues to the next agent when step ⓪ finds a changed row (`row changed`) or an open job (`job open`) - nothing was changed for it -, or when step ② finds no id (`no conversation id recorded`) or an id other than the recorded one or one that does not verify for the row's folder (`id not verified`), or step ③ exits `4` (`purge refused`, in-use gate) - it is stopped but resumable (its row keeps its id) and its conversation files are untouched. Rerun once the cause is gone, or resume it. |
|
||||
|
||||
@@ -150,12 +150,16 @@ fi
|
||||
fence_out=""
|
||||
fence_rc=0
|
||||
fence_out="$(MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset fence --workspace "$WS")" || fence_rc=$?
|
||||
fence_refusal=""
|
||||
while IFS= read -r line; do
|
||||
[ -n "$line" ] && FENCED+=("$line")
|
||||
case "$line" in
|
||||
"fenced "*) FENCED+=("${line#fenced }") ;;
|
||||
"refused "*) fence_refusal="${line#refused }" ;;
|
||||
esac
|
||||
done <<< "$fence_out"
|
||||
if [ "$fence_rc" -ne 0 ]; then
|
||||
echo "ERROR: reset refused; nothing was changed:" >&2
|
||||
echo " - the job registries could not be fenced against new registrations (lib_py.reset fence rc $fence_rc)" >&2
|
||||
echo " - ${fence_refusal:-the job registries could not be fenced against new registrations (lib_py.reset fence rc $fence_rc)}" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
|
||||
+122
-10
@@ -1790,7 +1790,7 @@ def test_mam_reset_admit_refuses_only_a_live_reset(mam_sandbox: Path):
|
||||
assert res_stale.returncode == 0, res_stale.stderr
|
||||
|
||||
|
||||
@pytest.mark.parametrize("registry", ["canonical", "alias", "custom"])
|
||||
@pytest.mark.parametrize("registry", ["canonical", "alias", "custom", "row-absent"])
|
||||
def test_registry_cli_is_refused_after_the_target_recheck(
|
||||
mam_sandbox: Path, mock_herdr: Path, mock_agents, registry: str
|
||||
):
|
||||
@@ -1798,14 +1798,15 @@ def test_registry_cli_is_refused_after_the_target_recheck(
|
||||
that is not a child of the reset, after the target's open-job re-check and
|
||||
before its stop, is refused and registers nothing - whether it names the
|
||||
registry by its canonical path, by a symlink alias, or the registry is the
|
||||
MAM_JOBS_DIR the reset runs with; the reset's own create still registers
|
||||
its onboarding job."""
|
||||
MAM_JOBS_DIR the reset runs with, or a row's control-plane registry that
|
||||
did not exist when the reset started (round 4 R2); the reset's own create
|
||||
still registers its onboarding job."""
|
||||
import threading
|
||||
|
||||
name = "reviewer-claude-01"
|
||||
_claude_transcript(mam_sandbox, mam_sandbox, UA)
|
||||
_mock_herdr_agent(mock_herdr, name, mam_sandbox, agent="claude")
|
||||
_write_rows(mam_sandbox, [_row(name, "claude", mam_sandbox, own_id=UA)])
|
||||
row = _row(name, "claude", mam_sandbox, own_id=UA)
|
||||
jobs_dir = mam_sandbox / ".mam" / "jobs"
|
||||
jobs_dir.mkdir(parents=True, exist_ok=True)
|
||||
env_extra = None
|
||||
@@ -1816,6 +1817,10 @@ def test_registry_cli_is_refused_after_the_target_recheck(
|
||||
elif registry == "custom":
|
||||
jobs_dir = cli_dir = mam_sandbox / "elsewhere" / "jobs"
|
||||
env_extra = {"MAM_JOBS_DIR": str(jobs_dir)}
|
||||
elif registry == "row-absent":
|
||||
cli_dir = mam_sandbox / "elsewhere" / "jobs"
|
||||
row["control_plane"] = {"mam_jobs_dir": str(cli_dir)}
|
||||
_write_rows(mam_sandbox, [row])
|
||||
stop = mam_sandbox / ".agents/skills/multi-agent-mux-stop/scripts/stop_session.sh"
|
||||
original = stop.with_name("stop-original.sh")
|
||||
stop.rename(original)
|
||||
@@ -1844,7 +1849,7 @@ os.execv('/bin/bash', ['bash', {str(original)!r}, *sys.argv[1:]])
|
||||
while not entered.exists() and time.monotonic() < end:
|
||||
time.sleep(0.02)
|
||||
record["marker"] = (mam_sandbox / ".mam" / "reset-in-progress").is_file()
|
||||
before = sorted(p.name for p in jobs_dir.glob("*.json"))
|
||||
before = sorted(p.name for p in cli_dir.glob("*.json"))
|
||||
cli = mam_sandbox / ".agents/skills/multi-agent-mux-delegate-job/scripts/registry.py"
|
||||
res = subprocess.run(
|
||||
[sys.executable, str(cli), "--registry-dir", str(cli_dir), "register",
|
||||
@@ -1852,7 +1857,7 @@ os.execv('/bin/bash', ['bash', {str(original)!r}, *sys.argv[1:]])
|
||||
capture_output=True, text=True, timeout=30,
|
||||
)
|
||||
record.update(rc=res.returncode, stdout=res.stdout, stderr=res.stderr, before=before,
|
||||
after=sorted(p.name for p in jobs_dir.glob("*.json")))
|
||||
after=sorted(p.name for p in cli_dir.glob("*.json")))
|
||||
finally:
|
||||
release.touch()
|
||||
|
||||
@@ -1873,6 +1878,7 @@ os.execv('/bin/bash', ['bash', {str(original)!r}, *sys.argv[1:]])
|
||||
onboarding = [json.loads(p.read_text(encoding="utf-8")) for p in jobs_dir.glob("*.json")]
|
||||
assert [j["agent_session"] for j in onboarding] == ["herdr:" + name], onboarding
|
||||
assert not (jobs_dir / ".reset-in-progress").exists()
|
||||
assert not (cli_dir / ".reset-in-progress").exists()
|
||||
|
||||
|
||||
def _registry_marker(jobs_dir: Path, holder: subprocess.Popen, ws: Path, sessions: Optional[str] = None) -> None:
|
||||
@@ -1916,7 +1922,8 @@ def test_fence_waits_for_a_registration_and_covers_every_registry_read(tmp_path:
|
||||
registry lock until its record is written; fence writes the marker under
|
||||
that lock, so it waits, and the precheck after it sees the job. Fenced are
|
||||
.mam/jobs, MAM_JOBS_DIR (one marker for an alias of the same directory) and
|
||||
a row's other control-plane registry (for this registry's sessions only)."""
|
||||
a row's other control-plane registry (for this registry's sessions only),
|
||||
also one that does not exist yet (created); a stale marker is replaced."""
|
||||
import fcntl
|
||||
|
||||
ws = tmp_path / "ws"
|
||||
@@ -1924,10 +1931,15 @@ def test_fence_waits_for_a_registration_and_covers_every_registry_read(tmp_path:
|
||||
jobs_dir.mkdir(parents=True)
|
||||
(ws / "alias").symlink_to(jobs_dir)
|
||||
other = tmp_path / "cp" / "jobs"
|
||||
absent = tmp_path / "cp2" / "jobs"
|
||||
other.mkdir(parents=True)
|
||||
# A stale marker (a killed reset) is replaced
|
||||
(other / ".reset-in-progress").write_text("pid=999999\nlstart=x\nworkspace=/gone\n", encoding="utf-8")
|
||||
row = _row("reviewer-claude-01", "claude", ws, own_id=UA)
|
||||
row["control_plane"] = {"mam_jobs_dir": str(other)}
|
||||
state = {"herdr_sessions": [row]}
|
||||
row2 = _row("reviewer-claude-02", "claude", ws, own_id=UB)
|
||||
row2["control_plane"] = {"mam_jobs_dir": str(absent)}
|
||||
state = {"herdr_sessions": [row, row2]}
|
||||
env = dict(os.environ, PYTHONPATH=str(SKILLS), PYTHONDONTWRITEBYTECODE="1", MAM_STATE_JSON=json.dumps(state),
|
||||
MAM_JOBS_DIR=str(ws / "alias"))
|
||||
holder = subprocess.Popen(["sleep", "120"])
|
||||
@@ -1952,11 +1964,14 @@ def test_fence_waits_for_a_registration_and_covers_every_registry_read(tmp_path:
|
||||
lock.close()
|
||||
out, err = proc.communicate(timeout=30)
|
||||
assert proc.returncode == 0, out + err
|
||||
assert out.split() == [os.path.realpath(jobs_dir), os.path.realpath(other)]
|
||||
assert out.splitlines() == ["fenced " + os.path.realpath(p) for p in (jobs_dir, other, absent)]
|
||||
own = (jobs_dir / ".reset-in-progress").read_text(encoding="utf-8")
|
||||
assert "pid=%d\n" % holder.pid in own and "workspace=%s\n" % os.path.realpath(ws) in own
|
||||
assert "sessions=" not in own
|
||||
assert "sessions=herdr:reviewer-claude-01\n" in (other / ".reset-in-progress").read_text(encoding="utf-8")
|
||||
for d in (other, absent):
|
||||
text = (d / ".reset-in-progress").read_text(encoding="utf-8")
|
||||
assert "pid=%d\n" % holder.pid in text
|
||||
assert "sessions=herdr:reviewer-claude-01,herdr:reviewer-claude-02\n" in text
|
||||
pre = subprocess.run([sys.executable, "-m", "lib_py.reset", "precheck", "--workspace", str(ws)],
|
||||
env=env, capture_output=True, text=True, timeout=30)
|
||||
assert pre.returncode == 1 and "job job78locked" in pre.stdout, pre.stdout + pre.stderr
|
||||
@@ -2010,3 +2025,100 @@ def test_post_discovered_id_is_purged_only_when_it_verifies_for_its_folder(
|
||||
kept = _read_rows(mam_sandbox)[name]
|
||||
assert kept["status"] == "stopped" and kept["codex_session_id_own"] == UB
|
||||
assert (mam_sandbox / ".mam" / "reset-spec.json").is_file()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("other_has_target", [False, True])
|
||||
def test_other_reset_cannot_take_a_live_fence(
|
||||
mam_sandbox: Path, mock_herdr: Path, mock_agents, other_has_target: bool
|
||||
):
|
||||
"""Review round 4 R1 (codex): while reset A holds the fence of a registry
|
||||
shared with workspace B (both MAM_JOBS_DIR), B's reset is refused before
|
||||
any change and leaves A's fence as it is; after B has ended, a registration
|
||||
for A's target is still refused and A completes."""
|
||||
import shutil
|
||||
import threading
|
||||
|
||||
name = "reviewer-claude-01"
|
||||
shared = mam_sandbox / "shared" / "jobs"
|
||||
_claude_transcript(mam_sandbox, mam_sandbox, UA)
|
||||
_mock_herdr_agent(mock_herdr, name, mam_sandbox, agent="claude")
|
||||
_write_rows(mam_sandbox, [_row(name, "claude", mam_sandbox, own_id=UA)])
|
||||
other = mam_sandbox / "other-workspace"
|
||||
shutil.copytree(SKILLS, other / ".agents" / "skills")
|
||||
(other / ".mam").mkdir()
|
||||
other_rows = []
|
||||
other_tf = None
|
||||
if other_has_target:
|
||||
other_tf = _claude_transcript(mam_sandbox, other, UB)
|
||||
other_rows = [_row("reviewer-claude-09", "claude", other, own_id=UB)]
|
||||
(other / ".mam" / "agent-sessions.yaml").write_text(
|
||||
yaml.safe_dump({"herdr_sessions": other_rows, "delegation_jobs": []}), encoding="utf-8")
|
||||
|
||||
stop = mam_sandbox / ".agents/skills/multi-agent-mux-stop/scripts/stop_session.sh"
|
||||
original = stop.with_name("stop-original.sh")
|
||||
stop.rename(original)
|
||||
entered, release = mam_sandbox / "stop-entered", mam_sandbox / "stop-release"
|
||||
gate = mam_sandbox / "stop-gate.py"
|
||||
gate.write_text(
|
||||
f"""import os, pathlib, sys, time
|
||||
entered = pathlib.Path({str(entered)!r})
|
||||
release = pathlib.Path({str(release)!r})
|
||||
if not entered.exists():
|
||||
entered.touch()
|
||||
end = time.monotonic() + 90
|
||||
while not release.exists() and time.monotonic() < end:
|
||||
time.sleep(0.02)
|
||||
os.execv('/bin/bash', ['bash', {str(original)!r}, *sys.argv[1:]])
|
||||
""",
|
||||
encoding="utf-8",
|
||||
)
|
||||
stop.write_text("#!/bin/bash\nexec %s %s \"$@\"\n" % (shlex.quote(sys.executable), shlex.quote(str(gate))))
|
||||
stop.chmod(0o755)
|
||||
record: dict = {}
|
||||
|
||||
def writer():
|
||||
try:
|
||||
end = time.monotonic() + 60
|
||||
while not entered.exists() and time.monotonic() < end:
|
||||
time.sleep(0.02)
|
||||
fence = shared / ".reset-in-progress"
|
||||
record["fence_before"] = fence.read_text(encoding="utf-8")
|
||||
env_b = dict(os.environ, WORKSPACE_ROOT=str(other), AGENT_SESSIONS_YAML=str(other / ".mam/agent-sessions.yaml"),
|
||||
MAM_JOBS_DIR=str(shared), DELEGATE_JOB_LOGS_DIR=str(other / ".mam/delegate_job_logs"),
|
||||
MAM_ENV_FILE=str(other / ".mam.env.absent"), PYTHONDONTWRITEBYTECODE="1")
|
||||
b = subprocess.run(["/bin/bash", str(other / ".agents/skills/multi-agent-mux-reset/scripts/reset.sh"), "--yes"],
|
||||
cwd=str(other), env=env_b, stdin=subprocess.DEVNULL, capture_output=True, text=True,
|
||||
timeout=60)
|
||||
record.update(b_rc=b.returncode, b_out=b.stdout, b_err=b.stderr,
|
||||
fence_after=fence.read_text(encoding="utf-8") if fence.exists() else None,
|
||||
b_markers=[p.name for p in (other / ".mam").iterdir() if "progress" in p.name or "guard" in p.name])
|
||||
cli = mam_sandbox / ".agents/skills/multi-agent-mux-delegate-job/scripts/registry.py"
|
||||
res = subprocess.run(
|
||||
[sys.executable, str(cli), "--registry-dir", str(shared), "register",
|
||||
"--prompt", "external concurrent work", "--agent", "claude-code", "--agent-session", "herdr:" + name],
|
||||
capture_output=True, text=True, timeout=30,
|
||||
)
|
||||
record.update(rc=res.returncode, stderr=res.stderr, jobs=sorted(p.name for p in shared.glob("*.json")))
|
||||
finally:
|
||||
release.touch()
|
||||
|
||||
worker = threading.Thread(target=writer)
|
||||
worker.start()
|
||||
try:
|
||||
res = _run_reset(mam_sandbox, "--yes", env_extra={"MAM_JOBS_DIR": str(shared)})
|
||||
finally:
|
||||
worker.join(timeout=150)
|
||||
_wait_reconcile_idle(mam_sandbox)
|
||||
assert not worker.is_alive()
|
||||
assert record["b_rc"] == 1, record
|
||||
assert "fenced by a reset of %s" % mam_sandbox in record["b_err"] and "nothing was changed" in record["b_err"], record
|
||||
assert record["fence_after"] == record["fence_before"], record
|
||||
assert record["b_markers"] == [], record
|
||||
assert record["rc"] == 1 and "a reset of %s is in progress" % mam_sandbox in record["stderr"], record
|
||||
assert record["jobs"] == [], record
|
||||
if other_has_target:
|
||||
assert other_tf.is_file()
|
||||
assert yaml.safe_load((other / ".mam/agent-sessions.yaml").read_text())["herdr_sessions"][0]["status"] == "running"
|
||||
assert res.returncode == 0, res.stdout + res.stderr
|
||||
assert _result_of(res.stdout, name) == "done"
|
||||
assert not (shared / ".reset-in-progress").exists()
|
||||
Reference in new issue
Block a user