fix(issue-57): the audit log follows the registry; the disk fallback reads the audit status #58

Merged
Godopu merged 16 commits from fix/issue-57 into main 2026-10-15 13:52:10 +00:00
8 changed files with 143 additions and 58 deletions
Showing only changes of commit 75c004222c - Show all commits

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="${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() {
+3 -3
View File
@@ -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
@@ -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 typing import Any, Callable, Dict, Iterable, List, Optional
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/<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}.
# 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 ``<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+)
"""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,
@@ -14,7 +14,10 @@ import logging
import time
from typing import Any, Optional, Tuple
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
@@ -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;
@@ -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 += [
@@ -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