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)