fix(issue-57): review nits - a damaged audit sequence or an unresolvable log root never aborts the check

_check_disk_fallback converted transition_seq outside its try, so a
status.json whose sequence is not an integer would raise out of the
subscriber's main loop; such a status is now ignored with a warning, like a
stale one. delegate_started_recorded returned 1 before looking at the
registry's events.log when the audit-log root could not be resolved; it now
skips only the audit check.

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

.agents/skills/lib.sh

@@ -2322,7 +2322,7 @@ delegate_publish_cmd() {
   pub_reg="${MAM_JOBS_DIR:-${WORKSPACE_ROOT:+$WORKSPACE_ROOT/.mam/jobs}}"
   pub_reg="${pub_reg:-.mam/jobs}"
   mkdir -p "$pub_reg" && pub_reg="$(cd -P "$pub_reg" && pwd -P)"
-  pub_logs="$(PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$py_bin" -c 'import os; from lib_py.delegation.mqtt_common import get_logs_dir; print(os.path.abspath(get_logs_dir()))' 2>/dev/null || true)"
+  pub_logs="$(PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$py_bin" -c 'import os, sys; from lib_py.delegation.mqtt_common import get_logs_dir; print(os.path.abspath(get_logs_dir(sys.argv[1])))' "$pub_reg" 2>/dev/null || true)"
   local -a argv=("$py_bin" "$pub_script" --registry-dir "$pub_reg" --job "$job_id")
   [ -n "$tok" ] && argv+=(--expect-dispatch-token "$tok")
   [ -n "$att" ] && argv+=(--attempt-id "$att")
@@ -2372,12 +2372,21 @@ except Exception:
 # Never the registry status: a record leaves `pending` when another writer
 # closes it or another dispatch claims it. Lines are parsed as JSON, never
 # pattern-matched. rc 0 = recorded, 1 = not (yet).
+delegate_logs_dir() {
+  PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -c 'import os, sys; from lib_py.delegation.mqtt_common import get_logs_dir; print(os.path.abspath(get_logs_dir(sys.argv[1])))' "$1" 2>/dev/null
+}
+
 delegate_started_recorded() {
   local job_id="$1" tok="${2:-}" logs_dir="${3:-}"
   local reg_dir="${MAM_JOBS_DIR:-${WORKSPACE_ROOT:-$PWD}/.mam/jobs}"
-  [ -n "$logs_dir" ] || logs_dir="${DELEGATE_JOB_LOGS_DIR:-${WORKSPACE_ROOT:-$PWD}/.mam/delegate_job_logs}"
+  # Issue #57: the audit log is resolved by the one rule (next to the registry
+  # unless DELEGATE_JOB_LOGS_DIR says otherwise); no bash copy of the default.
+  # If that resolution fails, only the audit check is skipped: the registry's
+  # events.log is still evidence.
+  [ -n "$logs_dir" ] || logs_dir="$(delegate_logs_dir "$reg_dir")" || logs_dir=""
   local ev_log="$reg_dir/$job_id.events.log"
-  local audit="$logs_dir/$job_id/events.ndjson"
+  local audit=""
+  [ -n "$logs_dir" ] && audit="$logs_dir/$job_id/events.ndjson"
   [ -f "$ev_log" ] || [ -f "$audit" ] || return 1
   EV_LOG="$ev_log" EV_AUDIT="$audit" EV_TOK="$tok" python3 -c '
 import hashlib, json, os, sys
@@ -2431,10 +2440,12 @@ delegate_wait_started() {
   local job_id="$1" ack_s="${2:-180}" tok="${3:-}"
   [ -n "$job_id" ] || return 1
   ack_s=$(mam_int "$ack_s" 0) || { echo "WARNING: onboarding ack timeout ${2-} ignored (not a decimal integer >= 0); using 180" >&2; ack_s=180; }
-  local deadline
+  local deadline logs_dir
   deadline=$(( $(date +%s) + ack_s ))
+  # Issue #57: resolve the audit-log root once, not once per 1 s poll.
+  logs_dir="$(delegate_logs_dir "${MAM_JOBS_DIR:-${WORKSPACE_ROOT:-$PWD}/.mam/jobs}")" || logs_dir=""
   while :; do
-    if delegate_started_recorded "$job_id" "$tok"; then
+    if delegate_started_recorded "$job_id" "$tok" "$logs_dir"; then
       return 0
     fi
     if [ "$(date +%s)" -ge "$deadline" ]; then

.agents/skills/lib_py/control_plane.py

@@ -21,6 +21,8 @@ import shlex
 import sys
 from typing import Dict, Optional
 
+from lib_py.delegation import mqtt_common
+
 KEYS = ("workspace_root", "agent_sessions_yaml", "mam_jobs_dir", "delegate_job_logs_dir")
 ENV = {"workspace_root": "WORKSPACE_ROOT", "agent_sessions_yaml": "AGENT_SESSIONS_YAML",
        "mam_jobs_dir": "MAM_JOBS_DIR", "delegate_job_logs_dir": "DELEGATE_JOB_LOGS_DIR"}
@@ -45,6 +47,10 @@ def from_env(env=None) -> Dict[str, str]:
         v = (env.get(var) or "").strip()
         if v:
             b[k] = _abs(v)
+    # Issue #57: without an explicit DELEGATE_JOB_LOGS_DIR the audit log is
+    # the delegate_job_logs next to the registry in use, not the workspace's.
+    if not (env.get("DELEGATE_JOB_LOGS_DIR") or "").strip():
+        b["delegate_job_logs_dir"] = _abs(mqtt_common.sibling_logs_dir(b["mam_jobs_dir"]))
     return b
 
 

.agents/skills/lib_py/delegation/engine.py

@@ -374,13 +374,14 @@ def _flush(notifications, registry_dir) -> None:
     Best effort by design: the registry record is already durable, and an audit
     write must never be able to fail a transition that has committed.
     """
+    logs_dir = mqtt_common.get_logs_dir(registry_dir)
     for note in notifications:
         job_id = note["job_id"]
         kind = note["kind"]
         try:
             if kind == "status_changed":
                 mqtt_common.update_logged_status(
-                    job_id, note["to"], updated_at=note.get("timestamp"),
+                    job_id, note["to"], logs_dir=logs_dir, updated_at=note.get("timestamp"),
                     transition_seq=note.get("transition_seq"))
                 mqtt_common.append_event(job_id, {
                     "event": "status_changed",
@@ -389,7 +390,7 @@ def _flush(notifications, registry_dir) -> None:
                     "by": note.get("by"),
                     "timestamp": note.get("timestamp"),
                     "transition_seq": note.get("transition_seq"),
-                })
+                }, logs_dir=logs_dir)
             elif kind == "conflict_refused":
                 mqtt_common.append_event(job_id, {
                     "event": "conflict_refused",
@@ -398,6 +399,6 @@ def _flush(notifications, registry_dir) -> None:
                     "reason": note.get("reason"),
                     "by": note.get("by"),
                     "timestamp": note.get("timestamp"),
-                })
+                }, logs_dir=logs_dir)
         except Exception as exc:  # pragma: no cover - best effort
             logger.warning("audit mirror failed for %s: %s", job_id, exc)

.agents/skills/lib_py/delegation/job_subscriber.py

@@ -84,18 +84,41 @@ def _check_disk_fallback(
         disk_status = None
         rec_token = None
         rec_attempt = None
+        rec_seq = None
         try:
             job_rec = load_job(jid, registry_dir)
             disk_status = job_rec.get("status")
             rec_token = job_rec.get("dispatch_token")
             rec_attempt = job_rec.get("attempt_id", job_rec.get("iteration"))
+            rec_seq = job_rec.get("transition_seq")
         except Exception:
             job_rec = {}
         if not disk_status or disk_status not in ("completed", "error", "cancelled"):
+            # read_logged_status returns the status.json dict; the status is its
+            # "status" field (Issue #57, comment 5: comparing the dict matched nothing).
             try:
-                disk_status = mqtt_common.read_logged_status(jid, mqtt_common.get_logs_dir())
+                logged = mqtt_common.read_logged_status(jid, mqtt_common.get_logs_dir(registry_dir))
             except Exception:
-                pass
+                logged = None
+            if logged:
+                # The audit mirror is written after the registry commits, so a
+                # record that is not terminal may sit next to the previous
+                # attempt's terminal status. A versioned record (the engine's
+                # transition_seq) takes the audit status only when the audit is
+                # at least as new; an audit status without a sequence cannot
+                # prove which attempt it describes, so a versioned record
+                # ignores it. An unversioned record (never transitioned, or
+                # MAM_ENGINE_CAS=0) and a missing record keep the old rule.
+                aud_seq = logged.get("transition_seq")
+                try:
+                    stale = rec_seq is not None and (aud_seq is None or int(aud_seq) < int(rec_seq))
+                except (TypeError, ValueError):
+                    stale = True  # a damaged status.json proves nothing either
+                if stale:
+                    logger.warning("disk fallback ignored audit status %r for %s: transition_seq %r is older "
+                                   "than the record's %r", logged.get("status"), jid, aud_seq, rec_seq)
+                else:
+                    disk_status = logged.get("status")
 
         if disk_status in ("completed", "error", "cancelled"):
             if expected_dispatch_tokens:
@@ -130,8 +153,10 @@ class _Watcher:
 
     def __init__(self, expected_job_ids: Set[str], expected_tokens: Dict[str, Optional[str]], expected_seqs: Dict[str, int],
                  expected_dispatch_tokens: Optional[Dict[str, Optional[str]]] = None,
-                 expected_attempts: Optional[Dict[str, Optional[int]]] = None):
+                 expected_attempts: Optional[Dict[str, Optional[int]]] = None,
+                 logs_dir: Optional[str] = None):
         self.events: "queue.Queue[Tuple[str, Dict[str, Any]]]" = queue.Queue()
+        self.logs_dir = logs_dir  # Issue #57: the audit log next to the registry being watched
         self.expected = set(expected_job_ids)
         self.tokens = expected_tokens  # job_id -> expected auth_token (or None)
         self.last_seq = dict(expected_seqs)
@@ -163,7 +188,7 @@ class _Watcher:
             "seq": got,
             "via": via,
             "topic": topic,
-        })
+        }, logs_dir=self.logs_dir)
 
     def _on_state(self, msg, payload: Dict[str, Any], jid: str) -> None:
         """A retained `<prefix>/state` message (Issue #18): learn the job's
@@ -196,7 +221,7 @@ class _Watcher:
             "topic": msg.topic,
             "timestamp": payload.get("timestamp"),
             "detail": "status={} last_event={}".format(payload.get("status"), payload.get("last_event")),
-        })
+        }, logs_dir=self.logs_dir)
         self.events.put((msg.topic, dict(payload, _renews_liveness=renews)))
 
     def on_message(self, _client, _userdata, msg) -> None:
@@ -317,7 +342,7 @@ class _Watcher:
                                      if expected_disp_tok else None),
             "attempt_id": (data.get("attempt_id") if isinstance(data, dict) and data.get("attempt_id") is not None
                            else self.attempts.get(jid)),
-        })
+        }, logs_dir=self.logs_dir)
         self.events.put((msg.topic, payload))
 
 
@@ -375,7 +400,8 @@ def _run_subscriber(args) -> int:
     if args.attempt_id is not None and args.job:
         attempts[args.job] = args.attempt_id
 
-    watcher = _Watcher(expected_ids, tokens, seqs, dispatch_tokens, attempts)
+    logs_dir = mqtt_common.get_logs_dir(args.registry_dir)
+    watcher = _Watcher(expected_ids, tokens, seqs, dispatch_tokens, attempts, logs_dir=logs_dir)
 
     # Resolve timeouts from CLI, falling back to the (first) job's settings.
     base_job = jobs[0]
@@ -418,7 +444,8 @@ def _run_subscriber(args) -> int:
             # what makes the retained state message arrive again at once.
             logger.warning("broker reconnected (connect #%d); re-subscribing", connects["count"])
             for jid in expected_ids:
-                mqtt_common.append_event(jid, {"event": "reconnected", "connect_count": connects["count"]})
+                mqtt_common.append_event(jid, {"event": "reconnected", "connect_count": connects["count"]},
+                                         logs_dir=logs_dir)
         for topic in subscribed_topics + state_topics:
             _c.subscribe(topic, qos=1)
             logger.info("subscribed to %s", topic)

.agents/skills/lib_py/delegation/mqtt_common.py

@@ -183,20 +183,29 @@ EVENTS_FILENAME = "events.ndjson"
 STATUS_FILENAME = "status.json"
 
 
-def get_logs_dir() -> str:
+def sibling_logs_dir(registry_dir: str) -> str:
+    """The audit-log root that pairs with ``registry_dir``: its sibling
+    ``delegate_job_logs``. Pure; reads no environment. Issue #57."""
+    return os.path.join(os.path.dirname(os.path.abspath(registry_dir)), "delegate_job_logs")
+
+
+def get_logs_dir(registry_dir: Optional[str] = None) -> str:
     """Audit-log root, resolved at call time (B-9).
 
-    Overridable with ``DELEGATE_JOB_LOGS_DIR``; otherwise
-    ``$WORKSPACE_ROOT/.mam/delegate_job_logs`` (Issue #15, see
-    ``default_registry_dir``); otherwise ``<cwd>/.mam/delegate_job_logs``.
-    Resolved per call rather than at import so a chdir after import cannot
-    strand the audit trail in the old tree — the same reason
-    ``DEFAULT_REGISTRY_DIR`` stays a relative string.
+    ``DELEGATE_JOB_LOGS_DIR`` when set: it means the audit log was moved to a
+    place that cannot be derived from the registry. Otherwise the audit log
+    follows the registry (Issue #57): the ``delegate_job_logs`` next to
+    ``registry_dir``, or next to ``default_registry_dir()`` when the caller
+    does not know its registry. The default registry is
+    ``$WORKSPACE_ROOT/.mam/jobs`` (Issue #15), otherwise ``<cwd>/.mam/jobs``,
+    so the usual roots are unchanged. Resolved per call rather than at import
+    so a chdir after import cannot strand the audit trail in the old tree —
+    the same reason ``DEFAULT_REGISTRY_DIR`` stays a relative string.
     """
     env = os.environ.get("DELEGATE_JOB_LOGS_DIR")
     if env and env.strip():
         return env
-    return _workspace_mam_dir("delegate_job_logs") or os.path.join(os.getcwd(), ".mam", "delegate_job_logs")
+    return sibling_logs_dir(registry_dir or default_registry_dir())
 
 
 def __getattr__(name: str):          # PEP 562 (3.7+)
@@ -516,14 +525,15 @@ def _update_job_status_legacy(job_id: str, registry_dir: str = DEFAULT_REGISTRY_
         _atomic_write_record(job_id, registry_dir, record)
         if "status" in fields:
             new_status = record.get("status")
-            update_logged_status(job_id, new_status, updated_at=record["updated_at"])
+            logs_dir = get_logs_dir(registry_dir)
+            update_logged_status(job_id, new_status, logs_dir=logs_dir, updated_at=record["updated_at"])
             if old_status != new_status:
                 append_event(job_id, {
                     "event": "status_changed",
                     "from": old_status,
                     "to": new_status,
                     "timestamp": record["updated_at"],
-                })
+                }, logs_dir=logs_dir)
     return record
 
 

.agents/skills/lib_py/delegation/publish_event.py

@@ -241,7 +241,7 @@ def record_outcome(prepared: PreparedEvent, registry_dir: str,
         "payload": prepared.payload,
         "published": publish_ok,
         "publish_error": publish_error,
-    })
+    }, logs_dir=mqtt_common.get_logs_dir(registry_dir))
 
     if not local_event_recorded:
         registry.append_event(prepared.job_id, registry_dir, prepared.payload)

.agents/skills/lib_py/delegation/registry.py

@@ -175,7 +175,7 @@ def register_job(
         _atomic_write_record(job_id, registry_dir, record)
     # Seed the persistent audit log (meta.json + status.json + a "registered"
     # event). Best-effort inside init_job_log — never blocks registration.
-    mqtt_common.init_job_log(job_id, meta=record)
+    mqtt_common.init_job_log(job_id, meta=record, logs_dir=mqtt_common.get_logs_dir(registry_dir))
     logger.info("registered job %s (agent=%s session=%s)", job_id, agent, agent_session)
     return job_id
 
@@ -233,14 +233,15 @@ def _pick_pending_legacy(agent_session: str, registry_dir: str = DEFAULT_REGISTR
         updated_at = chosen["updated_at"]
     # Mirrors the pending->running transition into the audit log outside the
     # lock. This is the shape the engine now generalises to every transition.
-    mqtt_common.update_logged_status(job_id, "running", updated_at=updated_at)
+    logs_dir = mqtt_common.get_logs_dir(registry_dir)
+    mqtt_common.update_logged_status(job_id, "running", logs_dir=logs_dir, updated_at=updated_at)
     mqtt_common.append_event(job_id, {
         "event": "status_changed",
         "from": "pending",
         "to": "running",
         "by": agent_session,
         "timestamp": updated_at,
-    })
+    }, logs_dir=logs_dir)
     return job_id
 
 
@@ -328,7 +329,7 @@ def get_feedback(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR,
     # 1) Try the unified audit log first (ndjson) since it's written synchronously by the subscriber
     try:
         from lib_py.delegation import mqtt_common
-        logs_dir = mqtt_common.get_logs_dir()
+        logs_dir = mqtt_common.get_logs_dir(registry_dir)
         logged = list(mqtt_common.iter_logged_events(job_id, logs_dir))
         for e in reversed(logged):
             if e.get("source_event") in events:
@@ -450,7 +451,7 @@ def _build_parser() -> argparse.ArgumentParser:
                         help="summarise every job under the logs dir instead")
     p_logs.add_argument("--logs-dir", default=None,
                         help="override the audit-log root (default: $DELEGATE_JOB_LOGS_DIR "
-                             "or <cwd>/.mam/delegate_job_logs)")
+                             "or the delegate_job_logs next to --registry-dir)")
     p_logs.add_argument("--tail", type=int, default=0,
                         help="show only the last N events (0 = all)")
     p_logs.add_argument("--json", action="store_true",
@@ -698,7 +699,7 @@ def main(argv: Optional[List[str]] = None) -> int:
 
 def _cmd_logs(args) -> int:
     """Pretty-print one job's events.ndjson, or summarise all logged jobs."""
-    logs_dir = args.logs_dir or mqtt_common.get_logs_dir()
+    logs_dir = args.logs_dir or mqtt_common.get_logs_dir(args.registry_dir)
 
     if args.list_all:
         jobs = mqtt_common.list_logged_jobs(logs_dir)

.agents/skills/lib_py/delegation/remote.py

@@ -136,7 +136,8 @@ def publish_dispatch(job_id: str, registry_dir: str, session: str, instructions:
                            exceptions=(OSError, TimeoutError, ConnectionError, ValueError))(
         broker_config_from_job(job), topic, body, True)
     mqtt_common.append_event(job_id, {"event": "dispatched", "topic": topic, "session": session,
-                                      "attempt_id": payload["attempt_id"]})
+                                      "attempt_id": payload["attempt_id"]},
+                             logs_dir=mqtt_common.get_logs_dir(registry_dir))
     logger.info("dispatched job %s to remote session %s on %s", job_id, session, topic)
     return topic
 

.agents/skills/lib_py/loop/runner.py

@@ -26,6 +26,7 @@ from pathlib import Path
 from typing import Any, Dict, List, Optional, Set, Tuple
 
 from lib_py import ints
+from lib_py.delegation import mqtt_common
 from lib_py.delegation import registry as job_registry
 from lib_py.loop import checkpoint, eager, parser
 
@@ -1083,7 +1084,7 @@ class LoopRunner:
         if not head:
             return None
         root = self.worktree_root.resolve()
-        logs = Path(os.environ.get("DELEGATE_JOB_LOGS_DIR") or self.mam_dir / "delegate_job_logs")
+        logs = Path(mqtt_common.get_logs_dir(str(self.jobs_dir)))
         exclude = []
         for d in (self.mam_dir, self.jobs_dir, self.workspace_root / logs):
             try:

.agents/skills/multi-agent-mux-create/scripts/worktree_prepare.sh

@@ -234,8 +234,10 @@ if [ -n "$CREATOR_AGENT" ]; then
   env_add=(WORKSPACE_ROOT="$MAIN"
            AGENT_SESSIONS_YAML="${AGENT_SESSIONS_YAML:-$MAIN/.mam/agent-sessions.yaml}"
            MAM_JOBS_DIR="${MAM_JOBS_DIR:-$MAIN/.mam/jobs}"
-           DELEGATE_JOB_LOGS_DIR="${DELEGATE_JOB_LOGS_DIR:-$MAIN/.mam/delegate_job_logs}"
            MAM_NO_SUBSCRIBER_DAEMON=1)
+  # Issue #57: an explicit audit-log root is passed on; otherwise create's
+  # binding derives it from MAM_JOBS_DIR (the delegate_job_logs next to it).
+  if [ -n "${DELEGATE_JOB_LOGS_DIR:-}" ]; then env_add+=(DELEGATE_JOB_LOGS_DIR="$DELEGATE_JOB_LOGS_DIR"); fi
   if [ -z "${MAM_ENV_FILE:-}" ] && [ -f "$MAIN/.mam.env" ]; then env_add+=(MAM_ENV_FILE="$MAIN/.mam.env"); fi
   rc=0
   env "${env_add[@]}" bash "$SKILL_DIR/multi-agent-mux-create/scripts/create_session.sh" "${args[@]}" || rc=$?

.agents/skills/multi-agent-mux-delegate-job/SKILL.md

@@ -134,10 +134,11 @@ Which tree wins, by execution context:
 `python -m`: for `-m` the current directory precedes `PYTHONPATH`, so a frozen
 loop whose working directory happened to contain a `lib_py` package would have
 silently executed live code. The working directory is left untouched; the
-registry and audit-log paths resolve to `MAM_JOBS_DIR` / `DELEGATE_JOB_LOGS_DIR`
-when set, otherwise to `$WORKSPACE_ROOT/.mam/jobs` and
-`$WORKSPACE_ROOT/.mam/delegate_job_logs` (the workspace `lib.sh` exports), and
-only without either to the cwd-relative `.mam/...` (Issue #15 — this is what
+registry path resolves to `MAM_JOBS_DIR` when set, otherwise to
+`$WORKSPACE_ROOT/.mam/jobs` (the workspace `lib.sh` exports), and only without
+either to the cwd-relative `.mam/jobs`; the audit log is the
+`delegate_job_logs` next to the registry in use unless `DELEGATE_JOB_LOGS_DIR`
+names another root (Issue #57, Issue #15 — this is what
 stops a caller whose cwd is not the workspace, such as the pytest suite, from
 registering jobs into a foreign registry).
 

.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job

@@ -87,11 +87,12 @@ _abs_registry_dir() {
   export SKS_REGISTRY_DIR="$REGISTRY_DIR"
 }
 
-# The audit-log root this wrapper's own subscriber resolves (same resolver),
-# pinned into the injected commands so an agent publishing from another cwd
-# does not fall back to <cwd>/.mam/delegate_job_logs.
+# The audit-log root this wrapper's own subscriber resolves (same resolver:
+# next to $REGISTRY_DIR unless DELEGATE_JOB_LOGS_DIR says otherwise, Issue
+# #57), pinned into the injected commands so an agent publishing from another
+# cwd does not fall back to its own default.
 _logs_dir_abs() {
-  "$PY" -c 'import os, sys; sys.path.insert(0, sys.argv[1]); from lib_py.delegation.mqtt_common import get_logs_dir; print(os.path.abspath(get_logs_dir()))' "$SCRIPT_DIR/.."
+  "$PY" -c 'import os, sys; sys.path.insert(0, sys.argv[1]); from lib_py.delegation.mqtt_common import get_logs_dir; print(os.path.abspath(get_logs_dir(sys.argv[2])))' "$SCRIPT_DIR/.." "$REGISTRY_DIR"
 }
 
 # Issue #17 item 4: the wrapper's decisions are lib_py/delegation/wrapper.py;
@@ -184,7 +185,7 @@ multi-agent-mux-delegate-job <command> [options]
   list    [--registry-dir <dir>]
   verify  --job <id> --validate <script> [--registry-dir <dir>]
   wait    [--job <id>] [--timeout <sec>] [--registry-dir <dir>]
-  logs    <job_id> | --list      # persistent audit log (delegate_job_logs/)
+  logs    <job_id> | --list [--registry-dir <dir>]   # audit log next to the registry (or $DELEGATE_JOB_LOGS_DIR)
 EOF
 }
 
@@ -199,6 +200,7 @@ TIMEOUT=$(mam_int_knob MAM_JOB_TIMEOUT_S 3600 1)
 # Issue #18 liveness contract (N / K / T). An unset --idle-timeout derives K x N.
 HEARTBEAT_SEC=$(mam_int_knob MAM_HEARTBEAT_SEC 30 1); HEARTBEAT_MISSES=$(mam_int_knob MAM_HEARTBEAT_MISSES 4 1); ACK_TIMEOUT=$(mam_int_knob MAM_ACK_TIMEOUT_S 180 1)
 JOB_ID=""; REGISTRY_DIR="$REGISTRY_DIR_DEFAULT"; DELEGATE_ROLE="Worker"
+LOGS_DIR_ABS=""
 TYPE="direct"; REVIEWER="hermes"; REVIEWER_SESSION="herdr:hermes"; MAX_ITERATIONS=5
 DEFAULT_COUNTERPART_ROLE="Reviewer"
 COUNTERPART_ROLE="$DEFAULT_COUNTERPART_ROLE"
@@ -276,7 +278,10 @@ cmd_submit() {
   [[ "$PY" == */* && "$PY" != /* ]] && PY="$(pwd)/$PY"
   cd "$WORKDIR"
   _abs_registry_dir
-  LOGS_DIR_ABS="$(_logs_dir_abs 2>/dev/null || true)"
+  LOGS_DIR_ABS="$(_logs_dir_abs)" || { echo "ERROR: cannot resolve the audit-log root for $REGISTRY_DIR" >&2; exit 1; }
+  # Issue #57: every child (subscriber, registry.py, validate scripts) sees
+  # the same root this wrapper prints and pins into the injected commands.
+  export DELEGATE_JOB_LOGS_DIR="$LOGS_DIR_ABS"
 
   # Issue #34: an explicit --parent-job fails closed in the registry; the
   # $MAM_JOB_ID default may name a job of another host's registry (a remote
@@ -345,16 +350,14 @@ EOF
 
     if [[ "$DRY_RUN" == "1" ]]; then
       run_agent "$JOB_ID" "$instructions"
-      local logs_root_dry="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
-      echo "$logs_root_dry/$JOB_ID"
+      echo "$LOGS_DIR_ABS/$JOB_ID"
       return 0
     fi
 
     # 1) Preflight (a remote session has no local pane to check: Issue #18)
     if ! _is_remote "$AGENT_SESSION" && ! preflight_agent "$JOB_ID" "$AGENT_SESSION" 0; then
       finalize_unfinished "$JOB_ID" "error" "$SUB_RC_ERROR" "preflight"
-      local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
-      echo "$logs_root/$JOB_ID"
+      echo "$LOGS_DIR_ABS/$JOB_ID"
       exit 5
     fi
 
@@ -368,8 +371,7 @@ EOF
       export _SKS_IDLE_RC="$idle_rc"
       if [ $idle_rc -eq 1 ] || [ $idle_rc -eq 2 ]; then
         finalize_unfinished "$JOB_ID" "error" "$idle_rc" "idle_wait"
-        local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
-        echo "$logs_root/$JOB_ID"
+        echo "$LOGS_DIR_ABS/$JOB_ID"
         exit 5
       fi
     fi
@@ -409,8 +411,7 @@ EOF
           _decided mapped_st sub-status "$sub_exit" --job "$JOB_ID" --registry-dir "$REGISTRY_DIR"
           finalize_unfinished "$JOB_ID" "$mapped_st" "$sub_exit" "subscriber"
           mapped_st="${FINAL_JOB_STATUS:-$mapped_st}"
-          local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
-          echo "$logs_root/$JOB_ID"
+          echo "$LOGS_DIR_ABS/$JOB_ID"
           _exit_for "$mapped_st"
         fi
       fi
@@ -424,8 +425,7 @@ EOF
     if ! deliver_prompt "$JOB_ID" "$instructions" "$AGENT_SESSION" "$logf"; then
       reap_subscriber "$sub_pid"
       finalize_unfinished "$JOB_ID" "error" "$SUB_RC_ERROR" "inject"
-      local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
-      echo "$logs_root/$JOB_ID"
+      echo "$LOGS_DIR_ABS/$JOB_ID"
       exit 1
     fi
     start_stall_watch "$AGENT_SESSION" "$AGENT" "$_disp_tok" "$_attempt_id"
@@ -454,8 +454,7 @@ EOF
 
     # Last stdout line: the persistent audit-log dir for this job (see SKILL.md
     # "Audit Logs"). Callers can scrape `tail -n1` to find it.
-    local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
-    echo "$logs_root/$JOB_ID"
+    echo "$LOGS_DIR_ABS/$JOB_ID"
     _exit_for "$job_status"
   else
     # Implement loop/discuss orchestrator
@@ -470,8 +469,7 @@ EOF
     if [[ "$DRY_RUN" == "1" ]]; then
       echo "[dry-run] orchestrator loop would start for job: $JOB_ID type: $TYPE"
       echo "worker session: $AGENT_SESSION, reviewer session: $REVIEWER_SESSION"
-      local logs_root_dry="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
-      echo "$logs_root_dry/$JOB_ID"
+      echo "$LOGS_DIR_ABS/$JOB_ID"
       return 0
     fi
 
@@ -517,8 +515,7 @@ EOF
       # Preflight (skipped for a remote session: Issue #18)
       if ! _is_remote "$current_session" && ! preflight_agent "$JOB_ID" "$current_session" "$force_warn_only"; then
         finalize_unfinished "$JOB_ID" "error" "$SUB_RC_ERROR" "preflight"
-        local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
-        echo "$logs_root/$JOB_ID"
+        echo "$LOGS_DIR_ABS/$JOB_ID"
         exit 5
       fi
 
@@ -550,8 +547,7 @@ EOF
         export _SKS_IDLE_RC="$idle_rc"
         if [ $idle_rc -eq 1 ] || [ $idle_rc -eq 2 ]; then
           finalize_unfinished "$JOB_ID" "error" "$idle_rc" "idle_wait"
-          local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
-          echo "$logs_root/$JOB_ID"
+          echo "$LOGS_DIR_ABS/$JOB_ID"
           exit 5
         fi
       fi
@@ -587,8 +583,7 @@ EOF
             _decided mapped_st sub-status "$sub_exit" --job "$JOB_ID" --registry-dir "$REGISTRY_DIR"
             finalize_unfinished "$JOB_ID" "$mapped_st" "$sub_exit" "subscriber"
             mapped_st="${FINAL_JOB_STATUS:-$mapped_st}"
-            local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
-            echo "$logs_root/$JOB_ID"
+            echo "$LOGS_DIR_ABS/$JOB_ID"
             _exit_for "$mapped_st"
           fi
         fi
@@ -609,8 +604,7 @@ EOF
       if ! deliver_prompt "$JOB_ID" "$instructions" "$current_session" "$logf"; then
         reap_subscriber "$sub_pid"
         finalize_unfinished "$JOB_ID" "error" "$SUB_RC_ERROR" "inject"
-        local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
-        echo "$logs_root/$JOB_ID"
+        echo "$LOGS_DIR_ABS/$JOB_ID"
         exit 1
       fi
       start_stall_watch "$current_session" "$cur_agent" "$_disp_tok" "$attempt"
@@ -639,8 +633,7 @@ EOF
             echo "Worker completed during grace period. Continuing to reviewer."
             feedback="$("$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" get-feedback --job "$JOB_ID")"
           else
-            local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
-            echo "$logs_root/$JOB_ID"
+            echo "$LOGS_DIR_ABS/$JOB_ID"
             _exit_for "$job_status"
           fi
         fi
@@ -664,8 +657,7 @@ EOF
             echo "Reviewer completed during grace period. Continuing evaluation."
             feedback="$("$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" get-feedback --job "$JOB_ID")"
           else
-            local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
-            echo "$logs_root/$JOB_ID"
+            echo "$LOGS_DIR_ABS/$JOB_ID"
             _exit_for "$job_status"
           fi
         fi
@@ -720,8 +712,7 @@ EOF
     fi
 
     # Last stdout line: the persistent audit-log dir
-    local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
-    echo "$logs_root/$JOB_ID"
+    echo "$LOGS_DIR_ABS/$JOB_ID"
     exit "$final_loop_rc"
   fi
 }
@@ -868,14 +859,12 @@ _registry_wait() {
     clear) return 0 ;;
     superseded)
       echo "ERROR: job $job_id was redispatched, moved or closed by someone else while it waited for session '$sess'; this attempt is not dispatched and the record is left as it is" >&2
-      local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
-      echo "$logs_root/$job_id"
+      echo "$LOGS_DIR_ABS/$job_id"
       exit 5 ;;
     busy*)
       echo "ERROR: session '$sess' still has open jobs in the registry (${res#busy }); job $job_id is not dispatched" >&2
       finalize_unfinished "$job_id" "error" "$SUB_RC_ERROR" "registry_wait" "$sess"
-      local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
-      echo "$logs_root/$job_id"
+      echo "$LOGS_DIR_ABS/$job_id"
       exit 5 ;;
   esac
   _decision_failed wait-open-jobs "answer '$res'"
@@ -1099,16 +1088,24 @@ cmd_verify() {
 }
 
 cmd_logs() {
-  # logs <job_id> | logs --list — delegates to registry.py's logs CLI, which
-  # reads the persistent audit log under $DELEGATE_JOB_LOGS_DIR (or
-  # <cwd>/delegate_job_logs). Run from your project dir so the default resolves.
+  # logs <job_id> | logs --list [--registry-dir <dir>] — delegates to
+  # registry.py's logs CLI, which reads the audit log next to the registry
+  # (Issue #57; $DELEGATE_JOB_LOGS_DIR overrides). parse_opts is for submit's
+  # globals, so the few options are read here.
   PY="$(pick_python)"
-  if [[ "${1:-}" == "--list" ]]; then
-    "$PY" "$SCRIPT_DIR/scripts/registry.py" logs --list
+  local reg="$REGISTRY_DIR_DEFAULT" jid="" list=0
+  while [[ $# -gt 0 ]]; do
+    case "$1" in
+      --registry-dir) reg="$2"; shift 2;;
+      --list) list=1; shift;;
+      *) jid="$1"; shift;;
+    esac
+  done
+  if [[ "$list" == "1" ]]; then
+    "$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$reg" logs --list
   else
-    local jid="${1:-}"
     [[ -n "$jid" ]] || { echo "logs requires <job_id> or --list" >&2; exit 1; }
-    "$PY" "$SCRIPT_DIR/scripts/registry.py" logs "$jid"
+    "$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$reg" logs "$jid"
   fi
 }
 

.agents/skills/multi-agent-mux-delegate-job/registry.md

@@ -150,7 +150,8 @@ that session.
 
 Separate from the registry, every job is also mirrored to a durable append-only
 audit log at `.mam/delegate_job_logs/<job_id>/` (override with
-`DELEGATE_JOB_LOGS_DIR`, default `<cwd>/.mam/delegate_job_logs`). The registry
+`DELEGATE_JOB_LOGS_DIR`, an audit-log root that is **not** next to the
+registry; default: the `delegate_job_logs` next to the registry in use). The registry
 is **live state** mutated in place; the audit log is **history** that survives
 even after the registry dir is cleaned up. It is git-ignored.
 

.agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh

@@ -272,7 +272,7 @@ def on_message(_client, _userdata, msg):
             "topic": msg.topic,
             "timestamp": payload.get("timestamp"),
             "detail": payload.get("detail", ""),
-        })
+        }, logs_dir=mqtt_common.get_logs_dir(registry_dir))
         
         print(f"MQTT Monitor: recorded event {event} for job {jid} (seq={seq})", flush=True)
         

.agents/skills/multi-agent-mux-status/scripts/status.sh

@@ -167,6 +167,10 @@ def get_job_status(s):
                   for r in (os.environ.get('MAM_JOBS_DIR'), cp.get('mam_jobs_dir')) if r]
     candidates += [os.path.join(r, jid, 'status.json')
                    for r in (os.environ.get('DELEGATE_JOB_LOGS_DIR'), cp.get('delegate_job_logs_dir')) if r]
+    # Issue #57: the audit log lives next to the registry unless moved explicitly
+    from lib_py.delegation.mqtt_common import sibling_logs_dir
+    candidates += [os.path.join(sibling_logs_dir(r), jid, 'status.json')
+                   for r in (os.environ.get('MAM_JOBS_DIR'), cp.get('mam_jobs_dir')) if r]
     candidates += [
         os.path.join('.mam', 'jobs', f"{jid}.json"),
         os.path.join(project_root, '.mam', 'jobs', f"{jid}.json"),
@@ -375,6 +379,10 @@ def get_job_status(s):
                   for r in (os.environ.get('MAM_JOBS_DIR'), cp.get('mam_jobs_dir')) if r]
     candidates += [os.path.join(r, jid, 'status.json')
                    for r in (os.environ.get('DELEGATE_JOB_LOGS_DIR'), cp.get('delegate_job_logs_dir')) if r]
+    # Issue #57: the audit log lives next to the registry unless moved explicitly
+    from lib_py.delegation.mqtt_common import sibling_logs_dir
+    candidates += [os.path.join(sibling_logs_dir(r), jid, 'status.json')
+                   for r in (os.environ.get('MAM_JOBS_DIR'), cp.get('mam_jobs_dir')) if r]
     candidates += [
         os.path.join('.mam', 'jobs', f"{jid}.json"),
         os.path.join(project_root, '.mam', 'jobs', f"{jid}.json"),

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

@@ -0,0 +1,382 @@
+#!/usr/bin/env python3
+"""Issue #57 — the audit log follows the registry.
+
+`--registry-dir` / `MAM_JOBS_DIR` moved the registry, but the audit log kept
+its own root (`DELEGATE_JOB_LOGS_DIR`, else the workspace's), so an isolated
+experiment left audit-only directories in the real control plane. The rule is
+now one function, `mqtt_common.get_logs_dir(registry_dir)`: an explicit
+`DELEGATE_JOB_LOGS_DIR` wins, otherwise the audit log is the
+`delegate_job_logs` next to the registry in use. Every test ends by asserting
+that the workspace's own `.mam/delegate_job_logs` was never created.
+
+Comment 5 of the issue: `job_subscriber._check_disk_fallback` compared the
+dict `read_logged_status()` returns with status strings, so an audit-only
+terminal status was never recognised.
+"""
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+import _mam_state_guard  # noqa: F401
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+SKILLS = REPO_ROOT / ".agents" / "skills"
+WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
+REGISTRY_PY = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "registry.py"
+PUBLISH_PY = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.py"
+
+if str(SKILLS) not in sys.path:
+    sys.path.insert(0, str(SKILLS))
+
+from lib_py import control_plane  # noqa: E402
+from lib_py.delegation import engine, 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")
+
+
+@pytest.fixture
+def moved(tmp_path, monkeypatch):
+    """A workspace whose registry lives elsewhere, with neither override set:
+    conftest's autouse defaults would otherwise hide the fallback (comment 7)."""
+    for k in DROP:
+        monkeypatch.delenv(k, raising=False)
+    ws = tmp_path / "ws"
+    (ws / ".mam").mkdir(parents=True)
+    reg = tmp_path / "elsewhere" / "jobs"
+    monkeypatch.setenv("WORKSPACE_ROOT", str(ws))
+    monkeypatch.chdir(ws)
+    yield ws, reg, tmp_path / "elsewhere" / "delegate_job_logs"
+    assert not (ws / ".mam" / "delegate_job_logs").exists(), "the workspace's audit-log root was created"
+
+
+def _env(ws: Path, **extra) -> dict:
+    env = {k: v for k, v in os.environ.items() if k not in DROP}
+    env.update(WORKSPACE_ROOT=str(ws), PYTHONPATH=str(SKILLS), DELEGATE_JOB_PYTHON=sys.executable,
+               AGENT_PYTHON_BIN=sys.executable, MAM_ENV_FILE=str(ws / ".mam.env.absent"), **UNREACHABLE)
+    env.update({k: str(v) for k, v in extra.items()})
+    return env
+
+
+# ---------------------------------------------------------------------------
+# comment 5: the disk fallback reads status.json's status, not the dict
+# ---------------------------------------------------------------------------
+
+def test_disk_fallback_reads_the_audit_only_terminal_status(moved):
+    ws, reg, logs = moved
+    reg.mkdir(parents=True)
+    # no registry record at all: only the audit log says the job ended
+    jid = "a57fb001"
+    mqtt_common.update_logged_status(jid, "completed", logs_dir=mqtt_common.get_logs_dir(str(reg)))
+    pending, terminal = {jid}, {}
+    job_subscriber._check_disk_fallback(pending, str(reg), terminal, None, None)
+    assert terminal == {jid: "completed"}
+    assert pending == set()
+
+    # a pending record with a dispatch token, an audit-only `error`
+    tok = "0123456789abcdef0123456789abcdef"
+    jid2 = registry.register_job("fb2", registry_dir=str(reg), agent_session="herdr:t57", dispatch_token=tok)
+    mqtt_common.update_logged_status(jid2, "error", logs_dir=mqtt_common.get_logs_dir(str(reg)))
+    pending, terminal = {jid2}, {}
+    job_subscriber._check_disk_fallback(pending, str(reg), terminal, {jid2: tok}, None)
+    assert terminal == {jid2: "error"} and pending == set()
+    # a different expected token still ignores it (#14/#18 rule, unchanged)
+    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()
+
+
+# ---------------------------------------------------------------------------
+# review (codex P2): conflict refusals and legacy claims follow the registry too
+# ---------------------------------------------------------------------------
+
+@pytest.mark.parametrize("path", ["conflict", "legacy-claim"])
+def test_conflict_refusals_and_legacy_claims_follow_the_registry(moved, monkeypatch, path):
+    ws, reg, logs = moved
+    jid = registry.register_job("p2", registry_dir=str(reg), agent_session="herdr:t57")
+    if path == "conflict":
+        assert engine.transition(jid, "completed", registry_dir=str(reg), source="operator").committed
+        assert not engine.transition(jid, "error", registry_dir=str(reg), source="late").committed
+        expected = "conflict_refused"
+    else:
+        monkeypatch.setenv("MAM_ENGINE_CAS", "0")
+        assert registry.pick_pending("herdr:t57", registry_dir=str(reg)) == jid
+        expected = "status_changed"
+    assert expected in [e.get("event") for e in mqtt_common.iter_logged_events(jid, str(logs))]
+
+
+# ---------------------------------------------------------------------------
+# review (codex P1): an old audit terminal must not close a redispatched attempt
+# ---------------------------------------------------------------------------
+
+def test_old_audit_terminal_does_not_complete_the_redispatched_attempt(moved, monkeypatch):
+    ws, reg, logs = moved
+    jid = registry.register_job("p1", registry_dir=str(reg), agent_session="herdr:t57")
+    assert engine.transition(jid, "completed", registry_dir=str(reg), source="operator").committed
+    old_audit = mqtt_common.read_logged_status(jid, str(logs))
+    flush, observed = engine._flush, {}
+
+    def before_the_mirror(notes, registry_dir):
+        # the real gap: attempt 2 is committed, the audit still says attempt 1 completed
+        rec = mqtt_common.load_job(jid, registry_dir)
+        assert rec["status"] == "pending" and rec["attempt_id"] == 2
+        assert old_audit["transition_seq"] < rec["transition_seq"]
+        pending, terminal = {jid}, {}
+        job_subscriber._check_disk_fallback(pending, registry_dir, terminal, {jid: rec["dispatch_token"]}, {jid: 2})
+        observed.update(pending=pending, terminal=terminal)
+        flush(notes, registry_dir)
+
+    monkeypatch.setattr(engine, "_flush", before_the_mirror)
+    assert engine.transition(jid, "redispatch", registry_dir=str(reg), source="operator",
+                             fields={"attempt_id": 2}).committed
+    assert observed == {"pending": {jid}, "terminal": {}}, "the old completion was attributed to attempt 2"
+    rec = mqtt_common.load_job(jid, str(reg))
+    # a failed mirror leaves the old status.json behind (update_logged_status
+    # itself refuses to go backwards, so the file is written directly): still ignored
+    mqtt_common.job_log_path(jid, mqtt_common.STATUS_FILENAME, str(logs)).write_text(json.dumps(old_audit))
+    assert mqtt_common.read_logged_status(jid, str(logs))["status"] == "completed"
+    pending, terminal = {jid}, {}
+    job_subscriber._check_disk_fallback(pending, str(reg), terminal, {jid: rec["dispatch_token"]}, {jid: 2})
+    assert pending == {jid} and terminal == {}
+    # an audit status without a sequence cannot prove it is about this attempt
+    mqtt_common.update_logged_status(jid, "completed", logs_dir=str(logs))
+    pending, terminal = {jid}, {}
+    job_subscriber._check_disk_fallback(pending, str(reg), terminal, {jid: rec["dispatch_token"]}, {jid: 2})
+    assert pending == {jid} and terminal == {}
+    # a damaged sequence is ignored too, and does not take the subscriber down
+    mqtt_common.job_log_path(jid, mqtt_common.STATUS_FILENAME, str(logs)).write_text(
+        json.dumps({"job_id": jid, "status": "completed", "transition_seq": "n/a"}))
+    pending, terminal = {jid}, {}
+    job_subscriber._check_disk_fallback(pending, str(reg), terminal, {jid: rec["dispatch_token"]}, {jid: 2})
+    assert pending == {jid} and terminal == {}
+    # a newer audit-only terminal is still taken
+    mqtt_common.update_logged_status(jid, "completed", logs_dir=str(logs), transition_seq=rec["transition_seq"] + 1)
+    pending, terminal = {jid}, {}
+    job_subscriber._check_disk_fallback(pending, str(reg), terminal, {jid: rec["dispatch_token"]}, {jid: 2})
+    assert pending == set() and terminal == {jid: "completed"}

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