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>
This commit is contained in:
GodopuandClaude Fable 5.1 committed 2026-10-13 10:00:25 +00:00
1 parent cbd87734c2
commit a159a790df
2 files changed
+98 -2

No files matched your search

@@ -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:
@@ -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}