multi-agent-mux/tests/data/issue17_item4_old_decisions.sh

# Issue #17 item 4: the decision functions as they were in bash at base 7209750
# (v4.8.5), copied verbatim, for tests/test_issue17_item4_decisions.py to run
# side by side with the new wrapper. Never sourced by the skills.

mam_int() {
  # Spelled-out ASCII sets, not [[:space:]]/[0-9]: those follow the locale, and
  # the runner's _positive_int takes ASCII only (review R2: U+2003 30 U+2003
  # was 30 here under en_US.UTF-8 and invalid there).
  local _ws=$' \t\n\r\v\f' _re
  _re="^[${_ws}]*([0123456789]+)[${_ws}]*\$"
  [[ "${1-}" =~ $_re ]] || return 1
  local d="${BASH_REMATCH[1]}"
  d="${d#"${d%%[!0]*}"}"
  [ ${#d} -le 9 ] || return 1
  local n=$(( 10#${d:-0} ))
  [ "$n" -ge "$2" ] || return 1
  printf '%s\n' "$n"
}
mam_int_knob() {
  local raw="${!1-}"
  [ -n "$raw" ] || { printf '%s\n' "$2"; return 0; }
  mam_int "$raw" "$3" && return 0
  echo "WARNING: $1=$raw ignored (not a decimal integer >= $3); using $2" >&2
  printf '%s\n' "$2"
}

start_stall_watch() {
  local session="$1" agent="$2" tok="${3:-}" att="${4:-}"
  STALL_PID=""
  [ "${MAM_STALL_WATCH:-1}" = "0" ] && return 0
  if _is_remote "$session"; then
    echo "stall watch: remote job, no local pane - stall detection off" >&2; return 0
  fi
  if [ "$agent" = "human" ] || [ "${session#herdr:}" = "human" ]; then
    echo "stall watch: human job, no local pane - stall detection off" >&2; return 0
  fi
  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)"
}

_disk_status() {
  local jid="$1"
  "$PY" -c "import json, os; p=os.path.join('$REGISTRY_DIR', '$jid.json'); print(json.load(open(p)).get('status','')) if os.path.exists(p) else print('')" 2>/dev/null || true
}

_recent_disk_heartbeat() {
  local job_id="$1" window="$2" f="$REGISTRY_DIR/$job_id.events.log"
  [ -f "$f" ] || return 1
  "$PY" - "$f" "$window" <<'PYEOF'
import calendar, json, sys, time
path, window = sys.argv[1], float(sys.argv[2])
last = ""
with open(path) as fh:
    for line in fh:
        if line.strip():
            last = line
try:
    stamp = json.loads(last)["timestamp"]
    seen = calendar.timegm(time.strptime(stamp, "%Y-%m-%dT%H:%M:%SZ"))
except Exception:
    sys.exit(1)
sys.exit(0 if time.time() - seen <= window else 1)
PYEOF
}

_map_sub_rc() {
  local sub_rc="$1" job_id="$2"
  local job_status="running"
  if [[ $sub_rc -eq $SUB_RC_COMPLETED ]]; then
    job_status="completed"
  elif [[ $sub_rc -eq $SUB_RC_ERROR ]]; then
    job_status="error"
  elif [[ $sub_rc -eq $SUB_RC_BROKER_UNAVAILABLE ]]; then
    job_status="broker_unavailable"
    local disk_st
    disk_st="$(_disk_status "$job_id")"
    if [[ "$disk_st" == "completed" || "$disk_st" == "error" || "$disk_st" == "cancelled" ]]; then
      job_status="$disk_st"
    fi
  else
    job_status="timeout"  # SUB_RC_NO_EVENT, SUB_RC_WALL_CLOCK (and any other code)
  fi
  echo "$job_status"
}

_rc_for() {
  case "$1" in
    completed) echo 0 ;;
    error) echo 1 ;;
    timeout) echo 2 ;;
    broker_unavailable) echo 3 ;;
    cancelled) echo 5 ;;
    *) echo 1 ;;
  esac
}

finalize_unfinished() {
  local job_id="$1" job_status="$2" sub_rc="$3" phase="${4:-}" FINALIZE_CLOCK=""
  if [ "${MAM_DELEGATE_FAIL_CLOSED:-1}" = "0" ]; then
    return 0
  fi
  if [ "$job_status" = "completed" ]; then
    return 0
  fi
  # Issue #30: what the stall watcher saw goes with an unfinished job's close.
  if [ -s "$REGISTRY_DIR/$job_id.stall.out" ]; then
    echo "stall watch log for $job_id (last 3 lines):" >&2
    tail -n 3 "$REGISTRY_DIR/$job_id.stall.out" >&2 || true
  fi

  local cur_disk_st="$(_disk_status "$job_id")"
  if [ "$cur_disk_st" = "completed" ] || [ "$cur_disk_st" = "error" ] || [ "$cur_disk_st" = "cancelled" ]; then
    FINAL_JOB_STATUS="$cur_disk_st"
    [ "$cur_disk_st" = "completed" ] && return 0
  else
    FINAL_JOB_STATUS="$job_status"
  fi
  # broker_unavailable: re-observe the disk at 3 s intervals until this
  # dispatch's wall-clock deadline (START_TIME + TIMEOUT). The deadline is read
  # off the clock on every pass; summing the sleeps alone ignores the time the
  # disk probes themselves take and overruns TIMEOUT.
  if [ "$job_status" = "broker_unavailable" ]; then
    local deadline left disk_st
    deadline=$(( ${START_TIME:-$(date +%s)} + TIMEOUT ))
    while :; do
      left=$(( deadline - $(date +%s) ))
      [ "$left" -le 0 ] && break
      [ "$left" -gt 3 ] && left=3
      sleep "$left"
      disk_st="$(_disk_status "$job_id")"
      if [ "$disk_st" = "completed" ] || [ "$disk_st" = "error" ] || [ "$disk_st" = "cancelled" ]; then
        job_status="$disk_st"
        FINAL_JOB_STATUS="$disk_st"
        [ "$disk_st" = "completed" ] && return 0
        break
      fi
    done
    # Issue #33 (review R3): a wait that ran into the deadline is the wall clock.
    [ "$(date +%s)" -ge "$deadline" ] && FINALIZE_CLOCK="wall"
  fi

  # timeout: if subscriber timed out, check if agent is still actively working in herdr
  if [ "$job_status" = "timeout" ]; then
    local target_sess="${5:-${current_session:-${target_session:-$AGENT_SESSION}}}"
    local sess="${target_sess#herdr:}"
    local target_agent
    target_agent=$(_sanitize_herdr_agent_name "$sess")

    # ISSUE-14 / ISSUE-18: the subscriber's idle clock only counts events it
    # received over the wire. Keep re-observing the disk while the *contract*
    # says the agent is alive: an event in events.log at most IDLE_TIMEOUT
    # (K x N) old (_recent_disk_heartbeat - publish_event records every event
    # there before its wire publish, the automatic heartbeat timer included).
    # herdr status and the pane are NOT evidence any more; they are logged
    # once here as a diagnostic, so an operator can tell a dead agent from a
    # silent one on the same host. Window policy: entering this grace opens
    # one IDLE_TIMEOUT window and every disk event re-opens it from the time
    # of that poll, so an agent that leaves no event at all is closed about
    # 2 x IDLE_TIMEOUT after its last wire event. Evidence only ever extends
    # the idle window, never this dispatch's wall clock: the loop ends at
    # START_TIME + TIMEOUT read off the clock on every pass.
    local deadline evidence_until left disk_st
    deadline=$(( ${START_TIME:-$(date +%s)} + TIMEOUT ))
    evidence_until=$(( $(date +%s) + IDLE_TIMEOUT ))
    if [ "$AGENT" != "human" ] && [ "$sess" != "human" ] && [ -n "$sess" ] && ! _is_remote "$target_sess"; then
      local diag="no"
      _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
    while [ "$(date +%s)" -lt "$deadline" ]; do
      disk_st="$(_disk_status "$job_id")"
      if [ "$disk_st" = "completed" ] || [ "$disk_st" = "error" ] || [ "$disk_st" = "cancelled" ]; then
        job_status="$disk_st"
        FINAL_JOB_STATUS="$disk_st"
        [ "$disk_st" = "completed" ] && return 0
        break
      fi
      if _recent_disk_heartbeat "$job_id" "$IDLE_TIMEOUT"; then
        evidence_until=$(( $(date +%s) + IDLE_TIMEOUT ))
      elif [ "$(date +%s)" -ge "$evidence_until" ]; then
        break
      fi
      left=$(( deadline - $(date +%s) ))
      [ "$left" -le 0 ] && break
      sleep 1
    done
    [ "$(date +%s)" -ge "$deadline" ] && FINALIZE_CLOCK="wall"
  fi

  # Issue #33: the subscriber's rc says which clock ran out (4 wall clock,
  # 2 no event); the disk-heartbeat grace above that ends on the deadline is
  # the wall clock too. bash only carries the label.
  local clock="${FINALIZE_CLOCK:-idle}"
  [ "$sub_rc" = "$SUB_RC_WALL_CLOCK" ] && clock="wall"
  local close_req="deadline_timeout"
  [ "$clock" = "wall" ] && [ "$phase" != "inject" ] && close_req="wall_clock_timeout"
  local detail=""
  if [ "$phase" = "preflight" ] || [ "$phase" = "idle_wait" ]; then
    close_req="cancelled"
    detail="session_busy"
    # Issue #34: a target the preflight could not see was never busy
    [ "$phase" = "preflight" ] && [ -n "${PREFLIGHT_REASON:-}" ] && detail="$PREFLIGHT_REASON"
  fi

  [ "$close_req" != "cancelled" ] && echo "finalize: $job_id closed by $([ "$clock" = "wall" ] && echo wall_clock || echo no_event) clock (budget ${TIMEOUT}s / idle ${IDLE_TIMEOUT}s)" >&2
  local close_args=(--job "$job_id" --request "$close_req")
  [ -n "$detail" ] && close_args+=(--detail "$detail")
  [ -n "${_disp_tok:-}" ] && close_args+=(--expect-dispatch-token "$_disp_tok")

  local close_rc=0
  "$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" close "${close_args[@]}" 2>&1 || close_rc=$?

  # If close failed due to dispatch token mismatch (rc=4), suppress secondary event emission
  # to prevent CAS/ABA bypass
  if [ $close_rc -eq 4 ]; then
    echo "finalize_unfinished: dispatch token mismatch (stale attempt); suppressing event emission" >&2
    return 0
  fi

  # Check if job completed on disk (universal, for all TYPEs)
  local cur_disk_st
  cur_disk_st="$(_disk_status "$job_id")"
  if [ "$cur_disk_st" = "completed" ]; then
    FINAL_JOB_STATUS="completed"
    return 0
  fi

  # direct path only: publish MQTT error/cancelled event (loop/discuss forbidden: O-1)
  if [ "$TYPE" = "direct" ]; then
    local disp_arg=()
    [ -n "${_disp_tok:-}" ] && disp_arg=(--expect-dispatch-token "$_disp_tok")
    if [ "$close_req" = "cancelled" ]; then
      "$PY" "$SCRIPT_DIR/scripts/publish_event.py" --registry-dir "$REGISTRY_DIR" --job "$job_id" --event cancelled --detail "$detail" ${disp_arg[@]+"${disp_arg[@]}"} 2>/dev/null || true
    elif [ "$sub_rc" -ne "$SUB_RC_ERROR" ] || [ "$phase" = "inject" ]; then
      local err_detail="$close_req"
      [ "$phase" = "inject" ] && err_detail="agent_blocked"
      "$PY" "$SCRIPT_DIR/scripts/publish_event.py" --registry-dir "$REGISTRY_DIR" --job "$job_id" --event error --detail "$err_detail" ${disp_arg[@]+"${disp_arg[@]}"} 2>/dev/null || true
    fi
  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.
  [ "$rc" -eq "$SUB_RC_WALL_CLOCK" ] && rc=$SUB_RC_NO_EVENT
  return "$rc"
}