.agents/skills/lib_py/delegation/wrapper.py

@@ -5,12 +5,16 @@ injects; everything it used to decide in bash is here, and it branches only on
 what a subcommand prints or returns:
 
   sub-status <rc> --job J --registry-dir R   subscriber rc -> job status (was _map_sub_rc)
-  subscriber-up <rc>                         exit 0: a subscriber that ended before
+  subscriber-up <rc> [--print]               exit 0: a subscriber that ended before
                                              SUBSCRIBED still lets the job proceed
   exit-code <status>                         the wrapper's exit code (was _rc_for, with
                                              MAM_DELEGATE_FAIL_CLOSED=0 -> 0)
   wait-rc <rc>                               `wait`'s rc (either clock is a timeout, 2)
-  stall-watch <session> <agent>              exit 0: start the stall watcher (Issue #30)
+  stall-watch <session> <agent> [--print]    exit 0: start the stall watcher (Issue #30)
+
+The two yes/no decisions answer by exit code (0 yes, 1 no); with --print they
+print `yes`/`no` and exit 0, so a failure to decide (any other exit code, or
+no answer) cannot pass for "no". The wrapper uses --print.
   finalize-begin / finalize-end ...          finalize_unfinished, split around the one
                                              herdr diagnostic probe bash still runs
 
@@ -294,12 +298,15 @@ def _parser() -> argparse.ArgumentParser:
     s.add_argument("rc", type=int)
     s.add_argument("--job", required=True)
     s.add_argument("--registry-dir", required=True)
-    sub.add_parser("subscriber-up").add_argument("rc", type=int)
+    s = sub.add_parser("subscriber-up")
+    s.add_argument("rc", type=int)
+    s.add_argument("--print", action="store_true", help="print yes/no and exit 0")
     sub.add_parser("exit-code").add_argument("status")
     sub.add_parser("wait-rc").add_argument("rc", type=int)
     s = sub.add_parser("stall-watch")
     s.add_argument("session")
     s.add_argument("agent")
+    s.add_argument("--print", action="store_true", help="print yes/no and exit 0")
     for name in ("finalize-begin", "finalize-end"):
         s = sub.add_parser(name)
         s.add_argument("--job", required=True)
@@ -325,18 +332,25 @@ def _parser() -> argparse.ArgumentParser:
     return p
 
 
+def _yes_no(answer: bool, printed: bool) -> int:
+    if printed:
+        print("yes" if answer else "no")
+        return 0
+    return 0 if answer else 1
+
+
 def main(argv: Optional[List[str]] = None) -> int:
     a = _parser().parse_args(argv)
     if a.cmd == "sub-status":
         print(sub_status(a.rc, a.registry_dir, a.job))
     elif a.cmd == "subscriber-up":
-        return 0 if subscriber_up(a.rc) else 1
+        return _yes_no(subscriber_up(a.rc), a.print)
     elif a.cmd == "exit-code":
         print(exit_code(a.status))
     elif a.cmd == "wait-rc":
         print(wait_rc(a.rc))
     elif a.cmd == "stall-watch":
-        return 0 if stall_watch(a.session, a.agent) else 1
+        return _yes_no(stall_watch(a.session, a.agent), a.print)
     elif a.cmd == "finalize-begin":
         print(finalize_begin(a))
     else:

.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job

@@ -118,6 +118,18 @@ _decided() {
   printf -v "$__var" '%s' "$__out"
 }
 
+# _decided_yes <subcommand> <args...>: a yes/no decision (asked with --print,
+# so "no" is an answer and a failed call is not): returns 0 for yes, 1 for no.
+_decided_yes() {
+  local yn
+  _decided yn "$@" --print
+  case "$yn" in
+    yes) return 0 ;;
+    no) return 1 ;;
+  esac
+  _decision_failed "$1" "answer '$yn'"
+}
+
 # _exit_for <status>: exit with the wrapper exit code for a job status.
 _exit_for() {
   local code
@@ -381,7 +393,7 @@ EOF
       else
         local sub_exit=0
         wait "$sub_pid" 2>/dev/null || sub_exit=$?
-        if _decide subscriber-up "$sub_exit"; then
+        if _decided_yes subscriber-up "$sub_exit"; then
           sub_ready=1
           break
         else
@@ -553,7 +565,7 @@ EOF
         else
           local sub_exit=0
           wait "$sub_pid" 2>/dev/null || sub_exit=$?
-          if _decide subscriber-up "$sub_exit"; then
+          if _decided_yes subscriber-up "$sub_exit"; then
             sub_ready=1
             break
           else
@@ -715,7 +727,7 @@ reap_subscriber() {
 start_stall_watch() {
   local session="$1" agent="$2" tok="${3:-}" att="${4:-}"
   STALL_PID=""
-  _decide stall-watch "$session" "$agent" || return 0
+  _decided_yes stall-watch "$session" "$agent" || return 0
   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")

tests/test_issue17_item4_decisions.py

@@ -651,7 +651,7 @@ def _failing_py(tmp_path: Path, decision: str, mode: str) -> Path:
     return shim
 
 
-def _human_submit(tmp_path: Path, py: str) -> subprocess.CompletedProcess:
+def _human_submit(tmp_path: Path, py: str, typ: str = "direct", **extra) -> subprocess.CompletedProcess:
     """The real wrapper, one human job, an unreachable broker: the subscriber
     ends with rc 3, the job is closed at its 3 s wall clock and the wrapper
     exits through sub-status, finalize-begin/-end and exit-code (rc 3)."""
@@ -660,7 +660,10 @@ def _human_submit(tmp_path: Path, py: str) -> subprocess.CompletedProcess:
     env.update(HOME=str(tmp_path), WORKSPACE_ROOT=str(tmp_path), DELEGATE_JOB_PYTHON=py,
                AGENT_PYTHON_BIN=sys.executable, PYTHONPATH=str(SKILLS), DELEGATE_JOB_LOGS_DIR=str(tmp_path / "logs"),
                MAM_ENV_FILE=str(tmp_path / ".mam.env.absent"), MQTT_BROKER="127.0.0.1", MQTT_PORT="9", MQTT_TLS="0")
-    return subprocess.run(["bash", str(WRAPPER), "submit", "--agent", "human", "--registry-dir", str(tmp_path / "reg"),
+    env.update({k: str(v) for k, v in extra.items()})
+    loop = ["--reviewer", "human", "--reviewer-session", "herdr:human"] if typ != "direct" else []
+    return subprocess.run(["bash", str(WRAPPER), "submit", "--agent", "human", "--type", typ, *loop,
+                           "--registry-dir", str(tmp_path / "reg"),
                            "--workdir", str(tmp_path), "--prompt", "n2", "--timeout", "3", "--idle-timeout", "2"],
                           capture_output=True, text=True, env=env, cwd=str(tmp_path), timeout=120)
 
@@ -684,6 +687,40 @@ def test_f_a_failed_decision_ends_the_submit_with_1(tmp_path, decision, mode):
     assert f"ERROR: the wrapper could not decide '{decision}'" in res.stderr, res.stderr[-800:]
 
 
+@pytest.mark.parametrize("mode", ["error", "gone", "empty"])
+@pytest.mark.parametrize("decision,typ", [("subscriber-up", "direct"), ("subscriber-up", "loop"),
+                                          ("stall-watch", "direct")])
+def test_f_a_failed_yes_no_decision_ends_the_submit_with_1(tmp_path, decision, typ, mode):
+    """Review R1 (codex): a yes/no decision that fails must not pass for "no".
+    The subscriber stand-in ends with 0 before SUBSCRIBED: read as "not up",
+    the error path mapped rc 0 to completed and the submit ended with 0."""
+    res = _human_submit(tmp_path, str(_failing_py(tmp_path, decision, mode)), typ, SUB_STANDIN=1, SUB_RC=0)
+    assert res.returncode == 1, (res.returncode, res.stderr[-800:])
+    assert f"ERROR: the wrapper could not decide '{decision}'" in res.stderr, res.stderr[-800:]
+
+
+@pytest.mark.parametrize("typ", ["direct", "loop"])
+@pytest.mark.parametrize("sub_rc,up", [(0, True), (1, False)])
+def test_f_yes_and_no_keep_their_branches(tmp_path, typ, sub_rc, up):
+    """The same runs without a failure: "yes" goes on, "no" takes the error path."""
+    res = _human_submit(tmp_path, str(_failing_py(tmp_path, "none", "error")), typ, SUB_STANDIN=1, SUB_RC=sub_rc)
+    assert "could not decide" not in res.stderr, res.stderr[-800:]
+    assert ("subscriber died early" in res.stderr) is (not up), res.stderr[-800:]
+    if typ == "direct":
+        assert res.returncode == (0 if up else 1), (res.returncode, res.stderr[-800:])
+
+
+@pytest.mark.parametrize("rc,up", [(0, True), (3, True), (1, False), (2, False), (4, False)])
+def test_u_yes_no_print(capsys, rc, up):
+    assert decide.main(["subscriber-up", str(rc), "--print"]) == 0
+    assert capsys.readouterr().out == ("yes\n" if up else "no\n")
+    assert decide.main(["subscriber-up", str(rc)]) == (0 if up else 1)  # the rc form is unchanged
+    assert decide.main(["stall-watch", "herdr:c", "claude", "--print"]) == 0
+    assert capsys.readouterr().out == "yes\n"
+    assert decide.main(["stall-watch", "remote:r", "claude", "--print"]) == 0
+    assert capsys.readouterr().out == "no\n"
+
+
 @pytest.mark.parametrize("mode", ["error", "empty"])
 def test_f_a_failed_wait_rc_ends_wait_with_1(tmp_path, mode):
     py = _failing_py(tmp_path, "wait-rc", mode)
@@ -695,11 +732,12 @@ def test_f_a_failed_wait_rc_ends_wait_with_1(tmp_path, mode):
 
 
 def test_s_no_printed_decision_is_read_raw():
-    """Every printed answer goes through _decided/_exit_for; the boolean ones
+    """Every printed answer goes through _decided/_exit_for/_decided_yes (the
-    (subscriber-up, stall-watch) are read by their rc, and finalize-end is
+    yes/no ones with --print), and finalize-end is checked by its rc."""
-    checked by its rc."""
     text = "\n".join(l for l in WRAPPER.read_text().splitlines() if not l.lstrip().startswith("#"))
     raw = [l.strip() for l in text.splitlines() if "$(_decide" in l]
     assert raw == ['__out="$(_decide "$@")" || __rc=$?',
                    '{ out="$(_decide finalize-end --job "$job_id" --registry-dir "$REGISTRY_DIR" --status "$job_status" \\'], raw
     assert 'exit "$(' not in text
+    assert not re.search(r"\b_decide (subscriber-up|stall-watch)\b", text), "a yes/no decision read by its rc"
+    assert text.count("_decided_yes subscriber-up") == 2 and text.count("_decided_yes stall-watch") == 1