From a159a790dfeea60c3d4d35470bafddbb292a677e Mon Sep 17 00:00:00 2001 From: Godopu Date: Tue, 13 Oct 2026 10:00:25 +0000 Subject: [PATCH] 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 --- .../lib_py/delegation/job_subscriber.py | 8 +- tests/test_issue57_logs_follow_registry.py | 92 +++++++++++++++++++ 2 files changed, 98 insertions(+), 2 deletions(-) create mode 100644 tests/test_issue57_logs_follow_registry.py diff --git a/.agents/skills/lib_py/delegation/job_subscriber.py b/.agents/skills/lib_py/delegation/job_subscriber.py index 7c68d9f..838b646 100755 --- a/.agents/skills/lib_py/delegation/job_subscriber.py +++ b/.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: diff --git a/tests/test_issue57_logs_follow_registry.py b/tests/test_issue57_logs_follow_registry.py new file mode 100644 index 0000000..6830067 --- /dev/null +++ b/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}