GodopuandClaude Opus 5.5
c5cebe6c32
finalize-end ran the disk grace and the broker re-observation as a child of the wrapper; TERM to the wrapper left it running, and at the deadline it still closed the job and published the error (codex R1, reproduced with the real registry). The bash loops it replaced ended with the wrapper. The wrapper now passes its own pid (--watch-pid "$$", as for the stall watcher). finalize-end checks it every 0.2 s through both waits and before the close and the event, and leaves without either once it is gone. A comparison test sends TERM to the old and the new wrapper inside each grace: both exit at once, leave no process behind and make no call. Also: no blank line at the end of the old-decisions fixture, and the #42 inventory follows main() one line down. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
264 lines
11 KiB
Bash
264 lines
11 KiB
Bash
# 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"
|
|
}
|