From b8bed8124e19288d59db6f862f90f214074ec95c Mon Sep 17 00:00:00 2001 From: Godopu Date: Tue, 13 Oct 2026 09:55:09 +0000 Subject: [PATCH 01/16] fix(delegation): audit-log root follows the registry (Issue #57) get_logs_dir(registry_dir) resolves the audit-log root as the delegate_job_logs next to the registry in use when DELEGATE_JOB_LOGS_DIR is not set; sibling_logs_dir() is the pure sibling rule. Every writer and reader that knows its registry (registry, engine, publish_event, remote, job_subscriber, reconcile.sh, the loop runner's fingerprint exclusion) passes that root instead of falling back to the workspace default. Co-Authored-By: Claude Fable 5.1 --- .agents/skills/lib_py/delegation/engine.py | 5 ++-- .../lib_py/delegation/job_subscriber.py | 18 ++++++----- .../skills/lib_py/delegation/mqtt_common.py | 30 ++++++++++++------- .../skills/lib_py/delegation/publish_event.py | 2 +- .agents/skills/lib_py/delegation/registry.py | 11 +++---- .agents/skills/lib_py/delegation/remote.py | 3 +- .agents/skills/lib_py/loop/runner.py | 3 +- .../scripts/reconcile.sh | 2 +- 8 files changed, 46 insertions(+), 28 deletions(-) diff --git a/.agents/skills/lib_py/delegation/engine.py b/.agents/skills/lib_py/delegation/engine.py index a0a97d3..1f7fb92 100644 --- a/.agents/skills/lib_py/delegation/engine.py +++ b/.agents/skills/lib_py/delegation/engine.py @@ -374,13 +374,14 @@ def _flush(notifications, registry_dir) -> None: Best effort by design: the registry record is already durable, and an audit write must never be able to fail a transition that has committed. """ + logs_dir = mqtt_common.get_logs_dir(registry_dir) for note in notifications: job_id = note["job_id"] kind = note["kind"] try: if kind == "status_changed": mqtt_common.update_logged_status( - job_id, note["to"], updated_at=note.get("timestamp"), + job_id, note["to"], logs_dir=logs_dir, updated_at=note.get("timestamp"), transition_seq=note.get("transition_seq")) mqtt_common.append_event(job_id, { "event": "status_changed", @@ -389,7 +390,7 @@ def _flush(notifications, registry_dir) -> None: "by": note.get("by"), "timestamp": note.get("timestamp"), "transition_seq": note.get("transition_seq"), - }) + }, logs_dir=logs_dir) elif kind == "conflict_refused": mqtt_common.append_event(job_id, { "event": "conflict_refused", diff --git a/.agents/skills/lib_py/delegation/job_subscriber.py b/.agents/skills/lib_py/delegation/job_subscriber.py index 6c6b22b..7c68d9f 100755 --- a/.agents/skills/lib_py/delegation/job_subscriber.py +++ b/.agents/skills/lib_py/delegation/job_subscriber.py @@ -93,7 +93,7 @@ def _check_disk_fallback( job_rec = {} if not disk_status or disk_status not in ("completed", "error", "cancelled"): try: - disk_status = mqtt_common.read_logged_status(jid, mqtt_common.get_logs_dir()) + disk_status = mqtt_common.read_logged_status(jid, mqtt_common.get_logs_dir(registry_dir)) except Exception: pass @@ -130,8 +130,10 @@ class _Watcher: def __init__(self, expected_job_ids: Set[str], expected_tokens: Dict[str, Optional[str]], expected_seqs: Dict[str, int], expected_dispatch_tokens: Optional[Dict[str, Optional[str]]] = None, - expected_attempts: Optional[Dict[str, Optional[int]]] = None): + expected_attempts: Optional[Dict[str, Optional[int]]] = None, + logs_dir: Optional[str] = None): self.events: "queue.Queue[Tuple[str, Dict[str, Any]]]" = queue.Queue() + self.logs_dir = logs_dir # Issue #57: the audit log next to the registry being watched self.expected = set(expected_job_ids) self.tokens = expected_tokens # job_id -> expected auth_token (or None) self.last_seq = dict(expected_seqs) @@ -163,7 +165,7 @@ class _Watcher: "seq": got, "via": via, "topic": topic, - }) + }, logs_dir=self.logs_dir) def _on_state(self, msg, payload: Dict[str, Any], jid: str) -> None: """A retained `/state` message (Issue #18): learn the job's @@ -196,7 +198,7 @@ class _Watcher: "topic": msg.topic, "timestamp": payload.get("timestamp"), "detail": "status={} last_event={}".format(payload.get("status"), payload.get("last_event")), - }) + }, logs_dir=self.logs_dir) self.events.put((msg.topic, dict(payload, _renews_liveness=renews))) def on_message(self, _client, _userdata, msg) -> None: @@ -317,7 +319,7 @@ class _Watcher: if expected_disp_tok else None), "attempt_id": (data.get("attempt_id") if isinstance(data, dict) and data.get("attempt_id") is not None else self.attempts.get(jid)), - }) + }, logs_dir=self.logs_dir) self.events.put((msg.topic, payload)) @@ -375,7 +377,8 @@ def _run_subscriber(args) -> int: if args.attempt_id is not None and args.job: attempts[args.job] = args.attempt_id - watcher = _Watcher(expected_ids, tokens, seqs, dispatch_tokens, attempts) + logs_dir = mqtt_common.get_logs_dir(args.registry_dir) + watcher = _Watcher(expected_ids, tokens, seqs, dispatch_tokens, attempts, logs_dir=logs_dir) # Resolve timeouts from CLI, falling back to the (first) job's settings. base_job = jobs[0] @@ -418,7 +421,8 @@ def _run_subscriber(args) -> int: # what makes the retained state message arrive again at once. logger.warning("broker reconnected (connect #%d); re-subscribing", connects["count"]) for jid in expected_ids: - mqtt_common.append_event(jid, {"event": "reconnected", "connect_count": connects["count"]}) + mqtt_common.append_event(jid, {"event": "reconnected", "connect_count": connects["count"]}, + logs_dir=logs_dir) for topic in subscribed_topics + state_topics: _c.subscribe(topic, qos=1) logger.info("subscribed to %s", topic) diff --git a/.agents/skills/lib_py/delegation/mqtt_common.py b/.agents/skills/lib_py/delegation/mqtt_common.py index 8cc30b4..67f1ec7 100644 --- a/.agents/skills/lib_py/delegation/mqtt_common.py +++ b/.agents/skills/lib_py/delegation/mqtt_common.py @@ -183,20 +183,29 @@ EVENTS_FILENAME = "events.ndjson" STATUS_FILENAME = "status.json" -def get_logs_dir() -> str: +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). - Overridable with ``DELEGATE_JOB_LOGS_DIR``; otherwise - ``$WORKSPACE_ROOT/.mam/delegate_job_logs`` (Issue #15, see - ``default_registry_dir``); otherwise ``/.mam/delegate_job_logs``. - 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. + ``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 _workspace_mam_dir("delegate_job_logs") or os.path.join(os.getcwd(), ".mam", "delegate_job_logs") + return sibling_logs_dir(registry_dir or default_registry_dir()) def __getattr__(name: str): # PEP 562 (3.7+) @@ -516,14 +525,15 @@ def _update_job_status_legacy(job_id: str, registry_dir: str = DEFAULT_REGISTRY_ _atomic_write_record(job_id, registry_dir, record) if "status" in fields: new_status = record.get("status") - update_logged_status(job_id, new_status, updated_at=record["updated_at"]) + logs_dir = get_logs_dir(registry_dir) + update_logged_status(job_id, new_status, logs_dir=logs_dir, updated_at=record["updated_at"]) if old_status != new_status: append_event(job_id, { "event": "status_changed", "from": old_status, "to": new_status, "timestamp": record["updated_at"], - }) + }, logs_dir=logs_dir) return record diff --git a/.agents/skills/lib_py/delegation/publish_event.py b/.agents/skills/lib_py/delegation/publish_event.py index f294c27..52cc36d 100755 --- a/.agents/skills/lib_py/delegation/publish_event.py +++ b/.agents/skills/lib_py/delegation/publish_event.py @@ -241,7 +241,7 @@ def record_outcome(prepared: PreparedEvent, registry_dir: str, "payload": prepared.payload, "published": publish_ok, "publish_error": publish_error, - }) + }, logs_dir=mqtt_common.get_logs_dir(registry_dir)) if not local_event_recorded: registry.append_event(prepared.job_id, registry_dir, prepared.payload) diff --git a/.agents/skills/lib_py/delegation/registry.py b/.agents/skills/lib_py/delegation/registry.py index 18241cf..9a86297 100755 --- a/.agents/skills/lib_py/delegation/registry.py +++ b/.agents/skills/lib_py/delegation/registry.py @@ -175,7 +175,7 @@ def register_job( _atomic_write_record(job_id, registry_dir, record) # Seed the persistent audit log (meta.json + status.json + a "registered" # event). Best-effort inside init_job_log — never blocks registration. - mqtt_common.init_job_log(job_id, meta=record) + mqtt_common.init_job_log(job_id, meta=record, logs_dir=mqtt_common.get_logs_dir(registry_dir)) logger.info("registered job %s (agent=%s session=%s)", job_id, agent, agent_session) return job_id @@ -233,7 +233,8 @@ def _pick_pending_legacy(agent_session: str, registry_dir: str = DEFAULT_REGISTR updated_at = chosen["updated_at"] # Mirrors the pending->running transition into the audit log outside the # lock. This is the shape the engine now generalises to every transition. - mqtt_common.update_logged_status(job_id, "running", updated_at=updated_at) + logs_dir = mqtt_common.get_logs_dir(registry_dir) + mqtt_common.update_logged_status(job_id, "running", logs_dir=logs_dir, updated_at=updated_at) mqtt_common.append_event(job_id, { "event": "status_changed", "from": "pending", @@ -328,7 +329,7 @@ def get_feedback(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR, # 1) Try the unified audit log first (ndjson) since it's written synchronously by the subscriber try: from lib_py.delegation import mqtt_common - logs_dir = mqtt_common.get_logs_dir() + logs_dir = mqtt_common.get_logs_dir(registry_dir) logged = list(mqtt_common.iter_logged_events(job_id, logs_dir)) for e in reversed(logged): if e.get("source_event") in events: @@ -450,7 +451,7 @@ def _build_parser() -> argparse.ArgumentParser: help="summarise every job under the logs dir instead") p_logs.add_argument("--logs-dir", default=None, help="override the audit-log root (default: $DELEGATE_JOB_LOGS_DIR " - "or /.mam/delegate_job_logs)") + "or the delegate_job_logs next to --registry-dir)") p_logs.add_argument("--tail", type=int, default=0, help="show only the last N events (0 = all)") p_logs.add_argument("--json", action="store_true", @@ -698,7 +699,7 @@ def main(argv: Optional[List[str]] = None) -> int: def _cmd_logs(args) -> int: """Pretty-print one job's events.ndjson, or summarise all logged jobs.""" - logs_dir = args.logs_dir or mqtt_common.get_logs_dir() + logs_dir = args.logs_dir or mqtt_common.get_logs_dir(args.registry_dir) if args.list_all: jobs = mqtt_common.list_logged_jobs(logs_dir) diff --git a/.agents/skills/lib_py/delegation/remote.py b/.agents/skills/lib_py/delegation/remote.py index 31e63fc..47dcbdf 100644 --- a/.agents/skills/lib_py/delegation/remote.py +++ b/.agents/skills/lib_py/delegation/remote.py @@ -136,7 +136,8 @@ def publish_dispatch(job_id: str, registry_dir: str, session: str, instructions: exceptions=(OSError, TimeoutError, ConnectionError, ValueError))( broker_config_from_job(job), topic, body, True) mqtt_common.append_event(job_id, {"event": "dispatched", "topic": topic, "session": session, - "attempt_id": payload["attempt_id"]}) + "attempt_id": payload["attempt_id"]}, + logs_dir=mqtt_common.get_logs_dir(registry_dir)) logger.info("dispatched job %s to remote session %s on %s", job_id, session, topic) return topic diff --git a/.agents/skills/lib_py/loop/runner.py b/.agents/skills/lib_py/loop/runner.py index 5b1c46a..1ad23df 100644 --- a/.agents/skills/lib_py/loop/runner.py +++ b/.agents/skills/lib_py/loop/runner.py @@ -26,6 +26,7 @@ from pathlib import Path from typing import Any, Dict, List, Optional, Set, Tuple from lib_py import ints +from lib_py.delegation import mqtt_common from lib_py.delegation import registry as job_registry from lib_py.loop import checkpoint, eager, parser @@ -1083,7 +1084,7 @@ class LoopRunner: if not head: return None root = self.worktree_root.resolve() - logs = Path(os.environ.get("DELEGATE_JOB_LOGS_DIR") or self.mam_dir / "delegate_job_logs") + logs = Path(mqtt_common.get_logs_dir(str(self.jobs_dir))) exclude = [] for d in (self.mam_dir, self.jobs_dir, self.workspace_root / logs): try: diff --git a/.agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh b/.agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh index a1c3adf..7ede101 100755 --- a/.agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh +++ b/.agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh @@ -272,7 +272,7 @@ def on_message(_client, _userdata, msg): "topic": msg.topic, "timestamp": payload.get("timestamp"), "detail": payload.get("detail", ""), - }) + }, logs_dir=mqtt_common.get_logs_dir(registry_dir)) print(f"MQTT Monitor: recorded event {event} for job {jid} (seq={seq})", flush=True) -- 2.54.0 From b7a857d12fc0ad70e849b48efa9f44022516b1eb Mon Sep 17 00:00:00 2001 From: Godopu Date: Tue, 13 Oct 2026 09:57:42 +0000 Subject: [PATCH 02/16] fix(control-plane): derive the audit-log root from the registry in the binding and the shell defaults from_env() fills delegate_job_logs_dir with the delegate_job_logs next to the resolved mam_jobs_dir unless DELEGATE_JOB_LOGS_DIR is set, so a session created with only MAM_JOBS_DIR moved no longer records the workspace's old log root. worktree_prepare.sh passes DELEGATE_JOB_LOGS_DIR on only when the caller set it. lib.sh and the delegate-job wrapper no longer carry a bash copy of the default: delegate_publish_cmd, delegate_started_recorded and _logs_dir_abs resolve it with get_logs_dir(), the wrapper fails closed when it cannot, exports the resolved root to its children, prints it as submit's last line, and logs takes --registry-dir. Co-Authored-By: Claude Fable 5.1 --- .agents/skills/lib.sh | 16 +++- .agents/skills/lib_py/control_plane.py | 6 ++ .../scripts/worktree_prepare.sh | 4 +- .../multi-agent-mux-delegate-job | 87 +++++++++---------- 4 files changed, 63 insertions(+), 50 deletions(-) diff --git a/.agents/skills/lib.sh b/.agents/skills/lib.sh index ee120a6..59f587f 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; from lib_py.delegation.mqtt_common import get_logs_dir; print(os.path.abspath(get_logs_dir()))' 2>/dev/null || true)" + 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)" 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") @@ -2372,10 +2372,16 @@ except Exception: # Never the registry status: a record leaves `pending` when another writer # 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 +} + delegate_started_recorded() { local job_id="$1" tok="${2:-}" logs_dir="${3:-}" local reg_dir="${MAM_JOBS_DIR:-${WORKSPACE_ROOT:-$PWD}/.mam/jobs}" - [ -n "$logs_dir" ] || logs_dir="${DELEGATE_JOB_LOGS_DIR:-${WORKSPACE_ROOT:-$PWD}/.mam/delegate_job_logs}" + # Issue #57: the audit log is resolved by the one rule (next to the registry + # unless DELEGATE_JOB_LOGS_DIR says otherwise); no bash copy of the default. + [ -n "$logs_dir" ] || logs_dir="$(delegate_logs_dir "$reg_dir")" || return 1 local ev_log="$reg_dir/$job_id.events.log" local audit="$logs_dir/$job_id/events.ndjson" [ -f "$ev_log" ] || [ -f "$audit" ] || return 1 @@ -2431,10 +2437,12 @@ delegate_wait_started() { local job_id="$1" ack_s="${2:-180}" tok="${3:-}" [ -n "$job_id" ] || return 1 ack_s=$(mam_int "$ack_s" 0) || { echo "WARNING: onboarding ack timeout ${2-} ignored (not a decimal integer >= 0); using 180" >&2; ack_s=180; } - local deadline + local deadline logs_dir deadline=$(( $(date +%s) + ack_s )) + # Issue #57: resolve the audit-log root once, not once per 1 s poll. + logs_dir="$(delegate_logs_dir "${MAM_JOBS_DIR:-${WORKSPACE_ROOT:-$PWD}/.mam/jobs}")" || logs_dir="" while :; do - if delegate_started_recorded "$job_id" "$tok"; then + if delegate_started_recorded "$job_id" "$tok" "$logs_dir"; then return 0 fi if [ "$(date +%s)" -ge "$deadline" ]; then diff --git a/.agents/skills/lib_py/control_plane.py b/.agents/skills/lib_py/control_plane.py index 3fb41ee..2503588 100644 --- a/.agents/skills/lib_py/control_plane.py +++ b/.agents/skills/lib_py/control_plane.py @@ -21,6 +21,8 @@ import shlex import sys from typing import Dict, Optional +from lib_py.delegation import mqtt_common + KEYS = ("workspace_root", "agent_sessions_yaml", "mam_jobs_dir", "delegate_job_logs_dir") ENV = {"workspace_root": "WORKSPACE_ROOT", "agent_sessions_yaml": "AGENT_SESSIONS_YAML", "mam_jobs_dir": "MAM_JOBS_DIR", "delegate_job_logs_dir": "DELEGATE_JOB_LOGS_DIR"} @@ -45,6 +47,10 @@ def from_env(env=None) -> Dict[str, str]: v = (env.get(var) or "").strip() if v: b[k] = _abs(v) + # 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"])) return b diff --git a/.agents/skills/multi-agent-mux-create/scripts/worktree_prepare.sh b/.agents/skills/multi-agent-mux-create/scripts/worktree_prepare.sh index fc81607..1677478 100755 --- a/.agents/skills/multi-agent-mux-create/scripts/worktree_prepare.sh +++ b/.agents/skills/multi-agent-mux-create/scripts/worktree_prepare.sh @@ -234,8 +234,10 @@ if [ -n "$CREATOR_AGENT" ]; then env_add=(WORKSPACE_ROOT="$MAIN" AGENT_SESSIONS_YAML="${AGENT_SESSIONS_YAML:-$MAIN/.mam/agent-sessions.yaml}" MAM_JOBS_DIR="${MAM_JOBS_DIR:-$MAIN/.mam/jobs}" - DELEGATE_JOB_LOGS_DIR="${DELEGATE_JOB_LOGS_DIR:-$MAIN/.mam/delegate_job_logs}" MAM_NO_SUBSCRIBER_DAEMON=1) + # Issue #57: an explicit audit-log root is passed on; otherwise create's + # binding derives it from MAM_JOBS_DIR (the delegate_job_logs next to it). + if [ -n "${DELEGATE_JOB_LOGS_DIR:-}" ]; then env_add+=(DELEGATE_JOB_LOGS_DIR="$DELEGATE_JOB_LOGS_DIR"); fi if [ -z "${MAM_ENV_FILE:-}" ] && [ -f "$MAIN/.mam.env" ]; then env_add+=(MAM_ENV_FILE="$MAIN/.mam.env"); fi rc=0 env "${env_add[@]}" bash "$SKILL_DIR/multi-agent-mux-create/scripts/create_session.sh" "${args[@]}" || rc=$? 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 83531b0..6fc8075 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 @@ -87,11 +87,12 @@ _abs_registry_dir() { export SKS_REGISTRY_DIR="$REGISTRY_DIR" } -# The audit-log root this wrapper's own subscriber resolves (same resolver), -# pinned into the injected commands so an agent publishing from another cwd -# does not fall back to /.mam/delegate_job_logs. +# The audit-log root this wrapper's own subscriber resolves (same resolver: +# next to $REGISTRY_DIR unless DELEGATE_JOB_LOGS_DIR says otherwise, Issue +# #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()))' "$SCRIPT_DIR/.." + "$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" } # Issue #17 item 4: the wrapper's decisions are lib_py/delegation/wrapper.py; @@ -184,7 +185,7 @@ multi-agent-mux-delegate-job [options] list [--registry-dir ] verify --job --validate