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(<registry>), 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 <noreply@anthropic.com>
This commit is contained in:
GodopuandClaude Fable 5.1 committed 2026-10-13 09:57:42 +00:00
1 parent b8bed8124e
commit b7a857d12f
4 files changed
+63 -50

No files matched your search

+12 -4
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; 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
+6
View File
@@ -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
@@ -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=$?
@@ -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 <cwd>/.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 <command> [options]
list [--registry-dir <dir>]
verify --job <id> --validate <script> [--registry-dir <dir>]
wait [--job <id>] [--timeout <sec>] [--registry-dir <dir>]
logs <job_id> | --list # persistent audit log (delegate_job_logs/)
logs <job_id> | --list [--registry-dir <dir>] # audit log next to the registry (or $DELEGATE_JOB_LOGS_DIR)
EOF
}
@@ -199,6 +200,7 @@ TIMEOUT=$(mam_int_knob MAM_JOB_TIMEOUT_S 3600 1)
# Issue #18 liveness contract (N / K / T). An unset --idle-timeout derives K x N.
HEARTBEAT_SEC=$(mam_int_knob MAM_HEARTBEAT_SEC 30 1); HEARTBEAT_MISSES=$(mam_int_knob MAM_HEARTBEAT_MISSES 4 1); ACK_TIMEOUT=$(mam_int_knob MAM_ACK_TIMEOUT_S 180 1)
JOB_ID=""; REGISTRY_DIR="$REGISTRY_DIR_DEFAULT"; DELEGATE_ROLE="Worker"
LOGS_DIR_ABS=""
TYPE="direct"; REVIEWER="hermes"; REVIEWER_SESSION="herdr:hermes"; MAX_ITERATIONS=5
DEFAULT_COUNTERPART_ROLE="Reviewer"
COUNTERPART_ROLE="$DEFAULT_COUNTERPART_ROLE"
@@ -276,7 +278,10 @@ cmd_submit() {
[[ "$PY" == */* && "$PY" != /* ]] && PY="$(pwd)/$PY"
cd "$WORKDIR"
_abs_registry_dir
LOGS_DIR_ABS="$(_logs_dir_abs 2>/dev/null || true)"
LOGS_DIR_ABS="$(_logs_dir_abs)" || { echo "ERROR: cannot resolve the audit-log root for $REGISTRY_DIR" >&2; exit 1; }
# Issue #57: every child (subscriber, registry.py, validate scripts) sees
# the same root this wrapper prints and pins into the injected commands.
export DELEGATE_JOB_LOGS_DIR="$LOGS_DIR_ABS"
# Issue #34: an explicit --parent-job fails closed in the registry; the
# $MAM_JOB_ID default may name a job of another host's registry (a remote
@@ -345,16 +350,14 @@ EOF
if [[ "$DRY_RUN" == "1" ]]; then
run_agent "$JOB_ID" "$instructions"
local logs_root_dry="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
echo "$logs_root_dry/$JOB_ID"
echo "$LOGS_DIR_ABS/$JOB_ID"
return 0
fi
# 1) Preflight (a remote session has no local pane to check: Issue #18)
if ! _is_remote "$AGENT_SESSION" && ! preflight_agent "$JOB_ID" "$AGENT_SESSION" 0; then
finalize_unfinished "$JOB_ID" "error" "$SUB_RC_ERROR" "preflight"
local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
echo "$logs_root/$JOB_ID"
echo "$LOGS_DIR_ABS/$JOB_ID"
exit 5
fi
@@ -368,8 +371,7 @@ EOF
export _SKS_IDLE_RC="$idle_rc"
if [ $idle_rc -eq 1 ] || [ $idle_rc -eq 2 ]; then
finalize_unfinished "$JOB_ID" "error" "$idle_rc" "idle_wait"
local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
echo "$logs_root/$JOB_ID"
echo "$LOGS_DIR_ABS/$JOB_ID"
exit 5
fi
fi
@@ -409,8 +411,7 @@ EOF
_decided mapped_st sub-status "$sub_exit" --job "$JOB_ID" --registry-dir "$REGISTRY_DIR"
finalize_unfinished "$JOB_ID" "$mapped_st" "$sub_exit" "subscriber"
mapped_st="${FINAL_JOB_STATUS:-$mapped_st}"
local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
echo "$logs_root/$JOB_ID"
echo "$LOGS_DIR_ABS/$JOB_ID"
_exit_for "$mapped_st"
fi
fi
@@ -424,8 +425,7 @@ EOF
if ! deliver_prompt "$JOB_ID" "$instructions" "$AGENT_SESSION" "$logf"; then
reap_subscriber "$sub_pid"
finalize_unfinished "$JOB_ID" "error" "$SUB_RC_ERROR" "inject"
local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
echo "$logs_root/$JOB_ID"
echo "$LOGS_DIR_ABS/$JOB_ID"
exit 1
fi
start_stall_watch "$AGENT_SESSION" "$AGENT" "$_disp_tok" "$_attempt_id"
@@ -454,8 +454,7 @@ EOF
# Last stdout line: the persistent audit-log dir for this job (see SKILL.md
# "Audit Logs"). Callers can scrape `tail -n1` to find it.
local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
echo "$logs_root/$JOB_ID"
echo "$LOGS_DIR_ABS/$JOB_ID"
_exit_for "$job_status"
else
# Implement loop/discuss orchestrator
@@ -470,8 +469,7 @@ EOF
if [[ "$DRY_RUN" == "1" ]]; then
echo "[dry-run] orchestrator loop would start for job: $JOB_ID type: $TYPE"
echo "worker session: $AGENT_SESSION, reviewer session: $REVIEWER_SESSION"
local logs_root_dry="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
echo "$logs_root_dry/$JOB_ID"
echo "$LOGS_DIR_ABS/$JOB_ID"
return 0
fi
@@ -517,8 +515,7 @@ EOF
# Preflight (skipped for a remote session: Issue #18)
if ! _is_remote "$current_session" && ! preflight_agent "$JOB_ID" "$current_session" "$force_warn_only"; then
finalize_unfinished "$JOB_ID" "error" "$SUB_RC_ERROR" "preflight"
local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
echo "$logs_root/$JOB_ID"
echo "$LOGS_DIR_ABS/$JOB_ID"
exit 5
fi
@@ -550,8 +547,7 @@ EOF
export _SKS_IDLE_RC="$idle_rc"
if [ $idle_rc -eq 1 ] || [ $idle_rc -eq 2 ]; then
finalize_unfinished "$JOB_ID" "error" "$idle_rc" "idle_wait"
local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
echo "$logs_root/$JOB_ID"
echo "$LOGS_DIR_ABS/$JOB_ID"
exit 5
fi
fi
@@ -587,8 +583,7 @@ EOF
_decided mapped_st sub-status "$sub_exit" --job "$JOB_ID" --registry-dir "$REGISTRY_DIR"
finalize_unfinished "$JOB_ID" "$mapped_st" "$sub_exit" "subscriber"
mapped_st="${FINAL_JOB_STATUS:-$mapped_st}"
local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
echo "$logs_root/$JOB_ID"
echo "$LOGS_DIR_ABS/$JOB_ID"
_exit_for "$mapped_st"
fi
fi
@@ -609,8 +604,7 @@ EOF
if ! deliver_prompt "$JOB_ID" "$instructions" "$current_session" "$logf"; then
reap_subscriber "$sub_pid"
finalize_unfinished "$JOB_ID" "error" "$SUB_RC_ERROR" "inject"
local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
echo "$logs_root/$JOB_ID"
echo "$LOGS_DIR_ABS/$JOB_ID"
exit 1
fi
start_stall_watch "$current_session" "$cur_agent" "$_disp_tok" "$attempt"
@@ -639,8 +633,7 @@ EOF
echo "Worker completed during grace period. Continuing to reviewer."
feedback="$("$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" get-feedback --job "$JOB_ID")"
else
local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
echo "$logs_root/$JOB_ID"
echo "$LOGS_DIR_ABS/$JOB_ID"
_exit_for "$job_status"
fi
fi
@@ -664,8 +657,7 @@ EOF
echo "Reviewer completed during grace period. Continuing evaluation."
feedback="$("$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" get-feedback --job "$JOB_ID")"
else
local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
echo "$logs_root/$JOB_ID"
echo "$LOGS_DIR_ABS/$JOB_ID"
_exit_for "$job_status"
fi
fi
@@ -720,8 +712,7 @@ EOF
fi
# Last stdout line: the persistent audit-log dir
local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
echo "$logs_root/$JOB_ID"
echo "$LOGS_DIR_ABS/$JOB_ID"
exit "$final_loop_rc"
fi
}
@@ -868,14 +859,12 @@ _registry_wait() {
clear) return 0 ;;
superseded)
echo "ERROR: job $job_id was redispatched, moved or closed by someone else while it waited for session '$sess'; this attempt is not dispatched and the record is left as it is" >&2
local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
echo "$logs_root/$job_id"
echo "$LOGS_DIR_ABS/$job_id"
exit 5 ;;
busy*)
echo "ERROR: session '$sess' still has open jobs in the registry (${res#busy }); job $job_id is not dispatched" >&2
finalize_unfinished "$job_id" "error" "$SUB_RC_ERROR" "registry_wait" "$sess"
local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
echo "$logs_root/$job_id"
echo "$LOGS_DIR_ABS/$job_id"
exit 5 ;;
esac
_decision_failed wait-open-jobs "answer '$res'"
@@ -1099,16 +1088,24 @@ cmd_verify() {
}
cmd_logs() {
# logs <job_id> | logs --list — delegates to registry.py's logs CLI, which
# reads the persistent audit log under $DELEGATE_JOB_LOGS_DIR (or
# <cwd>/delegate_job_logs). Run from your project dir so the default resolves.
# logs <job_id> | logs --list [--registry-dir <dir>] — delegates to
# registry.py's logs CLI, which reads the audit log next to the registry
# (Issue #57; $DELEGATE_JOB_LOGS_DIR overrides). parse_opts is for submit's
# globals, so the few options are read here.
PY="$(pick_python)"
if [[ "${1:-}" == "--list" ]]; then
"$PY" "$SCRIPT_DIR/scripts/registry.py" logs --list
local reg="$REGISTRY_DIR_DEFAULT" jid="" list=0
while [[ $# -gt 0 ]]; do
case "$1" in
--registry-dir) reg="$2"; shift 2;;
--list) list=1; shift;;
*) jid="$1"; shift;;
esac
done
if [[ "$list" == "1" ]]; then
"$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$reg" logs --list
else
local jid="${1:-}"
[[ -n "$jid" ]] || { echo "logs requires <job_id> or --list" >&2; exit 1; }
"$PY" "$SCRIPT_DIR/scripts/registry.py" logs "$jid"
"$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$reg" logs "$jid"
fi
}