#!/usr/bin/env bash
# multi-agent-mux-delegate-job — user-facing orchestrator for the multi-agent-mux-delegate-job skill.
#
# Subcommands:
#   submit   register a job, start the subscriber FIRST, then run the agent,
#            then (optionally) run a validation script.
#   status   show one job record.
#   list     list all jobs.
#   verify   run a user-supplied --validate script against a job's artifacts.
#   wait     block until all running/pending jobs reach a terminal state.
#
# This is a reference wrapper: it shells out to the python scripts that live
# next to it. Copy it into your project and customise as needed. It never hard
# fails if `claude`/`codex`/`herdr` are missing — it prints what it would run.
set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"

# Load local env file (.mam.env preferred, .env fallback) from workspace root or explicit MAM_ENV_FILE
REPO_ROOT="$(cd "$SCRIPT_DIR/../../.." && pwd)"
TARGET_ENV="${MAM_ENV_FILE:-}"
if [[ -z "$TARGET_ENV" ]]; then
  if [[ -f "$REPO_ROOT/.mam.env" ]]; then
    TARGET_ENV="$REPO_ROOT/.mam.env"
  elif [[ -f "$REPO_ROOT/.env" ]]; then
    TARGET_ENV="$REPO_ROOT/.env"
    echo "WARNING: Loading deprecated config file '$TARGET_ENV'. Please migrate to '.mam.env'." >&2
  elif [[ -f .mam.env ]]; then
    TARGET_ENV=".mam.env"
    echo "WARNING: Loading config from cwd relative path '$TARGET_ENV'." >&2
  elif [[ -f .env ]]; then
    TARGET_ENV=".env"
    echo "WARNING: Loading deprecated config from cwd relative path '$TARGET_ENV'. Please migrate to '.mam.env'." >&2
  fi
fi

# Issue #26: parsed by envfile.py (the Python loader's rules; environment wins),
# never sourced. Values come back as shlex-quoted exports over the pipe. The
# variables this script has already set are handed over as environment, so
# the file cannot replace them either.
if [[ -n "$TARGET_ENV" && -f "$TARGET_ENV" ]]; then
  _env_exports="$(SCRIPT_DIR="$SCRIPT_DIR" REPO_ROOT="$REPO_ROOT" TARGET_ENV="$TARGET_ENV" \
    "${DELEGATE_JOB_PYTHON:-python3}" -I "$SCRIPT_DIR/../lib_py/delegation/envfile.py" --shell "$TARGET_ENV")" \
    || { echo "ERROR: cannot load $TARGET_ENV (see the line above)" >&2; exit 1; }
  eval "$_env_exports"; unset _env_exports
fi

# Source EARLY (before any herdr usage in run_agent) — this is what turns
# plain `herdr` into the tmux-compat shim (herdr() function) and provides
# resolve_herdr_session/send_keys_safe. Sourcing it late meant the
# has-session pre-flight check below used to hit the real herdr binary with
# a nonexistent subcommand and always fail.
source "$SCRIPT_DIR/../lib.sh"

# Pick an interpreter: prefer a project .venv, else python3.
pick_python() {
  local py_bin
  if [[ -n "${DELEGATE_JOB_PYTHON:-}" ]]; then
    py_bin="$DELEGATE_JOB_PYTHON"
  elif [[ -n "${AGENT_PYTHON_BIN:-}" ]] && [[ -x "$AGENT_PYTHON_BIN" ]]; then
    py_bin="$AGENT_PYTHON_BIN"
  elif [[ -x "${WORKDIR:-.}/.venv/bin/python" ]]; then
    py_bin="${WORKDIR}/.venv/bin/python"
  elif [[ -x ".venv/bin/python" ]]; then
    py_bin="$(pwd)/.venv/bin/python"
  else
    py_bin="python3"
  fi
  if ! "$py_bin" -c "import paho.mqtt" 2>/dev/null; then
    echo "ERROR: paho-mqtt package is missing for $py_bin." >&2
    echo "       Please create a virtual environment and install it:" >&2
    echo "       python3 -m venv .venv && .venv/bin/pip install -r \"$SCRIPT_DIR/requirements.txt\"" >&2
    exit 1
  fi
  echo "$py_bin"
}

REGISTRY_DIR_DEFAULT="${MAM_JOBS_DIR:-.mam/jobs}"

# Issue #21: the injected brief path, the start/success/error commands and the
# Output Report Path must work from any cwd - a creator in a git worktree runs
# them from there. Called right after `cd "$WORKDIR"`, so a relative registry
# still resolves against WORKDIR, as before.
_abs_registry_dir() {
  mkdir -p "$REGISTRY_DIR"
  REGISTRY_DIR="$(cd -P "$REGISTRY_DIR" && pwd -P)"
  export SKS_REGISTRY_DIR="$REGISTRY_DIR"
}

# 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.logroot 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;
# bash only branches on what a subcommand prints or returns.
_decide() {
  PYTHONPATH="$SCRIPT_DIR/..${PYTHONPATH:+:$PYTHONPATH}" "$PY" -m lib_py.delegation.wrapper "$@"
}

# A decision wrapper.py could not make (its module or $PY broken) ends the
# wrapper with 1 (fail-closed): never with its own code, 0, or the empty answer
# (`exit ""` is 255 on bash 3.2).
_decision_failed() {
  echo "ERROR: the wrapper could not decide '$1' (lib_py.delegation.wrapper: $2); exiting 1 (fail-closed)" >&2
  exit 1
}

# _decided <var> <subcommand> <args...>: a printed decision into $var.
_decided() {
  local __var="$1" __out __rc=0
  shift
  __out="$(_decide "$@")" || __rc=$?
  [ "$__rc" -eq 0 ] || _decision_failed "$1" "rc $__rc"
  [ -n "$__out" ] || _decision_failed "$1" "no answer"
  printf -v "$__var" '%s' "$__out"
}

# _decided_yes <subcommand> <args...>: a yes/no decision (asked with --print,
# so "no" is an answer and a failed call is not): returns 0 for yes, 1 for no.
_decided_yes() {
  local yn
  _decided yn "$@" --print
  case "$yn" in
    yes) return 0 ;;
    no) return 1 ;;
  esac
  _decision_failed "$1" "answer '$yn'"
}

# _exit_for <status>: exit with the wrapper exit code for a job status.
_exit_for() {
  local code
  _decided code exit-code "$1"
  [[ "$code" =~ ^[0-9]+$ ]] || _decision_failed exit-code "answer '$code'"
  exit "$code"
}

# _pub_cmd <extra...>: the publish command as one shell-quoted string.
_pub_cmd() {
  local s
  s="$(
    if [ -n "${LOGS_DIR_ABS:-}" ]; then printf 'DELEGATE_JOB_LOGS_DIR=%q ' "$LOGS_DIR_ABS"; fi
    printf '%q ' "$PY" "$SCRIPT_DIR/scripts/publish_event.py" --registry-dir "$REGISTRY_DIR" --job "$JOB_ID" "$@"
  )"
  printf '%s' "${s% }"
}

# _instructions <pub> [iteration]: the prompt injected into the agent (direct
# and loop jobs share it; test18 pins parts of this wording, so Issue #34's
# sub-delegation sentence goes at the end).
_instructions() {
  local pub="$1" iter="${2:-}" brief="$REGISTRY_DIR/$JOB_ID/brief.md"
  printf 'Your job_id is "%s". Detailed task requirements, instructions, and target output paths%s are documented in the task brief file at: %s. Please READ and follow %s to complete your work. Commands: start='"'"'%s --event started --heartbeat %s'"'"' (run it FIRST, within %s s; it also starts an automatic heartbeat timer), success='"'"'%s --event completed --detail <summary>'"'"', error='"'"'%s --event error --detail <reason>'"'"'. If you delegate any part of this work with multi-agent-mux-delegate-job submit, add --parent-job %s so the orchestrator waits for it.' \
    "$JOB_ID" "${iter:+ for iteration $iter}" "$brief" "$brief" "$pub" "$HEARTBEAT_SEC" "$ACK_TIMEOUT" "$pub" "$pub" "$JOB_ID"
}

usage() {
  cat <<'EOF'
multi-agent-mux-delegate-job <command> [options]

  submit  --agent <name> (--prompt <text> | --prompt-file <path>) [--workdir <dir>] [--agent-session <label>]
          [--timeout <sec> (positive decimal integer; default: $MAM_JOB_TIMEOUT_S or 3600)] [--idle-timeout <sec>] [--validate <script>]
          [--heartbeat <sec>] [--heartbeat-misses <count>] [--ack-timeout <sec>]
          [--registry-dir <dir>] [--dry-run] [--role <role_name>]
          [--type <direct|loop|discuss>] [--reviewer <reviewer_agent>]
          [--reviewer-session <reviewer_session>] [--max-iterations <count>]
          [--counterpart-role <role_name>] [--strict-role-check] [--parent-job <id>]
          # --parent-job names the job this one is delegated from, so a loop
          # waiting on that job waits for this one too (Issue #34). It must
          # exist in the registry. Default: $MAM_JOB_ID (set by the remote
          # runner), ignored with a warning when the registry lacks it.
          # --prompt-file passes the prompt by file: one argv string is capped
          # at 131,072 bytes on Linux (Issue #36). The file is read, not kept.
          # --agent-session herdr:<name> injects into a local herdr pane;
          # --agent-session remote:<name> dispatches over the broker (Issue #18).
          # Liveness contract: the agent heartbeats every --heartbeat N s
          # (default 30, automatic), the job is unresponsive after
          # --heartbeat-misses K x N s (default 4 x 30 = 120; --idle-timeout
          # overrides), `started` must arrive within --ack-timeout T s (180).
  status  --job <id> [--registry-dir <dir>]
  list    [--registry-dir <dir>]
  verify  --job <id> --validate <script> [--registry-dir <dir>]
  wait    [--job <id>] [--timeout <sec>] [--registry-dir <dir>]
  logs    <job_id> | --list [--registry-dir <dir>]   # audit log next to the registry (or $DELEGATE_JOB_LOGS_DIR)
EOF
}

# ---- arg parsing helpers --------------------------------------------------
AGENT="claude-code"; PROMPT=""; PROMPT_FILE=""; WORKDIR="$(pwd)"; AGENT_SESSION="herdr:claude"
TIMEOUT=3600; IDLE_TIMEOUT=""; VALIDATE=""; DRY_RUN=0
# job_subscriber.py exit codes (its RC_* constants; a test keeps the two equal).
SUB_RC_COMPLETED=0; SUB_RC_ERROR=1; SUB_RC_NO_EVENT=2; SUB_RC_BROKER_UNAVAILABLE=3; SUB_RC_WALL_CLOCK=4
# Issue #33: the job budget falls back to $MAM_JOB_TIMEOUT_S (--timeout > env > 3600).
# Issue #26: integer knobs are read as decimal by mam_int (lib.sh).
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"
STRICT_ROLE_CHECK=0
ROLE_ALIASES_JSON='{"worker": ["worker", "creator"], "planner": ["planner"], "reviewer": ["reviewer"]}'
COUNTERPART_ROLE_EXPLICIT=0
PARENT_JOB=""

parse_opts() {
  _INT_FLAGS_GIVEN=""
  while [[ $# -gt 0 ]]; do
    case "$1" in
      --agent) AGENT="$2"; shift 2;;
      --prompt) PROMPT="$2"; shift 2;;
      --prompt-file) PROMPT_FILE="$2"; shift 2;;
      --workdir) WORKDIR="$2"; shift 2;;
      --agent-session) AGENT_SESSION="$2"; shift 2;;
      --timeout) TIMEOUT="$2"; _INT_FLAGS_GIVEN+=" TIMEOUT"; shift 2;;
      --idle-timeout) IDLE_TIMEOUT="$2"; _INT_FLAGS_GIVEN+=" IDLE_TIMEOUT"; shift 2;;
      --heartbeat) HEARTBEAT_SEC="$2"; _INT_FLAGS_GIVEN+=" HEARTBEAT_SEC"; shift 2;;
      --heartbeat-misses) HEARTBEAT_MISSES="$2"; _INT_FLAGS_GIVEN+=" HEARTBEAT_MISSES"; shift 2;;
      --ack-timeout) ACK_TIMEOUT="$2"; _INT_FLAGS_GIVEN+=" ACK_TIMEOUT"; shift 2;;
      --validate) VALIDATE="$2"; shift 2;;
      --job) JOB_ID="$2"; shift 2;;
      --registry-dir) REGISTRY_DIR="$2"; shift 2;;
      --dry-run) DRY_RUN=1; shift;;
      --role) DELEGATE_ROLE="$2"; shift 2;;
      --type) TYPE="$2"; shift 2;;
      --reviewer) REVIEWER="$2"; shift 2;;
      --reviewer-session) REVIEWER_SESSION="$2"; shift 2;;
      --max-iterations) MAX_ITERATIONS="$2"; shift 2;;
      --counterpart-role) COUNTERPART_ROLE="$2"; COUNTERPART_ROLE_EXPLICIT=1; shift 2;;
      --strict-role-check) STRICT_ROLE_CHECK=1; shift;;
      --parent-job) PARENT_JOB="$2"; shift 2;;
      *) echo "unknown option: $1" >&2; usage; exit 1;;
    esac
  done
  # Issue #26: a flag value that is not a positive decimal integer is refused -
  # an explicitly given empty value too (review R3); only an absent flag keeps
  # its default.
  local _flag _var _n
  for _flag in timeout:TIMEOUT idle-timeout:IDLE_TIMEOUT heartbeat:HEARTBEAT_SEC heartbeat-misses:HEARTBEAT_MISSES ack-timeout:ACK_TIMEOUT; do
    _var="${_flag#*:}"
    [ -n "${!_var}" ] || [[ " $_INT_FLAGS_GIVEN " == *" $_var "* ]] || continue
    _n=$(mam_int "${!_var}" 1) || { echo "ERROR: --${_flag%%:*} ${!_var}: not a positive decimal integer" >&2; exit 1; }
    printf -v "$_var" '%s' "$_n"
  done
  [ -n "$IDLE_TIMEOUT" ] || IDLE_TIMEOUT=$(( HEARTBEAT_SEC * HEARTBEAT_MISSES ))
  # lib.sh's send_keys_safe reads the job's own `started` from this registry.
  export SKS_REGISTRY_DIR="$REGISTRY_DIR"
}

# Issue #18: `remote:<name>` sessions get the prompt over the broker, not a pane.
_is_remote() { case "${1:-}" in remote:*) return 0;; *) return 1;; esac; }

_to_lower() {
  printf '%s' "$1" | tr '[:upper:]' '[:lower:]'
}

cmd_submit() {
  parse_opts "$@"
  if [[ -n "$PROMPT_FILE" ]]; then
    [[ -z "$PROMPT" ]] || { echo "submit takes --prompt or --prompt-file, not both" >&2; exit 1; }
    [[ "$PROMPT_FILE" == /* ]] || PROMPT_FILE="$(pwd)/$PROMPT_FILE"
    [[ -f "$PROMPT_FILE" && -r "$PROMPT_FILE" ]] || { echo "submit: cannot read --prompt-file $PROMPT_FILE" >&2; exit 1; }
    [[ -s "$PROMPT_FILE" ]] || { echo "submit: --prompt-file $PROMPT_FILE is empty" >&2; exit 1; }
    PROMPT="$(<"$PROMPT_FILE")"
  fi
  [[ -n "$PROMPT" || -n "$PROMPT_FILE" ]] || { echo "submit requires --prompt or --prompt-file" >&2; exit 1; }
  # Issue #36: the prompt reaches the registry by file when it came by file,
  # never as one argv string (capped at 131,072 bytes).
  local reg_prompt=(--prompt "$PROMPT")
  [[ -n "$PROMPT_FILE" ]] && reg_prompt=(--prompt-file "$PROMPT_FILE")
  PY="$(pick_python)"
  [[ "$PY" == */* && "$PY" != /* ]] && PY="$(pwd)/$PY"
  cd "$WORKDIR"
  _abs_registry_dir
  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
  # mirror), so it is only used when this registry has it.
  local reg_parent=()
  if [[ -n "$PARENT_JOB" ]]; then
    reg_parent=(--parent-job "$PARENT_JOB")
  elif [[ -n "${MAM_JOB_ID:-}" ]]; then
    if [[ -f "$REGISTRY_DIR/$MAM_JOB_ID.json" ]]; then
      PARENT_JOB="$MAM_JOB_ID"; reg_parent=(--parent-job "$PARENT_JOB")
    else
      echo "WARNING: MAM_JOB_ID=$MAM_JOB_ID is not in $REGISTRY_DIR; registering without a parent job" >&2
    fi
  fi

  # Issue #50: a dry-run's record is never dispatched, so admission skips it
  local reg_dry=()
  if [[ "$DRY_RUN" == "1" ]]; then reg_dry=(--dry-run); fi

  # Issue #72: no job is registered while an update of the target's workspace
  # runs; held until the record is written (lib.sh mam_update_admit). A
  # dry-run record is never dispatched and is not counted by the update.
  if [[ "$DRY_RUN" != "1" ]]; then mam_update_admit "$WORKSPACE_ROOT" || exit 1; fi
  # 1) register job (prints the new job id)
  JOB_ID="$("$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" register \
      "${reg_prompt[@]}" --agent "$AGENT" --agent-session "$AGENT_SESSION" --role "$DELEGATE_ROLE" \
      --timeout "$TIMEOUT" --idle-timeout "$IDLE_TIMEOUT" \
      --heartbeat-sec "$HEARTBEAT_SEC" --heartbeat-misses "$HEARTBEAT_MISSES" --ack-timeout "$ACK_TIMEOUT" \
      --transport "$(_is_remote "$AGENT_SESSION" && echo remote || echo herdr)" \
      --job-type "$TYPE" --reviewer "$REVIEWER" --reviewer-session "$REVIEWER_SESSION" \
      --max-iterations "$MAX_ITERATIONS" ${reg_parent[@]+"${reg_parent[@]}"} ${reg_dry[@]+"${reg_dry[@]}"})"
  echo "registered job: $JOB_ID"
  [[ "$DRY_RUN" == "1" ]] || mam_update_admit_release

  # 1-1) Provision job directory and write direct brief.md (MAM Job Restructuring)
  local job_dir="$REGISTRY_DIR/$JOB_ID"
  if [[ "$DRY_RUN" != "1" ]]; then
    mkdir -p "$job_dir"
    cat <<EOF > "$job_dir/brief.md"
# 📋 Brief: Job $JOB_ID Delegation

- **Job ID**: $JOB_ID
- **Target Agent**: $AGENT (session: $AGENT_SESSION)
- **Role**: $DELEGATE_ROLE
- **Timeout**: $TIMEOUT s (Idle: $IDLE_TIMEOUT s)
- **Liveness contract**: heartbeat every $HEARTBEAT_SEC s (automatic once you publish \`started --heartbeat $HEARTBEAT_SEC\`), unresponsive after $HEARTBEAT_MISSES missed heartbeats, \`started\` expected within $ACK_TIMEOUT s
- **Output Report Path**: $REGISTRY_DIR/$JOB_ID/$AGENT-reports/report-final.md
- **Parent for sub-delegations**: pass \`--parent-job $JOB_ID\` to \`multi-agent-mux-delegate-job submit\`

## 🔎 Task Description
$PROMPT
EOF
    echo "provisioned job directory: $job_dir"
  fi

  local START_TIME
  START_TIME=$(date +%s)

  if [[ "$TYPE" == "direct" ]]; then
    # 1) Read dispatch token and attempt id for this dispatch
    local _disp_tok=""
    local _attempt_id=""
    _disp_tok="$("$PY" -c "import json, os; p=os.path.join('$REGISTRY_DIR', '$JOB_ID.json'); print(json.load(open(p)).get('dispatch_token','') or '') if os.path.exists(p) else print('')" 2>/dev/null || true)"
    _attempt_id="$("$PY" -c "import json, os; p=os.path.join('$REGISTRY_DIR', '$JOB_ID.json'); print(json.load(open(p)).get('attempt_id','') or '') if os.path.exists(p) else print('')" 2>/dev/null || true)"
    local pub_extra=()
    [ -n "$_disp_tok" ] && pub_extra+=(--expect-dispatch-token "$_disp_tok")
    [ -n "$_attempt_id" ] && pub_extra+=(--attempt-id "$_attempt_id")
    local pub instructions
    pub="$(_pub_cmd ${pub_extra[@]+"${pub_extra[@]}"})"
    instructions="$(_instructions "$pub")"

    if [[ "$DRY_RUN" == "1" ]]; then
      run_agent "$JOB_ID" "$instructions"
      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"
      echo "$LOGS_DIR_ABS/$JOB_ID"
      exit 5
    fi

    # 2) The registry, then the TUI idle handshake (Issue #50: the session's
    #    open jobs - an onboarding still running, a job sent by hand - end
    #    first, in order; herdr's agent_status is the screen, read after)
    if [ "$AGENT" != "human" ] && ! _is_remote "$AGENT_SESSION"; then
      _registry_wait "$JOB_ID" "$AGENT_SESSION" "$_disp_tok"
      local idle_rc=0
      wait_for_agent_idle "$AGENT_SESSION" || idle_rc=$?
      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"
        echo "$LOGS_DIR_ABS/$JOB_ID"
        exit 5
      fi
    fi

    # 3) START THE SUBSCRIBER
    local logf="$REGISTRY_DIR/$JOB_ID.subscriber.out"
    # Before the first event the subscriber is bounded by the wall clock only:
    # delivery (T from the injection, which happens after the subscriber is
    # up and can take a while itself) is judged by _wait_started, and a failed
    # injection reaps the subscriber. From the first event on, K x N applies.
    local sub_extra_args=(--ack-timeout "$TIMEOUT")
    [ -n "$_disp_tok" ] && sub_extra_args+=(--expect-dispatch-token "$_disp_tok")
    _is_remote "$AGENT_SESSION" && sub_extra_args+=(--remote)
    "$PY" "$SCRIPT_DIR/scripts/job_subscriber.py" --registry-dir "$REGISTRY_DIR" \
        --job "$JOB_ID" --timeout "$TIMEOUT" --idle-timeout "$IDLE_TIMEOUT" \
        ${sub_extra_args[@]+"${sub_extra_args[@]}"} \
        >"$logf" 2>&1 &
    local sub_pid=$!
    echo "subscriber pid: $sub_pid (log: $logf)"
    # Wait for the subscriber to CONNACK + SUBSCRIBE (reactive handshake)
    local sub_ready=0
    for ((i = 0; i < 25; i++)); do
      if kill -0 "$sub_pid" 2>/dev/null; then
        if grep -q '^SUBSCRIBED ' "$logf" 2>/dev/null; then
          sub_ready=1
          break
        fi
      else
        local sub_exit=0
        wait "$sub_pid" 2>/dev/null || sub_exit=$?
        if _decided_yes subscriber-up "$sub_exit"; then
          sub_ready=1
          break
        else
          echo "ERROR: subscriber died early (pid=$sub_pid, exit=$sub_exit, check $logf)" >&2
          local mapped_st
          _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}"
          echo "$LOGS_DIR_ABS/$JOB_ID"
          _exit_for "$mapped_st"
        fi
      fi
      sleep 0.2
    done
    if [ "$sub_ready" -ne 1 ]; then
      echo "WARNING: subscriber subscribe handshake timed out — falling back to proceed" >&2
    fi

    # 5) Deliver the prompt (herdr pane or remote dispatch) and wait for `started`
    if ! deliver_prompt "$JOB_ID" "$instructions" "$AGENT_SESSION" "$logf"; then
      reap_subscriber "$sub_pid"
      finalize_unfinished "$JOB_ID" "error" "$SUB_RC_ERROR" "inject"
      echo "$LOGS_DIR_ABS/$JOB_ID"
      exit 1
    fi
    start_stall_watch "$AGENT_SESSION" "$AGENT" "$_disp_tok" "$_attempt_id"

    # 6) optional validation hook
    if [[ -n "$VALIDATE" ]]; then
      echo "running validation: $VALIDATE"
      if JOB_ID="$JOB_ID" REGISTRY_DIR="$REGISTRY_DIR" bash "$VALIDATE"; then
        echo "validation: PASS"
      else
        local rc=$?
        echo "validation: FAIL (exit $rc)"
      fi
    fi

    local sub_rc=0
    wait "$sub_pid" || sub_rc=$?
    stop_stall_watch
    echo "subscriber output:"; cat "$logf" || true

    local job_status
    _decided job_status sub-status "$sub_rc" --job "$JOB_ID" --registry-dir "$REGISTRY_DIR"

    finalize_unfinished "$JOB_ID" "$job_status" "$sub_rc" "direct"
    job_status="${FINAL_JOB_STATUS:-$job_status}"
    # Issue #40: this subscriber verified the `completed` (or the disk did) -
    # pin the session's post-discovered id now, daemon or not.
    if [ "$job_status" = "completed" ]; then
      delegate_autopin_completed "$JOB_ID" "$_disp_tok" "$REGISTRY_DIR"
    fi

    # Last stdout line: the persistent audit-log dir for this job (see SKILL.md
    # "Audit Logs"). Callers can scrape `tail -n1` to find it.
    echo "$LOGS_DIR_ABS/$JOB_ID"
    _exit_for "$job_status"
  else
    # Implement loop/discuss orchestrator
    local iteration=1
    local current_prompt="$PROMPT"
    local current_session="$AGENT_SESSION"
    local _phase="worker"
    local display_role="$DELEGATE_ROLE"
    local attempt=0
    local final_loop_rc=0

    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"
      echo "$LOGS_DIR_ABS/$JOB_ID"
      return 0
    fi

    while true; do
      echo "=================================================="
      echo "Iteration $iteration - Role: $display_role"
      echo "Session: $current_session"
      echo "=================================================="

      # 1-1) Provision job directory and write iteration brief.md (MAM Job Restructuring)
      local job_dir="$REGISTRY_DIR/$JOB_ID"
      local clean_session="${current_session#herdr:}"
      mkdir -p "$job_dir"
      cat <<EOF > "$job_dir/brief.md"
# 📋 Brief: Job $JOB_ID Delegation (Iteration $iteration)
 
- **Job ID**: $JOB_ID
- **Target Agent/Session**: $current_session
- **Role**: $display_role
- **Iteration**: $iteration
- **Liveness contract**: heartbeat every $HEARTBEAT_SEC s (automatic once you publish \`started --heartbeat $HEARTBEAT_SEC\`), unresponsive after $HEARTBEAT_MISSES missed heartbeats, \`started\` expected within $ACK_TIMEOUT s
- **Output Report Path**: $REGISTRY_DIR/$JOB_ID/${clean_session}-reports/report-final.md
- **Parent for sub-delegations**: pass \`--parent-job $JOB_ID\` to \`multi-agent-mux-delegate-job submit\`
 
## 🔎 Task Description
$current_prompt
EOF
      # Issue #36: the registry gets the prompt by file, not as one argv string
      printf '%s' "$current_prompt" > "$job_dir/prompt.md"
      echo "provisioned iteration brief: $job_dir/brief.md"

      attempt=$((attempt + 1))

      local force_warn_only=0
      local cp_lower def_lower
      cp_lower=$(_to_lower "$COUNTERPART_ROLE")
      def_lower=$(_to_lower "$DEFAULT_COUNTERPART_ROLE")
      if [[ "$_phase" == "reviewer" && "$COUNTERPART_ROLE_EXPLICIT" -eq 1 \
            && "$cp_lower" != "$def_lower" ]]; then
        force_warn_only=1
      fi

      # 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"
        echo "$LOGS_DIR_ABS/$JOB_ID"
        exit 5
      fi

      # Update job details in registry
      "$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" update \
          --job "$JOB_ID" \
          --agent-session "$current_session" \
          --prompt-file "$job_dir/prompt.md" \
          --iteration "$iteration" \
          --attempt-id "$attempt" \
          --role "$display_role" \
          --status "pending"

      # The reset above is a redispatch: it issued this attempt's token. Read it
      # now, so a close from the waits below (busy, idle_wait) names this
      # attempt, not the previous phase's (review reviewer-grok-01)
      local _disp_tok=""
      _disp_tok="$("$PY" -c "import json, os; p=os.path.join('$REGISTRY_DIR', '$JOB_ID.json'); print(json.load(open(p)).get('dispatch_token','') or '') if os.path.exists(p) else print('')" 2>/dev/null || true)"

      # Registry wait (Issue #50), then idle wait - on the record just reset to
      # pending for this phase, so the waiter's signal is written and other
      # waiters see this job in the FIFO (review F1)
      local cur_agent="$AGENT"
      [ "$_phase" = "reviewer" ] && cur_agent="$REVIEWER"
      if [ "$cur_agent" != "human" ] && ! _is_remote "$current_session"; then
        _registry_wait "$JOB_ID" "$current_session" "$_disp_tok"
        local idle_rc=0
        wait_for_agent_idle "$current_session" || idle_rc=$?
        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"
          echo "$LOGS_DIR_ABS/$JOB_ID"
          exit 5
        fi
      fi

      # Start subscriber
      local logf="$REGISTRY_DIR/${JOB_ID}.iter_${iteration}_${display_role}.subscriber.out"
      local loop_sub_args=(--attempt-id "$attempt" --ack-timeout "$TIMEOUT")
      [ -n "$_disp_tok" ] && loop_sub_args+=(--expect-dispatch-token "$_disp_tok")
      _is_remote "$current_session" && loop_sub_args+=(--remote)
      "$PY" "$SCRIPT_DIR/scripts/job_subscriber.py" --registry-dir "$REGISTRY_DIR" \
          --job "$JOB_ID" --timeout "$TIMEOUT" --idle-timeout "$IDLE_TIMEOUT" \
          "${loop_sub_args[@]}" \
          >"$logf" 2>&1 &
      local sub_pid=$!
      echo "subscriber pid: $sub_pid (log: $logf)"
      # Wait for the subscriber to CONNACK + SUBSCRIBE (reactive handshake)
      local sub_ready=0
      for ((i = 0; i < 25; i++)); do
        if kill -0 "$sub_pid" 2>/dev/null; then
          if grep -q '^SUBSCRIBED ' "$logf" 2>/dev/null; then
            sub_ready=1
            break
          fi
        else
          local sub_exit=0
          wait "$sub_pid" 2>/dev/null || sub_exit=$?
          if _decided_yes subscriber-up "$sub_exit"; then
            sub_ready=1
            break
          else
            echo "ERROR: subscriber died early (pid=$sub_pid, exit=$sub_exit, check $logf)" >&2
            local mapped_st
            _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}"
            echo "$LOGS_DIR_ABS/$JOB_ID"
            _exit_for "$mapped_st"
          fi
        fi
        sleep 0.2
      done
      if [ "$sub_ready" -ne 1 ]; then
        echo "WARNING: subscriber subscribe handshake timed out — falling back to proceed" >&2
      fi

      # Format instruction block
      local loop_pub_extra=(--attempt-id "$attempt")
      [ -n "$_disp_tok" ] && loop_pub_extra+=(--expect-dispatch-token "$_disp_tok")
      local pub instructions
      pub="$(_pub_cmd "${loop_pub_extra[@]}")"
      instructions="$(_instructions "$pub" "$iteration")"

      # Trigger agent
      if ! deliver_prompt "$JOB_ID" "$instructions" "$current_session" "$logf"; then
        reap_subscriber "$sub_pid"
        finalize_unfinished "$JOB_ID" "error" "$SUB_RC_ERROR" "inject"
        echo "$LOGS_DIR_ABS/$JOB_ID"
        exit 1
      fi
      start_stall_watch "$current_session" "$cur_agent" "$_disp_tok" "$attempt"

      # Wait for subscriber
      local sub_rc=0
      wait "$sub_pid" || sub_rc=$?
      stop_stall_watch
      echo "subscriber output:"; cat "$logf" || true

      local job_status
      _decided job_status sub-status "$sub_rc" --job "$JOB_ID" --registry-dir "$REGISTRY_DIR"
      
      echo "Job role $display_role finished with status: $job_status"

      # Retrieve feedback from the last event
      local feedback
      feedback="$("$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" get-feedback --job "$JOB_ID")"
      echo "Feedback/Detail: $feedback"
      if [[ "$_phase" == "worker" ]]; then
        if [[ "$job_status" != "completed" ]]; then
          echo "Worker did not complete successfully (status: $job_status). Terminating workflow."
          finalize_unfinished "$JOB_ID" "$job_status" "$sub_rc" "worker" "$current_session"
          job_status="${FINAL_JOB_STATUS:-$job_status}"
          if [[ "$job_status" == "completed" ]]; then
            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
            echo "$LOGS_DIR_ABS/$JOB_ID"
            _exit_for "$job_status"
          fi
        fi

        # Worker completed successfully, now switch to reviewer
        _phase="reviewer"
        display_role="$COUNTERPART_ROLE"
        current_session="$REVIEWER_SESSION"
        # Build reviewer prompt based on type
        if [[ "$TYPE" == "loop" ]]; then
          current_prompt="Review the changes/artifacts generated for job $JOB_ID. Check if they meet the requirements. If correct, publish completed event with 'PASS'. If there are issues, publish error event with detailed feedback/nits. CRITICAL: When raising issues or giving a review, you MUST include the exact reason for the issue and a clear direction for improvement (문제 제시에 대한 이유와 확실한 개선 방향을 반드시 포함해야 합니다)."
        elif [[ "$TYPE" == "discuss" ]]; then
          current_prompt="Read draft/documents generated for job $JOB_ID. Review the feasibility and content. Write your feedback/objections. If you agree with the plan, reply with 'AGREE'."
        fi
      else
        if [[ "$job_status" != "completed" ]]; then
          echo "Reviewer did not complete successfully (status: $job_status). Terminating workflow."
          finalize_unfinished "$JOB_ID" "$job_status" "$sub_rc" "reviewer" "$current_session"
          job_status="${FINAL_JOB_STATUS:-$job_status}"
          if [[ "$job_status" == "completed" ]]; then
            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
            echo "$LOGS_DIR_ABS/$JOB_ID"
            _exit_for "$job_status"
          fi
        fi

        # Reviewer finished. Check if pass/agree
        local success=0
        local fb_lower
        fb_lower=$(_to_lower "$feedback")
        if [[ "$TYPE" == "loop" ]]; then
          if [[ "$fb_lower" == *"pass"* ]]; then
            success=1
          fi
        elif [[ "$TYPE" == "discuss" ]]; then
          if [[ "$fb_lower" == *"agree"* ]]; then
            success=1
          fi
        fi

        if [[ "$success" == "1" ]]; then
          echo "Reviewer approved the work. Finalizing job as completed."
          "$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" status --job "$JOB_ID" --set "completed"
          final_loop_rc=0
          break
        else
          # Reviewer rejected/provided feedback. Increment & check max iterations
          if [[ $iteration -ge $MAX_ITERATIONS ]]; then
            echo "Max iterations ($MAX_ITERATIONS) reached without approval. Terminating workflow."
            "$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" status --job "$JOB_ID" --set "error"
            final_loop_rc=1
            break
          fi
          
          iteration=$((iteration + 1))
          _phase="worker"
          display_role="$DELEGATE_ROLE"
          current_session="$AGENT_SESSION"
          current_prompt="The reviewer provided the following feedback for job $JOB_ID: $feedback. Please modify the code/artifacts to address these comments. CRITICAL: As the Developer Team Leader, you must thoroughly review the suggested modifications, verify their validity, adopt/implement them if valid, and if you judge any recommendation to be invalid, do NOT implement it but instead explain your reasons clearly in your response and send it back to the reviewer (수정안을 최대한 꼼꼼히 검토하여 타당성을 검증하고, 타당하다면 수렴하여 수정을 진행하되, 타당하지 않다고 판단되는 부분이 있다면 그 이유를 명확히 밝혀 리뷰어에게 전달하십시오)."
        fi
      fi
    done

    # 4) optional validation hook
    if [[ -n "$VALIDATE" ]]; then
      echo "running validation: $VALIDATE"
      if JOB_ID="$JOB_ID" REGISTRY_DIR="$REGISTRY_DIR" bash "$VALIDATE"; then
        echo "validation: PASS"
      else
        local rc=$?
        echo "validation: FAIL (exit $rc)"
        final_loop_rc=$rc
      fi
    fi

    # Last stdout line: the persistent audit-log dir
    echo "$LOGS_DIR_ABS/$JOB_ID"
    exit "$final_loop_rc"
  fi
}

reap_subscriber() {
  if [ -n "${1:-}" ]; then
    kill "$1" 2>/dev/null || true
    wait "$1" 2>/dev/null || true
  fi
}

# Issue #30: a notify-only stall watcher for a job with a local herdr pane
# (lib_py.stall_watch decides; it never closes the job). Whether to watch
# (MAM_STALL_WATCH, remote, human) is `_decide stall-watch`. Sets STALL_PID
# ("" when none started).
start_stall_watch() {
  local session="$1" agent="$2" tok="${3:-}" att="${4:-}"
  STALL_PID=""
  _decided_yes stall-watch "$session" "$agent" || return 0
  local args=(--job "$JOB_ID" --registry-dir "$REGISTRY_DIR" --target "${session#herdr:}"
              --agent "$agent" --herdr-session "${HERDR_SESSION_NAME:-}" --watch-pid "$$")
  [ -n "$tok" ] && args+=(--expect-dispatch-token "$tok")
  [ -n "$att" ] && args+=(--attempt-id "$att")
  PYTHONPATH="$SCRIPT_DIR/..${PYTHONPATH:+:$PYTHONPATH}" "$PY" -m lib_py.stall_watch "${args[@]}" \
      >"$REGISTRY_DIR/$JOB_ID.stall.out" 2>&1 &
  STALL_PID=$!
  echo "stall watch pid: $STALL_PID (log: $REGISTRY_DIR/$JOB_ID.stall.out)"
}

# A watcher that already ended on its own with an error (herdr unavailable)
# is not left silent: its last lines go to this wrapper's stderr.
# The stop is KILL, not TERM: a TERM can be swallowed (it hung this wrapper in
# `wait`, PR #41 review N1), and a TERM budget in front of the job's close
# clock delayed deadline_timeout (round-2 review). The watcher holds nothing
# to flush - its TERM handler is a bare os._exit - so KILL loses nothing, and
# `wait` returns at once.
stop_stall_watch() {
  [ -n "${STALL_PID:-}" ] || return 0
  local rc=0
  if kill -0 "$STALL_PID" 2>/dev/null; then
    kill -9 "$STALL_PID" 2>/dev/null || true
    wait "$STALL_PID" 2>/dev/null || true
  else
    wait "$STALL_PID" 2>/dev/null || rc=$?
  fi
  if [ "$rc" -ne 0 ]; then
    echo "stall watch for $JOB_ID exited rc=$rc:" >&2
    tail -n 3 "$REGISTRY_DIR/$JOB_ID.stall.out" >&2 2>/dev/null || true
  fi
  STALL_PID=""
}

# ISSUE-18: delivery is confirmed by the protocol, not the screen. After the
# prompt was handed over (herdr accepted the submission, or the dispatch was
# published for a remote session) the job's own `started` must arrive within
# the ack timeout T. The evidence is lib.sh's delegate_started_recorded - the
# same predicate create_session's onboarding ack uses, so the two waits share
# one contract (ack-finding req. 2, review R1): the local publisher's line in
# events.log, or the subscriber's HMAC-verified, dispatch-attributed audit
# record of a `started` it received (a remote agent has no shared disk). The
# registry status is not evidence - a record leaves `pending` when another
# writer closes it (error, cancelled) or another dispatch claims it, none of
# which says this agent saw this prompt. No `started` within T is a delivery
# failure: fail closed, never re-inject (single-injection contract). A job
# that is already terminal is handled by the terminal logic after this, not
# turned into a success here.
_wait_started() {
  local job_id="$1" ack_s="${2:-$ACK_TIMEOUT}" sub_log="${3:-}" deadline wall
  # sub_log is retained for callers' signatures; the subscriber's evidence is
  # read from its audit record, which is attributable, not from its console.
  deadline=$(( $(date +%s) + ack_s ))
  # The ack wait never outlives this dispatch's wall clock (START_TIME + TIMEOUT).
  wall=$(( ${START_TIME:-$(date +%s)} + ${TIMEOUT:-3600} ))
  [ "$deadline" -gt "$wall" ] && deadline=$wall
  local budget=$(( deadline - $(date +%s) ))
  while :; do
    # publish_event records `started` locally before its wire publish and
    # before the status sync, so a slow broker cannot delay the acknowledgement.
    if MAM_JOBS_DIR="$REGISTRY_DIR" delegate_started_recorded "$job_id" "${_disp_tok:-}" "${LOGS_DIR_ABS:-}"; then
      echo "inject: job $job_id recorded 'started' in events.log; treating the prompt as delivered and continuing." >&2
      return 0
    fi
    if [ "$(date +%s)" -ge "$deadline" ]; then
      ACK_WAITED_S="$budget"
      return 1
    fi
    sleep 1
  done
}

# Legacy name retained for callers/tests of the ISSUE-14 guard.
_inject_acknowledged() { _wait_started "$1" "$(mam_int_knob MAM_INJECT_ACK_GRACE_S "$ACK_TIMEOUT" 0)" "${2:-}"; }

# ISSUE-14 / ISSUE-18: herdr's agent_status and the pane busy token describe
# the screen, not the protocol. They are no longer liveness *evidence* (a
# headless or remote agent shows neither, and a CLI update can change both
# silently); finalize_unfinished logs them as a diagnostic so an operator can
# tell "dead" from "silent" on the same host. Liveness itself is the contract:
# an event on disk or on the wire within K x N seconds.
_agent_still_working() {
  local sess="$1" target="$2"
  [ "$(_sks_agent_status "$sess" "$target")" = "working" ] && return 0
  _pane_tail "$sess" 8 2>/dev/null | grep -Fq "esc to interrupt"
}

# _within_deadline <deadline_epoch> <cmd...>
# Run a probe that can block (a herdr RPC, a pane capture) without letting it
# outlive a wall-clock deadline: once the clock reaches <deadline> the probe
# and its children are killed and the result is rc 124, i.e. "no evidence".
_within_deadline() {
  local deadline="$1" pid
  shift
  "$@" &
  pid=$!
  while kill -0 "$pid" 2>/dev/null; do
    if [ "$(date +%s)" -ge "$deadline" ]; then
      pkill -P "$pid" 2>/dev/null || true
      kill "$pid" 2>/dev/null || true
      wait "$pid" 2>/dev/null || true
      return 124
    fi
    sleep 0.1
  done
  wait "$pid"
}

# _registry_wait <job_id> <session> <dispatch_token>  (Issue #50)
# Waits, by the registry alone, until no live open job of <session> remains
# and this job is the earliest pending one (lib_py.delegation.wrapper
# wait-open-jobs: FIFO, the waiter signals on its own record, a dead record is
# ignored). The waiter is pinned to this attempt's token and session (review
# 55219be1 R1): a record redispatched, moved or closed by someone else while
# it waited is `superseded` - this wrapper exits 5 without touching it, the
# record is its new owner's. A wait that runs out (MAM_DISPATCH_WAIT_S,
# default 600 s, bounded by this job's wall clock) closes this job as
# cancelled/session_busy and exits 5, like the preflight and the idle wait.
_registry_wait() {
  local job_id="$1" sess="$2" tok="$3" res
  _decided res wait-open-jobs --job "$job_id" --agent-session "$sess" --registry-dir "$REGISTRY_DIR" \
      ${tok:+--dispatch-token "$tok"} \
      --wait-s "$(mam_int_knob MAM_DISPATCH_WAIT_S 600 0)" ${START_TIME:+--start-time "$START_TIME"} \
      --timeout "$TIMEOUT" --watch-pid "$$" --hold
  case "$res" in
    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
      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"
      echo "$LOGS_DIR_ABS/$job_id"
      exit 5 ;;
  esac
  _decision_failed wait-open-jobs "answer '$res'"
}

# finalize_unfinished <job_id> <status> <sub_rc> [phase] [session]
# Closes a job that did not complete (Issue #33 labels, ISSUE-14/18 disk grace,
# Issue #34 cancel reasons) and sets FINAL_JOB_STATUS. The decisions, the disk
# waits, the close and the event are `_decide finalize-begin/-end`; bash runs
# only the herdr diagnostic probe between the two, when asked to.
finalize_unfinished() {
  local job_id="$1" job_status="$2" sub_rc="$3" phase="${4:-}"
  local target_sess="${5:-${current_session:-${target_session:-$AGENT_SESSION}}}"
  local step final evidence_until deadline out begin end_rc=0
  _decided begin finalize-begin --job "$job_id" \
      --registry-dir "$REGISTRY_DIR" --status "$job_status" ${START_TIME:+--start-time "$START_TIME"} \
      --timeout "$TIMEOUT" --idle-timeout "$IDLE_TIMEOUT" --agent "$AGENT" --session "$target_sess"
  read -r step final evidence_until deadline <<<"$begin"
  case "$step" in
    keep) return 0 ;;
    done) FINAL_JOB_STATUS="$final"; return 0 ;;
  esac
  FINAL_JOB_STATUS="$final"
  if [ "$step" = "probe" ]; then
    local sess="${target_sess#herdr:}" target_agent diag="no"
    target_agent=$(_sanitize_herdr_agent_name "$sess") || target_agent=""
    [ -n "$target_agent" ] && _within_deadline "$deadline" _agent_still_working "$sess" "$target_agent" && diag="yes"
    echo "liveness: no wire event for ${IDLE_TIMEOUT}s on job $job_id; herdr/pane says busy=$diag (diagnostic only, not evidence)" >&2
  fi
  # The answer comes back on fd 3: stdout carries registry.py's close output.
  # --watch-pid: the waits end, unclosed, with this wrapper (review R1).
  { out="$(_decide finalize-end --job "$job_id" --registry-dir "$REGISTRY_DIR" --status "$job_status" \
      ${START_TIME:+--start-time "$START_TIME"} --timeout "$TIMEOUT" --idle-timeout "$IDLE_TIMEOUT" \
      --sub-rc "$sub_rc" --phase "$phase" --type "$TYPE" --final "$final" --evidence-until "$evidence_until" \
      --dispatch-token "${_disp_tok:-}" --preflight-reason "${PREFLIGHT_REASON:-}" \
      --python "$PY" --scripts-dir "$SCRIPT_DIR/scripts" --watch-pid "$$" 3>&1 1>&4 4>&-)" || end_rc=$?; } 4>&1
  [ "$end_rc" -eq 0 ] || _decision_failed finalize-end "rc $end_rc"
  FINAL_JOB_STATUS="$out"
}

preflight_agent() {
  local job_id="$1" target_session="${2:-$AGENT_SESSION}" force_warn_only="${3:-0}"
  if [ "$AGENT" = "human" ]; then
    return 0
  fi
  local sess="${target_session#herdr:}"

  # B-3: must probe for the real binary, not the herdr() function / shim wrapper.
  if ! has_real_herdr; then
    echo "ERROR: this skill requires herdr (interactive agent sessions)." >&2
    echo "       Ensure herdr is installed and executable." >&2
    return 1
  fi

  export HERDR_SESSION_NAME="$(resolve_herdr_session "$sess" "$WORKDIR")"

  if ! herdr has-session -t "$sess" 2>/dev/null; then
    PREFLIGHT_REASON="target_not_found"  # Issue #34: not a busy session
    echo "ERROR: 에이전트 세션 '$sess'이 존재하지 않습니다. 작업을 위임하기 전에 먼저 에이전트 세션을 기동해 주세요." >&2
    echo "       팁: 'multi-agent-mux-resume' 또는 'multi-agent-mux-create'를 통해 에이전트를 먼저 생성할 수 있습니다." >&2
    return 1
  fi

  # Check role suitability
  local sess_role job_role
  sess_role=$(SESS_NAME="$sess" MAM_STATE_JSON="$(load_state_json)" "$PY" -c "
import os, json
d = json.loads(os.environ.get('MAM_STATE_JSON', '{}'))
name = os.environ.get('SESS_NAME')
for s in d.get('herdr_sessions', []):
    if s.get('name') == name:
        print(s.get('role', ''))
        break
" 2>/dev/null || echo "")

  job_role=$("$PY" -c "
import json
try:
    with open('$REGISTRY_DIR/$job_id.json') as f:
        print(json.load(f).get('role', ''))
except Exception:
    pass
" 2>/dev/null || echo "")

  if [[ -n "$job_role" && -n "$sess_role" ]]; then
    local check_result
    check_result=$(JOB_ROLE="$job_role" SESS_ROLE="$sess_role" ROLE_ALIASES_JSON="$ROLE_ALIASES_JSON" "$PY" -c "
import os, json
job = os.environ.get('JOB_ROLE', '').lower()
sess = os.environ.get('SESS_ROLE', '').lower()
aliases = json.loads(os.environ.get('ROLE_ALIASES_JSON', '{}'))
candidates = aliases.get(job, [job])
if any(c in sess for c in candidates):
    print('OK')
else:
    print('MISMATCH')
" 2>/dev/null || echo "OK")

    if [[ "$check_result" == "MISMATCH" ]]; then
      local mismatch_msg="Target session '$sess' has role '$sess_role' which does not match job role '$job_role'."
      if [[ "$STRICT_ROLE_CHECK" -eq 1 && "$force_warn_only" -ne 1 ]]; then
        echo "ERROR: role suitability mismatch. $mismatch_msg" >&2
        return 1
      else
        echo "WARNING: $mismatch_msg" >&2
      fi
    fi
  fi
  return 0
}

inject_agent() {
  local job_id="$1" instructions="$2" target_session="${3:-$AGENT_SESSION}"
  # $4: subscriber log (a started line there also confirms delivery)
  if [ "$AGENT" = "human" ]; then
    echo "[human agent] complete the task, then run publish_event.py --event completed"
    return 0
  fi
  local sess="${target_session#herdr:}"

  # Before launching the agent, set up error trap to publish error event
  if [ -n "${job_id:-}" ] && [ -n "${PY:-}" ]; then
    pub_script="$SCRIPT_DIR/scripts/publish_event.py"
    trap "rc=\$?; if [ \$rc -ne 0 ]; then \"$PY\" \"$pub_script\" --registry-dir '$REGISTRY_DIR' --job '$job_id' --event error --detail 'agent bootstrap failed (exit '\$rc')'; fi" EXIT
  fi

  echo "살아있는 에이전트 세션 '$sess'에 작업을 위임합니다..."
  # ISSUE-18: send_keys_safe is the injection *means*. rc 0 (accepted, herdr
  # saw it working) and rc 4 (accepted, no screen signal) both hand over to
  # the protocol: the job's own `started` within ACK_TIMEOUT decides. Any
  # other rc means the input never reached the agent (pane never quiesced,
  # dialog, paste failure) - fail at once, never re-inject.
  local sks_rc=0
  send_keys_safe "$sess" "$instructions" "$job_id" || sks_rc=$?
  case "$sks_rc" in
    0|4) ;;
    *)
      trap - EXIT
      echo "ERROR: 프롬프트 주입 실패 (send_keys_safe rc=$sks_rc) — 세션 '$sess'" >&2
      return 1 ;;
  esac
  local ack_s="$(mam_int_knob MAM_INJECT_ACK_GRACE_S "$ACK_TIMEOUT" 0)"
  if ! _wait_started "$job_id" "$ack_s" "${4:-}"; then
    trap - EXIT
    echo "ERROR: 프롬프트 주입 실패 — 세션 '$sess': no 'started' within ${ACK_WAITED_S:-$ack_s}s (ack timeout ${ack_s}s, bounded by the wall clock); not re-injecting" >&2
    return 1
  fi

  echo "작업이 세션 '$sess'에 전송되었습니다. (연결하려면: $(session_attach_command "$sess" "${HERDR_SESSION_NAME:-default}"))"
  trap - EXIT
  return 0
}

# remote_dispatch <job_id> <instructions> <remote:session> <subscriber_log>
# ISSUE-18: no pane, no shared disk. Publish the retained dispatch on the job's
# topic and wait for the agent's own `started` (seen by the subscriber) within
# ACK_TIMEOUT. The dispatch is signed with MAM_DISPATCH_SECRET (.mam.env).
remote_dispatch() {
  local job_id="$1" instructions="$2" target_session="$3" sub_log="${4:-}"
  local sess="${target_session#remote:}"
  echo "원격 세션 '$sess'에 작업을 dispatch 합니다 (broker)..."
  local d_args=(--job "$job_id" --registry-dir "$REGISTRY_DIR" --session "$sess" --instructions "$instructions")
  [ "${MAM_ALLOW_UNSIGNED_DISPATCH:-0}" = "1" ] && d_args+=(--allow-unsigned)
  if ! "$PY" "$SCRIPT_DIR/scripts/remote.py" dispatch "${d_args[@]}" >/dev/null; then
    echo "ERROR: dispatch publish failed for job $job_id (remote:$sess)" >&2
    return 1
  fi
  local ack_s="$(mam_int_knob MAM_INJECT_ACK_GRACE_S "$ACK_TIMEOUT" 0)"
  if ! _wait_started "$job_id" "$ack_s" "$sub_log"; then
    echo "ERROR: remote session '$sess' published no 'started' within ${ACK_WAITED_S:-$ack_s}s (ack timeout ${ack_s}s, bounded by the wall clock); not re-dispatching" >&2
    return 1
  fi
  echo "작업이 원격 세션 '$sess'에 전달되었습니다."
  return 0
}

# deliver_prompt <job_id> <instructions> <session> <subscriber_log>
deliver_prompt() {
  if _is_remote "$3"; then
    remote_dispatch "$1" "$2" "$3" "${4:-}"
  else
    inject_agent "$1" "$2" "$3" "${4:-}"
  fi
}

run_agent() {
  local job_id="$1" instructions="$2" target_session="${3:-$AGENT_SESSION}" force_warn_only="${4:-0}"
  local sess="${target_session#herdr:}"
  if [[ "$DRY_RUN" == "1" ]]; then
    echo "[dry-run] would delegate task to running agent '$AGENT' in herdr session '$sess' with instructions:"
    echo "----"; echo "$instructions"; echo "----"
    return 0
  fi
  preflight_agent "$job_id" "$target_session" "$force_warn_only" || return $?
  inject_agent "$job_id" "$instructions" "$target_session" || return $?
}

cmd_status() {
  parse_opts "$@"
  [[ -n "$JOB_ID" ]] || { echo "status requires --job" >&2; exit 1; }
  PY="$(pick_python)"
  "$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" get --job "$JOB_ID"
}

cmd_list() {
  parse_opts "$@"
  PY="$(pick_python)"
  "$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" list
}

cmd_verify() {
  parse_opts "$@"
  [[ -n "$JOB_ID" ]] || { echo "verify requires --job" >&2; exit 1; }
  [[ -n "$VALIDATE" ]] || { echo "verify requires --validate <script>" >&2; exit 1; }
  echo "verifying job $JOB_ID with $VALIDATE"
  if JOB_ID="$JOB_ID" REGISTRY_DIR="$REGISTRY_DIR" bash "$VALIDATE"; then
    echo "verify: PASS (exit 0)"; exit 0
  else
    rc=$?; echo "verify: FAIL (exit $rc)"; exit "$rc"
  fi
}

cmd_logs() {
  # 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)"
  # --registry-dir is passed on only when given: registry.py's own default
  # follows MAM_JOBS_DIR, then WORKSPACE_ROOT, so the usual run works from any
  # cwd (the wrapper's REGISTRY_DIR_DEFAULT is cwd-relative). bash 3.2 + set -u:
  # an empty array expands through ${reg[@]+"${reg[@]}"}.
  local -a reg=()
  local jid="" list=0
  while [[ $# -gt 0 ]]; do
    case "$1" in
      --registry-dir) reg=(--registry-dir "$2"); shift 2;;
      --list) list=1; shift;;
      *) jid="$1"; shift;;
    esac
  done
  if [[ "$list" == "1" ]]; then
    "$PY" "$SCRIPT_DIR/scripts/registry.py" ${reg[@]+"${reg[@]}"} logs --list
  else
    [[ -n "$jid" ]] || { echo "logs requires <job_id> or --list" >&2; exit 1; }
    "$PY" "$SCRIPT_DIR/scripts/registry.py" ${reg[@]+"${reg[@]}"} logs "$jid"
  fi
}

cmd_wait() {
  parse_opts "$@"
  PY="$(pick_python)"
  local rc=0
  if [[ -n "$JOB_ID" ]]; then
    "$PY" "$SCRIPT_DIR/scripts/job_subscriber.py" --registry-dir "$REGISTRY_DIR" \
        --job "$JOB_ID" --timeout "$TIMEOUT" || rc=$?
  else
    "$PY" "$SCRIPT_DIR/scripts/job_subscriber.py" --registry-dir "$REGISTRY_DIR" \
        --wait-any --timeout "$TIMEOUT" || rc=$?
  fi
  # Issue #33: either clock is a timeout (rc 2) to callers of `wait`, as before;
  # the subscriber's own log says which clock it was.
  _decided rc wait-rc "$rc"
  [[ "$rc" =~ ^[0-9]+$ ]] || _decision_failed wait-rc "answer '$rc'"
  return "$rc"
}

main() {
  local sub="${1:-}"; shift || true
  case "$sub" in
    submit) cmd_submit "$@";;
    status) cmd_status "$@";;
    list)   cmd_list "$@";;
    verify) cmd_verify "$@";;
    wait)   cmd_wait "$@";;
    logs)   cmd_logs "$@";;
    ""|-h|--help|help) usage;;
    *) echo "unknown command: $sub" >&2; usage; exit 1;;
  esac
}

# Tests source this file to exercise single helpers (MAM_DELEGATE_JOB_NO_MAIN=1).
if [ "${MAM_DELEGATE_JOB_NO_MAIN:-0}" != "1" ]; then
  main "$@"
fi