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 <noreply@anthropic.com>
This commit is contained in:
GodopuandClaude Fable 5.1 committed 2026-10-15 00:02:42 +00:00
1 parent 63d816468e
commit 75c004222c
8 files changed
+143 -58

No files matched your search

+2 -2
View File
@@ -2322,7 +2322,7 @@ delegate_publish_cmd() {
pub_reg="${MAM_JOBS_DIR:-${WORKSPACE_ROOT:+$WORKSPACE_ROOT/.mam/jobs}}" pub_reg="${MAM_JOBS_DIR:-${WORKSPACE_ROOT:+$WORKSPACE_ROOT/.mam/jobs}}"
pub_reg="${pub_reg:-.mam/jobs}" pub_reg="${pub_reg:-.mam/jobs}"
mkdir -p "$pub_reg" && pub_reg="$(cd -P "$pub_reg" && pwd -P)" 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") local -a argv=("$py_bin" "$pub_script" --registry-dir "$pub_reg" --job "$job_id")
[ -n "$tok" ] && argv+=(--expect-dispatch-token "$tok") [ -n "$tok" ] && argv+=(--expect-dispatch-token "$tok")
[ -n "$att" ] && argv+=(--attempt-id "$att") [ -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 # closes it or another dispatch claims it. Lines are parsed as JSON, never
# pattern-matched. rc 0 = recorded, 1 = not (yet). # pattern-matched. rc 0 = recorded, 1 = not (yet).
delegate_logs_dir() { 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() { delegate_started_recorded() {
+3 -3
View File
@@ -25,7 +25,7 @@ import shlex
import sys import sys
from typing import Dict, Optional 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") KEYS = ("workspace_root", "agent_sessions_yaml", "mam_jobs_dir", "delegate_job_logs_dir")
ENV = {"workspace_root": "WORKSPACE_ROOT", "agent_sessions_yaml": "AGENT_SESSIONS_YAML", 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 # 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. # the delegate_job_logs next to the registry in use, not the workspace's.
if not (env.get("DELEGATE_JOB_LOGS_DIR") or "").strip(): 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 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). when it is the sibling of the registry and so derivable there (Issue #57).
An explicit, non-sibling root is forwarded as before.""" An explicit, non-sibling root is forwarded as before."""
keys = list(KEYS) 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") keys.remove("delegate_job_logs_dir")
return keys return keys
@@ -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/<leaf>` 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 ``<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 sibling_logs_dir(registry_dir or default_registry_dir())
@@ -30,9 +30,14 @@ from dataclasses import asdict, dataclass
from pathlib import Path from pathlib import Path
from typing import Any, Callable, Dict, Iterable, List, Optional from typing import Any, Callable, Dict, Iterable, List, Optional
try:
import paho.mqtt.client as mqtt 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 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") logger = logging.getLogger("delegate_job.mqtt_common")
@@ -145,35 +150,10 @@ def load_env_or_report() -> bool:
# Constants # Constants
# -------------------------------------------------------------------------- # --------------------------------------------------------------------------
SCHEMA_VERSION = 1 SCHEMA_VERSION = 1
DEFAULT_REGISTRY_DIR = ".mam/jobs"
DEFAULT_TOPIC_ROOT = "python/mqtt/jobs" DEFAULT_TOPIC_ROOT = "python/mqtt/jobs"
LOCK_FILENAME = ".lock" LOCK_FILENAME = ".lock"
def _workspace_mam_dir(leaf: str) -> Optional[str]:
"""`$WORKSPACE_ROOT/.mam/<leaf>` 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/<job_id>/{meta,events,status}. # Persistent audit-log layout: .mam/delegate_job_logs/<job_id>/{meta,events,status}.
# This is a *separate* artifact from the registry: the registry is the live job # 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 # 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" 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 ``<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 sibling_logs_dir(registry_dir or default_registry_dir())
def __getattr__(name: str): # PEP 562 (3.7+) def __getattr__(name: str): # PEP 562 (3.7+)
"""Keep ``mqtt_common.LOGS_DIR`` working for external consumers """Keep ``mqtt_common.LOGS_DIR`` working for external consumers
(documented in registry.md) while resolving it dynamically.""" (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() config = config or broker_config_from_env()
client_id = f"{config.client_id_prefix}-{role}-{uuid.uuid4().hex[:8]}" 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( client = mqtt.Client(
callback_api_version=mqtt.CallbackAPIVersion.VERSION2, callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
client_id=client_id, client_id=client_id,
@@ -14,7 +14,10 @@ import logging
import time import time
from typing import Any, Optional, Tuple from typing import Any, Optional, Tuple
try:
import paho.mqtt.client as mqtt 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 from lib_py.delegation import mqtt_common
@@ -92,7 +92,7 @@ _abs_registry_dir() {
# #57), pinned into the injected commands so an agent publishing from another # #57), pinned into the injected commands so an agent publishing from another
# cwd does not fall back to its own default. # cwd does not fall back to its own default.
_logs_dir_abs() { _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; # Issue #17 item 4: the wrapper's decisions are lib_py/delegation/wrapper.py;
@@ -168,7 +168,7 @@ def get_job_status(s):
candidates += [os.path.join(r, jid, 'status.json') 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] 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 # 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') 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] for r in (os.environ.get('MAM_JOBS_DIR'), cp.get('mam_jobs_dir')) if r]
candidates += [ candidates += [
@@ -380,7 +380,7 @@ def get_job_status(s):
candidates += [os.path.join(r, jid, 'status.json') 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] 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 # 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') 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] for r in (os.environ.get('MAM_JOBS_DIR'), cp.get('mam_jobs_dir')) if r]
candidates += [ candidates += [
@@ -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 res.stdout.strip().splitlines()[-1] == str(Path(os.path.realpath(main)) / ".mam" / "delegate_job_logs" / jid2)
assert os.listdir(real_logs) == [jid2] assert os.listdir(real_logs) == [jid2]
assert not (wt / ".mam" / "delegate_job_logs").exists() 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