From 253a0aa56ee3e7f51602b65fddf2346662ff91e6 Mon Sep 17 00:00:00 2001 From: Godopu Date: Tue, 27 Oct 2026 11:09:30 +0000 Subject: [PATCH] fix(delegate): admission covers a rowless target and the installed parent of an incidental .mam (Issue #72, review round 2) R1: a target with no registry row skipped the target gate. Its cwd is now read from the live agent on the server delivery uses (resolve_herdr_session), through the real herdr only when that server is running, so no server is started; a herdr that cannot answer refuses before registration. R2: any .mam counted as a workspace and stopped the walk, so B/sub/.mam hid an installed B and the row's control_plane root was dropped. Workspaces are now update.sh's installed trees (.mam_deploy/remove.sh or remove.sh): every installed ancestor of the cwd, plus the recorded control_plane root. lib_py.update target-workspaces prints them one per line; the wrapper holds each on fds 20.. until the record is written. Co-Authored-By: Claude Opus 5.5 --- .agents/skills/lib.sh | 2 +- .agents/skills/lib_py/update.py | 77 +++++++++++++------ .../multi-agent-mux-delegate-job | 24 ++++-- tests/test_issue72_admission.py | 58 ++++++++++---- 4 files changed, 115 insertions(+), 46 deletions(-) diff --git a/.agents/skills/lib.sh b/.agents/skills/lib.sh index 90d08ff..85be750 100644 --- a/.agents/skills/lib.sh +++ b/.agents/skills/lib.sh @@ -3266,7 +3266,7 @@ mam_control_plane_prefix() { # # mam_update_admit [workspace [fd]]: rc 0 admitted (fd 7, or the given one, # held unless the workspace has no .mam), 1 refused (stderr). Delegate submit -# also takes its target's workspace on fd 8. +# also takes the workspaces its target is in on fds 20 and up. mam_update_admit() { local root="${1:-$WORKSPACE_ROOT}" fd="${2:-7}" local lock="$root/.mam/update-admission.lock" marker="$root/.mam/update-in-progress" st diff --git a/.agents/skills/lib_py/update.py b/.agents/skills/lib_py/update.py index 99fe496..a88796a 100644 --- a/.agents/skills/lib_py/update.py +++ b/.agents/skills/lib_py/update.py @@ -16,11 +16,13 @@ this module what needs parsing. Standard library only, Python 3.9. python -m lib_py.update monitor --workspace W JSON {"pids": [...], "lock": "free"|"held"}: the monitor daemons daemon.find_running recognises, and whether .mam/monitor.lock is held. - python -m lib_py.update target-workspace --workspace W --agent-session S - The installed workspace a delegate target `herdr:` belongs to - (decision 8), when it is not W: the nearest directory holding a .mam - at or above its row's pane.cwd, else the row's control_plane - workspace_root. Nothing for a rowless or non-herdr target. + python -m lib_py.update target-workspaces --workspace W --agent-session S --server V --herdr H + The workspaces other than W whose update admission a delegate submit + to `herdr:` must take (decision 8), one per line: every installed + workspace at or above the target's cwd (its row's pane.cwd; with no row + the live agent's on server V) and its row's control_plane + workspace_root. Nothing for a non-herdr target. rc 1: herdr cannot + answer for a rowless target. python -m lib_py.update backup-registry --workspace W --dest D Copies the registry into D under the registry's own write lock (agent-sessions.db.lock, as atomic_yaml takes it): the DB through the @@ -264,27 +266,46 @@ def precheck(workspace: str, state: dict, herdr: str, env: Mapping[str, str]) -> # --------------------------------------------------------------------------- delegate target -def target_workspace(workspace: str, state: dict, agent_session: str) -> str: +def _installed(d: str) -> bool: + """update.sh's own test of an installed workspace.""" + return os.path.isfile(os.path.join(d, ".mam_deploy", "remove.sh")) or os.path.isfile(os.path.join(d, "remove.sh")) + + +def target_workspaces(workspace: str, state: dict, agent_session: str, server: str, + herdr: str, env: Mapping[str, str]) -> List[str]: + """Every installed workspace at or above the target's cwd - an update of + any of them counts that agent (decision 1) - and its row's + control_plane.workspace_root; the caller's own workspace left out. With + no row the cwd is the live agent's on (the one delivery uses), + read with the real herdr. No herdr, a server that is not running, or no + such agent gives nothing (delivery then fails on its own); a herdr that + cannot answer raises HerdrError.""" if not agent_session.startswith("herdr:"): - return "" + return [] name = agent_session[len("herdr:"):] row = next((s for s in state.get("herdr_sessions") or [] if isinstance(s, dict) and s.get("name") == name), None) - if row is None: - return "" - found = "" - cwd = (row.get("pane") or {}).get("cwd") or "" - d = _real(cwd) if cwd else "" - while d and not found: - if os.path.isdir(os.path.join(d, ".mam")): - found = d - elif os.path.dirname(d) == d: - break - d = os.path.dirname(d) - if not found: + cwds, out = [], [] + if row is not None: + cwds.append((row.get("pane") or {}).get("cwd") or "") root = (row.get("control_plane") or {}).get("workspace_root") or "" - found = _real(root) if root else "" - return "" if found == _real(workspace) else found + if root: + out.append(_real(root)) + elif herdr and server in running_servers(herdr, env): + for a in server_agents(herdr, server, env): + if a.get("name") in (name, sanitize_herdr_agent_name(name)): + cwds += [a.get("cwd") or "", a.get("foreground_cwd") or ""] + break + for cwd in cwds: + d = _real(cwd) if cwd else "" + while d: + if _installed(d): + out.append(d) + if os.path.dirname(d) == d: + break + d = os.path.dirname(d) + me = _real(workspace) + return [d for i, d in enumerate(out) if d != me and d not in out[:i]] # --------------------------------------------------------------------------- monitor @@ -344,9 +365,11 @@ def main(argv: Optional[List[str]] = None) -> int: s.add_argument("--herdr", required=True) s = sub.add_parser("monitor") s.add_argument("--workspace", required=True) - s = sub.add_parser("target-workspace") + s = sub.add_parser("target-workspaces") s.add_argument("--workspace", required=True) s.add_argument("--agent-session", required=True) + s.add_argument("--server", default="default") + s.add_argument("--herdr", default="") s = sub.add_parser("backup-registry") s.add_argument("--workspace", required=True) s.add_argument("--dest", required=True) @@ -362,12 +385,18 @@ def main(argv: Optional[List[str]] = None) -> int: for line in lines: print(line) return 1 if lines else 0 - if a.cmd == "target-workspace": + if a.cmd == "target-workspaces": try: state = _state() except ValueError: state = {} - print(target_workspace(ws, state, a.agent_session)) + try: + roots = target_workspaces(ws, state, a.agent_session, a.server, a.herdr, os.environ) + except HerdrError as exc: + print("cannot find %s live: %s" % (a.agent_session, exc), file=sys.stderr) + return 1 + for root in roots: + print(root) return 0 if a.cmd == "monitor": print(json.dumps(monitor_state(ws))) diff --git a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job index 1b1e488..b3e97f6 100755 --- a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job +++ b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job @@ -302,17 +302,24 @@ cmd_submit() { if [[ "$DRY_RUN" == "1" ]]; then reg_dry=(--dry-run); fi # Issue #72: no job is registered while an update of this registry's - # workspace or of the target's (its row's workspace, lib_py.update - # target-workspace) runs; held until the record is written (lib.sh + # workspace or of one the target is in (lib_py.update target-workspaces: + # its row's, or with no row the live agent's on the server delivery uses) + # runs; each is held, fd 7 then 20.., until the record is written (lib.sh # mam_update_admit). A dry-run record is never dispatched and is not # counted by the update. + local admit_fds=() target_ws tws fd=20 if [[ "$DRY_RUN" != "1" ]]; then - local target_ws target_ws="$(MAM_STATE_JSON="$(load_state_json)" PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$PY" -m lib_py.update \ - target-workspace --workspace "$WORKSPACE_ROOT" --agent-session "$AGENT_SESSION")" \ - || { echo "ERROR: cannot resolve the workspace of $AGENT_SESSION" >&2; exit 1; } + target-workspaces --workspace "$WORKSPACE_ROOT" --agent-session "$AGENT_SESSION" \ + --server "$(_is_remote "$AGENT_SESSION" || resolve_herdr_session "${AGENT_SESSION#herdr:}" "$WORKDIR" 2>/dev/null)" \ + --herdr "$(_resolve_real_herdr_path || true)")" \ + || { echo "ERROR: cannot resolve the workspace of $AGENT_SESSION; nothing was registered" >&2; exit 1; } mam_update_admit "$WORKSPACE_ROOT" || exit 1 - if [[ -n "$target_ws" ]]; then mam_update_admit "$target_ws" 8 || exit 1; fi + while IFS= read -r tws; do + [[ -n "$tws" ]] || continue + mam_update_admit "$tws" "$fd" || exit 1 + admit_fds+=("$fd"); fd=$((fd + 1)) + done <<< "$target_ws" fi # 1) register job (prints the new job id) JOB_ID="$("$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" register \ @@ -323,7 +330,10 @@ cmd_submit() { --job-type "$TYPE" --reviewer "$REVIEWER" --reviewer-session "$REVIEWER_SESSION" \ --max-iterations "$MAX_ITERATIONS" ${reg_parent[@]+"${reg_parent[@]}"} ${reg_dry[@]+"${reg_dry[@]}"})" echo "registered job: $JOB_ID" - [[ "$DRY_RUN" == "1" ]] || { mam_update_admit_release; mam_update_admit_release 8; } + if [[ "$DRY_RUN" != "1" ]]; then + mam_update_admit_release + for fd in ${admit_fds[@]+"${admit_fds[@]}"}; do mam_update_admit_release "$fd"; done + fi # 1-1) Provision job directory and write direct brief.md (MAM Job Restructuring) local job_dir="$REGISTRY_DIR/$JOB_ID" diff --git a/tests/test_issue72_admission.py b/tests/test_issue72_admission.py index a8a395f..0e573c1 100644 --- a/tests/test_issue72_admission.py +++ b/tests/test_issue72_admission.py @@ -146,9 +146,12 @@ def test_delegate_submit_registers_no_job(mam_sandbox, mock_herdr, marker): @pytest.fixture def other_ws(tmp_path): - """A second installed workspace B, outside the sandbox (A).""" + """A second installed workspace B (update.sh's own test: .mam_deploy/remove.sh), + outside the sandbox (A).""" b = Path(os.path.realpath(tmp_path)) / "ws-b" (b / ".mam").mkdir(parents=True) + (b / ".mam_deploy").mkdir() + (b / ".mam_deploy" / "remove.sh").write_text("#!/bin/sh\n") (b / "sub").mkdir() return b @@ -163,27 +166,54 @@ def _no_job(sandbox): return not jobs.exists() or not list(jobs.glob("*.json")) -@pytest.mark.parametrize("binding", ["pane-cwd", "control-plane"]) -def test_delegate_submit_takes_the_target_workspace_marker(mam_sandbox, mock_herdr, marker, other_ws, binding): - """R1: the target row's workspace (the .mam at or above its pane.cwd, else its - control_plane.workspace_root) is checked, not only the caller's.""" +def _server_for(sandbox, name): + """The server the wrapper delivers a rowless target on (resolve_herdr_session).""" + res = subprocess.run(["bash", "-c", 'source "$1" && resolve_herdr_session "$2" "$3"', "_", + str(sandbox / ".agents" / "skills" / "lib.sh"), name, str(sandbox)], + capture_output=True, text=True, cwd=str(sandbox)) + assert res.returncode == 0 and res.stdout.strip(), res.stderr + return res.stdout.strip() + + +def _target_in(kind, sandbox, mock_herdr, b): + """Put the delegate target `tgt` in workspace B as ; the env for submit.""" + if kind == "rowless-live": + _set_mock(mock_herdr, agents={"tgt": _agent(b / "sub", "w1:p6")}) + return {"MAM_SESS": _server_for(sandbox, "tgt")} r = _row("tgt") - if binding == "pane-cwd": - r["pane"]["cwd"] = str(other_ws / "sub") - else: + if kind == "row-pane-cwd": + r["pane"]["cwd"] = str(b / "sub") + elif kind == "row-control-plane": r["pane"]["cwd"] = "/nonexistent/tgt" - r["control_plane"] = {"workspace_root": str(other_ws)} - _seed(mam_sandbox, [r]) + r["control_plane"] = {"workspace_root": str(b)} + elif kind == "row-under-incidental-mam": + (b / "sub" / ".mam").mkdir() # runtime state only: not an installed tree + r["pane"]["cwd"] = str(b / "sub") + r["control_plane"] = {"workspace_root": str(b)} + _seed(sandbox, [r]) + return {} + + +KINDS = ["row-pane-cwd", "row-control-plane", "row-under-incidental-mam", "rowless-live"] + + +@pytest.mark.parametrize("kind", KINDS) +def test_delegate_submit_takes_the_target_workspace_marker(mam_sandbox, mock_herdr, marker, other_ws, kind): + """R1 (rounds 1, 2): every installed workspace the target is in - by its row, + or for a rowless target by the live agent's cwd - is checked, not only the + caller's; an incidental .mam below B does not hide B.""" + env = _target_in(kind, mam_sandbox, mock_herdr, other_ws) marker(True, other_ws) - res = _submit_to(mam_sandbox, "herdr:tgt") + res = _submit_to(mam_sandbox, "herdr:tgt", env=env) assert res.returncode != 0, res.stdout + res.stderr assert "an update of %s is in progress" % other_ws in res.stderr, res.stdout + res.stderr assert "registered job" not in res.stdout assert _no_job(mam_sandbox) -def test_delegate_submit_waits_for_the_target_workspace_admission_lock(mam_sandbox, mock_herdr, other_ws): - _seed(mam_sandbox, [dict(_row("tgt"), pane={"cwd": str(other_ws), "cmd": "claude"})]) +@pytest.mark.parametrize("kind", KINDS) +def test_delegate_submit_waits_for_the_target_workspace_admission_lock(mam_sandbox, mock_herdr, other_ws, kind): + env = _target_in(kind, mam_sandbox, mock_herdr, other_ws) holder = subprocess.Popen([sys.executable, "-c", "import fcntl, sys, time\n" "f = open(sys.argv[1], 'a'); fcntl.flock(f, fcntl.LOCK_EX)\n" "print('locked', flush=True); time.sleep(120)", @@ -191,7 +221,7 @@ def test_delegate_submit_waits_for_the_target_workspace_admission_lock(mam_sandb stdout=subprocess.PIPE, text=True, start_new_session=True) try: assert holder.stdout.readline().strip() == "locked" - res = _submit_to(mam_sandbox, "herdr:tgt", env={"MAM_UPDATE_ADMISSION_WAIT_S": "1"}) + res = _submit_to(mam_sandbox, "herdr:tgt", env=dict(env, MAM_UPDATE_ADMISSION_WAIT_S="1")) finally: holder.kill() holder.wait()