From 75c004222cbbde2d89ae17540031c3465d24bc61 Mon Sep 17 00:00:00 2001 From: Godopu Date: Thu, 15 Oct 2026 00:02:42 +0000 Subject: [PATCH] fix(delegation): the path rules live in a stdlib-only module; the package imports without paho (Issue #57, Mac F1) control_plane and status.sh imported sibling_logs_dir from mqtt_common, which imports paho at module level, so create/resume's binding and status died on an interpreter without paho-mqtt (_delegate_py_bin's last fallback is the PATH's python3). The registry/audit-log path rules move to lib_py/delegation/logroot.py (standard library only) and mqtt_common re-exports them. The package's own import chain also reached paho through service -> publish_event -> publisher, so mqtt_common and publisher import paho tolerantly and make_client() refuses with a clear error when a broker is actually needed. control_plane, status.sh, lib.sh's delegate_publish_cmd and delegate_logs_dir, and the wrapper's _logs_dir_abs import from logroot. Co-Authored-By: Claude Fable 5.1 --- .agents/skills/lib.sh | 4 +- .agents/skills/lib_py/control_plane.py | 6 +- .agents/skills/lib_py/delegation/logroot.py | 63 +++++++++++++++++++ .../skills/lib_py/delegation/mqtt_common.py | 60 +++--------------- .agents/skills/lib_py/delegation/publisher.py | 5 +- .../multi-agent-mux-delegate-job | 2 +- .../multi-agent-mux-status/scripts/status.sh | 4 +- tests/test_issue57_logs_follow_registry.py | 61 ++++++++++++++++++ 8 files changed, 145 insertions(+), 60 deletions(-) create mode 100644 .agents/skills/lib_py/delegation/logroot.py diff --git a/.agents/skills/lib.sh b/.agents/skills/lib.sh index c8e3ab0..327d157 100644 --- a/.agents/skills/lib.sh +++ b/.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, 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)" + pub_logs="$(PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$py_bin" -c 'import os, sys; from lib_py.delegation.logroot 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") @@ -2373,7 +2373,7 @@ except Exception: # 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 + PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -c 'import os, sys; from lib_py.delegation.logroot import get_logs_dir; print(os.path.abspath(get_logs_dir(sys.argv[1])))' "$1" 2>/dev/null } delegate_started_recorded() { diff --git a/.agents/skills/lib_py/control_plane.py b/.agents/skills/lib_py/control_plane.py index 959b3d2..01b139e 100644 --- a/.agents/skills/lib_py/control_plane.py +++ b/.agents/skills/lib_py/control_plane.py @@ -25,7 +25,7 @@ import shlex import sys from typing import Dict, Optional -from lib_py.delegation import mqtt_common +from lib_py.delegation.logroot import sibling_logs_dir KEYS = ("workspace_root", "agent_sessions_yaml", "mam_jobs_dir", "delegate_job_logs_dir") ENV = {"workspace_root": "WORKSPACE_ROOT", "agent_sessions_yaml": "AGENT_SESSIONS_YAML", @@ -54,7 +54,7 @@ def from_env(env=None) -> Dict[str, str]: # 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"])) + b["delegate_job_logs_dir"] = _abs(sibling_logs_dir(b["mam_jobs_dir"])) return b @@ -79,7 +79,7 @@ def forwarded_keys(desc: dict) -> list: when it is the sibling of the registry and so derivable there (Issue #57). An explicit, non-sibling root is forwarded as before.""" keys = list(KEYS) - if desc.get("delegate_job_logs_dir") == _abs(mqtt_common.sibling_logs_dir(desc["mam_jobs_dir"])): + if desc.get("delegate_job_logs_dir") == _abs(sibling_logs_dir(desc["mam_jobs_dir"])): keys.remove("delegate_job_logs_dir") return keys diff --git a/.agents/skills/lib_py/delegation/logroot.py b/.agents/skills/lib_py/delegation/logroot.py new file mode 100644 index 0000000..b6d6492 --- /dev/null +++ b/.agents/skills/lib_py/delegation/logroot.py @@ -0,0 +1,63 @@ +"""Where the job registry and its audit log live (Issue #15, Issue #57). + +Standard library only, on purpose: create/resume's control-plane binding, +status.sh and lib.sh resolve these paths through ``_delegate_py_bin``, whose +last fallback is the PATH's ``python3``, which may not have paho-mqtt. +``mqtt_common`` re-exports every name here, so existing callers are unchanged. +""" +from __future__ import annotations + +import os +from typing import Optional + +DEFAULT_REGISTRY_DIR = ".mam/jobs" + + +def _workspace_mam_dir(leaf: str) -> Optional[str]: + """`$WORKSPACE_ROOT/.mam/` when lib.sh has told us which workspace we + serve; None otherwise.""" + root = os.environ.get("WORKSPACE_ROOT") + if root and root.strip() and os.path.isdir(root): + return os.path.join(root, ".mam", leaf) + return None + + +def default_registry_dir() -> str: + """Registry root, resolved at call time (Issue #15). + + ``MAM_JOBS_DIR`` wins; otherwise the workspace lib.sh exported as + ``WORKSPACE_ROOT``; otherwise the historical cwd-relative ``.mam/jobs``. + Anchoring on the workspace is what keeps a test sandbox (or any caller + whose cwd is not the workspace) from registering jobs into whichever + ``.mam/jobs`` happens to sit under the current directory - the pytest + suite used to leave hundreds of onboarding records in the real registry. + """ + env = os.environ.get("MAM_JOBS_DIR") + if env and env.strip(): + return env + return _workspace_mam_dir("jobs") or DEFAULT_REGISTRY_DIR + + +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). + + ``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 ``/.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 sibling_logs_dir(registry_dir or default_registry_dir()) diff --git a/.agents/skills/lib_py/delegation/mqtt_common.py b/.agents/skills/lib_py/delegation/mqtt_common.py index 67f1ec7..d2cc427 100644 --- a/.agents/skills/lib_py/delegation/mqtt_common.py +++ b/.agents/skills/lib_py/delegation/mqtt_common.py @@ -30,9 +30,14 @@ from dataclasses import asdict, dataclass from pathlib import Path from typing import Any, Callable, Dict, Iterable, List, Optional -import paho.mqtt.client as mqtt +try: + import paho.mqtt.client as mqtt +except ImportError: # the registry and the path rules work without a broker library (Issue #57) + mqtt = None # make_client() says so when a broker is actually needed from lib_py.delegation import envfile +from lib_py.delegation.logroot import ( # noqa: F401 (re-exported: this module is the documented home) + DEFAULT_REGISTRY_DIR, _workspace_mam_dir, default_registry_dir, get_logs_dir, sibling_logs_dir) logger = logging.getLogger("delegate_job.mqtt_common") @@ -145,35 +150,10 @@ def load_env_or_report() -> bool: # Constants # -------------------------------------------------------------------------- SCHEMA_VERSION = 1 -DEFAULT_REGISTRY_DIR = ".mam/jobs" DEFAULT_TOPIC_ROOT = "python/mqtt/jobs" LOCK_FILENAME = ".lock" -def _workspace_mam_dir(leaf: str) -> Optional[str]: - """`$WORKSPACE_ROOT/.mam/` when lib.sh has told us which workspace we - serve; None otherwise.""" - root = os.environ.get("WORKSPACE_ROOT") - if root and root.strip() and os.path.isdir(root): - return os.path.join(root, ".mam", leaf) - return None - - -def default_registry_dir() -> str: - """Registry root, resolved at call time (Issue #15). - - ``MAM_JOBS_DIR`` wins; otherwise the workspace lib.sh exported as - ``WORKSPACE_ROOT``; otherwise the historical cwd-relative ``.mam/jobs``. - Anchoring on the workspace is what keeps a test sandbox (or any caller - whose cwd is not the workspace) from registering jobs into whichever - ``.mam/jobs`` happens to sit under the current directory - the pytest - suite used to leave hundreds of onboarding records in the real registry. - """ - env = os.environ.get("MAM_JOBS_DIR") - if env and env.strip(): - return env - return _workspace_mam_dir("jobs") or DEFAULT_REGISTRY_DIR - # Persistent audit-log layout: .mam/delegate_job_logs//{meta,events,status}. # This is a *separate* artifact from the registry: the registry is the live job # record (mutated in place), the audit log is an append-only history that @@ -183,31 +163,6 @@ EVENTS_FILENAME = "events.ndjson" STATUS_FILENAME = "status.json" -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). - - ``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 ``/.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 sibling_logs_dir(registry_dir or default_registry_dir()) - - def __getattr__(name: str): # PEP 562 (3.7+) """Keep ``mqtt_common.LOGS_DIR`` working for external consumers (documented in registry.md) while resolving it dynamically.""" @@ -328,6 +283,9 @@ def make_client(role: str, config: Optional[BrokerConfig] = None) -> mqtt.Client """ config = config or broker_config_from_env() client_id = f"{config.client_id_prefix}-{role}-{uuid.uuid4().hex[:8]}" + if mqtt is None: + raise ImportError("paho-mqtt is required to talk to a broker; install it into the virtualenv " + "(see multi-agent-mux-delegate-job/requirements.txt)") client = mqtt.Client( callback_api_version=mqtt.CallbackAPIVersion.VERSION2, client_id=client_id, diff --git a/.agents/skills/lib_py/delegation/publisher.py b/.agents/skills/lib_py/delegation/publisher.py index a921fe1..b8a4669 100644 --- a/.agents/skills/lib_py/delegation/publisher.py +++ b/.agents/skills/lib_py/delegation/publisher.py @@ -14,7 +14,10 @@ import logging import time from typing import Any, Optional, Tuple -import paho.mqtt.client as mqtt +try: + import paho.mqtt.client as mqtt +except ImportError: # importable without a broker library; mqtt_common.make_client() refuses to connect + mqtt = None from lib_py.delegation import mqtt_common diff --git a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job index 6fc8075..9d0b882 100755 --- a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job +++ b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job @@ -92,7 +92,7 @@ _abs_registry_dir() { # #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(sys.argv[2])))' "$SCRIPT_DIR/.." "$REGISTRY_DIR" + "$PY" -c 'import os, sys; sys.path.insert(0, sys.argv[1]); from lib_py.delegation.logroot 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; diff --git a/.agents/skills/multi-agent-mux-status/scripts/status.sh b/.agents/skills/multi-agent-mux-status/scripts/status.sh index f291b9c..249f906 100755 --- a/.agents/skills/multi-agent-mux-status/scripts/status.sh +++ b/.agents/skills/multi-agent-mux-status/scripts/status.sh @@ -168,7 +168,7 @@ def get_job_status(s): 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 + from lib_py.delegation.logroot 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 += [ @@ -380,7 +380,7 @@ def get_job_status(s): 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 + from lib_py.delegation.logroot 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 += [ diff --git a/tests/test_issue57_logs_follow_registry.py b/tests/test_issue57_logs_follow_registry.py index 36af5c3..c7bc625 100644 --- a/tests/test_issue57_logs_follow_registry.py +++ b/tests/test_issue57_logs_follow_registry.py @@ -472,3 +472,64 @@ def test_a_forwarded_pane_moves_its_logs_with_the_registry(tmp_path, monkeypatch assert res.stdout.strip().splitlines()[-1] == str(Path(os.path.realpath(main)) / ".mam" / "delegate_job_logs" / jid2) assert os.listdir(real_logs) == [jid2] assert not (wt / ".mam" / "delegate_job_logs").exists() + + +# --------------------------------------------------------------------------- +# Mac review F1: the path helpers must not need paho +# --------------------------------------------------------------------------- + +@pytest.fixture +def nopaho(tmp_path): + """A `paho` package that fails to import, first on PYTHONPATH: the real + import path then behaves like an interpreter without paho-mqtt.""" + d = tmp_path / "nopaho" + (d / "paho").mkdir(parents=True) + (d / "paho" / "__init__.py").write_text("raise ModuleNotFoundError(\"No module named 'paho'\")\n") + probe = subprocess.run([sys.executable, "-c", "import paho.mqtt.client"], capture_output=True, text=True, + env={**os.environ, "PYTHONPATH": f"{d}:{SKILLS}"}, timeout=60) + assert probe.returncode != 0 and "No module named 'paho'" in probe.stderr, probe.stderr + return d + + +def test_control_plane_binding_works_without_paho(moved, nopaho): + """_delegate_py_bin's last fallback is the PATH's python3, which may lack + paho: create/resume's binding and lib.sh's publish-command prefix must still + resolve the audit-log root.""" + ws, reg, logs = moved + env = _env(ws, MAM_JOBS_DIR=str(reg), PYTHONPATH=f"{nopaho}:{SKILLS}", AGENT_PYTHON_BIN=sys.executable) + res = subprocess.run([sys.executable, "-m", "lib_py.control_plane", "json", str(ws / "wt")], + capture_output=True, text=True, env=env, cwd=str(ws), timeout=60) + assert res.returncode == 0, res.stderr + desc = json.loads(res.stdout) + assert desc["delegate_job_logs_dir"] == os.path.realpath(logs) + res = subprocess.run([sys.executable, "-m", "lib_py.control_plane", "prefix", json.dumps(desc)], + capture_output=True, text=True, env=env, cwd=str(ws), timeout=60) + assert res.returncode == 0 and "MAM_JOBS_DIR=" in res.stdout and "DELEGATE_JOB_LOGS_DIR=" not in res.stdout, res.stdout + # lib.sh's injected publish command still pins the root (it resolved it with the same interpreter) + res = subprocess.run(["bash", "-c", f"source '{SKILLS}/lib.sh' && delegate_publish_cmd j57"], + capture_output=True, text=True, env=env, cwd=str(ws), timeout=60) + assert res.returncode == 0, res.stderr + assert res.stdout.startswith(f"DELEGATE_JOB_LOGS_DIR={os.path.realpath(logs)} "), res.stdout + res.stderr + + +@pytest.mark.parametrize("mode", ["json", "text"]) +def test_status_works_without_paho(mam_sandbox, nopaho, monkeypatch, mode): + tmp = mam_sandbox + reg = tmp / ".mam" / "jobs" + jid = registry.register_job("s57 nopaho", registry_dir=str(reg), agent_session="herdr:s57-nopaho") + registry.update_status(jid, str(reg), "completed") + _mutate(tmp, f""" +d['herdr_sessions'] = [{{'name': 's57-creator-claude', 'status': 'running', 'delegate_job_id': {jid!r}, + 'pane': {{'cwd': {str(tmp)!r}}}}}] +""") + monkeypatch.setenv("PYTHONPATH", str(nopaho)) + monkeypatch.setenv("AGENT_PYTHON_BIN", sys.executable) + 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.stdout[-500:] + res.stderr[-800:] + 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 "completed" in line, line