From 1673818bb5fed14ed6fee87884e70d39e81fe91f Mon Sep 17 00:00:00 2001 From: Godopu Date: Sat, 26 Sep 2026 00:22:43 +0000 Subject: [PATCH] fix(issue-33): subscriber exit codes are named on both sides; `wait` keeps a spent budget at rc 2; G-8 is deterministic - job_subscriber.py returns only named codes (RC_COMPLETED 0, RC_ERROR 1, RC_NO_EVENT 2, RC_BROKER_UNAVAILABLE 3, RC_WALL_CLOCK 4). The wrapper mirrors them as SUB_RC_* and compares subscriber codes only by name; a test keeps the mirror equal to the Python constants and forbids numeric comparisons of $sub_rc/$sub_exit and numeric `return` in the subscriber. - `delegate-job wait` passed the subscriber's code straight through, so a spent budget would have reached its callers as 4; it is 2 again (B7 fails without the mapping). - tier1 G-8 gave both clocks 0.5 s and raced; the wall clock now runs out first by construction (idle 30 s) and the test asserts RC_WALL_CLOCK. Other subscriber rc asserts use the names. Co-Authored-By: Claude Opus 5.5 --- .../lib_py/delegation/job_subscriber.py | 24 +++++++------ .../multi-agent-mux-delegate-job | 35 +++++++++++-------- tests/test_issue18_liveness_contract.py | 2 +- tests/test_issue33_budget.py | 15 ++++++++ tests/test_issue5_phase1_delegation.py | 22 +++++++++--- tests/test_tier1_unit.py | 16 +++++---- 6 files changed, 77 insertions(+), 37 deletions(-) diff --git a/.agents/skills/lib_py/delegation/job_subscriber.py b/.agents/skills/lib_py/delegation/job_subscriber.py index f4244de..ef8cb83 100755 --- a/.agents/skills/lib_py/delegation/job_subscriber.py +++ b/.agents/skills/lib_py/delegation/job_subscriber.py @@ -49,11 +49,15 @@ logger = logging.getLogger("delegate_job.job_subscriber") TERMINAL_EVENTS = ("completed", "error") TERMINAL_STATUSES = ("completed", "error", "cancelled") STATE_EVENT = "state" +# Exit codes. The wrapper mirrors them as SUB_RC_* (a test keeps the two equal). +RC_COMPLETED = 0 +RC_ERROR = 1 # Issue #33: which clock ended the wait. The wrapper maps both to a timeout # (caller rc 2) and labels the close from them: rc 2 -> deadline_timeout # (no event for the idle window, K x N), rc 4 -> wall_clock_timeout (the job # budget, --timeout, ran out while events may still have been arriving). RC_NO_EVENT = 2 +RC_BROKER_UNAVAILABLE = 3 RC_WALL_CLOCK = 4 @@ -350,9 +354,9 @@ def _run_subscriber(args) -> int: jobs = _collect_jobs(args) except FileNotFoundError as exc: logger.error("%s", exc) - return 2 + return RC_NO_EVENT # nothing to watch: reported like silence, as before #33 if not jobs: - return 2 + return RC_NO_EVENT expected_ids: Set[str] = {j["job_id"] for j in jobs} tokens = {j["job_id"]: j.get("auth_token") for j in jobs} @@ -451,8 +455,8 @@ def _run_subscriber(args) -> int: _check_disk_fallback(pending, args.registry_dir, terminal, dispatch_tokens, attempts) if not pending: if any(state == "error" for state in terminal.values()): - return 1 - return 0 + return RC_ERROR + return RC_COMPLETED try: mqtt_common.with_retry( @@ -467,10 +471,10 @@ def _run_subscriber(args) -> int: _check_disk_fallback(pending, args.registry_dir, terminal, dispatch_tokens, attempts) if not pending: if any(state == "error" for state in terminal.values()): - return 1 - return 0 + return RC_ERROR + return RC_COMPLETED logger.error("broker connection failed: %s", exc) - return 3 + return RC_BROKER_UNAVAILABLE try: while pending: @@ -552,8 +556,8 @@ def _run_subscriber(args) -> int: # All jobs reached a terminal state. error wins over completed. if any(state == "error" for state in terminal.values()): - return 1 - return 0 + return RC_ERROR + return RC_COMPLETED def main(argv=None) -> int: @@ -590,7 +594,7 @@ def main(argv=None) -> int: return _run_subscriber(args) except Exception as exc: logger.error("subscriber fatal error: %s", exc) - return 3 + return RC_BROKER_UNAVAILABLE if __name__ == "__main__": diff --git a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job index a798d69..08b54e0 100755 --- a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job +++ b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job @@ -140,6 +140,8 @@ 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). if [[ -n "${MAM_JOB_TIMEOUT_S:-}" ]]; then if [[ "$MAM_JOB_TIMEOUT_S" =~ ^[0-9]+$ && "$MAM_JOB_TIMEOUT_S" -gt 0 ]]; then @@ -289,7 +291,7 @@ EOF # 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" 1 "preflight" + finalize_unfinished "$JOB_ID" "error" "$SUB_RC_ERROR" "preflight" local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}" echo "$logs_root/$JOB_ID" exit 5 @@ -334,7 +336,7 @@ EOF else local sub_exit=0 wait "$sub_pid" 2>/dev/null || sub_exit=$? - if [ $sub_exit -eq 0 ] || [ $sub_exit -eq 3 ]; then + if [ $sub_exit -eq $SUB_RC_COMPLETED ] || [ $sub_exit -eq $SUB_RC_BROKER_UNAVAILABLE ]; then sub_ready=1 break else @@ -360,7 +362,7 @@ EOF # 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" 1 "inject" + finalize_unfinished "$JOB_ID" "error" "$SUB_RC_ERROR" "inject" local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}" echo "$logs_root/$JOB_ID" exit 1 @@ -456,7 +458,7 @@ EOF # 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" 1 "preflight" + finalize_unfinished "$JOB_ID" "error" "$SUB_RC_ERROR" "preflight" local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}" echo "$logs_root/$JOB_ID" exit 5 @@ -512,7 +514,7 @@ EOF else local sub_exit=0 wait "$sub_pid" 2>/dev/null || sub_exit=$? - if [ $sub_exit -eq 0 ] || [ $sub_exit -eq 3 ]; then + if [ $sub_exit -eq $SUB_RC_COMPLETED ] || [ $sub_exit -eq $SUB_RC_BROKER_UNAVAILABLE ]; then sub_ready=1 break else @@ -545,7 +547,7 @@ EOF # Trigger agent if ! deliver_prompt "$JOB_ID" "$instructions" "$current_session" "$logf"; then reap_subscriber "$sub_pid" - finalize_unfinished "$JOB_ID" "error" 1 "inject" + finalize_unfinished "$JOB_ID" "error" "$SUB_RC_ERROR" "inject" local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}" echo "$logs_root/$JOB_ID" exit 1 @@ -827,11 +829,11 @@ PYEOF _map_sub_rc() { local sub_rc="$1" job_id="$2" local job_status="running" - if [[ $sub_rc -eq 0 ]]; then + if [[ $sub_rc -eq $SUB_RC_COMPLETED ]]; then job_status="completed" - elif [[ $sub_rc -eq 1 ]]; then + elif [[ $sub_rc -eq $SUB_RC_ERROR ]]; then job_status="error" - elif [[ $sub_rc -eq 3 ]]; then + elif [[ $sub_rc -eq $SUB_RC_BROKER_UNAVAILABLE ]]; then job_status="broker_unavailable" local disk_st disk_st="$(_disk_status "$job_id")" @@ -839,7 +841,7 @@ _map_sub_rc() { job_status="$disk_st" fi else - job_status="timeout" + job_status="timeout" # SUB_RC_NO_EVENT, SUB_RC_WALL_CLOCK (and any other code) fi echo "$job_status" } @@ -952,7 +954,7 @@ finalize_unfinished() { # 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" = "4" ] && clock="wall" + [ "$sub_rc" = "$SUB_RC_WALL_CLOCK" ] && clock="wall" local close_req="deadline_timeout" [ "$clock" = "wall" ] && [ "$phase" != "inject" ] && close_req="wall_clock_timeout" local detail="" @@ -992,7 +994,7 @@ finalize_unfinished() { [ -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 1 ] || [ "$phase" = "inject" ]; then + 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 @@ -1199,13 +1201,18 @@ cmd_logs() { 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" + --job "$JOB_ID" --timeout "$TIMEOUT" || rc=$? else "$PY" "$SCRIPT_DIR/scripts/job_subscriber.py" --registry-dir "$REGISTRY_DIR" \ - --wait-any --timeout "$TIMEOUT" + --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" } main() { diff --git a/tests/test_issue18_liveness_contract.py b/tests/test_issue18_liveness_contract.py index 6131960..8bd4b02 100644 --- a/tests/test_issue18_liveness_contract.py +++ b/tests/test_issue18_liveness_contract.py @@ -1030,7 +1030,7 @@ def test_a_replayed_retained_state_does_not_keep_a_silent_job_alive(tmp_path, mo "--timeout", "20"]) elapsed = time.monotonic() - started fake.stopped.set() - assert rc == 2 + assert rc == js.RC_NO_EVENT assert elapsed < 8, f"the replayed state kept the silent job alive for {elapsed:.1f}s" diff --git a/tests/test_issue33_budget.py b/tests/test_issue33_budget.py index 5965080..269c002 100644 --- a/tests/test_issue33_budget.py +++ b/tests/test_issue33_budget.py @@ -341,3 +341,18 @@ def test_r3_a_budget_spent_waiting_on_an_unreachable_broker_is_wall_clock_timeou if p.poll() is None: p.kill() p.communicate(timeout=10) + + + +# Issue #33 follow-up (job 60da22f6): `delegate-job wait` passes the subscriber's +# code through; a spent budget stays rc 2 for its callers, as before the split. +@pytest.mark.parametrize("state,rc", [("running", 2), ("completed", 0)]) +def test_b7_wait_keeps_a_spent_budget_at_rc_2(tmp_path, state, rc): + env = _env(tmp_path, broker=True) + reg = tmp_path / "reg" + jid = _jid(_dry_submit(env, reg)) + if state == "completed": + _publish(env, reg, jid, "completed") + res = subprocess.run(["bash", str(WRAPPER), "wait", "--job", jid, "--registry-dir", str(reg), "--timeout", "2"], + capture_output=True, text=True, env=env, cwd=env["HOME"], timeout=60) + assert res.returncode == rc, res.stdout + res.stderr diff --git a/tests/test_issue5_phase1_delegation.py b/tests/test_issue5_phase1_delegation.py index e09b30f..1ae9b57 100644 --- a/tests/test_issue5_phase1_delegation.py +++ b/tests/test_issue5_phase1_delegation.py @@ -361,7 +361,11 @@ def test_job_subscriber_cli_contract_snapshot(mam_sandbox): for doc in (" 0 all watched jobs reached", " 1 any watched job reached", " 2 no event for the idle window", " 4 the wall-clock budget"): assert doc in src, doc - assert 'return 3' in src, "fatal-error exit code 3 must survive" + # Issue #33: codes are named; their values are the frozen contract. + assert (job_subscriber.RC_COMPLETED, job_subscriber.RC_ERROR, job_subscriber.RC_NO_EVENT, + job_subscriber.RC_BROKER_UNAVAILABLE, job_subscriber.RC_WALL_CLOCK) == (0, 1, 2, 3, 4) + assert 'return RC_BROKER_UNAVAILABLE' in src, "fatal-error exit code 3 must survive" + assert not re.search(r"^\s+return [0-9]+\s*(#.*)?$", src, re.M), "exit codes are returned by name" # ============================================================================== @@ -378,17 +382,25 @@ def test_event_to_status_mapping_is_frozen(mam_sandbox): } -def test_wrapper_exit_code_mapping_literals_are_frozen(): +def test_wrapper_exit_code_mapping_literals_are_frozen(mam_sandbox): wrapper = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job" text = wrapper.read_text(encoding="utf-8") for literal in ( - 'if [[ $sub_rc -eq 0 ]]; then', - 'elif [[ $sub_rc -eq 1 ]]; then', - 'elif [[ $sub_rc -eq 3 ]]; then', + 'if [[ $sub_rc -eq $SUB_RC_COMPLETED ]]; then', + 'elif [[ $sub_rc -eq $SUB_RC_ERROR ]]; then', + 'elif [[ $sub_rc -eq $SUB_RC_BROKER_UNAVAILABLE ]]; then', + '[ "$sub_rc" = "$SUB_RC_WALL_CLOCK" ] && clock="wall"', 'job_status="completed"', 'job_status="error"', ): assert literal in text, literal + # Issue #33: the wrapper's SUB_RC_* mirror job_subscriber's RC_* exactly, + # and the wrapper compares subscriber codes only by those names. + job_subscriber = _import_delegation(mam_sandbox, "job_subscriber") + mirror = dict(re.findall(r"\bSUB_RC_([A-Z_]+)=([0-9]+)", text)) + assert mirror == {name[3:]: str(getattr(job_subscriber, name)) + for name in dir(job_subscriber) if name.startswith("RC_")}, mirror + assert not re.search(r"\$(sub_rc|sub_exit)\"? +(-eq|-ne|=) +\"?[0-9]", text), "subscriber codes compared by value" def test_reconcile_terminal_mutation_literals_are_frozen(): diff --git a/tests/test_tier1_unit.py b/tests/test_tier1_unit.py index 091cc47..e20f379 100644 --- a/tests/test_tier1_unit.py +++ b/tests/test_tier1_unit.py @@ -721,7 +721,7 @@ def test_g5_subscriber_disk_fallback_on_broker_down_exits_0(mam_sandbox, monkeyp "--timeout", "5", "--idle-timeout", "2" ]) - assert rc == 0 + assert rc == job_subscriber.RC_COMPLETED def test_g6_subscriber_disk_fallback_outputs_source_tag(mam_sandbox, capsys, monkeypatch): @@ -744,7 +744,7 @@ def test_g6_subscriber_disk_fallback_outputs_source_tag(mam_sandbox, capsys, mon "--job", "g6test01", "--timeout", "5" ]) - assert rc == 0 + assert rc == job_subscriber.RC_COMPLETED captured = capsys.readouterr() assert "disk-fallback" in captured.out @@ -769,7 +769,7 @@ def test_g7_subscriber_disk_fallback_error_status_exits_1(mam_sandbox, monkeypat "--job", "g7test01", "--timeout", "5" ]) - assert rc == 1 + assert rc == job_subscriber.RC_ERROR def test_g8_subscriber_wait_any_does_not_exit_early_if_partial_pending(mam_sandbox, monkeypatch): @@ -806,13 +806,15 @@ def test_g8_subscriber_wait_any_does_not_exit_early_if_partial_pending(mam_sandb monkeypatch.setattr(job_subscriber, "make_client", lambda *args, **kwargs: FakeClient()) monkeypatch.setattr(mqtt_common, "with_retry", lambda fn, **kwargs: fn) + # The wall clock (0.5 s) runs out well before the idle window (30 s), so + # which timeout ends the wait is decided, not a race (Issue #33 rc split). rc = job_subscriber.main([ "--registry-dir", reg_dir, "--wait-any", "--timeout", "0.5", - "--idle-timeout", "0.5" + "--idle-timeout", "30" ]) - assert rc == 2, "Must timeout waiting for incomplete job2 rather than exiting 0" + assert rc == job_subscriber.RC_WALL_CLOCK, "Must timeout waiting for incomplete job2 rather than exiting 0" def test_g9_subscriber_broker_down_no_disk_terminal_exits_3(mam_sandbox, monkeypatch): @@ -835,14 +837,14 @@ def test_g9_subscriber_broker_down_no_disk_terminal_exits_3(mam_sandbox, monkeyp "--job", "g9test01", "--timeout", "5" ]) - assert rc == 3, f"Expected rc=3 on infrastructure broker down without disk terminal, got {rc}" + assert rc == job_subscriber.RC_BROKER_UNAVAILABLE, f"Expected rc=3 on infrastructure broker down without disk terminal, got {rc}" def test_g10_delegate_job_rc3_not_mistaken_for_error(mam_sandbox): """G-10: multi-agent-mux-delegate-job maps rc=3 to broker_unavailable and checks disk status.""" delegate_script = mam_sandbox / "skills" / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job" content = delegate_script.read_text() - assert "elif [[ $sub_rc -eq 3 ]]; then" in content + assert "elif [[ $sub_rc -eq $SUB_RC_BROKER_UNAVAILABLE ]]; then" in content assert 'job_status="broker_unavailable"' in content