fix(subscriber): disk fallback reads status.json's status, not the dict

read_logged_status() returns the status.json dict; _check_disk_fallback
compared that dict with the terminal status strings, so a terminal status
recorded only in the audit log was never recognised (Issue #57, comment 5).
The dispatch-token and attempt checks are unchanged.

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

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

@@ -92,10 +92,14 @@ def _check_disk_fallback(
         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(registry_dir))
+                logged = mqtt_common.read_logged_status(jid, mqtt_common.get_logs_dir(registry_dir))
             except Exception:
-                pass
+                logged = None
+            if logged:
+                disk_status = logged.get("status")
 
         if disk_status in ("completed", "error", "cancelled"):
             if expected_dispatch_tokens:

tests/test_issue57_logs_follow_registry.py

@@ -0,0 +1,92 @@
+#!/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 job_subscriber, mqtt_common, registry  # noqa: E402
+
+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}