diff --git a/.agents/skills/lib.sh b/.agents/skills/lib.sh index c65686a..90d08ff 100644 --- a/.agents/skills/lib.sh +++ b/.agents/skills/lib.sh @@ -3264,26 +3264,27 @@ mam_control_plane_prefix() { # fd 7 (`7>&-` on such a child, as pane_lock's fd 9). The wait is bounded by # MAM_UPDATE_ADMISSION_WAIT_S (default 120). # -# mam_update_admit [workspace]: rc 0 admitted (fd 7 held, unless the workspace -# has no .mam), 1 refused (stderr). +# 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. mam_update_admit() { - local root="${1:-$WORKSPACE_ROOT}" + local root="${1:-$WORKSPACE_ROOT}" fd="${2:-7}" local lock="$root/.mam/update-admission.lock" marker="$root/.mam/update-in-progress" st # No .mam: not an installed workspace, which is all the update runs in (and a # checkout running its own scripts gets no .mam from this). [ -d "$root/.mam" ] || return 0 - if ! exec 7>>"$lock"; then + if ! eval "exec $fd>>\"\$lock\""; then echo "ERROR: cannot open the update admission lock $lock" >&2 return 1 fi - case "$(PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" python3 -m lib_py.pane_lock acquire --fd 7 --shared --wait-env MAM_UPDATE_ADMISSION_WAIT_S)" in + case "$(PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" python3 -m lib_py.pane_lock acquire --fd "$fd" --shared --wait-env MAM_UPDATE_ADMISSION_WAIT_S)" in locked) ;; timeout) - exec 7>&- + mam_update_admit_release "$fd" echo "ERROR: an update of $root holds its admission lock (MAM_UPDATE_ADMISSION_WAIT_S passed); try again when it has finished" >&2 return 1 ;; *) - exec 7>&- + mam_update_admit_release "$fd" echo "ERROR: the update admission lock $lock could not be decided (lib_py.pane_lock)" >&2 return 1 ;; esac @@ -3296,11 +3297,11 @@ mam_update_admit() { echo "NOTE: stale update marker $marker (its update is gone); ignored" >&2 return 0 ;; esac - exec 7>&- + mam_update_admit_release "$fd" echo "ERROR: an update of $root is in progress (marker $marker, holder $st); try again when it has finished" >&2 return 1 } mam_update_admit_release() { - exec 7>&- + eval "exec ${1:-7}>&-" } diff --git a/.agents/skills/lib_py/update.py b/.agents/skills/lib_py/update.py index b9372e4..9d326c8 100644 --- a/.agents/skills/lib_py/update.py +++ b/.agents/skills/lib_py/update.py @@ -16,6 +16,11 @@ 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 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 @@ -231,6 +236,31 @@ def precheck(workspace: str, state: dict, herdr: str, env: Mapping[str, str]) -> return blockers +# --------------------------------------------------------------------------- delegate target + +def target_workspace(workspace: str, state: dict, agent_session: str) -> str: + if not agent_session.startswith("herdr:"): + return "" + name = agent_session[len("herdr:"):] + row = next((s for s in state.get("herdr_sessions") or [] + if isinstance(s, dict) and s.get("name") == name), None) + 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: + root = (row.get("control_plane") or {}).get("workspace_root") or "" + found = _real(root) if root else "" + return "" if found == _real(workspace) else found + + # --------------------------------------------------------------------------- monitor def monitor_state(workspace: str) -> dict: @@ -288,6 +318,9 @@ 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.add_argument("--workspace", required=True) + s.add_argument("--agent-session", required=True) s = sub.add_parser("backup-registry") s.add_argument("--workspace", required=True) s.add_argument("--dest", required=True) @@ -303,6 +336,13 @@ 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": + try: + state = _state() + except ValueError: + state = {} + print(target_workspace(ws, state, a.agent_session)) + return 0 if a.cmd == "monitor": print(json.dumps(monitor_state(ws))) return 0 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 6223140..1b1e488 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 @@ -301,10 +301,19 @@ cmd_submit() { local reg_dry=() if [[ "$DRY_RUN" == "1" ]]; then reg_dry=(--dry-run); fi - # Issue #72: no job is registered while an update of the target's workspace - # runs; held until the record is written (lib.sh mam_update_admit). A - # dry-run record is never dispatched and is not counted by the update. - if [[ "$DRY_RUN" != "1" ]]; then mam_update_admit "$WORKSPACE_ROOT" || exit 1; 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 + # mam_update_admit). A dry-run record is never dispatched and is not + # counted by the update. + 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; } + mam_update_admit "$WORKSPACE_ROOT" || exit 1 + if [[ -n "$target_ws" ]]; then mam_update_admit "$target_ws" 8 || exit 1; fi + fi # 1) register job (prints the new job id) JOB_ID="$("$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" register \ "${reg_prompt[@]}" --agent "$AGENT" --agent-session "$AGENT_SESSION" --role "$DELEGATE_ROLE" \ @@ -314,7 +323,7 @@ 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 + [[ "$DRY_RUN" == "1" ]] || { mam_update_admit_release; mam_update_admit_release 8; } # 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 7df3e44..a8a395f 100644 --- a/tests/test_issue72_admission.py +++ b/tests/test_issue72_admission.py @@ -12,6 +12,7 @@ import hashlib import json import os import subprocess +import sys from pathlib import Path import pytest @@ -26,8 +27,8 @@ ORC_OLD = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee" def marker(mam_sandbox): procs = [] - def make(live: bool): - p = mam_sandbox / ".mam" / "update-in-progress" + def make(live: bool, root: Path = None): + p = (root or mam_sandbox) / ".mam" / "update-in-progress" if live: proc = subprocess.Popen(["sleep", "120"], start_new_session=True) procs.append(proc) @@ -143,6 +144,62 @@ def test_delegate_submit_registers_no_job(mam_sandbox, mock_herdr, marker): assert not list(jobs.glob("*.json")) if jobs.exists() else True +@pytest.fixture +def other_ws(tmp_path): + """A second installed workspace B, outside the sandbox (A).""" + b = Path(os.path.realpath(tmp_path)) / "ws-b" + (b / ".mam").mkdir(parents=True) + (b / "sub").mkdir() + return b + + +def _submit_to(sandbox, session, env=None): + return _run(sandbox, ["bash", _script(sandbox, "multi-agent-mux-delegate-job/multi-agent-mux-delegate-job"), + "submit", "--agent", "claude-code", "--agent-session", session, "--prompt", "hi"], env=env) + + +def _no_job(sandbox): + jobs = sandbox / ".mam" / "jobs" + 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.""" + r = _row("tgt") + if binding == "pane-cwd": + r["pane"]["cwd"] = str(other_ws / "sub") + else: + r["pane"]["cwd"] = "/nonexistent/tgt" + r["control_plane"] = {"workspace_root": str(other_ws)} + _seed(mam_sandbox, [r]) + marker(True, other_ws) + res = _submit_to(mam_sandbox, "herdr:tgt") + 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"})]) + 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)", + str(other_ws / ".mam" / "update-admission.lock")], + 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"}) + finally: + holder.kill() + holder.wait() + assert res.returncode != 0, res.stdout + res.stderr + assert "an update of %s holds its admission lock" % other_ws in res.stderr, res.stdout + res.stderr + assert _no_job(mam_sandbox) + + @pytest.mark.parametrize("purge", [False, True]) def test_stop_refuses_and_changes_nothing(mam_sandbox, mock_herdr, marker, purge): _seed(mam_sandbox, [_row("gated-stop")])