fix(#26): .mam.env 를 Python 파서 하나로 읽고, 정수 환경변수를 10진수로, 테스트를 로컬 MQTT 브로커로 #43
No files matched your search
+33
-3
@@ -2335,6 +2335,35 @@ sys.exit(1)
|
|||||||
' 2>/dev/null
|
' 2>/dev/null
|
||||||
}
|
}
|
||||||
|
|
||||||
|
# mam_int <value> <min>
|
||||||
|
#
|
||||||
|
# Issue #26: the one integer rule for knobs that reach bash arithmetic (the
|
||||||
|
# Python twin is lib_py/loop/runner.py::_positive_int). Surrounding blanks are
|
||||||
|
# allowed, then digits only, read as DECIMAL (bash would read 0700 as octal
|
||||||
|
# 448 and fail on 0900), and min <= n <= 999999999. Prints n and returns 0, or
|
||||||
|
# returns 1. The per-knob minimum lives in tests/test_issue26_ints.py.
|
||||||
|
mam_int() {
|
||||||
|
[[ "${1-}" =~ ^[[:space:]]*([0-9]+)[[:space:]]*$ ]] || 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 <ENV_NAME> <default> <min>
|
||||||
|
#
|
||||||
|
# The value of $ENV_NAME through mam_int; unset or empty gives <default>, an
|
||||||
|
# invalid value gives <default> with a warning on stderr.
|
||||||
|
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"
|
||||||
|
}
|
||||||
|
|
||||||
# delegate_wait_started <job_id> [ack_timeout_s] [dispatch_token]
|
# delegate_wait_started <job_id> [ack_timeout_s] [dispatch_token]
|
||||||
#
|
#
|
||||||
# ISSUE-23 / ISSUE-18: delivery is confirmed by the protocol, not by the
|
# ISSUE-23 / ISSUE-18: delivery is confirmed by the protocol, not by the
|
||||||
@@ -2347,6 +2376,7 @@ sys.exit(1)
|
|||||||
delegate_wait_started() {
|
delegate_wait_started() {
|
||||||
local job_id="$1" ack_s="${2:-180}" tok="${3:-}"
|
local job_id="$1" ack_s="${2:-180}" tok="${3:-}"
|
||||||
[ -n "$job_id" ] || return 1
|
[ -n "$job_id" ] || return 1
|
||||||
|
ack_s=$(mam_int "$ack_s" 0) || { echo "WARNING: onboarding ack timeout ${2-} ignored (not a decimal integer >= 0); using 180" >&2; ack_s=180; }
|
||||||
local deadline
|
local deadline
|
||||||
deadline=$(( $(date +%s) + ack_s ))
|
deadline=$(( $(date +%s) + ack_s ))
|
||||||
while :; do
|
while :; do
|
||||||
@@ -2759,7 +2789,7 @@ send_keys_safe() {
|
|||||||
fi
|
fi
|
||||||
|
|
||||||
if [ "$_q_rc" != "2" ]; then
|
if [ "$_q_rc" != "2" ]; then
|
||||||
deadline=$(( $(date +%s) + ${SKS_DIALOG_TIMEOUT:-30} ))
|
deadline=$(( $(date +%s) + $(mam_int_knob SKS_DIALOG_TIMEOUT 30 0) ))
|
||||||
while _pane_dialog_open "$sess"; do
|
while _pane_dialog_open "$sess"; do
|
||||||
if [ "${SKS_DIALOG_ESCAPE:-0}" = "1" ]; then
|
if [ "${SKS_DIALOG_ESCAPE:-0}" = "1" ]; then
|
||||||
_sks_herdr send-keys -t "$sess" Escape
|
_sks_herdr send-keys -t "$sess" Escape
|
||||||
@@ -2778,7 +2808,7 @@ send_keys_safe() {
|
|||||||
|
|
||||||
# Busy gate: wait up to SKS_BUSY_TIMEOUT only if wait_for_agent_idle returned 3 (unknown)
|
# Busy gate: wait up to SKS_BUSY_TIMEOUT only if wait_for_agent_idle returned 3 (unknown)
|
||||||
if [ "${_SKS_IDLE_RC:-0}" = "3" ]; then
|
if [ "${_SKS_IDLE_RC:-0}" = "3" ]; then
|
||||||
local b_deadline=$(( $(date +%s) + ${SKS_BUSY_TIMEOUT:-120} ))
|
local b_deadline=$(( $(date +%s) + $(mam_int_knob SKS_BUSY_TIMEOUT 120 0) ))
|
||||||
local still_busy=1
|
local still_busy=1
|
||||||
while [ "$(date +%s)" -lt "$b_deadline" ]; do
|
while [ "$(date +%s)" -lt "$b_deadline" ]; do
|
||||||
local tail_pane
|
local tail_pane
|
||||||
@@ -2847,7 +2877,7 @@ send_keys_safe() {
|
|||||||
# still fails closed (rc 4); the wrapper then waits for `started` up to
|
# still fails closed (rc 4); the wrapper then waits for `started` up to
|
||||||
# the job's ack timeout T before finalising the inject as failed.
|
# the job's ack timeout T before finalising the inject as failed.
|
||||||
if printf '%s\n' "$p_out" | grep -Eq '"code": *"(timeout|agent_prompt_stalled)"'; then
|
if printf '%s\n' "$p_out" | grep -Eq '"code": *"(timeout|agent_prompt_stalled)"'; then
|
||||||
local g_deadline=$(( $(date +%s) + ${SKS_DELIVERY_GRACE_S:-15} ))
|
local g_deadline=$(( $(date +%s) + $(mam_int_knob SKS_DELIVERY_GRACE_S 15 0) ))
|
||||||
while [ "$(date +%s)" -lt "$g_deadline" ]; do
|
while [ "$(date +%s)" -lt "$g_deadline" ]; do
|
||||||
[ "$(_sks_agent_status "$sess" "$agent_target")" = "working" ] && return 0
|
[ "$(_sks_agent_status "$sess" "$agent_target")" = "working" ] && return 0
|
||||||
_sks_job_started "$job_id" && return 0
|
_sks_job_started "$job_id" && return 0
|
||||||
|
|||||||
@@ -67,6 +67,16 @@ def log_error(msg: str) -> None:
|
|||||||
print(f"{_COLOR_ERROR} {msg}", file=sys.stderr)
|
print(f"{_COLOR_ERROR} {msg}", file=sys.stderr)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
def _positive_int(text: Optional[str], minimum: int = 1) -> Optional[int]:
|
||||||
|
"""Issue #26: lib.sh mam_int's rule - blanks around, digits only, decimal
|
||||||
|
(0700 is 700), minimum <= n <= 999999999; else None."""
|
||||||
|
s = (text or "").strip(" \t\n\r\v\f")
|
||||||
|
if not s or not s.isascii() or not s.isdigit():
|
||||||
|
return None
|
||||||
|
n = int(s)
|
||||||
|
return n if minimum <= n <= 999_999_999 else None
|
||||||
|
|
||||||
class LoopRunner:
|
class LoopRunner:
|
||||||
"""Orchestrates planning, code implementation, and multi-agent peer reviews."""
|
"""Orchestrates planning, code implementation, and multi-agent peer reviews."""
|
||||||
|
|
||||||
@@ -142,7 +152,7 @@ class LoopRunner:
|
|||||||
" --all-reviewer Enforce PASS verdict from all active reviewer sessions",
|
" --all-reviewer Enforce PASS verdict from all active reviewer sessions",
|
||||||
" --max-loop N Max execution-review corrective loop runs (default: 3)",
|
" --max-loop N Max execution-review corrective loop runs (default: 3)",
|
||||||
" --max-rebut N Max rebuttal attempts per review iteration (default: 1, 0: disabled)",
|
" --max-rebut N Max rebuttal attempts per review iteration (default: 1, 0: disabled)",
|
||||||
" --job-timeout <sec> Budget of each delegated job (default: $MAM_JOB_TIMEOUT_S or 3600)",
|
" --job-timeout <sec> Budget of each delegated job (positive decimal integer; default: $MAM_JOB_TIMEOUT_S or 3600)",
|
||||||
" --verbose Print detailed execution timeline traces",
|
" --verbose Print detailed execution timeline traces",
|
||||||
" --cleanup Purge temporary job directories upon success",
|
" --cleanup Purge temporary job directories upon success",
|
||||||
" --worktree <path> Source worktree for diff/base/head (control plane stays here)",
|
" --worktree <path> Source worktree for diff/base/head (control plane stays here)",
|
||||||
@@ -214,10 +224,10 @@ class LoopRunner:
|
|||||||
self.max_loop = int(args[i + 1])
|
self.max_loop = int(args[i + 1])
|
||||||
i += 2
|
i += 2
|
||||||
elif arg == "--job-timeout":
|
elif arg == "--job-timeout":
|
||||||
if i + 1 >= len(args) or not re.match(r"^[0-9]+$", args[i + 1]) or int(args[i + 1]) <= 0:
|
self.job_timeout = _positive_int(args[i + 1]) if i + 1 < len(args) else None
|
||||||
print("ERROR: --job-timeout requires a positive non-zero integer.", file=sys.stderr)
|
if self.job_timeout is None:
|
||||||
|
print("ERROR: --job-timeout requires a positive decimal integer.", file=sys.stderr)
|
||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
self.job_timeout = int(args[i + 1])
|
|
||||||
i += 2
|
i += 2
|
||||||
elif arg == "--max-rebut":
|
elif arg == "--max-rebut":
|
||||||
if i + 1 >= len(args) or not re.match(r"^[0-9]+$", args[i + 1]):
|
if i + 1 >= len(args) or not re.match(r"^[0-9]+$", args[i + 1]):
|
||||||
@@ -471,9 +481,9 @@ class LoopRunner:
|
|||||||
for env_name, flag in (("MAM_HEARTBEAT_SEC", "--heartbeat"),
|
for env_name, flag in (("MAM_HEARTBEAT_SEC", "--heartbeat"),
|
||||||
("MAM_HEARTBEAT_MISSES", "--heartbeat-misses"),
|
("MAM_HEARTBEAT_MISSES", "--heartbeat-misses"),
|
||||||
("MAM_ACK_TIMEOUT_S", "--ack-timeout")):
|
("MAM_ACK_TIMEOUT_S", "--ack-timeout")):
|
||||||
val = os.environ.get(env_name)
|
val = _positive_int(os.environ.get(env_name))
|
||||||
if val and val.isdigit() and int(val) > 0:
|
if val is not None:
|
||||||
args += [flag, val]
|
args += [flag, str(val)]
|
||||||
return args
|
return args
|
||||||
|
|
||||||
def resolve_all_reviewers(self, state: Dict[str, Any]) -> List[str]:
|
def resolve_all_reviewers(self, state: Dict[str, Any]) -> List[str]:
|
||||||
@@ -593,11 +603,17 @@ class LoopRunner:
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def job_budget(self) -> int:
|
def job_budget(self) -> int:
|
||||||
"""Issue #33: --job-timeout > $MAM_JOB_TIMEOUT_S (positive integer) > 3600."""
|
"""Issue #33: --job-timeout > $MAM_JOB_TIMEOUT_S (positive decimal integer) > 3600."""
|
||||||
if self.job_timeout:
|
if self.job_timeout:
|
||||||
return self.job_timeout
|
return self.job_timeout
|
||||||
env = os.environ.get("MAM_JOB_TIMEOUT_S", "")
|
env = os.environ.get("MAM_JOB_TIMEOUT_S", "")
|
||||||
return int(env) if env.isdigit() and int(env) > 0 else 3600
|
n = _positive_int(env)
|
||||||
|
if n is None and env:
|
||||||
|
if not getattr(self, "_warned_job_timeout", False):
|
||||||
|
self._warned_job_timeout = True
|
||||||
|
print(f"WARNING: MAM_JOB_TIMEOUT_S={env} ignored (not a decimal integer >= 1); using 3600", file=sys.stderr)
|
||||||
|
return 3600
|
||||||
|
return n or 3600
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def job_wait_s(self) -> int:
|
def job_wait_s(self) -> int:
|
||||||
|
|||||||
@@ -115,7 +115,7 @@ usage() {
|
|||||||
multi-agent-mux-delegate-job <command> [options]
|
multi-agent-mux-delegate-job <command> [options]
|
||||||
|
|
||||||
submit --agent <name> (--prompt <text> | --prompt-file <path>) [--workdir <dir>] [--agent-session <label>]
|
submit --agent <name> (--prompt <text> | --prompt-file <path>) [--workdir <dir>] [--agent-session <label>]
|
||||||
[--timeout <sec> (default: $MAM_JOB_TIMEOUT_S or 3600)] [--idle-timeout <sec>] [--validate <script>]
|
[--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>]
|
[--heartbeat <sec>] [--heartbeat-misses <count>] [--ack-timeout <sec>]
|
||||||
[--registry-dir <dir>] [--dry-run] [--role <role_name>]
|
[--registry-dir <dir>] [--dry-run] [--role <role_name>]
|
||||||
[--type <direct|loop|discuss>] [--reviewer <reviewer_agent>]
|
[--type <direct|loop|discuss>] [--reviewer <reviewer_agent>]
|
||||||
@@ -147,15 +147,10 @@ TIMEOUT=3600; IDLE_TIMEOUT=""; VALIDATE=""; DRY_RUN=0
|
|||||||
# job_subscriber.py exit codes (its RC_* constants; a test keeps the two equal).
|
# 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
|
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 #33: the job budget falls back to $MAM_JOB_TIMEOUT_S (--timeout > env > 3600).
|
||||||
if [[ -n "${MAM_JOB_TIMEOUT_S:-}" ]]; then
|
# Issue #26: integer knobs are read as decimal by mam_int (lib.sh).
|
||||||
if [[ "$MAM_JOB_TIMEOUT_S" =~ ^[0-9]+$ && "$MAM_JOB_TIMEOUT_S" -gt 0 ]]; then
|
TIMEOUT=$(mam_int_knob MAM_JOB_TIMEOUT_S 3600 1)
|
||||||
TIMEOUT="$MAM_JOB_TIMEOUT_S"
|
|
||||||
else
|
|
||||||
echo "WARNING: MAM_JOB_TIMEOUT_S=$MAM_JOB_TIMEOUT_S ignored (not a positive integer); using 3600" >&2
|
|
||||||
fi
|
|
||||||
fi
|
|
||||||
# Issue #18 liveness contract (N / K / T). An unset --idle-timeout derives K x N.
|
# Issue #18 liveness contract (N / K / T). An unset --idle-timeout derives K x N.
|
||||||
HEARTBEAT_SEC="${MAM_HEARTBEAT_SEC:-30}"; HEARTBEAT_MISSES="${MAM_HEARTBEAT_MISSES:-4}"; ACK_TIMEOUT="${MAM_ACK_TIMEOUT_S:-180}"
|
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"
|
JOB_ID=""; REGISTRY_DIR="$REGISTRY_DIR_DEFAULT"; DELEGATE_ROLE="Worker"
|
||||||
TYPE="direct"; REVIEWER="hermes"; REVIEWER_SESSION="herdr:hermes"; MAX_ITERATIONS=5
|
TYPE="direct"; REVIEWER="hermes"; REVIEWER_SESSION="herdr:hermes"; MAX_ITERATIONS=5
|
||||||
DEFAULT_COUNTERPART_ROLE="Reviewer"
|
DEFAULT_COUNTERPART_ROLE="Reviewer"
|
||||||
@@ -193,6 +188,13 @@ parse_opts() {
|
|||||||
*) echo "unknown option: $1" >&2; usage; exit 1;;
|
*) echo "unknown option: $1" >&2; usage; exit 1;;
|
||||||
esac
|
esac
|
||||||
done
|
done
|
||||||
|
# Issue #26: a flag value that is not a positive decimal integer is refused.
|
||||||
|
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}" ] || 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 ))
|
[ -n "$IDLE_TIMEOUT" ] || IDLE_TIMEOUT=$(( HEARTBEAT_SEC * HEARTBEAT_MISSES ))
|
||||||
# lib.sh's send_keys_safe reads the job's own `started` from this registry.
|
# lib.sh's send_keys_safe reads the job's own `started` from this registry.
|
||||||
export SKS_REGISTRY_DIR="$REGISTRY_DIR"
|
export SKS_REGISTRY_DIR="$REGISTRY_DIR"
|
||||||
@@ -772,7 +774,7 @@ _wait_started() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
# Legacy name retained for callers/tests of the ISSUE-14 guard.
|
# Legacy name retained for callers/tests of the ISSUE-14 guard.
|
||||||
_inject_acknowledged() { _wait_started "$1" "${MAM_INJECT_ACK_GRACE_S:-$ACK_TIMEOUT}" "${2:-}"; }
|
_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
|
# 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
|
# the screen, not the protocol. They are no longer liveness *evidence* (a
|
||||||
@@ -1112,7 +1114,7 @@ inject_agent() {
|
|||||||
echo "ERROR: 프롬프트 주입 실패 (send_keys_safe rc=$sks_rc) — 세션 '$sess'" >&2
|
echo "ERROR: 프롬프트 주입 실패 (send_keys_safe rc=$sks_rc) — 세션 '$sess'" >&2
|
||||||
return 1 ;;
|
return 1 ;;
|
||||||
esac
|
esac
|
||||||
local ack_s="${MAM_INJECT_ACK_GRACE_S:-$ACK_TIMEOUT}"
|
local ack_s="$(mam_int_knob MAM_INJECT_ACK_GRACE_S "$ACK_TIMEOUT" 0)"
|
||||||
if ! _wait_started "$job_id" "$ack_s" "${4:-}"; then
|
if ! _wait_started "$job_id" "$ack_s" "${4:-}"; then
|
||||||
trap - EXIT
|
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
|
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
|
||||||
@@ -1138,7 +1140,7 @@ remote_dispatch() {
|
|||||||
echo "ERROR: dispatch publish failed for job $job_id (remote:$sess)" >&2
|
echo "ERROR: dispatch publish failed for job $job_id (remote:$sess)" >&2
|
||||||
return 1
|
return 1
|
||||||
fi
|
fi
|
||||||
local ack_s="${MAM_INJECT_ACK_GRACE_S:-$ACK_TIMEOUT}"
|
local ack_s="$(mam_int_knob MAM_INJECT_ACK_GRACE_S "$ACK_TIMEOUT" 0)"
|
||||||
if ! _wait_started "$job_id" "$ack_s" "$sub_log"; then
|
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
|
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
|
return 1
|
||||||
|
|||||||
@@ -0,0 +1,295 @@
|
|||||||
|
"""Issue #26 (B): integer knobs that reach bash arithmetic are read as decimal.
|
||||||
|
|
||||||
|
bash reads 0700 as octal 448 and fails on 0900; the runner reads 700. Both now
|
||||||
|
go through one rule (lib.sh mam_int / runner._positive_int) with a per-knob
|
||||||
|
minimum, and INT_KNOBS below is the one place that minimum is written.
|
||||||
|
|
||||||
|
- B1: one input table through the real wrapper and the real runner.
|
||||||
|
- B2: flags are normalised or refused.
|
||||||
|
- B3: the derived idle timeout uses decimal values.
|
||||||
|
- B4: the ack waits read 0900 as 900 and keep 0 as "check once".
|
||||||
|
- B5: static guard - no env default reaches arithmetic directly, and every
|
||||||
|
normalised knob uses the minimum in INT_KNOBS.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import re
|
||||||
|
import subprocess
|
||||||
|
import sys
|
||||||
|
import time
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||||
|
SKILLS = REPO_ROOT / ".agents" / "skills"
|
||||||
|
WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
|
||||||
|
LIB_SH = SKILLS / "lib.sh"
|
||||||
|
|
||||||
|
# knob -> (minimum, default). The minimum is 0 where 0 already means "do not
|
||||||
|
# wait / check once" (SKS_BUSY_TIMEOUT=0 is test_issue6_turn_handshake T2-12's
|
||||||
|
# contract); 1 where 0 would close a job at once.
|
||||||
|
INT_KNOBS = {
|
||||||
|
"MAM_JOB_TIMEOUT_S": (1, "3600"),
|
||||||
|
"MAM_HEARTBEAT_SEC": (1, "30"),
|
||||||
|
"MAM_HEARTBEAT_MISSES": (1, "4"),
|
||||||
|
"MAM_ACK_TIMEOUT_S": (1, "180"),
|
||||||
|
"MAM_INJECT_ACK_GRACE_S": (0, "$ACK_TIMEOUT"),
|
||||||
|
"SKS_DIALOG_TIMEOUT": (0, "30"),
|
||||||
|
"SKS_BUSY_TIMEOUT": (0, "120"),
|
||||||
|
"SKS_DELIVERY_GRACE_S": (0, "15"),
|
||||||
|
# delegate_wait_started's ack_s (MAM_ONBOARD_ACK_S / DELEGATE_JOB_ACK_S via create_session.sh)
|
||||||
|
"delegate_wait_started:ack_s": (0, "180"),
|
||||||
|
# wrapper flags
|
||||||
|
"--timeout": (1, None),
|
||||||
|
"--idle-timeout": (1, None),
|
||||||
|
"--heartbeat": (1, None),
|
||||||
|
"--heartbeat-misses": (1, None),
|
||||||
|
"--ack-timeout": (1, None),
|
||||||
|
}
|
||||||
|
|
||||||
|
INPUTS = ["30", "0700", "0900", " 30 ", "0", "-5", "abc", "", "1e3", "9999999999", "000000000030"]
|
||||||
|
|
||||||
|
|
||||||
|
def _expected(value: str, minimum: int):
|
||||||
|
s = value.strip()
|
||||||
|
if not s.isdigit():
|
||||||
|
return None
|
||||||
|
n = int(s)
|
||||||
|
return n if minimum <= n <= 999_999_999 else None
|
||||||
|
|
||||||
|
|
||||||
|
def _base_env(home: Path, extra=None) -> dict:
|
||||||
|
env = {k: v for k, v in os.environ.items() if k not in INT_KNOBS and not k.startswith("MAM_JOB")}
|
||||||
|
env.update({"HOME": str(home), "WORKSPACE_ROOT": str(home), "MAM_ENV_FILE": str(home / ".mam.env.absent"),
|
||||||
|
"DELEGATE_JOB_PYTHON": sys.executable, "AGENT_PYTHON_BIN": sys.executable,
|
||||||
|
"PYTHONPATH": str(SKILLS), "DELEGATE_JOB_LOGS_DIR": str(home / "logs")})
|
||||||
|
if extra:
|
||||||
|
env.update(extra)
|
||||||
|
return env
|
||||||
|
|
||||||
|
|
||||||
|
def _dry_submit(home: Path, extra_env=None, *flags) -> subprocess.CompletedProcess:
|
||||||
|
return subprocess.run(["bash", str(WRAPPER), "submit", "--agent", "human", "--dry-run",
|
||||||
|
"--registry-dir", str(home / "reg"), "--prompt", "p", *flags],
|
||||||
|
capture_output=True, text=True, env=_base_env(home, extra_env), cwd=str(home), timeout=120)
|
||||||
|
|
||||||
|
|
||||||
|
def _record(home: Path, res: subprocess.CompletedProcess) -> dict:
|
||||||
|
assert res.returncode == 0, res.stdout + res.stderr
|
||||||
|
jid = re.search(r"registered job:\s*(\S+)", res.stdout).group(1)
|
||||||
|
return json.loads((home / "reg" / f"{jid}.json").read_text())
|
||||||
|
|
||||||
|
|
||||||
|
def _runner(home: Path, extra_env=None, *argv) -> subprocess.CompletedProcess:
|
||||||
|
"""The real runner in its own process: parse_args, then the budget and the
|
||||||
|
contract flags it would hand the wrapper."""
|
||||||
|
code = ("import json, sys; from lib_py.loop import runner as lr; "
|
||||||
|
"r = lr.LoopRunner(['--creator', 'c', '--task', 't'] + sys.argv[1:]); r.parse_args(); "
|
||||||
|
"print(json.dumps({'budget': r.job_budget, 'contract': r.contract_args()}))")
|
||||||
|
return subprocess.run([sys.executable, "-c", code, *argv], capture_output=True, text=True,
|
||||||
|
env=_base_env(home, extra_env), cwd=str(home), timeout=60)
|
||||||
|
|
||||||
|
|
||||||
|
def _bash(snippet: str, home: Path, extra_env=None) -> subprocess.CompletedProcess:
|
||||||
|
return subprocess.run(["bash", "-c", f'source "{LIB_SH}" >/dev/null 2>&1; {snippet}'], capture_output=True,
|
||||||
|
text=True, env=_base_env(home, extra_env), cwd=str(home), timeout=60)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# B1 - one table, the real wrapper and the real runner agree
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("value", INPUTS)
|
||||||
|
def test_b1_wrapper_and_runner_give_the_same_budget(tmp_path, value):
|
||||||
|
minimum, default = INT_KNOBS["MAM_JOB_TIMEOUT_S"]
|
||||||
|
want = _expected(value, minimum)
|
||||||
|
budget = want if want is not None else int(default)
|
||||||
|
warns = want is None and value != ""
|
||||||
|
|
||||||
|
wres = _dry_submit(tmp_path, {"MAM_JOB_TIMEOUT_S": value})
|
||||||
|
assert _record(tmp_path, wres)["timeout_sec"] == budget
|
||||||
|
assert ("MAM_JOB_TIMEOUT_S=" in wres.stderr and "ignored" in wres.stderr) is warns, wres.stderr
|
||||||
|
# The record goes through registry.py's int(); the wrapper's own deadlines
|
||||||
|
# (START_TIME + TIMEOUT) are bash arithmetic on TIMEOUT - check that value.
|
||||||
|
ares = subprocess.run(["bash", "-c", f'source "{WRAPPER}"; parse_opts; echo "arith=$(( 0 + TIMEOUT ))"'],
|
||||||
|
capture_output=True, text=True, cwd=str(tmp_path), timeout=60,
|
||||||
|
env=_base_env(tmp_path, {"MAM_JOB_TIMEOUT_S": value, "MAM_DELEGATE_JOB_NO_MAIN": "1"}))
|
||||||
|
assert f"arith={budget}" in ares.stdout, (ares.stdout, ares.stderr)
|
||||||
|
|
||||||
|
rres = _runner(tmp_path, {"MAM_JOB_TIMEOUT_S": value})
|
||||||
|
assert rres.returncode == 0, rres.stderr
|
||||||
|
got = json.loads(rres.stdout.strip().splitlines()[-1])
|
||||||
|
assert got["budget"] == budget
|
||||||
|
assert got["contract"][:2] == ["--timeout", str(budget)]
|
||||||
|
assert ("MAM_JOB_TIMEOUT_S=" in rres.stderr and "ignored" in rres.stderr) is warns, rres.stderr
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("minimum", [0, 1])
|
||||||
|
def test_b1_mam_int_and_positive_int_are_one_rule(tmp_path, minimum):
|
||||||
|
if str(SKILLS) not in sys.path:
|
||||||
|
sys.path.insert(0, str(SKILLS))
|
||||||
|
from lib_py.loop.runner import _positive_int
|
||||||
|
|
||||||
|
script = "; ".join(f'printf "%s\\n" "$(mam_int {json.dumps(v)} {minimum} || echo invalid)"' for v in INPUTS)
|
||||||
|
res = _bash(script, tmp_path)
|
||||||
|
bash = [None if l == "invalid" else int(l) for l in res.stdout.splitlines()]
|
||||||
|
want = [_expected(v, minimum) for v in INPUTS]
|
||||||
|
assert bash == want, (res.stdout, res.stderr)
|
||||||
|
assert [_positive_int(v, minimum) for v in INPUTS] == want
|
||||||
|
assert want[INPUTS.index("0")] == (0 if minimum == 0 else None)
|
||||||
|
assert want[INPUTS.index("0700")] == 700
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("knob", ["SKS_DIALOG_TIMEOUT", "SKS_BUSY_TIMEOUT", "SKS_DELIVERY_GRACE_S"])
|
||||||
|
def test_b1_sks_knobs_keep_zero_and_read_decimal(tmp_path, knob):
|
||||||
|
minimum, default = INT_KNOBS[knob]
|
||||||
|
for value, want, warns in (("0", "0", False), ("0900", "900", False), ("", default, False),
|
||||||
|
("abc", default, True), ("-1", default, True)):
|
||||||
|
res = _bash(f"mam_int_knob {knob} {default} {minimum}", tmp_path, {knob: value})
|
||||||
|
assert res.stdout.strip() == want, (value, res.stdout, res.stderr)
|
||||||
|
assert ("ignored" in res.stderr) is warns, (value, res.stderr)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# B2 - flags
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def test_b2_wrapper_flags_are_decimal_or_refused(tmp_path):
|
||||||
|
rec = _record(tmp_path, _dry_submit(tmp_path, None, "--timeout", "0700", "--heartbeat", "010",
|
||||||
|
"--heartbeat-misses", "04", "--ack-timeout", "0090", "--idle-timeout", "0900"))
|
||||||
|
assert (rec["timeout_sec"], rec["heartbeat_sec"], rec["heartbeat_misses"], rec["ack_timeout_sec"],
|
||||||
|
rec["idle_timeout_sec"]) == (700, 10, 4, 90, 900)
|
||||||
|
ok = _dry_submit(tmp_path, None, "--timeout", "30")
|
||||||
|
assert _record(tmp_path, ok)["timeout_sec"] == 30
|
||||||
|
for flag in ("--timeout", "--idle-timeout", "--heartbeat", "--heartbeat-misses", "--ack-timeout"):
|
||||||
|
for bad in ("abc", "0", "-5", "9999999999"):
|
||||||
|
res = _dry_submit(tmp_path, None, flag, bad)
|
||||||
|
assert res.returncode == 1, (flag, bad, res.stdout, res.stderr)
|
||||||
|
assert f"ERROR: {flag} {bad}: not a positive decimal integer" in res.stderr, res.stderr
|
||||||
|
assert "registered job" not in res.stdout
|
||||||
|
|
||||||
|
|
||||||
|
def test_b2_runner_job_timeout_is_decimal_or_refused(tmp_path):
|
||||||
|
res = _runner(tmp_path, None, "--job-timeout", "0700")
|
||||||
|
assert res.returncode == 0, res.stderr
|
||||||
|
assert json.loads(res.stdout.strip().splitlines()[-1])["budget"] == 700
|
||||||
|
for bad in ("0", "abc", "-5", "9999999999"):
|
||||||
|
res = _runner(tmp_path, None, "--job-timeout", bad)
|
||||||
|
assert res.returncode == 1, (bad, res.stderr)
|
||||||
|
assert "--job-timeout requires a positive decimal integer" in res.stderr
|
||||||
|
|
||||||
|
|
||||||
|
def test_b2_runner_passes_normalised_contract_flags(tmp_path):
|
||||||
|
res = _runner(tmp_path, {"MAM_HEARTBEAT_SEC": "0030", "MAM_HEARTBEAT_MISSES": " 04 ", "MAM_ACK_TIMEOUT_S": "0"})
|
||||||
|
assert res.returncode == 0, res.stderr
|
||||||
|
contract = json.loads(res.stdout.strip().splitlines()[-1])["contract"]
|
||||||
|
assert contract[2:] == ["--heartbeat", "30", "--heartbeat-misses", "4"], contract
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# B3 - the derived idle timeout
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def test_b3_idle_timeout_derives_from_decimal_values(tmp_path):
|
||||||
|
rec = _record(tmp_path, _dry_submit(tmp_path, {"MAM_HEARTBEAT_SEC": "010", "MAM_HEARTBEAT_MISSES": "04"}))
|
||||||
|
assert (rec["heartbeat_sec"], rec["heartbeat_misses"], rec["idle_timeout_sec"]) == (10, 4, 40)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("knob,default", [("MAM_HEARTBEAT_SEC", 30), ("MAM_HEARTBEAT_MISSES", 4),
|
||||||
|
("MAM_ACK_TIMEOUT_S", 180)])
|
||||||
|
def test_b3_invalid_contract_env_falls_back_with_a_warning(tmp_path, knob, default):
|
||||||
|
field = {"MAM_HEARTBEAT_SEC": "heartbeat_sec", "MAM_HEARTBEAT_MISSES": "heartbeat_misses",
|
||||||
|
"MAM_ACK_TIMEOUT_S": "ack_timeout_sec"}[knob]
|
||||||
|
res = _dry_submit(tmp_path, {knob: "0"})
|
||||||
|
assert _record(tmp_path, res)[field] == default
|
||||||
|
assert f"WARNING: {knob}=0 ignored" in res.stderr
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# B4 - the ack waits
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def _job(reg: Path, jid: str, tok: str, started: bool) -> None:
|
||||||
|
reg.mkdir(parents=True, exist_ok=True)
|
||||||
|
(reg / f"{jid}.json").write_text(json.dumps({"job_id": jid, "status": "pending", "dispatch_token": tok}))
|
||||||
|
if started:
|
||||||
|
(reg / f"{jid}.events.log").write_text(
|
||||||
|
json.dumps({"event": "started", "seq": 1, "data": {"dispatch_token": tok}}) + "\n")
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("grace,started,rc", [("0900", True, 0), ("0", True, 0), ("0", False, 1)])
|
||||||
|
def test_b4_wrapper_inject_ack_grace(tmp_path, grace, started, rc):
|
||||||
|
reg = tmp_path / "jobs"
|
||||||
|
_job(reg, "abcd1234", "tok", started)
|
||||||
|
script = (f'source "{WRAPPER}"; set +e; REGISTRY_DIR="{reg}"; PY="{sys.executable}"; TIMEOUT=60; '
|
||||||
|
f'START_TIME=$(date +%s); _disp_tok="tok"; _inject_acknowledged abcd1234; echo "rc=$?"')
|
||||||
|
t0 = time.time()
|
||||||
|
res = subprocess.run(["bash", "-c", script], capture_output=True, text=True, cwd=str(tmp_path), timeout=60,
|
||||||
|
env=_base_env(tmp_path, {"MAM_DELEGATE_JOB_NO_MAIN": "1", "MAM_INJECT_ACK_GRACE_S": grace}))
|
||||||
|
assert f"rc={rc}" in res.stdout, (res.stdout, res.stderr)
|
||||||
|
assert "value too great for base" not in res.stderr and "ignored" not in res.stderr, res.stderr
|
||||||
|
assert time.time() - t0 < 10, "0 must mean check once, not wait"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("ack,started,rc", [("0900", True, 0), ("0", True, 0), ("0", False, 1)])
|
||||||
|
def test_b4_delegate_wait_started(tmp_path, ack, started, rc):
|
||||||
|
reg = tmp_path / "jobs"
|
||||||
|
_job(reg, "abcd1234", "tok", started)
|
||||||
|
t0 = time.time()
|
||||||
|
res = _bash(f'delegate_wait_started abcd1234 {ack} tok; echo "rc=$?"', tmp_path, {"MAM_JOBS_DIR": str(reg)})
|
||||||
|
assert f"rc={rc}" in res.stdout, (res.stdout, res.stderr)
|
||||||
|
assert "value too great for base" not in res.stderr and "ignored" not in res.stderr, res.stderr
|
||||||
|
assert time.time() - t0 < 10
|
||||||
|
|
||||||
|
|
||||||
|
def test_b4_delegate_wait_started_rejects_garbage_with_a_warning(tmp_path):
|
||||||
|
reg = tmp_path / "jobs"
|
||||||
|
_job(reg, "abcd1234", "tok", True)
|
||||||
|
res = _bash('delegate_wait_started abcd1234 abc tok; echo "rc=$?"', tmp_path, {"MAM_JOBS_DIR": str(reg)})
|
||||||
|
assert "rc=0" in res.stdout and "onboarding ack timeout abc ignored" in res.stderr, res.stderr
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# B5 - static guard
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
_ARITH = re.compile(r"\$\(\((.*?)\)\)|\(\((.*?)\)\)|\[\[(.*?)\]\]")
|
||||||
|
_ENV_DEFAULT = re.compile(r"\$\{(MAM|SKS|DELEGATE_JOB)_[A-Z0-9_]*:-")
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("path", [WRAPPER, LIB_SH], ids=lambda p: p.name)
|
||||||
|
def test_b5_no_env_default_inside_arithmetic(path):
|
||||||
|
bad = []
|
||||||
|
for n, line in enumerate(path.read_text().splitlines(), 1):
|
||||||
|
for m in _ARITH.finditer(line):
|
||||||
|
arith, cmd, test = m.groups()
|
||||||
|
body = arith if arith is not None else cmd if cmd is not None else test
|
||||||
|
is_arith = test is None or re.search(r"\s-(gt|ge|lt|le|eq|ne)\s", body)
|
||||||
|
if is_arith and _ENV_DEFAULT.search(body):
|
||||||
|
bad.append(f"{path.name}:{n}: {line.strip()}")
|
||||||
|
assert not bad, "normalise with mam_int_knob first:\n" + "\n".join(bad)
|
||||||
|
|
||||||
|
|
||||||
|
def test_b5_every_normalised_knob_uses_the_table_minimum():
|
||||||
|
src = WRAPPER.read_text() + "\n" + LIB_SH.read_text()
|
||||||
|
calls = re.findall(r"mam_int_knob ([A-Z_]+) (\S+) (\d+)\)", src)
|
||||||
|
seen = {}
|
||||||
|
for name, default, minimum in calls:
|
||||||
|
assert name in INT_KNOBS, f"{name} is normalised but missing from INT_KNOBS"
|
||||||
|
want_min, want_default = INT_KNOBS[name]
|
||||||
|
assert int(minimum) == want_min, (name, minimum)
|
||||||
|
assert default.strip('"') == want_default, (name, default)
|
||||||
|
seen[name] = True
|
||||||
|
env_knobs = {k for k in INT_KNOBS if re.fullmatch(r"[A-Z_]+", k)}
|
||||||
|
assert set(seen) == env_knobs, f"in INT_KNOBS but never normalised: {env_knobs - set(seen)}"
|
||||||
|
|
||||||
|
# delegate_wait_started's argument and the wrapper's flags
|
||||||
|
assert re.search(r'ack_s=\$\(mam_int "\$ack_s" 0\)', src)
|
||||||
|
loop = re.search(r"for _flag in ([^;]+); do\n(.*?)done", WRAPPER.read_text(), re.S)
|
||||||
|
flags = {f.split(":")[0] for f in loop.group(1).split()}
|
||||||
|
assert {f"--{f}" for f in flags} == {k for k in INT_KNOBS if k.startswith("--")}
|
||||||
|
assert 'mam_int "${!_var}" 1' in loop.group(2)
|
||||||
Reference in new issue
Block a user