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 <noreply@anthropic.com>

.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 RC_ERROR
-        return 0
+        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 RC_ERROR
-            return 0
+            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 RC_ERROR
-    return 0
+    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__":

.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() {

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"
 
 

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

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',
+        'if [[ $sub_rc -eq $SUB_RC_COMPLETED ]]; then',
-        'elif [[ $sub_rc -eq 1 ]]; then',
+        'elif [[ $sub_rc -eq $SUB_RC_ERROR ]]; then',
-        'elif [[ $sub_rc -eq 3 ]]; 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():

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