test(issue-57): the audit log follows a moved registry

New file tests/test_issue57_logs_follow_registry.py: the resolver (L1),
registry.py (L2), the wrapper's submit/wait/logs (L3), publish_event.py
(L4), control_plane.from_env, worktree_prepare, status.sh, the comment-5
disk fallback, and the invariant that an explicit DELEGATE_JOB_LOGS_DIR
still wins. Every test ends by asserting the workspace's own
.mam/delegate_job_logs was never created.

Two existing tests gain monkeypatch.delenv("MAM_JOBS_DIR"): conftest's
autouse _isolate_mam_state_dirs gives every test a MAM_JOBS_DIR, and under
the sibling rule the WORKSPACE_ROOT/cwd fallbacks they assert would
otherwise resolve next to that registry. The order they assert (explicit
root, workspace, cwd) is unchanged.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

tests/test_issue15_registry_isolation.py

@@ -57,6 +57,9 @@ def test_registry_dir_falls_back_to_cwd_relative(monkeypatch, tmp_path):
 
 
 def test_logs_dir_follows_the_same_order(monkeypatch, tmp_path):
+    # Issue #57: the audit log follows the registry; conftest's MAM_JOBS_DIR
+    # would otherwise decide the fallback this test asserts.
+    monkeypatch.delenv("MAM_JOBS_DIR", raising=False)
     monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
     monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path))
     assert mqtt_common.get_logs_dir() == str(tmp_path / "logs")

tests/test_issue57_logs_follow_registry.py

@@ -36,6 +36,8 @@ if str(SKILLS) not in sys.path:
 
 from lib_py import control_plane  # noqa: E402
 from lib_py.delegation import job_subscriber, mqtt_common, registry  # noqa: E402
+from test_issue21_phase2_loop_a import _mutate, _prepare, _rows, _skill, _wt  # noqa: E402
+from test_issue21_phase2_loop_a import ctl  # noqa: E402,F401
 
 UNREACHABLE = {"MQTT_BROKER": "127.0.0.1", "MQTT_PORT": "9", "MQTT_TLS": "0"}
 DROP = ("MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR")
@@ -90,3 +92,221 @@ def test_disk_fallback_reads_the_audit_only_terminal_status(moved):
     pending, terminal = {jid2}, {}
     job_subscriber._check_disk_fallback(pending, str(reg), terminal, {jid2: "f" * 32}, None)
     assert terminal == {} and pending == {jid2}
+
+
+# ---------------------------------------------------------------------------
+# L1 — the resolver
+# ---------------------------------------------------------------------------
+
+def test_resolver_pairs_the_logs_with_the_registry(moved, monkeypatch):
+    ws, reg, logs = moved
+    # the usual roots are unchanged
+    assert mqtt_common.get_logs_dir() == str(ws / ".mam" / "delegate_job_logs")
+    assert mqtt_common.get_logs_dir(str(ws / ".mam" / "jobs")) == str(ws / ".mam" / "delegate_job_logs")
+    # a registry the caller names
+    assert mqtt_common.get_logs_dir(str(reg)) == str(logs)
+    # a registry moved by MAM_JOBS_DIR, and the compat alias
+    monkeypatch.setenv("MAM_JOBS_DIR", str(reg))
+    assert mqtt_common.get_logs_dir() == str(logs)
+    assert mqtt_common.LOGS_DIR == str(logs)
+    # an explicit root still wins
+    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(ws / "x"))
+    assert mqtt_common.get_logs_dir() == str(ws / "x")
+    assert mqtt_common.get_logs_dir(str(reg)) == str(ws / "x")
+    monkeypatch.delenv("DELEGATE_JOB_LOGS_DIR")
+    # a relative registry resolves against the cwd, as registry.py opens it
+    assert mqtt_common.get_logs_dir("r/jobs") == os.path.join(os.getcwd(), "r", "delegate_job_logs")
+    # the sibling rule itself reads no environment
+    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(ws / "x"))
+    assert mqtt_common.sibling_logs_dir(str(reg)) == str(logs)
+
+
+# ---------------------------------------------------------------------------
+# L2 — registry.py writes and reads next to --registry-dir
+# ---------------------------------------------------------------------------
+
+def _registry(ws: Path, reg: Path, *args) -> subprocess.CompletedProcess:
+    return subprocess.run([sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg), *args],
+                          capture_output=True, text=True, env=_env(ws), cwd=str(ws), timeout=60)
+
+
+def test_registry_cli_writes_and_reads_next_to_the_registry(moved):
+    ws, reg, logs = moved
+    res = _registry(ws, reg, "register", "--prompt", "p57", "--agent", "human", "--agent-session", "herdr:t57")
+    assert res.returncode == 0, res.stderr
+    jid = res.stdout.strip()
+    assert (logs / jid / "meta.json").is_file() and (logs / jid / "status.json").is_file()
+    assert _registry(ws, reg, "pick", "--agent-session", "herdr:t57").stdout.strip() == jid
+    assert _registry(ws, reg, "update", "--job", jid, "--status", "completed").returncode == 0
+    assert json.loads((logs / jid / "status.json").read_text())["status"] == "completed"
+    assert _registry(ws, reg, "get-feedback", "--job", jid).returncode == 0
+    res = _registry(ws, reg, "logs", jid)
+    assert res.returncode == 0, res.stderr
+    assert "registered" in res.stdout and "status_changed" in res.stdout
+    res = _registry(ws, reg, "logs", "--list")
+    assert res.returncode == 0 and jid in res.stdout, res.stdout + res.stderr
+    events = [json.loads(l) for l in (logs / jid / "events.ndjson").read_text().splitlines() if l.strip()]
+    assert [e["event"] for e in events][:2] == ["registered", "status_changed"]
+
+
+# ---------------------------------------------------------------------------
+# L3 — the wrapper's last line names the directory it wrote
+# ---------------------------------------------------------------------------
+
+def _wrapper(ws: Path, *args, **extra) -> subprocess.CompletedProcess:
+    return subprocess.run(["bash", str(WRAPPER), *args], capture_output=True, text=True,
+                          env=_env(ws, **extra), cwd=str(ws), timeout=120)
+
+
+def _only_job_id(reg: Path) -> str:
+    ids = [p.stem for p in reg.glob("*.json")]
+    assert len(ids) == 1, ids
+    return ids[0]
+
+
+def test_wrapper_output_names_the_dir_it_wrote(moved):
+    ws, reg, logs = moved
+    real_logs = Path(os.path.realpath(logs.parent)) / "delegate_job_logs"
+    # (a) dry-run: registered and seeded, both next to the registry
+    res = _wrapper(ws, "submit", "--dry-run", "--agent", "human", "--registry-dir", str(reg),
+                   "--workdir", str(ws), "--prompt", "p57")
+    assert res.returncode == 0, res.stdout[-800:] + res.stderr[-800:]
+    jid = _only_job_id(reg)
+    assert res.stdout.strip().splitlines()[-1] == str(real_logs / jid)
+    assert (logs / jid / "meta.json").is_file()
+    # (b) a real human job against an unreachable broker: rc 3, same rule
+    res = _wrapper(ws, "submit", "--agent", "human", "--registry-dir", str(reg), "--workdir", str(ws),
+                   "--prompt", "p57b", "--timeout", "3", "--idle-timeout", "2")
+    assert res.returncode == 3, res.stdout[-800:] + res.stderr[-800:]
+    jid2 = [p.stem for p in reg.glob("*.json") if p.stem != jid]
+    assert len(jid2) == 1, jid2
+    jid2 = jid2[0]
+    assert res.stdout.strip().splitlines()[-1] == str(real_logs / jid2)
+    events = [json.loads(l) for l in (logs / jid2 / "events.ndjson").read_text().splitlines() if l.strip()]
+    assert "published" in {e.get("event") for e in events}, events
+    # (c) wait and (d) logs find the same registry and its audit log
+    res = _wrapper(ws, "wait", "--registry-dir", str(reg), "--job", str(jid2), "--timeout", "2")
+    assert res.returncode in (0, 1, 2, 3), res.stdout[-400:] + res.stderr[-400:]
+    res = _wrapper(ws, "logs", "--registry-dir", str(reg), jid2)
+    assert res.returncode == 0 and "registered" in res.stdout, res.stdout + res.stderr
+    res = _wrapper(ws, "logs", "--registry-dir", str(reg), "--list")
+    assert res.returncode == 0 and jid2 in res.stdout, res.stdout + res.stderr
+
+
+# ---------------------------------------------------------------------------
+# L4 — publish_event.py records next to --registry-dir
+# ---------------------------------------------------------------------------
+
+def test_publish_event_records_next_to_the_registry(moved):
+    ws, reg, logs = moved
+    jid = registry.register_job("p4", registry_dir=str(reg), agent_session="herdr:t57")
+    (logs / jid / "events.ndjson").unlink()  # only what publish_event writes counts
+
+    def publish(*args):
+        return subprocess.run([sys.executable, str(PUBLISH_PY), "--registry-dir", str(reg), "--job", jid,
+                               "--attempts", "1", *args],
+                              capture_output=True, text=True, env=_env(ws), cwd=str(ws), timeout=60)
+
+    publish("--event", "progress", "--detail", "x")
+    events = [json.loads(l) for l in (logs / jid / "events.ndjson").read_text().splitlines() if l.strip()]
+    assert [e for e in events if e.get("event") == "published" and e.get("source_event") == "progress"], events
+    publish("--event", "completed", "--detail", "done")
+    assert json.loads((logs / jid / "status.json").read_text())["status"] == "completed"
+
+
+# ---------------------------------------------------------------------------
+# the session binding derives the audit-log root from the registry
+# ---------------------------------------------------------------------------
+
+def test_from_env_derives_the_logs_dir_from_a_moved_registry(moved):
+    ws, reg, logs = moved
+    b = control_plane.from_env({"WORKSPACE_ROOT": str(ws), "MAM_JOBS_DIR": str(reg)})
+    assert b["mam_jobs_dir"] == os.path.realpath(reg)
+    assert b["delegate_job_logs_dir"] == os.path.realpath(logs)
+    # an explicit root is kept as it is
+    b = control_plane.from_env({"WORKSPACE_ROOT": str(ws), "MAM_JOBS_DIR": str(reg),
+                                "DELEGATE_JOB_LOGS_DIR": str(ws / "custom-logs")})
+    assert b["delegate_job_logs_dir"] == os.path.realpath(ws / "custom-logs")
+    # the usual binding is unchanged
+    assert control_plane.from_env({"WORKSPACE_ROOT": str(ws)}) == control_plane.defaults(str(ws))
+    # and the CLI lib.sh calls agrees
+    res = subprocess.run([sys.executable, "-m", "lib_py.control_plane", "json", str(ws / "wt")],
+                         capture_output=True, text=True, env=_env(ws, MAM_JOBS_DIR=str(reg)), cwd=str(ws), timeout=60)
+    assert res.returncode == 0, res.stderr
+    assert json.loads(res.stdout)["delegate_job_logs_dir"] == os.path.realpath(logs)
+
+
+def test_an_explicit_logs_dir_still_wins_over_a_moved_registry(tmp_path, monkeypatch):
+    """Comment 1: DELEGATE_JOB_LOGS_DIR means the audit log was moved to a place
+    that cannot be derived; it wins over --registry-dir as before."""
+    for k in DROP:
+        monkeypatch.delenv(k, raising=False)
+    main, custom, exp = tmp_path / "main", tmp_path / "custom-logs", tmp_path / "exp" / "jobs"
+    (main / ".mam").mkdir(parents=True)
+    monkeypatch.setenv("WORKSPACE_ROOT", str(main))
+    monkeypatch.chdir(main)
+    env = dict(MAM_JOBS_DIR=str(main / ".mam" / "jobs"), DELEGATE_JOB_LOGS_DIR=str(custom))
+    res = _wrapper(main, "submit", "--dry-run", "--agent", "human", "--registry-dir", str(exp),
+                   "--workdir", str(main), "--prompt", "p57", **env)
+    assert res.returncode == 0, res.stdout[-800:] + res.stderr[-800:]
+    jid = _only_job_id(exp)
+    assert res.stdout.strip().splitlines()[-1] == str(custom / jid)
+    assert (custom / jid / "meta.json").is_file()
+    assert not (tmp_path / "exp" / "delegate_job_logs").exists()
+    res = subprocess.run([sys.executable, str(PUBLISH_PY), "--registry-dir", str(exp), "--job", jid,
+                          "--attempts", "1", "--event", "progress", "--detail", "x"],
+                         capture_output=True, text=True, env=_env(main, **env), cwd=str(main), timeout=60)
+    events = [json.loads(l) for l in (custom / jid / "events.ndjson").read_text().splitlines() if l.strip()]
+    assert [e for e in events if e.get("event") == "published"], events
+    assert not (tmp_path / "exp" / "delegate_job_logs").exists()
+    assert not (main / ".mam" / "delegate_job_logs").exists()
+
+
+# ---------------------------------------------------------------------------
+# worktree_prepare keeps the creator's audit log next to the control root's registry
+# ---------------------------------------------------------------------------
+
+def test_worktree_prepare_keeps_the_logs_next_to_the_control_root_registry(ctl, mock_herdr, mock_agents, monkeypatch):
+    main = ctl
+    reg = main / "elsewhere" / "jobs"
+    monkeypatch.setenv("MAM_JOBS_DIR", str(reg))
+    monkeypatch.delenv("DELEGATE_JOB_LOGS_DIR", raising=False)
+    wt = _wt(main)
+    res = _prepare(main, wt, "--creator-agent", "claude", "--onboard")
+    assert res.returncode == 0, res.stdout + res.stderr
+    (row,) = _rows(main)
+    assert row["control_plane"]["mam_jobs_dir"] == os.path.realpath(reg)
+    assert row["control_plane"]["delegate_job_logs_dir"] == os.path.realpath(main / "elsewhere" / "delegate_job_logs")
+    jid = row["delegate_job_id"]
+    assert (main / "elsewhere" / "delegate_job_logs" / jid / "status.json").is_file()
+    assert not (main / ".mam" / "delegate_job_logs").exists()
+
+
+# ---------------------------------------------------------------------------
+# status.sh reads the audit log next to a moved registry
+# ---------------------------------------------------------------------------
+
+@pytest.mark.parametrize("mode", ["json", "text"])
+def test_status_reads_the_audit_log_next_to_a_moved_registry(mam_sandbox, monkeypatch, mode):
+    tmp = mam_sandbox
+    reg, logs = tmp / "elsewhere" / "jobs", tmp / "elsewhere" / "delegate_job_logs"
+    jid = "s57aa001"
+    (logs / jid).mkdir(parents=True)
+    (logs / jid / "status.json").write_text('{"job_id": "s57aa001", "status": "completed"}')
+    reg.mkdir(parents=True)  # the registry exists, but the record was cleaned up
+    _mutate(tmp, f"""
+d['herdr_sessions'] = [{{'name': 's57-creator-claude', 'status': 'running', 'delegate_job_id': {jid!r},
+                         'pane': {{'cwd': {str(tmp)!r}}}}}]
+""")
+    monkeypatch.setenv("MAM_JOBS_DIR", str(reg))
+    monkeypatch.delenv("DELEGATE_JOB_LOGS_DIR", raising=False)
+    args = ["--json"] if mode == "json" else []
+    res = subprocess.run(["bash", _skill(tmp, "multi-agent-mux-status", "scripts", "status.sh"), *args],
+                         capture_output=True, text=True, timeout=120)
+    assert res.returncode == 0, res.stderr
+    if mode == "json":
+        assert json.loads(res.stdout)["sessions_detail"][0]["job_status"] == "completed"
+    else:
+        (line,) = [l for l in res.stdout.splitlines() if l.startswith("s57-creator-claude")]
+        assert jid in line and "completed" in line, line
+    assert not (tmp / ".mam" / "delegate_job_logs").exists()

tests/test_tier1_unit.py

@@ -495,6 +495,9 @@ def test_b9_logs_dir_follows_cwd_changes(mam_sandbox, tmp_path, monkeypatch):
     monkeypatch.delenv("DELEGATE_JOB_LOGS_DIR", raising=False)
     # Issue #15: with WORKSPACE_ROOT set the root is anchored on the workspace
     # instead; the cwd only decides when neither override is present.
+    # Issue #57: the audit log follows the registry, so the sandbox's
+    # MAM_JOBS_DIR must go too for the cwd to decide.
+    monkeypatch.delenv("MAM_JOBS_DIR", raising=False)
     monkeypatch.delenv("WORKSPACE_ROOT", raising=False)
     a = tmp_path / "a"; b = tmp_path / "b"
     a.mkdir(); b.mkdir()