# 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" }