fix(delegate): submit also takes the target workspace's update admission (Issue #72, review R1)

Decision 8: delegate submit uses the admission lock of the workspace the
target session belongs to. The wrapper resolved only its own
WORKSPACE_ROOT, so a job could be registered and delivered to an agent of
another installed workspace while that workspace was being updated.

lib_py.update target-workspace resolves a herdr:<name> row's workspace (the
.mam at or above its pane.cwd, else its control_plane.workspace_root);
mam_update_admit takes an fd, and submit holds the caller root on fd 7 and
the target on fd 8 until the record is written. Rowless and remote targets
keep the caller-root gate.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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")])