multi-agent-mux/.agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh

#!/usr/bin/env bash
# reconcile.sh — multi-agent-mux-monitor 의 부속 스크립트
# YAML ↔ herdr ↔ 디스크 artifact 간 drift 감지 (+ YAML 자동 갱신).
#
# Usage:
#   bash reconcile.sh --once --emit-diff            # drift 감지 + 갱신
#   bash reconcile.sh --once --emit-diff --dry-run  # drift 만 계산, 쓰기 안 함 (P1-E)
#
# --dry-run: 부수효과 없는 read-only. "지금 뭐 돌고 있지?" 질문에 안전.
#            multi-agent-mux-status 스킬이 이걸 재사용.
#            어떤 분기로도 디스크/DB 쓰기가 발생하지 않음을 보장함.
#
# 출력 (JSON): {timestamp, yaml_path, herdr_sessions_alive, herdr_confirmed, drifts, actions}
#
# Exit codes: 0 = ok | 1 = YAML not found | 2 = error
set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
SKILLS_DIR="$(cd "$SCRIPT_DIR/../.." && pwd)"
LIB_SH="$SKILLS_DIR/lib.sh"
[ -f "$LIB_SH" ] || LIB_SH="${WORKSPACE_ROOT:-$PWD}/.agents/skills/lib.sh"
source "$LIB_SH"

# ISSUE-9: the reconciler only observes and marks - it never creates a pane -
# so no herdr call it makes (the per-server `ls` sweep on every heartbeat, in
# --dry-run and in --subscribe alike) may bootstrap a server as a side effect.
# A server that is down simply has no live sessions.
export MAM_NO_AUTOSTART=1
if [ -n "${AGENT_SESSIONS_STATE_DIR:-}" ]; then
  echo "Notice: AGENT_SESSIONS_STATE_DIR is set but has no effect; the monitor keeps no state directory (C-2)." >&2
fi

ONCE=0
EMIT_DIFF=0
DRY_RUN=0
SUBSCRIBE=0
# --subscribe controls (review item 4): 0 = no overall timeout; idle default 3600s
# (raised from 600s to align with job timeout defaults); idle 0 = never idle-out.
SUB_TIMEOUT=0
SUB_IDLE_TIMEOUT=3600
POLL_INTERVAL="${RECONCILE_POLL_INTERVAL:-15}"

while [ $# -gt 0 ]; do
  case "$1" in
    --once) ONCE=1; shift ;;
    --emit-diff) EMIT_DIFF=1; shift ;;
    --dry-run) DRY_RUN=1; shift ;;
    --subscribe) SUBSCRIBE=1; shift ;;
    --timeout) SUB_TIMEOUT="$2"; shift 2 ;;
    --idle-timeout) SUB_IDLE_TIMEOUT="$2"; shift 2 ;;
    -h|--help)
      cat <<EOF
Usage: $0 [--once] [--emit-diff] [--dry-run] [--subscribe [--timeout N] [--idle-timeout N]]

Options:
  --dry-run   Runs in read-only mode, guaranteeing no database or file writes are performed.
EOF
      exit 0
      ;;
    *) echo "ERROR: unknown arg: $1" >&2; exit 2 ;;
  esac
done

[ -f "$AGENT_SESSIONS_YAML" ] || { echo "ERROR: $AGENT_SESSIONS_YAML not found" >&2; exit 1; }

if [ "$SUBSCRIBE" = "1" ]; then
  # Paths resolved relative to this script (review item 6): skills/ dir + lib.sh.
  SKILLS_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
  LIB_SH="$SKILLS_DIR/lib.sh"
  # MQTT client lives in the project venv (has paho). All YAML work is delegated
  # to lib.sh::atomic_dump_yaml, which runs the system python3 (has PyYAML) — so
  # no single interpreter needs both paho and PyYAML (review items 4/5/6).
  PYBIN="$(_delegate_py_bin)"

  # Issue #19: the subscriber runs as a background child and this shell waits
  # on it, so SIGTERM/SIGINT reach the daemon promptly. With the child in the
  # foreground bash deferred the signal until the child exited on its own,
  # which for --idle-timeout 0 was never - only SIGKILL worked.
  _sub_pid=""
  _sub_stop() {
    trap - TERM INT
    if [ -n "$_sub_pid" ]; then
      kill -TERM "$_sub_pid" 2>/dev/null || true
      wait "$_sub_pid" 2>/dev/null || true
    fi
    exit 143
  }
  trap _sub_stop TERM INT

  # The MQTT subscribe loop exits 3 to signal "broker unavailable → poll instead".
  set +e
  YAML_PATH="$AGENT_SESSIONS_YAML" HOME_DIR="$HOME_DIR" CLAUDE_PROJECT_DIR="$CLAUDE_PROJECT_DIR" LOCAL_BIN="$LOCAL_BIN" \
    WORKSPACE_ROOT="$WORKSPACE_ROOT" SUB_TIMEOUT="$SUB_TIMEOUT" SUB_IDLE_TIMEOUT="$SUB_IDLE_TIMEOUT" \
    SKILLS_DIR="$SKILLS_DIR" LIB_SH="$LIB_SH" \
    "$PYBIN" - <<'PYEOF' &
import os, sys, json, time, subprocess

lib_sh = os.environ.get('LIB_SH', '')
skills_dir = os.environ.get('SKILLS_DIR', '')
yaml_path = os.environ.get('YAML_PATH', '')
workspace_root = os.environ.get('WORKSPACE_ROOT', '')
timeout = int(os.environ.get('SUB_TIMEOUT', '0') or '0')          # 0 = no overall timeout
idle_timeout = int(os.environ.get('SUB_IDLE_TIMEOUT', '3600') or '0')  # 0 = no idle timeout

# Prevent duplicate wildcard subscribers for this workspace (concurrency race)
import fcntl
lock_file_path = os.path.join(workspace_root or '.', '.mam', 'monitor.lock')
try:
    os.makedirs(os.path.dirname(lock_file_path), exist_ok=True)
    lock_file = open(lock_file_path, 'w')
    fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
    print("MQTT Monitor: another subscriber is already running for this workspace. Exiting.", flush=True)
    sys.exit(0)
except Exception as e:
    print(f"MQTT Monitor: failed to acquire monitor lock ({e}). Exiting.", flush=True)
    sys.exit(1)
lock_ino = os.fstat(lock_file.fileno()).st_ino


def _workspace_gone():
    """Issue #19: the daemon serves exactly one workspace. Once that workspace
    is deleted (a purged sandbox, an rm -rf'd project) or its `.mam/monitor.lock`
    is no longer the inode we hold (a re-created `.mam/`, where a fresh daemon
    could now take a new lock), there is nothing left to reconcile - exit
    instead of holding a broker subscription forever."""
    try:
        root = os.path.abspath(workspace_root) if workspace_root else os.getcwd()
    except OSError:
        return "cwd no longer exists"
    if not os.path.isdir(root):
        return f"workspace {root} no longer exists"
    try:
        if os.stat(lock_file_path).st_ino != lock_ino:
            return f"{lock_file_path} was replaced"
    except OSError:
        return f"{lock_file_path} no longer exists"
    return None

# Locate skills/multi-agent-mux-delegate-job/scripts to import mqtt_common — relative first, then
# an upward walk from cwd. No hardcoded absolute path (review item 6).
cand = os.path.join(skills_dir, 'multi-agent-mux-delegate-job', 'scripts') if skills_dir else ''
if cand and os.path.isdir(cand):
    sys.path.append(cand)
else:
    d = os.getcwd()
    while d and d != '/':
        hit = None
        for sub in (('.agents', 'skills', 'multi-agent-mux-delegate-job', 'scripts'), ('skills', 'multi-agent-mux-delegate-job', 'scripts'), ('multi-agent-mux-delegate-job', 'scripts')):
            p = os.path.join(d, *sub)
            if os.path.isdir(p):
                hit = p
                break
        if hit:
            sys.path.append(hit)
            break
        d = os.path.dirname(d)

import mqtt_common
import registry
mqtt_common.ensure_env_loaded()   # D6: env now loads at entry points, not at import

# Executed INSIDE lib.sh::atomic_dump_yaml (system python3 + PyYAML), under the
# YAML flock with schema-validate + .bak (review item 5). Marks matching running
# sessions terminated and kills their herdr (review item 3 behaviour preserved),
# or aborts the write entirely when nothing matches. The untrusted MQTT job id /
# event arrive via env (MQTT_JID / MQTT_EVENT) — never spliced into source (P1-B).
_MUTATION = r'''
import os, subprocess
from datetime import datetime, timezone
_jid = os.environ['MQTT_JID']
_event = os.environ['MQTT_EVENT']
_now = datetime.now(timezone.utc)
_changed = False
for s in d.get('herdr_sessions', []):
    if s.get('delegate_job_id') == _jid and s.get('status') == 'running':
        _name = s.get('name')
        # herdr_workspace 는 워크스페이스 *라벨* 이지 소켓 이름이 아니다.
        _srv = s.get('herdr_session') or s.get('herdr_server') or 'default'
        if _event in ('completed', 'cancelled'):
            s['delegate_job_id'] = None
            print('MQTT Monitor: job ' + _event + ' on ' + str(_name) + ' — session kept alive', flush=True)
            _changed = True
        else:
            s['status'] = 'terminated'
            s['terminated_at'] = _now.strftime('%Y-%m-%dT%H:%M:%SZ')
            s['terminated_at_epoch'] = int(_now.timestamp())
            s['termination_mode'] = 'auto-detected (MQTT ' + _event + ')'
            _cmd = ['herdr'] + (['-L', _srv] if _srv != 'default' else []) + ['kill-session', '-t', _name]
            subprocess.run(_cmd, capture_output=True)
            print('MQTT Monitor: terminated + killed ' + str(_name) + ' on ' + str(_srv) + ' due to MQTT ' + _event, flush=True)
            _changed = True
if not _changed:
    raise SystemExit(0)  # nothing matched — skip the write entirely
'''


def handle_terminal(jid, event):
    if not lib_sh or not os.path.isfile(lib_sh):
        print('MQTT Monitor: lib.sh not found, cannot update YAML', flush=True)
        return
    env = dict(os.environ)
    env['MQTT_JID'] = jid
    env['MQTT_EVENT'] = event
    cmd = ['bash', '-c',
           'source "$LIB_SH"; atomic_dump_yaml "$YAML_PATH" MQTT_JID="$MQTT_JID" MQTT_EVENT="$MQTT_EVENT"']
    r = subprocess.run(cmd, input=_MUTATION, text=True, env=env, capture_output=True)
    if (r.stdout or '').strip():
        print(r.stdout.strip(), flush=True)
    if r.returncode != 0 and (r.stderr or '').strip():
        print('MQTT Monitor: atomic_dump_yaml stderr: ' + r.stderr.strip(), flush=True)


state = {'last_msg': time.time(), 'connected': False, 'failed': False}
last_seqs = {}


def on_message(_client, _userdata, msg):
    state['last_msg'] = time.time()
    try:
        payload = json.loads(msg.payload.decode("utf-8"))
        jid = payload.get("job_id")
        event = payload.get("event")
        if not jid or not event:
            return
        
        if workspace_root:
            registry_dir = os.path.join(workspace_root, '.mam', 'jobs')
        else:
            yaml_dir = os.path.dirname(yaml_path) if yaml_path else ""
            registry_dir = os.path.join(yaml_dir, 'jobs') if yaml_dir else '.mam/jobs'
        
        try:
            job = registry.load_job(jid, registry_dir)
        except FileNotFoundError:
            # Silently ignore events for jobs not in the local registry
            return
            
        expected_token = job.get("auth_token")
        if not mqtt_common.verify_hmac(payload, expected_token):
            print(f"MQTT Monitor: drop event for job {jid}: HMAC verify failed", flush=True)
            return
            
        seq = payload.get("seq")
        if seq is None or not isinstance(seq, int):
            print(f"MQTT Monitor: drop event for job {jid}: missing or invalid seq", flush=True)
            return
        if seq <= last_seqs.get(jid, 0):
            print(f"MQTT Monitor: drop event for job {jid}: seq {seq} not monotonic (last {last_seqs.get(jid, 0)})", flush=True)
            return
        last_seqs[jid] = seq
        
        # Append the event to events.ndjson audit trail
        mqtt_common.append_event(jid, {
            "event": "received",
            "source_event": event,
            "seq": seq,
            "topic": msg.topic,
            "timestamp": payload.get("timestamp"),
            "detail": payload.get("detail", ""),
        })
        
        print(f"MQTT Monitor: recorded event {event} for job {jid} (seq={seq})", flush=True)
        
        if event in ("completed", "error", "cancelled"):
            print(f"MQTT Monitor: received terminal event {event} for job {jid}", flush=True)
            handle_terminal(jid, event)
    except Exception as e:
        print(f"MQTT Monitor error parsing message: {e}", flush=True)


def on_connect(_c, _u, _flags, reason_code, _props):
    rc = mqtt_common.reason_code_value(reason_code)
    if rc == 0:
        state['connected'] = True
        ws_path = os.path.abspath(workspace_root) if workspace_root else os.getcwd()
        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:
        state['failed'] = True
        print(f"MQTT Monitor connection failed: rc={rc}", flush=True)


cfg = mqtt_common.broker_config_from_env()
client = mqtt_common.make_client("monitor_sub", cfg)
client.on_message = on_message
client.on_connect = on_connect
print(f"MQTT Monitor: connecting to {cfg.host}:{cfg.port} (TLS={cfg.tls})...", flush=True)

# Connection failure → fall back to polling (review item 4).
try:
    client.connect(cfg.host, cfg.port, cfg.keepalive)
except Exception as e:
    print(f"MQTT Monitor: connect failed ({e}); falling back to polling", flush=True)
    sys.exit(3)

client.loop_start()
_wait = time.time()
while time.time() - _wait < 5 and not state['connected'] and not state['failed']:
    time.sleep(0.1)
if not state['connected']:
    print("MQTT Monitor: broker did not accept connection; falling back to polling", flush=True)
    client.loop_stop()
    sys.exit(3)

import threading, signal
stop_event = threading.Event()
state['stop_signal'] = None


def _on_stop_signal(signum, _frame):
    state['stop_signal'] = signum
    stop_event.set()


signal.signal(signal.SIGTERM, _on_stop_signal)
signal.signal(signal.SIGINT, _on_stop_signal)
start = time.time()
try:
    while True:
        if state['stop_signal'] is not None:
            print(f"MQTT Monitor: signal {state['stop_signal']} received, exiting", flush=True)
            break
        gone = _workspace_gone()
        if gone:
            print(f"MQTT Monitor: {gone}; exiting", flush=True)
            break
        now = time.time()
        wall_left = (timeout - (now - start)) if timeout else None
        idle_left = (idle_timeout - (now - state['last_msg'])) if idle_timeout else None
        
        next_timeout = 5.0
        if wall_left is not None:
            next_timeout = min(next_timeout, wall_left)
        if idle_left is not None:
            next_timeout = min(next_timeout, idle_left)
            
        if next_timeout <= 0:
            if wall_left is not None and wall_left <= 0:
                print(f"MQTT Monitor: --timeout {timeout}s reached, exiting", flush=True)
            else:
                print(f"MQTT Monitor: --idle-timeout {idle_timeout}s reached, exiting", flush=True)
            break
            
        stop_event.wait(timeout=next_timeout)
finally:
    client.loop_stop()
    try:
        client.disconnect()
    except Exception:
        pass
sys.exit(0)
PYEOF
  _sub_pid=$!
  wait "$_sub_pid"
  sub_rc=$?
  _sub_pid=""
  set -e

  if [ "$sub_rc" = "3" ]; then
    echo "MQTT Monitor: broker unavailable — falling back to polling (interval ${POLL_INTERVAL}s)" >&2
    _self="$SKILLS_DIR/multi-agent-mux-monitor/scripts/reconcile.sh"
    _start=$(date +%s)
    # Issue #19: same lifecycle as the MQTT loop - stop when the workspace is
    # gone, and sleep in a waited child so SIGTERM/SIGINT are not deferred.
    trap 'trap - TERM INT; exit 143' TERM INT
    while :; do
      if [ ! -d "$WORKSPACE_ROOT" ] || [ ! -f "$WORKSPACE_ROOT/.mam/monitor.lock" ]; then
        echo "MQTT Monitor: workspace $WORKSPACE_ROOT or its monitor.lock is gone; exiting" >&2
        break
      fi
      bash "$_self" --once --emit-diff >/dev/null 2>&1 || true
      if [ "$SUB_TIMEOUT" != "0" ] && [ "$(( $(date +%s) - _start ))" -ge "$SUB_TIMEOUT" ]; then
        break
      fi
      sleep "$POLL_INTERVAL" &
      wait $! || true
    done
  fi
  exit 0
fi

# 모든 비교 로직을 단일 소스로 둔다. dry-run 은 env_python(읽기전용), 그 외엔
# atomic_dump_yaml(flock + temp+rename) 로 같은 소스를 돌린다. atomic 래퍼에서는
# 'actions' 가 없으면 SystemExit(0) 으로 쓰기를 건너뛴다 (불필요한 재포맷 방지).
read -r -d '' RECON_SRC <<'PYEOF' || true
import os, json, glob, subprocess, time, sqlite3, re, shutil
from datetime import datetime, timezone
import yaml

from lib_py.verify_session import verify_session_uuid, workspace_key
from lib_py.agents.registry import get_adapter
from lib_py.agents.base import DiscoveryContext

def _slug(path):
    if not path:
        return ''
    a = os.path.abspath(path)
    parent = os.path.basename(os.path.dirname(a)) or 'workspace'
    work = os.path.basename(a) or 'root'
    if parent in ('/', '.'): parent = 'workspace'
    if work in ('/', '.'): work = 'root'
    s = f'{parent}-{work}'.lower().replace('_', '-')
    return re.sub(r'[^a-zA-Z0-9-]', '', s).lstrip('-')

yaml_path = os.environ['YAML_PATH']
home = os.environ['HOME_DIR']
skills_dir = os.environ.get('SKILLS_DIR', '')
if not skills_dir:
    _ws_root = os.environ.get('WORKSPACE_ROOT', '')
    if _ws_root:
        skills_dir = os.path.join(_ws_root, '.agents/skills')
    else:
        skills_dir = ''
claude_project_dir = os.environ.get('CLAUDE_PROJECT_DIR', f"{home}/.claude/projects")

now_iso = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')

# Bound unconditionally. This used to be assigned only inside the "except
# NameError" branch below, which the write path never enters because
# atomic_dump_yaml predefines `d` -- so drift C's pin raised
# NameError: name 'lib_sh' is not defined and aborted the whole sweep,
# in write mode only.
lib_sh = os.environ.get('LIB_SH', '')
if not lib_sh:
    if skills_dir:
        lib_sh = os.path.join(skills_dir, 'lib.sh')
    else:
        _ws_root = os.environ.get('WORKSPACE_ROOT', '')
        if _ws_root:
            lib_sh = os.path.join(_ws_root, '.agents/skills/lib.sh')
        else:
            lib_sh = ''

try:
    d
except NameError:
    try:
        from lib_py.state import load_state_json
        d = load_state_json(yaml_path)
    except Exception:
        import subprocess
        d = {}
        try:
            script = f"source '{lib_sh}' && load_state_json"
            out = subprocess.check_output(['bash', '-c', script], stderr=subprocess.DEVNULL)
            d = json.loads(out.decode('utf-8'))
        except Exception:
            pass

# Any herdr_session_epoch at or below this is not a real session time.
# 1000000000 = 2001-09-09; it is below every plausible MAM session and far
# above the 0 / 999999 sentinels this guard exists to reject.
MAM_EPOCH_FLOOR = 1000000000

drifts = []
actions = []

# === 현재 herdr 상태 — transient 실패를 'no sessions' 와 구분 (P1-E) ===
herdr_sessions = []
herdr_confirmed = True

# YAML 에 등록된 고유한 herdr_session 목록 수집 + 환경변수 HERDR_SESSION_NAME 포함
unique_servers = {'default'}
if 'HERDR_SESSION_NAME' in os.environ:
    unique_servers.add(os.environ['HERDR_SESSION_NAME'])
elif 'HERDR_SERVER_NAME' in os.environ:
    unique_servers.add(os.environ['HERDR_SERVER_NAME'])
for s in d.get('herdr_sessions', []):
    # herdr_workspace 는 워크스페이스 *라벨* 이지 소켓 이름이 아니다.
    srv = s.get('herdr_session') or s.get('herdr_server') or 'default'
    unique_servers.add(srv)

try:
    for srv in sorted(unique_servers):
        cmd = ['herdr']
        if srv != 'default':
            cmd += ['-L', srv]
        cmd += ['ls', '-F', '#{session_name}|#{session_created}']
        r = subprocess.run(cmd, capture_output=True, text=True)
        if r.returncode == 0:
            for line in r.stdout.strip().split('\n'):
                if not line or '|' not in line:
                    continue
                name, created = line.split('|', 1)
                # A malformed field must not abort the whole sweep: the old
                # bare int() raised out of the enclosing try and flipped
                # herdr_confirmed to False for every server, which reads as
                # "herdr is down" and suppresses drift detection entirely.
                try:
                    created_i = int(created.strip())
                except ValueError:
                    created_i = 0
                # Below the floor means "no real creation time" -- 0, or the
                # 999999 sentinel the shim used to emit (B-4). Both are far
                # below any live transcript's mtime, so they silently disable
                # the stale-transcript guard in verify_session_uuid.
                if created_i < MAM_EPOCH_FLOOR:
                    created_i = 0
                herdr_sessions.append({'name': name, 'created': created_i, 'server': srv})
        else:
            err = (r.stderr or '').lower()
            is_empty = ('no server running' in err) or ('no sessions' in err) or ('failed to connect' in err)
            if not is_empty:
                herdr_confirmed = False
except Exception:
    herdr_confirmed = False


def pane_meta(session, srv):
    try:
        cmd = ['herdr']
        if srv != 'default':
            cmd += ['-L', srv]
        cmd += ['list-panes', '-t', session, '-F',
                '#{pane_pid}|#{pane_current_path}|#{pane_current_command}']
        out = subprocess.check_output(cmd, text=True)
        parts = out.strip().split('\n')[0].split('|')
        return {'pid': int(parts[0]), 'cwd': parts[1], 'cmd': parts[2]}
    except Exception:
        return None


from lib_py.agents.registry import get_adapter, own_key as _get_own_key, agent_of_row

def _pin_and_verify_resume(s, agent, cwd, uuid, degraded=False):
    own_key_field = _get_own_key(agent)
    if own_key_field:
        s[own_key_field] = uuid
    s['last_visible_status'] = 'pinned'
    resume_cmd = ['bash', os.path.join(skills_dir, 'multi-agent-mux-resume', 'scripts', 'resume_session.sh'),
                  '--workspace', cwd, '--agent', agent, '--session', s['name'], '--dry-run']
    res = subprocess.run(resume_cmd, capture_output=True, text=True)
    if res.returncode == 0:
        s['last_visible_status'] = 'resume_verified'
    else:
        s['last_visible_status'] = f"resume dry-run failed: {res.stderr.strip() or res.stdout.strip()}"
    
    id_name = 'session' if agent == 'claude' else 'conversation'
    if not degraded:
        drifts.append({'class': 'C', 'name': s['name'], 'msg': f"{s['name']}: {id_name} id materialized: {uuid}"})
    else:
        drifts.append({'class': 'C-degraded', 'name': s['name'], 'msg': f"{s['name']}: TUI viewport check unavailable, pinned via stage 1-3 only"})
    actions.append(f"updated {id_name} id: {uuid}")


from lib_py.agents.sanitize import sanitize_herdr_agent_name as _sanitize

yaml_sessions = d.get('herdr_sessions', [])
yaml_session_names = {s['name'] for s in yaml_sessions if s.get('name')}
alive_set = {(t['name'], t.get('server', 'default')) for t in herdr_sessions}

# === drift A: YAML running + herdr dead → mark terminated ===
if herdr_confirmed:
    for s in yaml_sessions:
        name = s.get('name')
        if not name:
            continue
        # 'stopped' 도 deliberate한 종료 상태 — drift 로 보지 않고 그대로 둔다.
        # (없으면 herdr-dead stopped 세션을 'terminated' 로 덮어써 resumable 플래그가 소실됨)
        if s.get('status') in ('terminated', 'archived', 'stopped'):
            continue
        # herdr_workspace 는 워크스페이스 *라벨* 이지 소켓 이름이 아니다.
        srv = s.get('herdr_session') or s.get('herdr_server') or 'default'
        if (name, srv) not in alive_set and (_sanitize(name), srv) not in alive_set:
            s['status'] = 'terminated'
            s['terminated_at'] = now_iso
            s['terminated_at_epoch'] = int(datetime.now(timezone.utc).timestamp())
            s['termination_mode'] = 'auto-detected (herdr gone)'
            pane = s.get('pane') or {}
            drifts.append({'class': 'A', 'name': name,
                           'msg': f"{name}: herdr gone (was pane {pane.get('pid')}, cmd {pane.get('cmd')}). Marked terminated."})
            actions.append(f"terminated: {name}")

# === drift B: herdr alive + not in YAML → auto-register ===
if herdr_confirmed:
    for t in herdr_sessions:
        name = t['name']
        srv = t.get('server', 'default')
        if (name, srv) in yaml_session_names or any(_sanitize(y) == name for y in yaml_session_names):
            continue
        workspace_root = os.environ.get('WORKSPACE_ROOT')
        if not workspace_root:
            workspace_root = os.path.abspath(os.path.join(os.path.dirname(yaml_path), '..'))
        if os.path.exists(os.path.join(workspace_root, '.mam', f"purging-{name}")):
            continue
        agent = None
        role = 'creator'
        for r_name in ('creator', 'planner', 'reviewer'):
            for a_name in ('claude', 'agy', 'hermes', 'grok', 'opencode', 'codex'):
                if name.endswith(f"-{r_name}-{a_name}"):
                    role = r_name
                    agent = a_name
                    break
            if agent:
                break

        if not agent:
            # Check MAM_MANAGED env marker from pane process environment if available
            pm_check = pane_meta(name, srv)
            if pm_check and pm_check.get('pid'):
                try:
                    pid_val = pm_check['pid']
                    env_output = subprocess.check_output(['ps', 'eww', '-p', str(pid_val)], text=True, stderr=subprocess.DEVNULL)
                    for tok in env_output.split():
                        if tok.startswith('MAM_MANAGED='):
                            managed_path = tok.split('=', 1)[1]
                            if os.path.realpath(managed_path) == os.path.realpath(workspace_root):
                                for a_name in ('claude', 'agy', 'hermes', 'grok', 'opencode', 'codex'):
                                    if a_name in pm_check.get('cmd', '') or a_name in pm_check.get('cmd_full', ''):
                                        agent = a_name
                                        break
                            break
                except Exception:
                    pass

        if not agent:
            continue

        pm = pane_meta(name, srv)
        if not pm:
            continue
        pane_cwd_abs = os.path.realpath(pm['cwd']) if pm.get('cwd') else ''
        ws_root_abs = os.path.realpath(workspace_root)
        if not pane_cwd_abs or not (pane_cwd_abs == ws_root_abs or pane_cwd_abs.startswith(ws_root_abs + os.sep)):
            continue
        _adapter = get_adapter(agent)
        cmd_full = _adapter.spawn_spec(agent) if _adapter else agent
        server_opt = f"-L {srv} " if srv != 'default' else ""
        # The shim resolves this from the pane's root process. Fall back to now
        # only if that failed: 'now' can merely over-estimate creation time,
        # which tightens the stale-transcript guard, whereas 0 disables it and
        # stamps a 1970-01-01 herdr_session_created_at into the YAML.
        created_epoch = t.get('created') or 0
        if created_epoch < MAM_EPOCH_FLOOR:
            created_epoch = int(time.time())
        entry = {
            'name': name,
            'status': 'running',
            'role': role,
            'herdr_session_created_at': datetime.fromtimestamp(created_epoch, tz=timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'),
            'herdr_session_epoch': created_epoch,
            'herdr_session': srv,
            'herdr_server': srv,
            'herdr_workspace': _slug(pm['cwd']),
            'pane': {'index': 0, 'pid': pm['pid'], 'cmd': agent, 'cmd_full': cmd_full, 'cwd': pm['cwd']},
            'start_command': f'HERDR_SESSION_NAME={srv} herdr new-session -d -s "{name}" -x 140 -y 40 -c "{pm["cwd"]}" "{cmd_full}"',
            'attach_command': f'HERDR_SESSION_NAME={srv} herdr agent attach {name}',
            'kill_command': f'herdr {server_opt}kill-session -t {name}',
            'last_visible_status': 'running',
            'last_visible_note': 'auto-registered by monitor',
        }
        if agent == 'claude':
            entry['tui'] = {'model': '(unknown — capture after first message)', 'provider': 'anthropic',
                            'plan': '(unknown)', 'account': '(unknown)', 'version': '(unknown)'}
            entry['claude_session_id_own'] = None
        elif agent == 'agy':
            entry['child_pid'] = 0
            entry['agy_conversation_id_own'] = None
            entry['mcp_attachments'] = [
                {
                    'name': 'stitch',
                    'transport': 'mcp-remote',
                    'endpoint': 'https://stitch.googleapis.com/mcp'
                }
            ]
        elif agent == 'hermes':
            entry['child_pid'] = 0
            entry['hermes_conversation_id_own'] = None
        elif agent == 'opencode':
            entry['child_pid'] = 0
            entry['opencode_session_id_own'] = None
        elif agent == 'codex':
            entry['child_pid'] = 0
            entry['codex_session_id_own'] = None
        d.setdefault('herdr_sessions', []).append(entry)
        yaml_session_names.add(name)
        drifts.append({'class': 'B', 'name': name,
                       'msg': f"{name}: herdr found but not in YAML. Auto-registered (pane {pm['pid']}, cmd {pm['cmd']}, cwd {pm['cwd']})."})
        actions.append(f"registered: {name}")

def row_agent(s):
    return agent_of_row(s)

OWN_KEY_BY_AGENT = {
    a: _get_own_key(a) for a in ('claude', 'agy', 'hermes', 'grok', 'opencode', 'codex')
}

# === drift C0: 지정된 ID 는 발견이 아니라 '확인'만 필요하다 ===
for s in d.get('herdr_sessions', []):
    if s.get('status') != 'running':
        continue
    if s.get('session_id_source') != 'assigned' or s.get('session_id_verified'):
        continue
    agent = row_agent(s)
    if not agent:
        continue
    uuid = s.get(OWN_KEY_BY_AGENT[agent])
    cwd = (s.get('pane') or {}).get('cwd', '')
    if not uuid or not cwd:
        continue
    if verify_session_uuid(cwd, agent, uuid, s, mode="discover"):
        s['session_id_verified'] = True
        s['last_visible_status'] = 'pinned'
        drifts.append({'class': 'C', 'name': s['name'],
                       'msg': f"{s['name']}: assigned session id confirmed on disk: {uuid}"})
        actions.append(f"confirmed session id: {uuid}")

# === drift C: claude 새 session id materialize (per-row own id) ===
for s in d.get('herdr_sessions', []):
    if row_agent(s) != 'claude':
        continue
    if s.get('status') != 'running':
        continue
    if s.get('claude_session_id_own'):
        continue
    cwd = (s.get('pane') or {}).get('cwd', '')
    if not cwd:
        continue
    key = workspace_key(cwd)
    proj_dir = f"{claude_project_dir}/{key}"
    if not os.path.isdir(proj_dir):
        continue
    jsonls = sorted(glob.glob(f"{proj_dir}/*.jsonl"), key=os.path.getmtime, reverse=True)
    
    valid_candidates = []
    for latest in jsonls:
        uuid = os.path.basename(latest)[:-6]
        if verify_session_uuid(cwd, 'claude', uuid, s, mode="discover"):
            valid_candidates.append(uuid)
            
    if len(valid_candidates) > 1:
        drifts.append({'class': 'C-ambiguous', 'name': s['name'],
                       'msg': f"{s['name']}: {len(valid_candidates)} candidate transcripts newer than session epoch; not pinning"})
        s['last_visible_status'] = f"ambiguous: {len(valid_candidates)} candidates"
        actions.append(f"ambiguous candidates: {s['name']}")
    if len(valid_candidates) == 1:
        uuid = valid_candidates[0]
        cmd = ['bash', '-c', f'source "{lib_sh}" && verify_tui_viewport "{s["name"]}" "claude" "{cwd}"']
        rc = subprocess.run(cmd).returncode
        if rc == 0:
            _pin_and_verify_resume(s, 'claude', cwd, uuid, degraded=False)
        elif rc == 1:
            drifts.append({'class': 'C-warn', 'name': s['name'], 'msg': f"{s['name']}: TUI viewport mismatch for candidate {uuid} — not pinned, will retry next cycle"})
        else:
            _pin_and_verify_resume(s, 'claude', cwd, uuid, degraded=True)

# === drift C (agy): agy 새 session id materialize (per-row own id) ===
for s in d.get('herdr_sessions', []):
    if row_agent(s) != 'agy':
        continue
    if s.get('status') != 'running':
        continue
    if s.get('agy_conversation_id_own'):
        continue
    cwd = (s.get('pane') or {}).get('cwd', '')
    if not cwd:
        continue
    conv_dir = f"{home}/.gemini/antigravity-cli/conversations"
    if not os.path.isdir(conv_dir):
        continue
    dbs = sorted(glob.glob(f"{conv_dir}/*.db"), key=os.path.getmtime, reverse=True)
    
    sibling_claimed = [
        other.get('agy_conversation_id_own')
        for other in d.get('herdr_sessions', [])
        if other is not s
        and (other.get('pane') or {}).get('cwd') == cwd
        and other.get('status') not in ('stopped', 'terminated')
        and other.get('agy_conversation_id_own')
    ]
    s_eval = dict(s)
    s_eval['_sibling_claimed_uuids'] = sibling_claimed

    valid_candidates = []
    for db in dbs:
        uuid = os.path.basename(db)[:-3]
        if verify_session_uuid(cwd, 'agy', uuid, s_eval, mode="discover"):
            valid_candidates.append(uuid)
            
    if len(valid_candidates) > 1:
        drifts.append({'class': 'C-ambiguous', 'name': s['name'],
                       'msg': f"{s['name']}: {len(valid_candidates)} candidate transcripts newer than session epoch; not pinning"})
        s['last_visible_status'] = f"ambiguous: {len(valid_candidates)} candidates"
        actions.append(f"ambiguous candidates: {s['name']}")
    if len(valid_candidates) == 1:
        uuid = valid_candidates[0]
        cmd = ['bash', '-c', f'source "{lib_sh}" && verify_tui_viewport "{s["name"]}" "agy" "{cwd}"']
        rc = subprocess.run(cmd).returncode
        if rc == 0:
            _pin_and_verify_resume(s, 'agy', cwd, uuid, degraded=False)
        elif rc == 1:
            drifts.append({'class': 'C-warn', 'name': s['name'], 'msg': f"{s['name']}: TUI viewport mismatch for candidate {uuid} — not pinned, will retry next cycle"})
        else:
            _pin_and_verify_resume(s, 'agy', cwd, uuid, degraded=True)

# === drift C (hermes): hermes 새 session id materialize (per-row own id) ===
for s in d.get('herdr_sessions', []):
    if row_agent(s) != 'hermes':
        continue
    if s.get('status') != 'running':
        continue
    if s.get('hermes_conversation_id_own'):
        continue
    cwd = (s.get('pane') or {}).get('cwd', '')
    if not cwd:
        continue
    hdb = f"{home}/.hermes/state.db"
    if not os.path.exists(hdb):
        continue

    epoch_threshold = s.get('herdr_session_epoch', 0)

    sibling_claimed = [
        other.get('hermes_conversation_id_own')
        for other in d.get('herdr_sessions', [])
        if other is not s
        and (other.get('pane') or {}).get('cwd') == cwd
        and other.get('status') not in ('stopped', 'terminated')
        and other.get('hermes_conversation_id_own')
    ]
    s_eval = dict(s)
    s_eval['_sibling_claimed_uuids'] = sibling_claimed

    valid_candidates = []
    try:
        conn = sqlite3.connect(hdb)
        rows = conn.execute(
            "SELECT id FROM sessions WHERE cwd=? AND started_at >= ? ORDER BY started_at DESC LIMIT 20",
            (cwd, epoch_threshold)
        ).fetchall()
        conn.close()
        for (uuid,) in rows:
            if uuid in (sibling_claimed or []):
                continue
            if verify_session_uuid(cwd, 'hermes', uuid, s_eval, mode="discover"):
                valid_candidates.append(uuid)
    except Exception:
        pass

    if len(valid_candidates) > 1:
        drifts.append({'class': 'C-ambiguous', 'name': s['name'],
                       'msg': f"{s['name']}: {len(valid_candidates)} candidate transcripts newer than session epoch; not pinning"})
        s['last_visible_status'] = f"ambiguous: {len(valid_candidates)} candidates"
        actions.append(f"ambiguous candidates: {s['name']}")
    if len(valid_candidates) == 1:
        uuid = valid_candidates[0]
        cmd = ['bash', '-c', f'source "{lib_sh}" && verify_tui_viewport "{s["name"]}" "hermes" "{cwd}"']
        rc = subprocess.run(cmd).returncode
        if rc == 0:
            _pin_and_verify_resume(s, 'hermes', cwd, uuid, degraded=False)
        elif rc == 1:
            drifts.append({'class': 'C-warn', 'name': s['name'], 'msg': f"{s['name']}: TUI viewport mismatch for candidate {uuid} — not pinned, will retry next cycle"})
        else:
            _pin_and_verify_resume(s, 'hermes', cwd, uuid, degraded=True)

# === drift C (opencode): opencode 새 session id materialize (per-row own id) ===
for s in d.get('herdr_sessions', []):
    if row_agent(s) != 'opencode':
        continue
    if s.get('status') != 'running':
        continue
    if s.get('opencode_session_id_own'):
        continue
    cwd = (s.get('pane') or {}).get('cwd', '')
    if not cwd:
        continue

    # Resolve DB path via shared OpenCodeAgentAdapter helper with fallbacks
    odb = None
    try:
        op_adapter = get_adapter('opencode')
        if op_adapter:
            odb = op_adapter._resolve_db(DiscoveryContext(workspace=cwd, agent_name='opencode', home_dir=home))
    except Exception:
        pass

    if not odb:
        opencode_base = f"{home}/.local/share/opencode"
        candidates = [
            f"{opencode_base}/opencode.db",
            f"{opencode_base}/storage.db",
        ]
        for cand in candidates:
            if os.path.exists(cand):
                odb = cand
                break
        if not odb and os.path.isdir(f"{opencode_base}/project"):
            for root, _, files in os.walk(f"{opencode_base}/project"):
                for f in files:
                    if f.endswith(".db"):
                        odb = os.path.join(root, f)
                        break
                if odb and os.path.exists(odb):
                    break

    if not odb or not os.path.exists(odb):
        continue

    epoch_threshold = s.get('herdr_session_epoch', 0)

    sibling_claimed = [
        other.get('opencode_session_id_own')
        for other in d.get('herdr_sessions', [])
        if other is not s
        and (other.get('pane') or {}).get('cwd') == cwd
        and other.get('status') not in ('stopped', 'terminated')
        and other.get('opencode_session_id_own')
    ]
    s_eval = dict(s)
    s_eval['_sibling_claimed_uuids'] = sibling_claimed

    valid_candidates = []
    try:
        conn = sqlite3.connect(odb)
        cursor = conn.cursor()
        cursor.execute("PRAGMA table_info(session)")
        cols = [col[1] for col in cursor.fetchall()]
        if cols:
            cwd_col = "directory" if "directory" in cols else ("cwd" if "cwd" in cols else None)
            time_col = "time_created" if "time_created" in cols else ("created_at" if "created_at" in cols else ("started_at" if "started_at" in cols else None))

            query = "SELECT id FROM session"
            params = []
            where_clauses = []
            if cwd_col and cwd:
                where_clauses.append(f"{cwd_col} = ?")
                params.append(cwd)
            if time_col and epoch_threshold:
                if time_col == "time_created":
                    where_clauses.append(f"{time_col} >= ?")
                    params.append(int(float(epoch_threshold) * 1000))
                else:
                    where_clauses.append(f"{time_col} >= ?")
                    params.append(epoch_threshold)
            if where_clauses:
                query += " WHERE " + " AND ".join(where_clauses)
            if time_col:
                query += f" ORDER BY {time_col} DESC"
            query += " LIMIT 20"
            rows = cursor.execute(query, params).fetchall()
            conn.close()
            for (uuid,) in rows:
                if uuid in (sibling_claimed or []):
                    continue
                if verify_session_uuid(cwd, 'opencode', uuid, s_eval, mode="discover"):
                    valid_candidates.append(uuid)
        else:
            conn.close()
    except Exception:
        pass

    if len(valid_candidates) > 1:
        drifts.append({'class': 'C-ambiguous', 'name': s['name'],
                       'msg': f"{s['name']}: {len(valid_candidates)} candidate transcripts newer than session epoch; not pinning"})
        s['last_visible_status'] = f"ambiguous: {len(valid_candidates)} candidates"
        actions.append(f"ambiguous candidates: {s['name']}")
    if len(valid_candidates) == 1:
        uuid = valid_candidates[0]
        cmd = ['bash', '-c', f'source "{lib_sh}" && verify_tui_viewport "{s["name"]}" "opencode" "{cwd}"']
        rc = subprocess.run(cmd).returncode
        if rc == 0:
            _pin_and_verify_resume(s, 'opencode', cwd, uuid, degraded=False)
        elif rc == 1:
            drifts.append({'class': 'C-warn', 'name': s['name'], 'msg': f"{s['name']}: TUI viewport mismatch for candidate {uuid} — not pinned, will retry next cycle"})
        else:
            _pin_and_verify_resume(s, 'opencode', cwd, uuid, degraded=True)

# === drift C (codex): codex 새 session id materialize (per-row own id) ===
for s in d.get('herdr_sessions', []):
    if row_agent(s) != 'codex':
        continue
    if s.get('status') != 'running':
        continue
    if s.get('codex_session_id_own'):
        continue
    cwd = (s.get('pane') or {}).get('cwd', '')
    if not cwd:
        continue

    codex_home = os.environ.get('CODEX_HOME') or f"{home}/.codex"
    epoch_threshold = s.get('herdr_session_epoch', 0)

    sibling_claimed = [
        other.get('codex_session_id_own')
        for other in d.get('herdr_sessions', [])
        if other is not s
        and (other.get('pane') or {}).get('cwd') == cwd
        and other.get('status') not in ('stopped', 'terminated')
        and other.get('codex_session_id_own')
    ]
    s_eval = dict(s)
    s_eval['_sibling_claimed_uuids'] = sibling_claimed

    valid_candidates = []
    try:
        # Rev.2: state DB first (indexed, single query) — dynamic filename,
        # no hardcoded suffix (see §1.5 / §2 _state_db()).
        state_matches = glob.glob(f"{codex_home}/state_*.sqlite")
        state_db = max(state_matches, key=os.path.getmtime) if state_matches else None
        rows = []
        if state_db and os.path.exists(state_db):
            try:
                conn = sqlite3.connect(state_db)
                cols = [c[1] for c in conn.execute("PRAGMA table_info(threads)").fetchall()]
                order_col = "updated_at" if "updated_at" in cols else ("created_at" if "created_at" in cols else None)
                if order_col:
                    rows = conn.execute(
                        f"SELECT id, {order_col} FROM threads WHERE cwd=? ORDER BY {order_col} DESC LIMIT 20",
                        (cwd,)
                    ).fetchall()
                else:
                    rows = [(r[0], int(time.time())) for r in conn.execute("SELECT id FROM threads WHERE cwd=? LIMIT 20", (cwd,)).fetchall()]
                conn.close()
            except Exception:
                rows = []
        if not rows:
            # Fallback: bounded glob over the date-nested tree
            paths = sorted(
                glob.glob(f"{codex_home}/sessions/*/*/*/rollout-*.jsonl"),
                key=os.path.getmtime, reverse=True
            )[:50]
            rows = [(os.path.basename(p)[:-6][-36:], os.path.getmtime(p)) for p in paths]

        for uuid, ts in rows:
            if epoch_threshold and ts and ts < epoch_threshold:
                continue
            if uuid in (sibling_claimed or []):
                continue
            if verify_session_uuid(cwd, 'codex', uuid, s_eval, mode="discover"):
                valid_candidates.append(uuid)
    except Exception:
        pass

    if len(valid_candidates) > 1:
        drifts.append({'class': 'C-ambiguous', 'name': s['name'],
                       'msg': f"{s['name']}: {len(valid_candidates)} candidate transcripts newer than session epoch; not pinning"})
        s['last_visible_status'] = f"ambiguous: {len(valid_candidates)} candidates"
        actions.append(f"ambiguous candidates: {s['name']}")
    elif len(valid_candidates) == 1:
        uuid = valid_candidates[0]
        cmd = ['bash', '-c', f'source "{lib_sh}" && verify_tui_viewport "{s["name"]}" "codex" "{cwd}"']
        rc = subprocess.run(cmd).returncode
        if rc == 0:
            _pin_and_verify_resume(s, 'codex', cwd, uuid, degraded=False)
        elif rc == 1:
            drifts.append({'class': 'C-warn', 'name': s['name'], 'msg': f"{s['name']}: TUI viewport mismatch for candidate {uuid} — not pinned, will retry next cycle"})
        else:
            _pin_and_verify_resume(s, 'codex', cwd, uuid, degraded=True)

result = {
    'timestamp': now_iso,
    'yaml_path': yaml_path,
    'herdr_sessions_alive': sorted(f"{t['name']}|{t.get('server', 'default')}" for t in herdr_sessions),
    'herdr_confirmed': herdr_confirmed,
    'drifts': drifts,
    'actions': actions,
}
print(json.dumps(result, indent=2, ensure_ascii=False))

# atomic 래퍼: actions 가 없으면 쓰기를 건너뛴다. env_python(dry-run)에선 무해.
if not actions:
    raise SystemExit(0)
PYEOF

if [ "$DRY_RUN" = "1" ]; then
  printf '%s' "$RECON_SRC" | SKILLS_DIR="$SKILLS_DIR" LIB_SH="$LIB_SH" env_python "$AGENT_SESSIONS_YAML"
else
  printf '%s' "$RECON_SRC" | SKILLS_DIR="$SKILLS_DIR" LIB_SH="$LIB_SH" atomic_dump_yaml "$AGENT_SESSIONS_YAML"
fi