fix(skills): resolve session resolution, add cancelled event status, and scope mqtt topic

This commit is contained in:
2026-08-05 09:11:35 +09:00
parent 68eff79810
commit ea36e81624
8 changed files with 71 additions and 44 deletions
+39 -11
View File
@@ -404,6 +404,7 @@ except Exception:
fi fi
;; ;;
set-buffer) set-buffer)
buf="tmp_buffer"
text="" text=""
while [ $# -gt 0 ]; do while [ $# -gt 0 ]; do
case "$1" in case "$1" in
@@ -412,18 +413,25 @@ except Exception:
echo "Error: -b requires a value" >&2 echo "Error: -b requires a value" >&2
exit 1 exit 1
fi fi
buf="$2" buf="$(echo "$2" | tr -cd 'A-Za-z0-9_.-')"
if [ -z "$buf" ]; then buf="tmp_buffer"; else buf="buf_$buf"; fi
shift 2 shift 2
;; ;;
*) text="$1"; shift ;; *) text="$1"; shift ;;
esac esac
done done
echo -n "$text" > "$wrapper_dir/tmp_buffer" echo -n "$text" > "$wrapper_dir/$buf"
;; ;;
paste-buffer) paste-buffer)
buf="tmp_buffer"
sess="" sess=""
while [ $# -gt 0 ]; do while [ $# -gt 0 ]; do
case "$1" in case "$1" in
-b)
buf="$(echo "$2" | tr -cd 'A-Za-z0-9_.-')"
if [ -z "$buf" ]; then buf="tmp_buffer"; else buf="buf_$buf"; fi
shift 2
;;
-t) -t)
if [ $# -lt 2 ]; then if [ $# -lt 2 ]; then
echo "Error: -t requires a value" >&2 echo "Error: -t requires a value" >&2
@@ -435,12 +443,26 @@ except Exception:
*) shift ;; *) shift ;;
esac esac
done done
if [ -f "$wrapper_dir/tmp_buffer" ]; then if [ -f "$wrapper_dir/$buf" ]; then
_real_herdr agent send "$sess" "$(cat "$wrapper_dir/tmp_buffer")" >/dev/null 2>&1 || true _real_herdr agent send "$sess" "$(cat "$wrapper_dir/$buf")" >/dev/null 2>&1 || true
else
echo "Error: buffer $buf not found" >&2
exit 1
fi fi
;; ;;
delete-buffer) delete-buffer)
rm -f "$wrapper_dir/tmp_buffer" buf="tmp_buffer"
while [ $# -gt 0 ]; do
case "$1" in
-b)
buf="$(echo "$2" | tr -cd 'A-Za-z0-9_.-')"
if [ -z "$buf" ]; then buf="tmp_buffer"; else buf="buf_$buf"; fi
shift 2
;;
*) shift ;;
esac
done
rm -f "$wrapper_dir/$buf"
;; ;;
ls) ls)
_real_herdr agent list 2>/dev/null | python3 -c " _real_herdr agent list 2>/dev/null | python3 -c "
@@ -451,7 +473,7 @@ try:
for a in res.get('agents', []): for a in res.get('agents', []):
try: try:
name = a.get('name') or a.get('agent') or 'unknown' name = a.get('name') or a.get('agent') or 'unknown'
print(f\"{name}|0\") print(f\"{name}|\")
except Exception: except Exception:
pass pass
except Exception: except Exception:
@@ -659,8 +681,11 @@ env_python() {
_check_is_nfs() { _check_is_nfs() {
local f="$1" local f="$1"
local mountpoint local mountpoint
mountpoint="$(df --output=target "$f" 2>/dev/null | tail -1)" || return 1 mountpoint="$(df --output=target "$f" 2>/dev/null | tail -1)"
if mount | grep -q "$mountpoint.*nfs\|$mountpoint.*cifs\|$mountpoint.*fuse.sshfs"; then if [ -z "$mountpoint" ]; then
mountpoint="$(df -P "$f" 2>/dev/null | tail -1 | awk '{print $6}')"
fi
if [ -n "$mountpoint" ] && mount | grep -i -q -E "$mountpoint.*(nfs|cifs|smb|sshfs)"; then
return 0 # is NFS return 0 # is NFS
fi fi
return 1 # not NFS return 1 # not NFS
@@ -1221,7 +1246,10 @@ elif agent == 'cline':
emit(cand) emit(cand)
ai = {} ai = {}
db_path = f"{mam_dir}/agent-sessions.db" if 'mam_dir' in locals() else os.path.join(ws, ".mam", "agent-sessions.db")
yaml_path = f"{mam_dir}/agent-sessions.yaml" if 'mam_dir' in locals() else os.path.join(ws, ".mam", "agent-sessions.yaml")
try: try:
import yaml
if os.path.exists(db_path): if os.path.exists(db_path):
conn = sqlite3.connect(db_path, timeout=60.0) conn = sqlite3.connect(db_path, timeout=60.0)
row = conn.execute('SELECT data FROM state WHERE id=1').fetchone() row = conn.execute('SELECT data FROM state WHERE id=1').fetchone()
@@ -1232,8 +1260,8 @@ try:
with open(yaml_path) as f: with open(yaml_path) as f:
d = yaml.safe_load(f) or {} d = yaml.safe_load(f) or {}
ai = d.get('agent_identities', {}) ai = d.get('agent_identities', {})
except Exception: except Exception as e:
pass print(f"WARN: tier-3 identity lookup failed: {e}", file=sys.stderr)
ai_agent = ai.get(agent) or {} ai_agent = ai.get(agent) or {}
if ai_agent.get('project_cwd') == ws: if ai_agent.get('project_cwd') == ws:
@@ -1242,7 +1270,7 @@ if ai_agent.get('project_cwd') == ws:
if cand and verify_session_uuid(ws, agent, cand, mode="revalidate"): if cand and verify_session_uuid(ws, agent, cand, mode="revalidate"):
emit(cand) emit(cand)
elif agent == 'agy': elif agent == 'agy':
cand = ai.get('conversation_id') cand = ai_agent.get('conversation_id')
if cand and verify_session_uuid(ws, agent, cand, mode="revalidate"): if cand and verify_session_uuid(ws, agent, cand, mode="revalidate"):
emit(cand) emit(cand)
elif agent == 'hermes': elif agent == 'hermes':
@@ -81,7 +81,7 @@ fi
[ -n "$AGENT" ] || { echo "ERROR: --agent required" >&2; usage; exit 2; } [ -n "$AGENT" ] || { echo "ERROR: --agent required" >&2; usage; exit 2; }
[ -n "$ROLE" ] || { echo "ERROR: --role required" >&2; usage; exit 2; } [ -n "$ROLE" ] || { echo "ERROR: --role required" >&2; usage; exit 2; }
[ -d "$WORKSPACE" ] || { echo "ERROR: workspace $WORKSPACE not a directory" >&2; exit 1; } [ -d "$WORKSPACE" ] || { echo "ERROR: workspace $WORKSPACE not a directory" >&2; exit 1; }
command -v herdr >/dev/null || { echo "ERROR: herdr not installed" >&2; exit 1; } command -v herdr >/dev/null || type -P herdr >/dev/null || { echo "ERROR: herdr not installed" >&2; exit 1; }
command -v "$AGENT" >/dev/null || { echo "ERROR: $AGENT CLI not in PATH" >&2; exit 1; } command -v "$AGENT" >/dev/null || { echo "ERROR: $AGENT CLI not in PATH" >&2; exit 1; }
# Auth Check (OAuth check for agy, loggedIn check for claude, status for hermes) # Auth Check (OAuth check for agy, loggedIn check for claude, status for hermes)
@@ -127,22 +127,15 @@ fi
LOCAL_BIN="${LOCAL_BIN:-$HOME/.local/bin}" LOCAL_BIN="${LOCAL_BIN:-$HOME/.local/bin}"
WRAPPER="$LOCAL_BIN/$SESSION_NAME" WRAPPER="$LOCAL_BIN/$SESSION_NAME"
ws_slug="$(derive_session_name "$WORKSPACE" "$AGENT" | sed 's/-creator-.*//')"
if [ -z "${HERDR_SERVER_NAME:-}" ] || [ "$HERDR_SERVER_NAME" = "default" ]; then if [ -z "${HERDR_SERVER_NAME:-}" ] || [ "$HERDR_SERVER_NAME" = "default" ]; then
export HERDR_SERVER_NAME="multi-agent-mux" export HERDR_SERVER_NAME="mam-$ws_slug"
fi fi
# Resolve absolute path of the agent command to prevent herdr PATH inheritance issues (especially on macOS) # Resolve absolute path of the agent command to prevent herdr PATH inheritance issues (especially on macOS)
RESOLVED_BIN="$AGENT" RESOLVED_BIN="$AGENT"
if [ "$AGENT" = "cline" ]; then if command -v "$AGENT" >/dev/null 2>&1; then
if command -v cline >/dev/null 2>&1; then RESOLVED_BIN="$(command -v "$AGENT")"
RESOLVED_BIN="$(command -v cline)"
fi
else
if [ "$AGENT" = "claude" ] && [ -x "$HOME/.nvm/versions/node/v24.15.0/bin/claude" ]; then
RESOLVED_BIN="$HOME/.nvm/versions/node/v24.15.0/bin/claude"
elif command -v "$AGENT" >/dev/null 2>&1; then
RESOLVED_BIN="$(command -v "$AGENT")"
fi
fi fi
# On macOS, clear quarantine attribute for the agent binary to prevent Gatekeeper hangs # On macOS, clear quarantine attribute for the agent binary to prevent Gatekeeper hangs
@@ -159,7 +152,7 @@ esac
spawn() { spawn() {
if [ -z "${HERDR_SERVER_NAME:-}" ] || [ "$HERDR_SERVER_NAME" = "default" ]; then if [ -z "${HERDR_SERVER_NAME:-}" ] || [ "$HERDR_SERVER_NAME" = "default" ]; then
export HERDR_SERVER_NAME="multi-agent-mux" export HERDR_SERVER_NAME="mam-$ws_slug"
fi fi
case "$AGENT" in case "$AGENT" in
claude) claude)
@@ -194,7 +187,8 @@ cleanup_herdr_on_error() {
} }
trap cleanup_herdr_on_error EXIT trap cleanup_herdr_on_error EXIT
export HERDR_SERVER_NAME="${HERDR_SERVER_NAME:-$(resolve_herdr_server "$SESSION_NAME")}" RESOLVED_SERVER="$(resolve_herdr_workspace "$SESSION_NAME")"
export HERDR_SERVER_NAME="${HERDR_SERVER_NAME:-$RESOLVED_SERVER}"
# TUI 준비 대기 # TUI 준비 대기
if ! wait_for_tui_ready "$SESSION_NAME" "$AGENT"; then if ! wait_for_tui_ready "$SESSION_NAME" "$AGENT"; then
@@ -45,13 +45,14 @@ from mqtt_common import (
logger = logging.getLogger("delegate_job.publish_event") logger = logging.getLogger("delegate_job.publish_event")
VALID_EVENTS = ("started", "permission_required", "progress", "completed", "error") VALID_EVENTS = ("started", "permission_required", "progress", "completed", "error", "cancelled")
TERMINAL_EVENTS = ("completed", "error") TERMINAL_EVENTS = ("completed", "error", "cancelled")
# event -> registry status to sync as a best-effort side effect # event -> registry status to sync as a best-effort side effect
EVENT_TO_STATUS = { EVENT_TO_STATUS = {
"started": "running", "started": "running",
"completed": "completed", "completed": "completed",
"error": "error", "error": "error",
"cancelled": "cancelled",
} }
CONNECT_ACK_TIMEOUT = 10 # seconds to wait for CONNACK CONNECT_ACK_TIMEOUT = 10 # seconds to wait for CONNACK
@@ -147,6 +147,9 @@ except Exception:
elif [ "$status" = "error" ]; then elif [ "$status" = "error" ]; then
log_error "Job '$job_id' finished with errors." log_error "Job '$job_id' finished with errors."
return 1 return 1
elif [ "$status" = "cancelled" ]; then
log_error "Job '$job_id' was cancelled."
return 1
fi fi
sleep "$check_interval" sleep "$check_interval"
done done
@@ -131,9 +131,9 @@ for s in d.get('herdr_sessions', []):
if s.get('delegate_job_id') == _jid and s.get('status') == 'running': if s.get('delegate_job_id') == _jid and s.get('status') == 'running':
_name = s.get('name') _name = s.get('name')
_srv = s.get('herdr_workspace') or s.get('herdr_server') or 'default' _srv = s.get('herdr_workspace') or s.get('herdr_server') or 'default'
if _event == 'completed': if _event in ('completed', 'cancelled'):
s['delegate_job_id'] = None s['delegate_job_id'] = None
print('MQTT Monitor: job completed on ' + str(_name) + ' — session kept alive', flush=True) print('MQTT Monitor: job ' + _event + ' on ' + str(_name) + ' — session kept alive', flush=True)
_changed = True _changed = True
else: else:
s['status'] = 'terminated' s['status'] = 'terminated'
@@ -216,7 +216,7 @@ def on_message(_client, _userdata, msg):
print(f"MQTT Monitor: recorded event {event} for job {jid} (seq={seq})", flush=True) print(f"MQTT Monitor: recorded event {event} for job {jid} (seq={seq})", flush=True)
if event in ("completed", "error"): if event in ("completed", "error", "cancelled"):
print(f"MQTT Monitor: received terminal event {event} for job {jid}", flush=True) print(f"MQTT Monitor: received terminal event {event} for job {jid}", flush=True)
handle_terminal(jid, event) handle_terminal(jid, event)
except Exception as e: except Exception as e:
@@ -227,8 +227,13 @@ def on_connect(_c, _u, _flags, reason_code, _props):
rc = mqtt_common.reason_code_value(reason_code) rc = mqtt_common.reason_code_value(reason_code)
if rc == 0: if rc == 0:
state['connected'] = True state['connected'] = True
_c.subscribe("python/mqtt/jobs/+/events", qos=1) ws_path = os.path.abspath(workspace_root) if workspace_root else os.getcwd()
print("MQTT Monitor: subscribed to python/mqtt/jobs/+/events", flush=True) import hashlib
fp = hashlib.sha256(ws_path.encode('utf-8')).hexdigest()[:12]
topic = f"mam/{fp}/jobs/+/events"
_c.subscribe(topic, qos=1)
_c.subscribe("python/mqtt/jobs/+/events", qos=1) # legacy fallback during transition
print(f"MQTT Monitor: subscribed to {topic}", flush=True)
else: else:
state['failed'] = True state['failed'] = True
print(f"MQTT Monitor connection failed: rc={rc}", flush=True) print(f"MQTT Monitor connection failed: rc={rc}", flush=True)
@@ -76,7 +76,7 @@ if [ -z "$UUID" ]; then
exit 1 exit 1
fi fi
export HERDR_SERVER_NAME="$(resolve_herdr_server "$SESSION_NAME")" export HERDR_SERVER_NAME="$(resolve_herdr_workspace "$SESSION_NAME")"
# 2. If herdr is alive, attach. Done. # 2. If herdr is alive, attach. Done.
if herdr has-session -t "$SESSION_NAME" 2>/dev/null; then if herdr has-session -t "$SESSION_NAME" 2>/dev/null; then
@@ -215,11 +215,12 @@ if [ "$PURGE" = "1" ] && [ "$HERDR_ALIVE" = "1" ]; then
echo " Refusing registry removal — records preserved (no state was modified)." >&2 echo " Refusing registry removal — records preserved (no state was modified)." >&2
echo " Diagnose the stuck TUI (herdr session attach '$SESSION_NAME'), then re-run" >&2 echo " Diagnose the stuck TUI (herdr session attach '$SESSION_NAME'), then re-run" >&2
echo " stop_session.sh --purge-conversation --yes (retry is safe/idempotent)." >&2 echo " stop_session.sh --purge-conversation --yes (retry is safe/idempotent)." >&2
delegate_publish_event "$DELEGATE_JOB_ID" error "purge aborted: herdr session still alive" delegate_publish_event "$DELEGATE_JOB_ID" cancelled "purge aborted: session still alive; no state was modified"
exit 4 exit 4
fi fi
fi fi
# INVARIANT: Terminal events for a job must be published AFTER atomic_dump_yaml updates the session status
atomic_dump_yaml "$AGENT_SESSIONS_YAML" \ atomic_dump_yaml "$AGENT_SESSIONS_YAML" \
SESSION_NAME="$SESSION_NAME" AGENT="$AGENT" PURGE="$PURGE" \ SESSION_NAME="$SESSION_NAME" AGENT="$AGENT" PURGE="$PURGE" \
NOW_ISO="$NOW_ISO" NOW_EPOCH="$NOW_EPOCH" LAST_STATUS="$LAST_STATUS" \ NOW_ISO="$NOW_ISO" NOW_EPOCH="$NOW_EPOCH" LAST_STATUS="$LAST_STATUS" \
@@ -358,7 +359,7 @@ else:
print(f"updated: {name} status={target['status']}", flush=True) print(f"updated: {name} status={target['status']}", flush=True)
PYEOF PYEOF
delegate_publish_event "$DELEGATE_JOB_ID" completed "session terminated" delegate_publish_event "$DELEGATE_JOB_ID" cancelled "session stopped by operator before job completion"
echo echo
echo "=== stop complete ===" echo "=== stop complete ==="
+4 -9
View File
@@ -13,6 +13,10 @@
# #
# SECURITY: never put real secrets in this template. Secret-bearing vars use a # SECURITY: never put real secrets in this template. Secret-bearing vars use a
# `replace_me` placeholder — fill them in only in your local `.mam.env`. # `replace_me` placeholder — fill them in only in your local `.mam.env`.
# SECURITY NOTE:
# The default MQTT broker (broker.hivemq.com) is public and unencrypted. Job event
# payloads (including task details) sent to the default broker can be read by anyone.
# For production or private workloads, configure a private broker with TLS and Auth below.
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# =========================================================================== # ===========================================================================
@@ -23,10 +27,6 @@
#default: <workspace>/.mam/agent-sessions.yaml #default: <workspace>/.mam/agent-sessions.yaml
# AGENT_SESSIONS_YAML=/path/to/workspace/.mam/agent-sessions.yaml # AGENT_SESSIONS_YAML=/path/to/workspace/.mam/agent-sessions.yaml
# Where the monitor (reconcile.sh) keeps its drift-state cache.
#default: <workspace>/.cache/multi-agent-mux-monitor
# AGENT_SESSIONS_STATE_DIR=/path/to/workspace/.cache/multi-agent-mux-monitor
# Root directory that holds Claude Code per-project conversation logs (*.jsonl). # Root directory that holds Claude Code per-project conversation logs (*.jsonl).
#default: $HOME/.claude/projects #default: $HOME/.claude/projects
# CLAUDE_PROJECT_DIR=$HOME/.claude/projects # CLAUDE_PROJECT_DIR=$HOME/.claude/projects
@@ -35,11 +35,6 @@
#default: $HOME/.local/bin #default: $HOME/.local/bin
# LOCAL_BIN=$HOME/.local/bin # LOCAL_BIN=$HOME/.local/bin
# tmux server socket name (`tmux -L <name>`). "default" = the normal tmux server
# (no -L). Set this to opt into an isolated server for all skill tmux calls.
#default: default
# TMUX_SERVER_NAME=default
# =========================================================================== # ===========================================================================
# delegate-job / MQTT broker # delegate-job / MQTT broker
# =========================================================================== # ===========================================================================