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 <noreply@anthropic.com>

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 <kind>; 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()