test(issue-26): C0 reads broker state only after the matching ack; the local broker acks a PUBLISH after routing it

Review (fa45b0e6): paho's connect() does not wait for the CONNACK, so C0
could read protocol_levels before the broker had handled the second
CONNECT, and the `before` slice could take in a previous test's connection.

- _Client waits for CONNACK (and UNSUBACK, instead of sleeping 0.3 s);
  C0 checks the levels of its own two client ids (stats() levels_by_client).
- The local broker now routes and stores a PUBLISH before its PUBACK, so a
  publisher holding the ack may rely on the retained state it asked for
  (late subscriber replay, retained delete).
- test_c0_holds_when_the_broker_is_slow runs C0 with CONNECT/UNSUBSCRIBE
  delayed and PUBLISH routed late; the old C0 fails it ([] == [4, 4]), as do
  an ack-before-route broker and a fixed-sleep unsubscribe.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

.agents/skills/lib.sh

@@ -2335,6 +2335,35 @@ sys.exit(1)
 ' 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]
 #
 # ISSUE-23 / ISSUE-18: delivery is confirmed by the protocol, not by the
@@ -2347,6 +2376,7 @@ sys.exit(1)
 delegate_wait_started() {
   local job_id="$1" ack_s="${2:-180}" tok="${3:-}"
   [ -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
   deadline=$(( $(date +%s) + ack_s ))
   while :; do
@@ -2752,14 +2782,14 @@ send_keys_safe() {
   fi
 
   local _q_rc=0
-  _pane_quiescent "$sess" "${SKS_QUIESCENT_TRIES:-20}" "${SKS_QUIESCENT_INTERVAL:-0.5}" || _q_rc=$?
+  _pane_quiescent "$sess" "$(mam_int_knob SKS_QUIESCENT_TRIES 20 0)" "${SKS_QUIESCENT_INTERVAL:-0.5}" || _q_rc=$?
   if [ "$_q_rc" = "1" ]; then
     echo "send_keys_safe: pane never quiesced ($sess)" >&2
     return 1
   fi
 
   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
       if [ "${SKS_DIALOG_ESCAPE:-0}" = "1" ]; then
         _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)
   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
     while [ "$(date +%s)" -lt "$b_deadline" ]; do
       local tail_pane
@@ -2847,7 +2877,7 @@ send_keys_safe() {
       # 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.
       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
           [ "$(_sks_agent_status "$sess" "$agent_target")" = "working" ] && return 0
           _sks_job_started "$job_id" && return 0

.agents/skills/lib_py/delegation/envfile.py

@@ -0,0 +1,103 @@
+"""The one `.mam.env` parser (Issue #26).
+
+Rules: blank lines and `#` comments are skipped; `KEY=VALUE` with the key and
+the value stripped; one matching pair of surrounding quotes is removed; no
+shell interpretation (`$(...)`, `${...}` and backticks stay literal). A key
+must match ``^[A-Za-z_][A-Za-z0-9_]*$`` and not be one of bash's own variables
+(RESERVED); a line that breaks either, or that has no `=`, is skipped with a
+warning naming the line (never the value — the file holds secrets).
+Variables already in the environment win over the file.
+
+Stdlib only: the delegate-job wrapper runs this file by path
+(``envfile.py --shell <path>``) and evals the ``export`` lines it prints, so
+values travel over a pipe, never on a command line.
+"""
+import os
+import re
+import shlex
+import sys
+from typing import List, MutableMapping, Tuple
+
+_KEY = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
+
+# Bash's own variables (bash(1) "Shell Variables"). Assigning some runs code
+# even from a quoted value (RANDOM/SECONDS/... are evaluated arithmetically,
+# so RANDOM='a[$(cmd)]' executes cmd), others steer the shell (IFS, PS4,
+# BASH_ENV, SHELLOPTS) or are readonly. A file key never assigns them - in
+# either loader. tests/test_issue26_envfile.py checks this covers every
+# variable a clean bash defines.
+RESERVED = frozenset("""
+    BASH BASHOPTS BASHPID CDPATH CHILD_MAX COLUMNS COMPREPLY COPROC DIRSTACK EMACS ENV
+    EPOCHREALTIME EPOCHSECONDS EUID EXECIGNORE FCEDIT FIGNORE FUNCNAME FUNCNEST GLOBIGNORE
+    GLOBSORT GROUPS HISTCMD HISTCONTROL HISTFILE HISTFILESIZE HISTIGNORE HISTSIZE
+    HISTTIMEFORMAT HOSTFILE HOSTNAME HOSTTYPE IFS IGNOREEOF INPUTRC INSIDE_EMACS LINENO LINES
+    MACHTYPE MAIL MAILCHECK MAILPATH MAPFILE OLDPWD OPTARG OPTERR OPTIND OSTYPE PIPESTATUS
+    POSIXLY_CORRECT PPID PROMPT_COMMAND PROMPT_DIRTRIM PS0 PS1 PS2 PS3 PS4 PWD RANDOM REPLY
+    SECONDS SHELL SHELLOPTS SHLVL SRANDOM TIMEFORMAT TMOUT UID auto_resume histchars _
+""".split())
+RESERVED_PREFIXES = ("BASH_", "COMP_", "READLINE_")
+
+
+def is_reserved(key: str) -> bool:
+    return key in RESERVED or key.startswith(RESERVED_PREFIXES)
+
+
+def parse(path: str) -> Tuple[List[Tuple[str, str]], List[str]]:
+    """Return ([(key, value)], [warning]) for the file at ``path``."""
+    pairs: List[Tuple[str, str]] = []
+    warnings: List[str] = []
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            lines = f.read().splitlines()
+    except (OSError, UnicodeDecodeError) as e:
+        return pairs, [f"{path}: cannot read ({type(e).__name__})"]
+    for n, raw in enumerate(lines, 1):
+        line = raw.strip()
+        if not line or line.startswith("#"):
+            continue
+        key, sep, val = line.partition("=")
+        key = key.strip()
+        if not sep:
+            warnings.append(f"{path}:{n}: skipped (no '=')")
+            continue
+        if not _KEY.fullmatch(key):
+            warnings.append(f"{path}:{n}: skipped (invalid key)")
+            continue
+        if is_reserved(key):
+            warnings.append(f"{path}:{n}: skipped (reserved shell variable {key})")
+            continue
+        val = val.strip()
+        if len(val) >= 2 and val[0] == val[-1] and val[0] in "\"'":
+            val = val[1:-1]
+        pairs.append((key, val))
+    return pairs, warnings
+
+
+def apply(path: str, environ: MutableMapping[str, str]) -> Tuple[List[Tuple[str, str]], List[str]]:
+    """Set each pair not already in ``environ``; return (the pairs set, warnings)."""
+    pairs, warnings = parse(path)
+    applied = []
+    for key, val in pairs:
+        if key not in environ:
+            environ[key] = val
+            applied.append((key, val))
+    return applied, warnings
+
+
+def main(argv: List[str]) -> int:
+    if len(argv) != 2 or argv[0] != "--shell":
+        print("usage: envfile.py --shell <path>", file=sys.stderr)
+        return 2
+    path = argv[1]
+    if not os.path.isfile(path):
+        return 0
+    applied, warnings = apply(path, dict(os.environ))
+    for w in warnings:
+        print(f"WARNING: {w}", file=sys.stderr)
+    for key, val in applied:
+        print(f"export {key}={shlex.quote(val)}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))

.agents/skills/lib_py/delegation/mqtt_common.py

@@ -31,6 +31,8 @@ from typing import Any, Callable, Dict, Iterable, List, Optional
 
 import paho.mqtt.client as mqtt
 
+from lib_py.delegation import envfile
+
 logger = logging.getLogger("delegate_job.mqtt_common")
 
 _warned_deprecated_env = False
@@ -94,20 +96,9 @@ def _load_dotenv(workspace_dir: Optional[str] = None) -> None:
         _parse_env_file(legacy_env_path)
 
 def _parse_env_file(path: str) -> None:
-    try:
-        with open(path, "r", encoding="utf-8") as f:
-            for line in f:
-                line = line.strip()
-                if not line or line.startswith("#"):
-                    continue
-                if "=" in line:
-                    key, _, val = line.partition("=")
-                    key = key.strip()
-                    val = val.strip().strip('"').strip("'")
-                    if key and key not in os.environ:
-                        os.environ[key] = val
-    except Exception as e:
-        logger.warning("Failed to parse env file %s: %s", path, e)
+    """Issue #26: the rules live in envfile.py, shared with the wrapper."""
+    for warning in envfile.apply(path, os.environ)[1]:
+        logger.warning("%s", warning)
 
 _env_loaded = False
 

.agents/skills/lib_py/loop/runner.py

@@ -67,6 +67,19 @@ def log_error(msg: str) -> None:
     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
+    s = s.lstrip("0") or "0"
+    if len(s) > 9:  # like mam_int: bounded before int(), which refuses > 4300 digits
+        return None
+    n = int(s)
+    return n if n >= minimum else None
+
 class LoopRunner:
     """Orchestrates planning, code implementation, and multi-agent peer reviews."""
 
@@ -142,7 +155,7 @@ class LoopRunner:
             "  --all-reviewer         Enforce PASS verdict from all active reviewer sessions",
             "  --max-loop N           Max execution-review corrective loop runs (default: 3)",
             "  --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",
             "  --cleanup              Purge temporary job directories upon success",
             "  --worktree <path>      Source worktree for diff/base/head (control plane stays here)",
@@ -214,10 +227,10 @@ class LoopRunner:
                 self.max_loop = int(args[i + 1])
                 i += 2
             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:
-                    print("ERROR: --job-timeout requires a positive non-zero integer.", file=sys.stderr)
+                self.job_timeout = _positive_int(args[i + 1]) if i + 1 < len(args) else None
+                if self.job_timeout is None:
+                    print("ERROR: --job-timeout requires a positive decimal integer.", file=sys.stderr)
                     sys.exit(1)
-                self.job_timeout = int(args[i + 1])
                 i += 2
             elif arg == "--max-rebut":
                 if i + 1 >= len(args) or not re.match(r"^[0-9]+$", args[i + 1]):
@@ -471,9 +484,9 @@ class LoopRunner:
         for env_name, flag in (("MAM_HEARTBEAT_SEC", "--heartbeat"),
                                ("MAM_HEARTBEAT_MISSES", "--heartbeat-misses"),
                                ("MAM_ACK_TIMEOUT_S", "--ack-timeout")):
-            val = os.environ.get(env_name)
-            if val and val.isdigit() and int(val) > 0:
-                args += [flag, val]
+            val = _positive_int(os.environ.get(env_name))
+            if val is not None:
+                args += [flag, str(val)]
         return args
 
     def resolve_all_reviewers(self, state: Dict[str, Any]) -> List[str]:
@@ -593,11 +606,17 @@ class LoopRunner:
 
     @property
     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:
             return self.job_timeout
         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
     def job_wait_s(self) -> int:

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

@@ -34,8 +34,15 @@ if [[ -z "$TARGET_ENV" ]]; then
   fi
 fi
 
+# Issue #26: parsed by envfile.py (the Python loader's rules; environment wins),
+# never sourced. Values come back as shlex-quoted exports over the pipe. The
+# variables this script has already set are handed over as environment, so
+# the file cannot replace them either.
 if [[ -n "$TARGET_ENV" && -f "$TARGET_ENV" ]]; then
-  set -a; source "$TARGET_ENV"; set +a
+  _env_exports="$(SCRIPT_DIR="$SCRIPT_DIR" REPO_ROOT="$REPO_ROOT" TARGET_ENV="$TARGET_ENV" \
+    "${DELEGATE_JOB_PYTHON:-python3}" -I "$SCRIPT_DIR/../lib_py/delegation/envfile.py" --shell "$TARGET_ENV")" \
+    || { echo "ERROR: cannot read $TARGET_ENV" >&2; exit 1; }
+  eval "$_env_exports"; unset _env_exports
 fi
 
 # Source EARLY (before any herdr usage in run_agent) — this is what turns
@@ -111,7 +118,7 @@ usage() {
 multi-agent-mux-delegate-job <command> [options]
 
   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>]
           [--registry-dir <dir>] [--dry-run] [--role <role_name>]
           [--type <direct|loop|discuss>] [--reviewer <reviewer_agent>]
@@ -143,15 +150,10 @@ 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
-    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 #26: integer knobs are read as decimal by mam_int (lib.sh).
+TIMEOUT=$(mam_int_knob MAM_JOB_TIMEOUT_S 3600 1)
 # 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"
 TYPE="direct"; REVIEWER="hermes"; REVIEWER_SESSION="herdr:hermes"; MAX_ITERATIONS=5
 DEFAULT_COUNTERPART_ROLE="Reviewer"
@@ -189,6 +191,13 @@ parse_opts() {
       *) echo "unknown option: $1" >&2; usage; exit 1;;
     esac
   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 ))
   # lib.sh's send_keys_safe reads the job's own `started` from this registry.
   export SKS_REGISTRY_DIR="$REGISTRY_DIR"
@@ -768,7 +777,7 @@ _wait_started() {
 }
 
 # 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
 # the screen, not the protocol. They are no longer liveness *evidence* (a
@@ -1108,7 +1117,7 @@ inject_agent() {
       echo "ERROR: 프롬프트 주입 실패 (send_keys_safe rc=$sks_rc) — 세션 '$sess'" >&2
       return 1 ;;
   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
     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
@@ -1134,7 +1143,7 @@ remote_dispatch() {
     echo "ERROR: dispatch publish failed for job $job_id (remote:$sess)" >&2
     return 1
   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
     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

.mam.env.example

@@ -29,11 +29,11 @@
 
 # Root directory that holds Claude Code per-project conversation logs (*.jsonl).
 #default: $HOME/.claude/projects
-# CLAUDE_PROJECT_DIR=$HOME/.claude/projects
+# CLAUDE_PROJECT_DIR=/home/you/.claude/projects
 
 # Directory scanned for per-session launcher wrappers (~/.local/bin/<session>).
 #default: $HOME/.local/bin
-# LOCAL_BIN=$HOME/.local/bin
+# LOCAL_BIN=/home/you/.local/bin
 
 # Isolated Herdr session socket name (`herdr --session <name>`).
 #default: mam-<workspace-slug>
@@ -46,7 +46,7 @@
 
 # Path to the Herdr Unix domain socket.
 #default: $HOME/.config/herdr/herdr.sock
-# HERDR_SOCKET_PATH=$HOME/.config/herdr/herdr.sock
+# HERDR_SOCKET_PATH=/home/you/.config/herdr/herdr.sock
 
 # Explicit Herdr workspace scope identifier (`herdr pane list --workspace <id>`).
 # Hard filters pane/agent resolution to this workspace. When unset, falls back
@@ -107,7 +107,7 @@
 
 # Optional override for Codex CLI state / session directory.
 #default: $HOME/.codex
-# CODEX_HOME=$HOME/.codex
+# CODEX_HOME=/home/you/.codex
 
 # Optional inline JSON permissions configuration for OpenCode.
 #default: (unset → defaults to --auto CLI bypass)

tests/_local_mqtt.py

@@ -0,0 +1,277 @@
+"""Issue #26 - a test-only, in-process MQTT 3.1.1 broker.
+
+The suite used to round-trip through the public default broker
+(mqtt_common.BrokerConfig.host). conftest starts one of these per session on
+127.0.0.1 and points MQTT_BROKER/MQTT_PORT/MQTT_TLS at it, so every real
+producer (publish_event, job_subscriber, the wrapper, the remote runner, the
+monitor daemon) talks to a broker the test process owns.
+
+Scope is what the repo uses: paho-mqtt 2.x clients with its defaults
+(protocol level 4 = 3.1.1, clean session, no will), QoS 0/1, retained
+messages. No TLS, no auth checks, no persistent sessions, no QoS 2, no
+redelivery of unacknowledged QoS 1 messages. The name does not start with
+`test_`, so pytest never collects it.
+"""
+from __future__ import annotations
+
+import collections
+import socket
+import struct
+import threading
+
+CONNECT, CONNACK, PUBLISH, PUBACK = 1, 2, 3, 4
+SUBSCRIBE, SUBACK, UNSUBSCRIBE, UNSUBACK = 8, 9, 10, 11
+PINGREQ, PINGRESP, DISCONNECT = 12, 13, 14
+NAMES = {CONNECT: "CONNECT", PUBLISH: "PUBLISH", PUBACK: "PUBACK", SUBSCRIBE: "SUBSCRIBE",
+         UNSUBSCRIBE: "UNSUBSCRIBE", PINGREQ: "PINGREQ", DISCONNECT: "DISCONNECT"}
+
+_CURRENT = None
+
+
+def topic_matches(filt: str, topic: str) -> bool:
+    f, t = filt.split("/"), topic.split("/")
+    if topic.startswith("$") and f[0] in ("+", "#"):
+        return False
+    for i, part in enumerate(f):
+        if part == "#":
+            return True
+        if i >= len(t) or (part != "+" and part != t[i]):
+            return False
+    return len(f) == len(t)
+
+
+def _varint(n: int) -> bytes:
+    out = bytearray()
+    while True:
+        n, digit = divmod(n, 128)
+        out.append(digit | (0x80 if n else 0))
+        if not n:
+            return bytes(out)
+
+
+def _packet(first: int, body: bytes) -> bytes:
+    return bytes([first]) + _varint(len(body)) + body
+
+
+def _str(data: bytes, i: int):
+    (n,) = struct.unpack_from("!H", data, i)
+    return data[i + 2:i + 2 + n], i + 2 + n
+
+
+class _Conn:
+    def __init__(self, sock):
+        self.sock = sock
+        self.lock = threading.Lock()
+        self.subs = {}          # filter -> granted qos
+        self.next_pid = 0
+        self.client_id = None
+
+    def send(self, data: bytes) -> None:
+        with self.lock:
+            try:
+                self.sock.sendall(data)
+            except OSError:
+                pass
+
+    def recv_exact(self, n: int) -> bytes:
+        buf = b""
+        while len(buf) < n:
+            chunk = self.sock.recv(n - len(buf))
+            if not chunk:
+                raise ConnectionError("closed")
+            buf += chunk
+        return buf
+
+
+class LocalBroker:
+    def __init__(self):
+        self._lsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self._lsock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self._lsock.bind(("127.0.0.1", 0))
+        self._lsock.listen(128)
+        self.host, self.port = self._lsock.getsockname()
+        self._lock = threading.Lock()
+        self._conns = set()
+        self.retained = {}                     # topic -> (payload, qos)
+        self.connections = 0
+        self.protocol_levels = []              # one entry per CONNECT
+        self.client_ids = []
+        self.packets = collections.Counter()   # inbound packets by name
+        self._stopped = False
+
+    # -- lifecycle -----------------------------------------------------------
+    def start(self) -> "LocalBroker":
+        threading.Thread(target=self._accept_loop, name="local-mqtt-accept", daemon=True).start()
+        return self
+
+    def stop(self) -> None:
+        self._stopped = True
+        try:
+            self._lsock.close()
+        except OSError:
+            pass
+        with self._lock:
+            conns = list(self._conns)
+        for c in conns:
+            try:
+                c.sock.shutdown(socket.SHUT_RDWR)
+                c.sock.close()
+            except OSError:
+                pass
+
+    def env(self) -> dict:
+        return {"MQTT_BROKER": self.host, "MQTT_PORT": str(self.port), "MQTT_TLS": "0"}
+
+    def stats(self) -> dict:
+        with self._lock:
+            return {"connections": self.connections, "protocol_levels": list(self.protocol_levels),
+                    "levels_by_client": dict(zip(self.client_ids, self.protocol_levels)),
+                    "packets": dict(self.packets), "open": len(self._conns)}
+
+    # -- network -------------------------------------------------------------
+    def _accept_loop(self) -> None:
+        while not self._stopped:
+            try:
+                sock, _ = self._lsock.accept()
+            except OSError:
+                return
+            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+            threading.Thread(target=self._serve, args=(_Conn(sock),), name="local-mqtt-conn", daemon=True).start()
+
+    def _serve(self, conn: _Conn) -> None:
+        with self._lock:
+            self._conns.add(conn)
+        try:
+            while True:
+                first = conn.recv_exact(1)[0]
+                length, mult = 0, 1
+                while True:
+                    b = conn.recv_exact(1)[0]
+                    length += (b & 0x7F) * mult
+                    mult *= 128
+                    if not b & 0x80:
+                        break
+                body = conn.recv_exact(length) if length else b""
+                ptype = first >> 4
+                with self._lock:
+                    self.packets[NAMES.get(ptype, str(ptype))] += 1
+                if conn.client_id is None and ptype != CONNECT:
+                    return
+                if not self._handle(conn, ptype, first & 0x0F, body):
+                    return
+        except (ConnectionError, OSError, struct.error, IndexError):
+            return
+        finally:
+            with self._lock:
+                self._conns.discard(conn)
+            try:
+                conn.sock.close()
+            except OSError:
+                pass
+
+    def _handle(self, conn: _Conn, ptype: int, flags: int, body: bytes) -> bool:
+        if ptype == CONNECT:
+            _name, i = _str(body, 0)
+            level, cflags = body[i], body[i + 1]
+            client_id, i = _str(body, i + 4)
+            with self._lock:
+                self.connections += 1
+                self.protocol_levels.append(level)
+                self.client_ids.append(client_id.decode("utf-8", "replace"))
+            if level not in (3, 4):   # 3.1 / 3.1.1 only (no MQTT v5 in the repo)
+                conn.send(_packet(CONNACK << 4, b"\x00\x01"))
+                return False
+            conn.client_id = client_id.decode("utf-8", "replace")
+            conn.send(_packet(CONNACK << 4, b"\x00\x00"))
+            return True
+        if ptype == PUBLISH:
+            qos, retain = (flags >> 1) & 3, bool(flags & 1)
+            topic, i = _str(body, 0)
+            if qos == 2:              # the repo publishes QoS 0/1 only
+                return False
+            pid = b""
+            if qos == 1:
+                pid = body[i:i + 2]
+                i += 2
+            # Route (and store a retained message) BEFORE the PUBACK: a publisher
+            # that has its ack may rely on the broker state it asked for.
+            self._route(topic.decode("utf-8"), body[i:], qos, retain)
+            if qos == 1:
+                conn.send(_packet(PUBACK << 4, pid))
+            return True
+        if ptype == SUBSCRIBE:
+            pid, i, granted, new = body[:2], 2, bytearray(), []
+            while i < len(body):
+                filt, i = _str(body, i)
+                q = min(body[i], 1)
+                i += 1
+                conn.subs[filt.decode("utf-8")] = q
+                granted.append(q)
+                new.append((filt.decode("utf-8"), q))
+            conn.send(_packet(SUBACK << 4, pid + bytes(granted)))
+            with self._lock:
+                retained = list(self.retained.items())
+            for filt, q in new:
+                for topic, (payload, rq) in retained:
+                    if topic_matches(filt, topic):
+                        self._deliver(conn, topic, payload, min(rq, q), True)
+            return True
+        if ptype == UNSUBSCRIBE:
+            pid, i = body[:2], 2
+            while i < len(body):
+                filt, i = _str(body, i)
+                conn.subs.pop(filt.decode("utf-8"), None)
+            conn.send(_packet(UNSUBACK << 4, pid))
+            return True
+        if ptype == PINGREQ:
+            conn.send(_packet(PINGRESP << 4, b""))
+            return True
+        if ptype == PUBACK:           # a subscriber acknowledging our QoS 1 delivery
+            return True
+        return False                  # DISCONNECT, or anything unsupported
+
+    def _route(self, topic: str, payload: bytes, qos: int, retain: bool) -> None:
+        with self._lock:
+            if retain:
+                if payload:
+                    self.retained[topic] = (payload, qos)
+                else:
+                    self.retained.pop(topic, None)
+            conns = list(self._conns)
+        for c in conns:
+            granted = [q for f, q in list(c.subs.items()) if topic_matches(f, topic)]
+            if granted:
+                self._deliver(c, topic, payload, min(qos, max(granted)), False)
+
+    def _deliver(self, conn: _Conn, topic: str, payload: bytes, qos: int, retain: bool) -> None:
+        t = topic.encode("utf-8")
+        body = struct.pack("!H", len(t)) + t
+        if qos:
+            with conn.lock:
+                conn.next_pid = conn.next_pid % 65535 + 1
+                pid = conn.next_pid
+            body += struct.pack("!H", pid)
+        conn.send(_packet((PUBLISH << 4) | (qos << 1) | int(retain), body + payload))
+
+
+def start_broker() -> LocalBroker:
+    global _CURRENT
+    _CURRENT = LocalBroker().start()
+    return _CURRENT
+
+
+def stop_broker() -> None:
+    global _CURRENT
+    if _CURRENT is not None:
+        _CURRENT.stop()
+        _CURRENT = None
+
+
+def current_broker() -> LocalBroker:
+    assert _CURRENT is not None, "the session's local MQTT broker is not running (conftest)"
+    return _CURRENT
+
+
+def local_broker_env() -> dict:
+    """MQTT_BROKER / MQTT_PORT / MQTT_TLS for this session's local broker."""
+    return current_broker().env()

tests/conftest.py

@@ -1550,3 +1550,35 @@ def delegation_tree_provenance_guard(tmp_path_factory):
         "(these would shadow the real tree for any later first-time import):\n  "
         + "\n  ".join(polluted)
     )
+
+
+# ==============================================================================
+# Issue #26: no test reaches a third-party MQTT broker. One in-process broker
+# (tests/_local_mqtt.py) per session on 127.0.0.1; MQTT_BROKER/MQTT_PORT/
+# MQTT_TLS name it, and every other MQTT_* that would redirect or authenticate
+# a client is dropped. Subprocesses inherit it through os.environ; a harness
+# that builds its own env takes _local_mqtt.local_broker_env(). Tests that
+# deliberately name an unreachable broker (127.0.0.1:9, :65534) still win.
+# ==============================================================================
+import _local_mqtt  # noqa: E402
+
+_MQTT_DROPPED = ("MQTT_HOST", "MQTT_USERNAME", "MQTT_PASSWORD", "MQTT_CA_CERTS", "MQTT_CERTFILE", "MQTT_KEYFILE")
+
+
+@pytest.fixture(scope="session", autouse=True)
+def local_mqtt_broker():
+    broker = _local_mqtt.start_broker()
+    keys = tuple(broker.env()) + _MQTT_DROPPED
+    previous = {k: os.environ.get(k) for k in keys}
+    for k in _MQTT_DROPPED:
+        os.environ.pop(k, None)
+    os.environ.update(broker.env())
+    try:
+        yield broker
+    finally:
+        for k, v in previous.items():
+            if v is None:
+                os.environ.pop(k, None)
+            else:
+                os.environ[k] = v
+        _local_mqtt.stop_broker()

tests/test_b19_headless_reconcile_fixes.py

@@ -88,7 +88,12 @@ def test_bug4_send_keys_safe_gating_order():
     sks_idx = content.find("send_keys_safe() {")
     assert sks_idx != -1
 
-    sks_body = content[sks_idx:sks_idx + 2500]
+    # The whole function body (up to the closing brace at column 0), not a fixed
+    # number of characters: lines added earlier in the function must not push
+    # a later call out of view.
+    sks_end = content.find("\n}\n", sks_idx)
+    assert sks_end != -1
+    sks_body = content[sks_idx:sks_end]
     quiescent_idx = sks_body.find("_pane_quiescent")
     dialog_idx = sks_body.find("_pane_dialog_open")
     prompt_idx = sks_body.find("agent prompt")

tests/test_issue18_liveness_contract.py

@@ -18,11 +18,12 @@ Contract (registry fields, wrapper options):
     its own registry and broker credentials runs it, and the orchestrator's
     subscriber mirrors its events into the orchestrator's registry.
 
-Tests marked "broker" talk to the configured broker (the public default in the
-sandbox); each stays under a minute.
+Tests marked "broker" talk to the session's in-process broker (tests/_local_mqtt.py,
+Issue #26); each stays under a minute.
 """
 from __future__ import annotations
 
+import hashlib
 import json
 import os
 import re
@@ -36,6 +37,8 @@ from pathlib import Path
 
 import pytest
 
+from _local_mqtt import local_broker_env
+
 REPO_ROOT = Path(__file__).resolve().parent.parent
 SKILLS = REPO_ROOT / ".agents" / "skills"
 LIB_SH = SKILLS / "lib.sh"
@@ -59,6 +62,7 @@ def _env(cwd: Path, extra=None) -> dict:
     env["PYTHONPATH"] = str(SKILLS)
     for k in ("MQTT_BROKER", "MQTT_HOST", "MQTT_PORT", "MQTT_TLS", "MQTT_USERNAME", "MQTT_PASSWORD"):
         env.pop(k, None)
+    env.update(local_broker_env())  # Issue #26: the session's in-process broker
     if extra:
         env.update({k: str(v) for k, v in extra.items()})
     return env
@@ -1091,6 +1095,12 @@ def test_acceptance1_headless_agent_survives_on_heartbeats_alone(tmp_path, mock_
     assert _prompt_calls(Path(mock_herdr)) == 1
     beats = [e for e in _events(reg, jid) if e["event"] == "progress"]
     assert len(beats) >= 5, len(beats)
+    # Issue #26: the wrapper's own subscriber saw them on the wire (its console
+    # record, written by job_subscriber, not by the test)
+    sub = (reg / f"{jid}.subscriber.out").read_text()
+    assert f"SUBSCRIBED {_job(reg, jid)['topic_prefix']}/events" in sub, sub
+    assert len(re.findall(rf"job={jid}  seq=\d+  progress\s+heartbeat$", sub, re.M)) >= 5, sub
+    assert re.search(rf"job={jid}  seq=\d+  completed\s+PASS$", sub, re.M), sub
     brief = (reg / jid / "brief.md").read_text()
     assert "heartbeat every 2 s" in brief and "after 2 missed heartbeats" in brief and "within 40 s" in brief
 
@@ -1161,31 +1171,6 @@ def test_acceptance3_lost_heartbeats_close_the_job_and_a_late_completed_is_refus
 # 6. remote transport (broker): no shared disk, no pane, only the wire
 # ---------------------------------------------------------------------------
 
-def _skip_if_broker_did_not_deliver(reg: Path, runner_log: Path) -> None:
-    """acceptance5 runs over a public broker. The remote runner already covers
-    the idle window with `started --heartbeat` (remote.py handle_dispatch). The
-    observed failures were of one kind: the runner published started and
-    completed successfully, the orchestrator had subscribed to the right topic,
-    and it received nothing at all for the job - the broker did not deliver.
-    Only that proven non-delivery skips; a drop (HMAC, identity), a wrong
-    topic, or a runner that failed to publish still fails."""
-    log = runner_log.read_text(errors="replace")
-    m = re.search(r"published started seq=1 job=(\w+)", log)
-    if not m:
-        return
-    jid = m.group(1)
-    rec = reg / f"{jid}.json"
-    if not rec.exists():
-        return
-    topic = json.loads(rec.read_text()).get("topic_prefix") or ""
-    out = reg / f"{jid}.subscriber.out"
-    sub = out.read_text(errors="replace") if out.exists() else ""
-    if (topic and f"SUBSCRIBED {topic}/events" in sub
-            and f"job={jid}" not in sub and f"drop event for job {jid}" not in sub):
-        pytest.skip(f"public broker did not deliver the runner's events to the orchestrator "
-                    f"(runner published started for {jid}; subscriber received nothing for it)")
-
-
 def test_acceptance5_remote_runner_completes_a_job_over_the_broker_only(tmp_path):
     orch = tmp_path / "orchestrator"
     remote = tmp_path / "remote-host"
@@ -1210,8 +1195,7 @@ def test_acceptance5_remote_runner_completes_a_job_over_the_broker_only(tmp_path
         deadline = time.time() + 45
         while time.time() < deadline and "SUBSCRIBED" not in runner_log.read_text(errors="replace"):
             time.sleep(0.2)
-        if "SUBSCRIBED" not in runner_log.read_text(errors="replace"):
-            pytest.skip("broker unreachable: " + runner_log.read_text(errors="replace")[-300:])
+        assert "SUBSCRIBED" in runner_log.read_text(errors="replace"), runner_log.read_text(errors="replace")
 
         reg = orch / "jobs"
         proc, env = _submit(orch, reg, f"remote:{session}", "remote",
@@ -1219,11 +1203,6 @@ def test_acceptance5_remote_runner_completes_a_job_over_the_broker_only(tmp_path
                             env_extra={"MAM_DISPATCH_SECRET": secret}, timeout_s=120)
         stdout, stderr = proc.communicate(timeout=180)
         jid = _wait_job_id(reg)
-        if proc.returncode != 0:
-            if ("no CONNACK from broker" in stderr
-                    and "published started" not in runner_log.read_text(errors="replace")):
-                pytest.skip("broker unreachable when publishing the dispatch (no CONNACK); the runner never saw the job")
-            _skip_if_broker_did_not_deliver(reg, runner_log)
         assert proc.returncode == 0, f"rc={proc.returncode}\n{stdout}\n{stderr}\nrunner:\n{runner_log.read_text()}"
         # delivery is confirmed by the protocol: the subscriber's attributed
         # audit record of the remote `started` (review R1: the registry status
@@ -1246,6 +1225,21 @@ def test_acceptance5_remote_runner_completes_a_job_over_the_broker_only(tmp_path
         # the agent command's last stdout line became the completed detail
         assert "remote work done for " + jid in stdout, stdout
         assert runner.returncode == 0
+        # Issue #26: both ends' own logs. The runner (and the publish_event it
+        # runs) wrote runner.log; the orchestrator's subscriber wrote the audit
+        # record of what it received, attributed to this dispatch.
+        rlog = runner_log.read_text(errors="replace")
+        assert f"DISPATCH job={jid} " in rlog, rlog
+        assert f"published started seq=1 job={jid} " in rlog, rlog
+        assert re.search(rf"published completed seq=\d+ job={jid} ", rlog), rlog
+        assert f"DONE job={jid} rc=0" in rlog, rlog
+        audit = [json.loads(l) for l in (orch / "logs" / jid / "events.ndjson").read_text().splitlines() if l.strip()]
+        fp = hashlib.sha256(_job(reg, jid)["dispatch_token"].encode("utf-8")).hexdigest()
+        for ev in ("started", "completed"):
+            got = [a for a in audit if a.get("event") == "received" and a.get("source_event") == ev]
+            assert got and got[0]["dispatch_verified"] is True and got[0]["dispatch_fingerprint"] == fp, (ev, audit)
+        sub = (reg / f"{jid}.subscriber.out").read_text()
+        assert f"SUBSCRIBED {_job(reg, jid)['topic_prefix']}/events" in sub, sub
         assert "살아있는 에이전트 세션" not in stdout, "no pane injection path for a remote session"
     finally:
         if runner.poll() is None:

tests/test_issue19_subscriber_daemon_lifecycle.py

@@ -28,6 +28,8 @@ from pathlib import Path
 
 import pytest
 
+from _local_mqtt import local_broker_env
+
 FIXTURES = Path(__file__).parent / "fixtures" / "issue_5" / "phase1"
 DELEGATION_MODES = ("0", "1")
 
@@ -415,11 +417,12 @@ def test_sub_stop_trap_exits_143_before_the_first_background_child():
 # ---------------------------------------------------------------------------
 
 def _start_connected(mam_sandbox, log):
-    proc = _start_subscriber(mam_sandbox, log, {})
-    if not _wait_for(log, r"MQTT Monitor: subscribed to ", 45):
+    # Issue #26: the session's in-process broker is always up - not reaching it fails.
+    proc = _start_subscriber(mam_sandbox, log, local_broker_env())
+    if not _wait_for(log, r"MQTT Monitor: subscribed to mam/[0-9a-f]{12}/jobs/\+/events", 45):
         proc.kill()
         proc.wait(timeout=10)
-        pytest.skip("broker unreachable from this host: " + log.read_text()[-300:])
+        pytest.fail("the monitor daemon never subscribed on the local broker: " + log.read_text()[-300:])
     return proc
 
 

tests/test_issue26_envfile.py

@@ -0,0 +1,319 @@
+"""Issue #26 (A): one `.mam.env` parser for the wrapper and the Python loader.
+
+- A1: the issue's file no longer kills the real wrapper, runs no shell, warns
+  per bad line without leaking values.
+- A2: the real wrapper and the real Python loader read one table the same way.
+- A3: an environment variable beats the file, in both loaders.
+- A4: the parser's command line carries the path only.
+- A5: the files this repo ships or generates parse without a warning.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import subprocess
+import sys
+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"
+ENVFILE = SKILLS / "lib_py" / "delegation" / "envfile.py"
+
+if str(SKILLS) not in sys.path:
+    sys.path.insert(0, str(SKILLS))
+
+from lib_py.delegation import envfile  # noqa: E402
+
+ISSUE_FILE = (
+    "AGENT_SESSIONS_YAML= .mam/agent-sessions.yaml\n"
+    "X=$(touch pwned)\n"
+    "BAD KEY=1\n"
+    "noequals\n"
+    'MQTT_PASSWORD="s3cr3t-a1"\n'
+)
+CLEAN_FILE = (
+    "# comment\n"
+    "\n"
+    "AGENT_SESSIONS_YAML=.mam/agent-sessions.yaml\n"
+    'MQTT_PASSWORD="s3cr3t-a1"\n'
+)
+# key -> value both loaders must produce (A2)
+TABLE_FILE = (
+    "  SPACED  =  padded value  \n"
+    "# ignored=1\n"
+    "\n"
+    "DQ=\"double quoted\"\n"
+    "SQ='single quoted'\n"
+    "HALF=\"unmatched\n"
+    "MIXED=\"a'\n"
+    "EQ=a=b=c\n"
+    "EMPTY=\n"
+    "LIT=$(echo no) ${HOME} `echo no`\n"
+    "export EXPORTED=1\n"
+    "1BAD=1\n"
+)
+TABLE = {
+    "SPACED": "padded value",
+    "DQ": "double quoted",
+    "SQ": "single quoted",
+    "HALF": '"unmatched',
+    "MIXED": "\"a'",
+    "EQ": "a=b=c",
+    "EMPTY": "",
+    "LIT": "$(echo no) ${HOME} `echo no`",
+}
+TABLE_SKIPPED = {"EXPORTED", "export EXPORTED", "1BAD"}
+
+
+def _env(home: Path, env_file: Path, extra=None) -> dict:
+    env = {k: v for k, v in os.environ.items() if k not in TABLE and k not in ("MAM_JOB_TIMEOUT_S", "MQTT_PASSWORD",
+                                                                               "AGENT_SESSIONS_YAML", "X")}
+    env.update({"HOME": str(home), "WORKSPACE_ROOT": str(home), "MAM_ENV_FILE": str(env_file),
+                "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 _wrapper_values(env: dict, keys, cwd: Path) -> subprocess.CompletedProcess:
+    """The values the real wrapper exported after loading its env file."""
+    probe = "; ".join(f'printf "%s=%s\\0" {k} "${{{k}-<unset>}}"' for k in keys)
+    return subprocess.run(["bash", "-c", f'source "{WRAPPER}"; {probe}'], capture_output=True, text=True,
+                          env={**env, "MAM_DELEGATE_JOB_NO_MAIN": "1"}, cwd=str(cwd), timeout=60)
+
+
+def _python_values(env: dict, keys) -> dict:
+    """The values the real Python loader (mqtt_common) set from the same file."""
+    code = ("import json, os; from lib_py.delegation import mqtt_common; mqtt_common.ensure_env_loaded(); "
+            f"print(json.dumps({{k: os.environ.get(k, '<unset>') for k in {list(keys)!r}}}))")
+    res = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, env=env, timeout=60)
+    assert res.returncode == 0, res.stderr
+    return json.loads(res.stdout.strip().splitlines()[-1])
+
+
+def _split(out: str) -> dict:
+    return dict(item.split("=", 1) for item in out.split("\0") if item)
+
+
+# ---------------------------------------------------------------------------
+# A1 - the issue's file
+# ---------------------------------------------------------------------------
+
+def test_a1_the_issue_file_does_not_kill_the_wrapper_nor_run_shell(tmp_path):
+    f = tmp_path / "issue.env"
+    f.write_text(ISSUE_FILE)
+    res = subprocess.run(["bash", str(WRAPPER), "--help"], capture_output=True, text=True,
+                         env=_env(tmp_path, f), cwd=str(tmp_path), timeout=60)
+    assert res.returncode == 0, res.stderr
+    assert "submit" in res.stdout
+    assert not (tmp_path / "pwned").exists(), "a value must never be run as shell"
+    for n, reason in ((3, "invalid key"), (4, "no '='")):
+        assert f"{f}:{n}: skipped ({reason})" in res.stderr, res.stderr
+    assert ":1: skipped" not in res.stderr and ":2: skipped" not in res.stderr
+    assert "s3cr3t-a1" not in res.stdout + res.stderr
+
+    pairs, warnings = envfile.parse(str(f))
+    assert dict(pairs) == {"AGENT_SESSIONS_YAML": ".mam/agent-sessions.yaml", "X": "$(touch pwned)",
+                           "MQTT_PASSWORD": "s3cr3t-a1"}
+    assert len(warnings) == 2 and not any("s3cr3t" in w for w in warnings)
+
+
+def test_a1_a_clean_file_loads_silently(tmp_path):
+    f = tmp_path / "clean.env"
+    f.write_text(CLEAN_FILE)
+    res = subprocess.run(["bash", str(WRAPPER), "--help"], capture_output=True, text=True,
+                         env=_env(tmp_path, f), cwd=str(tmp_path), timeout=60)
+    assert res.returncode == 0, res.stderr
+    assert "skipped" not in res.stderr and "WARNING" not in res.stderr, res.stderr
+    got = _wrapper_values(_env(tmp_path, f), ["AGENT_SESSIONS_YAML", "MQTT_PASSWORD"], tmp_path)
+    assert _split(got.stdout) == {"AGENT_SESSIONS_YAML": ".mam/agent-sessions.yaml", "MQTT_PASSWORD": "s3cr3t-a1"}
+
+
+def test_a1_an_unreadable_parser_fails_loudly(tmp_path):
+    """The wrapper stops when the parser cannot run instead of going on unconfigured."""
+    f = tmp_path / "clean.env"
+    f.write_text(CLEAN_FILE)
+    res = subprocess.run(["bash", str(WRAPPER), "--help"], capture_output=True, text=True,
+                         env=_env(tmp_path, f, {"DELEGATE_JOB_PYTHON": str(tmp_path / "no-python")}),
+                         cwd=str(tmp_path), timeout=60)
+    assert res.returncode == 1
+    assert f"ERROR: cannot read {f}" in res.stderr
+
+
+# ---------------------------------------------------------------------------
+# A2 - one table, two real loaders, same values
+# ---------------------------------------------------------------------------
+
+def test_a2_the_wrapper_and_the_python_loader_read_the_same_values(tmp_path):
+    f = tmp_path / "table.env"
+    f.write_text(TABLE_FILE)
+    env = _env(tmp_path, f)
+    keys = list(TABLE) + ["EXPORTED"]
+    res = _wrapper_values(env, keys, tmp_path)
+    assert res.returncode == 0, res.stderr
+    wrapper = _split(res.stdout)
+    python = _python_values(env, keys)
+    assert wrapper == python == {**TABLE, "EXPORTED": "<unset>"}
+    for n in (11, 12):
+        assert f"{f}:{n}: skipped (invalid key)" in res.stderr, res.stderr
+    pairs, _ = envfile.parse(str(f))
+    assert not TABLE_SKIPPED & {k for k, _ in pairs}
+
+
+# ---------------------------------------------------------------------------
+# A3 - the environment beats the file, in both loaders
+# ---------------------------------------------------------------------------
+
+def _dry_budget(env: dict, cwd: Path) -> int:
+    reg = cwd / "reg"
+    res = subprocess.run(["bash", str(WRAPPER), "submit", "--agent", "human", "--dry-run",
+                          "--registry-dir", str(reg), "--prompt", "p"],
+                         capture_output=True, text=True, env=env, cwd=str(cwd), timeout=120)
+    assert res.returncode == 0, res.stdout + res.stderr
+    jid = re.search(r"registered job:\s*(\S+)", res.stdout).group(1)
+    return json.loads((reg / f"{jid}.json").read_text())["timeout_sec"]
+
+
+def test_a3_an_environment_value_beats_the_file(tmp_path):
+    f = tmp_path / "budget.env"
+    f.write_text("MAM_JOB_TIMEOUT_S=50\n")
+    (tmp_path / "file").mkdir()
+    (tmp_path / "both").mkdir()
+    assert _dry_budget(_env(tmp_path, f), tmp_path / "file") == 50
+    both = _env(tmp_path, f, {"MAM_JOB_TIMEOUT_S": "70"})
+    assert _dry_budget(both, tmp_path / "both") == 70
+    assert _python_values(both, ["MAM_JOB_TIMEOUT_S"]) == {"MAM_JOB_TIMEOUT_S": "70"}
+    assert _python_values(_env(tmp_path, f), ["MAM_JOB_TIMEOUT_S"]) == {"MAM_JOB_TIMEOUT_S": "50"}
+
+
+# ---------------------------------------------------------------------------
+# A4 - the parser sees the path only
+# ---------------------------------------------------------------------------
+
+def test_a4_the_parser_command_line_carries_the_path_only():
+    calls = [l for l in WRAPPER.read_text().splitlines() if "envfile.py" in l and not l.lstrip().startswith("#")]
+    assert len(calls) == 1, calls
+    assert re.search(r'envfile\.py" --shell "\$TARGET_ENV"\)"', calls[0]), calls[0]
+    prefix = [l for l in WRAPPER.read_text().splitlines() if "_env_exports=" in l]
+    assert prefix == ['  _env_exports="$(SCRIPT_DIR="$SCRIPT_DIR" REPO_ROOT="$REPO_ROOT" TARGET_ENV="$TARGET_ENV" \\'], prefix
+    src = WRAPPER.read_text()
+    assert not re.search(r'^\s*(source|\.)\s+"?\$TARGET_ENV', src, re.M), "the env file must never be sourced"
+    assert "set -a" not in src
+
+
+# ---------------------------------------------------------------------------
+# A5 - the files the repo ships or generates follow the format
+# ---------------------------------------------------------------------------
+
+def _install_seed(tmp_path: Path) -> Path:
+    """The .mam.env install.sh writes, from its own heredoc run by bash."""
+    src = (REPO_ROOT / "deploy" / "install.sh").read_text()
+    m = re.search(r'cat <<EOF > "\$MAM_ENV"\n(.*?\nEOF)\n', src, re.S)
+    assert m, "install.sh no longer seeds .mam.env with a heredoc"
+    script = f'MAM_ENV="{tmp_path / "seed.env"}"; MAM_PORT=1883; MAM_CLIENT_PREFIX=hermes\ncat <<EOF > "$MAM_ENV"\n{m.group(1)}\n'
+    subprocess.run(["bash", "-c", script], check=True, env={**os.environ, "HOME": str(tmp_path)}, timeout=30)
+    return tmp_path / "seed.env"
+
+
+def test_a5_shipped_and_generated_env_files_parse_without_warnings(tmp_path):
+    ws = tmp_path / "ws"
+    (ws / "deploy").mkdir(parents=True)
+    (ws / ".mam.env.example").write_text((REPO_ROOT / ".mam.env.example").read_text())
+    (ws / "deploy" / "generate-env.sh").write_text((REPO_ROOT / "deploy" / "generate-env.sh").read_text())
+    subprocess.run(["bash", str(ws / "deploy" / "generate-env.sh")], check=True, capture_output=True, timeout=30)
+    for f in (REPO_ROOT / ".mam.env.example", ws / ".mam.env", _install_seed(tmp_path)):
+        pairs, warnings = envfile.parse(str(f))
+        assert warnings == [], (f, warnings)
+        assert not any("$" in v or "`" in v for _, v in pairs), (f, pairs)
+
+    # The template's commented assignments are what a user uncomments: they must
+    # parse too, and hold no shell syntax the parser would keep literally.
+    commented = tmp_path / "uncommented.env"
+    lines = [l[2:] for l in (REPO_ROOT / ".mam.env.example").read_text().splitlines()
+             if re.match(r"# [A-Za-z_][A-Za-z0-9_]*=", l)]
+    assert len(lines) > 20
+    commented.write_text("\n".join(lines) + "\n")
+    pairs, warnings = envfile.parse(str(commented))
+    assert warnings == [] and len(pairs) == len(lines)
+    assert not [(k, v) for k, v in pairs if "$" in v or "`" in v]
+
+
+@pytest.mark.parametrize("missing", [True, False])
+def test_a5_shell_mode_exit_code_is_zero_for_absent_and_present_files(tmp_path, missing):
+    f = tmp_path / "x.env"
+    if not missing:
+        f.write_text("A=1\n")
+    res = subprocess.run([sys.executable, "-I", str(ENVFILE), "--shell", str(f)], capture_output=True, text=True,
+                         env={k: v for k, v in os.environ.items() if k != "A"}, timeout=30)
+    assert res.returncode == 0
+    assert res.stdout == ("" if missing else "export A=1\n")
+
+
+# ---------------------------------------------------------------------------
+# A6 - bash's own variables are never assigned from the file (review R1)
+# ---------------------------------------------------------------------------
+
+# RANDOM/SECONDS/LINENO/OPTIND/SRANDOM are evaluated arithmetically on
+# assignment, so even a shlex-quoted value runs its $(...) in the wrapper.
+ARITH_SPECIALS = ["RANDOM", "SECONDS", "LINENO", "OPTIND", "SRANDOM"]
+
+
+@pytest.mark.parametrize("key", ARITH_SPECIALS)
+def test_a6_a_reserved_key_runs_nothing_in_the_real_wrapper(tmp_path, key):
+    f = tmp_path / "special.env"
+    f.write_text(f"{key}=BASH_VERSINFO[$(touch marker)0]\nSAFE=BASH_VERSINFO[$(touch marker2)0]\n")
+    env = _env(tmp_path, f)
+    env.pop(key, None)
+    res = subprocess.run(["bash", str(WRAPPER), "--help"], capture_output=True, text=True, env=env,
+                         cwd=str(tmp_path), timeout=60)
+    assert res.returncode == 0, res.stderr
+    assert not (tmp_path / "marker").exists(), f"{key}= from the file ran shell"
+    assert not (tmp_path / "marker2").exists()
+    assert f"{f}:1: skipped (reserved shell variable {key})" in res.stderr, res.stderr
+    assert "touch" not in res.stderr, "the warning must not echo the value"
+    # the same value under an ordinary key stays literal (positive case)
+    got = _wrapper_values(env, ["SAFE"], tmp_path)
+    assert _split(got.stdout) == {"SAFE": "BASH_VERSINFO[$(touch marker2)0]"}
+    assert not (tmp_path / "marker").exists() and not (tmp_path / "marker2").exists()
+
+
+def test_a6_both_loaders_skip_reserved_keys(tmp_path):
+    f = tmp_path / "reserved.env"
+    keys = ["BASH_ENV", "PS4", "IFS", "SHELLOPTS", "COMP_WORDBREAKS", "READLINE_LINE", "RANDOM"]
+    f.write_text("".join(f"{k}=x\n" for k in keys) + "PLAIN=ok\n")
+    env = {k: v for k, v in _env(tmp_path, f).items() if k not in keys}
+    python = _python_values(env, keys + ["PLAIN"])
+    assert python == {**{k: "<unset>" for k in keys}, "PLAIN": "ok"}
+    pairs, warnings = envfile.parse(str(f))
+    assert pairs == [("PLAIN", "ok")]
+    assert [w.rsplit("(", 1)[1] for w in warnings] == [f"reserved shell variable {k})" for k in keys]
+    res = subprocess.run([sys.executable, "-I", str(ENVFILE), "--shell", str(f)], capture_output=True,
+                         text=True, env=env, timeout=30)
+    assert res.stdout == "export PLAIN=ok\n"
+
+
+def test_a6_the_reserved_set_covers_every_variable_a_clean_bash_defines():
+    res = subprocess.run(["env", "-i", "bash", "--norc", "--noprofile", "-c", "compgen -v"],
+                         capture_output=True, text=True, check=True, timeout=30)
+    ordinary = {"PATH", "TERM"}  # environment, not shell state; .mam.env may set them as before
+    uncovered = [v for v in res.stdout.split() if v not in ordinary and not envfile.is_reserved(v)]
+    assert not uncovered, f"bash defines these but envfile.RESERVED misses them: {uncovered}"
+
+
+def test_a6_the_file_cannot_replace_the_wrappers_own_variables(tmp_path):
+    f = tmp_path / "own.env"
+    f.write_text(f"SCRIPT_DIR={tmp_path}/evil\nREPO_ROOT={tmp_path}/evil\nTARGET_ENV={tmp_path}/evil\n")
+    (tmp_path / "evil").mkdir()
+    (tmp_path / "evil" / "lib.sh").write_text("touch pwned-lib\n")
+    res = subprocess.run(["bash", str(WRAPPER), "--help"], capture_output=True, text=True,
+                         env=_env(tmp_path, f), cwd=str(tmp_path), timeout=60)
+    assert res.returncode == 0, res.stderr
+    assert not (tmp_path / "pwned-lib").exists()
+    got = _wrapper_values(_env(tmp_path, f), ["SCRIPT_DIR", "TARGET_ENV"], tmp_path)
+    assert _split(got.stdout) == {"SCRIPT_DIR": str(WRAPPER.parent), "TARGET_ENV": str(f)}

tests/test_issue26_ints.py

@@ -0,0 +1,368 @@
+"""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"),
+    # _pane_quiescent's `for ((i = 0; i < tries; i++))` (review R2); 0 keeps "headless" (rc 2)
+    "SKS_QUIESCENT_TRIES": (0, "20"),
+    # 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),
+}
+
+# The last two are longer than Python's int() digit limit (4300): the rule must
+# bound the significant digits before converting (review R3).
+INPUTS = ["30", "0700", "0900", " 30 ", "0", "-5", "abc", "", "1e3", "9999999999", "000000000030",
+          "9" * 5000, "0" * 5000 + "30"]
+
+# Integer-looking env defaults in the shell scripts that do NOT reach bash
+# arithmetic, each with where the value goes. B5 fails on any `${MAM_/SKS_/
+# DELEGATE_JOB_/RECONCILE_…:-<digit>…}` that is neither here nor normalised
+# through INT_KNOBS, so a new knob has to be classified to land.
+NOT_ARITHMETIC = {
+    "MAM_DELEGATE_FAIL_CLOSED": "string compare", "MAM_STALL_WATCH": "string compare",
+    "MAM_ALLOW_UNSIGNED_DISPATCH": "string compare", "MAM_DELEGATE_JOB_NO_MAIN": "string compare",
+    "MAM_NO_AUTOSTART": "string compare", "MAM_SHIM_DEBUG": "string compare",
+    "MAM_NO_SUBSCRIBER_DAEMON": "string compare", "MAM_DELEGATION_PY": "string compare",
+    "MAM_AUTO_TRUST": "string compare", "MAM_KILL_ON_READY_TIMEOUT": "string compare",
+    "MAM_LOOP_PY": "string compare", "MAM_LOOP_FREEZE": "string compare", "MAM_LOOP_NO_FREEZE": "string compare",
+    "MAM_LOOP_FREEZE_OWNED": "string compare", "MAM_AUTODETECT_FORCE_FAIL": "string compare",
+    "MAM_WS_LABEL_EXPLICIT": "passed to python", "SKS_IDLE_HANDSHAKE": "string compare",
+    "SKS_DIALOG_ESCAPE": "string compare", "SKS_PROMPT_WAIT": "string compare",
+    "MAM_MIN_PANE_COLS": "argv of python -m lib_py.layout", "MAM_MIN_PANE_ROWS": "argv of python -m lib_py.layout",
+    "MAM_MAX_PANE_COLS": "argv of python -m lib_py.layout", "MAM_MAX_PANE_ROWS": "argv of python -m lib_py.layout",
+    "MAM_BUFFER_GC_MINUTES": "find -mmin", "MAM_DIALOG_TAIL_LINES": "tail -n",
+    "SKS_IDLE_TIMEOUT_MS": "argv of the python idle watcher", "MAM_DIALOG_RESEND_S": "python int()",
+    "MAM_READY_EMPTY_GIVEUP": "[ -ge ] (decimal)", "SKS_EMPTY_GIVEUP": "[ -ge ] (decimal)",
+    "SKS_IDLE_RETRIES": "[ -le ] (decimal)", "SKS_IDLE_RETRY_BACKOFF_S": "sleep",
+    "SKS_QUIESCENT_INTERVAL": "sleep", "MAM_LOCK_STEAL_RETRIES": "[ -le ] (decimal)",
+    "MAM_DIFF_MAX_BYTES": "[ -gt ] (decimal)", "MAM_DIFF_MAX_LINES": "[ -gt ] (decimal)",
+    "RECONCILE_POLL_INTERVAL": "sleep",
+    "DELEGATE_JOB_ACK_S": "create_session.sh -> delegate_wait_started, which runs mam_int on it",
+}
+
+
+def _expected(value: str, minimum: int):
+    s = value.strip()
+    if not s.isdigit():
+        return None
+    s = s.lstrip("0") or "0"
+    if len(s) > 9:
+        return None
+    n = int(s)
+    return n if n >= minimum else None
+
+
+def _id(value: str) -> str:
+    return repr(value) if len(value) <= 12 else f"{value[:3]}...x{len(value)}"
+
+
+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, ids=_id)
+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 and "Traceback" not in rres.stderr, rres.stderr[-500:]
+    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", "9" * 5000):
+            res = _dry_submit(tmp_path, None, flag, bad)
+            assert res.returncode == 1, (flag, bad[:20], res.stdout, res.stderr[-500:])
+            assert f"ERROR: {flag} {bad}: not a positive decimal integer" in res.stderr, res.stderr[-500:]
+            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", "9" * 5000):
+        res = _runner(tmp_path, None, "--job-timeout", bad)
+        assert res.returncode == 1, (bad[:20], res.stderr[-500:])
+        assert "--job-timeout requires a positive decimal integer" in res.stderr
+        assert "Traceback" not in res.stderr, res.stderr[-500:]
+    res = _runner(tmp_path, None, "--job-timeout", "0" * 5000 + "30")
+    assert res.returncode == 0 and "Traceback" not in res.stderr, res.stderr[-500:]
+    assert json.loads(res.stdout.strip().splitlines()[-1])["budget"] == 30
+
+
+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)
+
+
+def test_b5_every_integer_env_default_is_classified():
+    """Review R2: a knob can reach arithmetic through a function argument, which
+    the direct-pattern guard above cannot see. Every integer-looking env default
+    in the shell code is therefore either normalised (INT_KNOBS) or listed in
+    NOT_ARITHMETIC with where it goes."""
+    files = [WRAPPER, *sorted(SKILLS.glob("**/*.sh"))]
+    found = {}
+    for f in files:
+        for m in re.finditer(r"\$\{((?:MAM|SKS|DELEGATE_JOB|RECONCILE)_[A-Z0-9_]*):-[0-9]", f.read_text()):
+            found.setdefault(m.group(1), f.relative_to(REPO_ROOT))
+    raw_normalised = {k: v for k, v in found.items() if k in INT_KNOBS}
+    assert not raw_normalised, f"normalised knobs read raw (use mam_int_knob): {raw_normalised}"
+    unclassified = {k: str(v) for k, v in found.items() if k not in NOT_ARITHMETIC}
+    assert not unclassified, f"classify in INT_KNOBS or NOT_ARITHMETIC: {unclassified}"
+    stale = set(NOT_ARITHMETIC) - set(found)
+    assert not stale, f"no longer in the code: {stale}"
+
+
+@pytest.mark.parametrize("value,rc", [("900", 0), ("0900", 0), ("0", 2)])
+def test_b5_pane_quiescent_reads_the_tries_knob_as_decimal(tmp_path, value, rc):
+    """The consuming function with the call site's normalisation (review R2):
+    0900 used to raise 'value too great for base' and return 2 (headless)."""
+    line = next(l for l in LIB_SH.read_text().splitlines() if l.strip().startswith('_pane_quiescent "$sess"'))
+    assert '"$(mam_int_knob SKS_QUIESCENT_TRIES 20 0)"' in line
+    res = _bash('_pane_capture() { echo stable; }; '
+                '_pane_quiescent s "$(mam_int_knob SKS_QUIESCENT_TRIES 20 0)" 0; echo "rc=$?"',
+                tmp_path, {"SKS_QUIESCENT_TRIES": value})
+    assert f"rc={rc}" in res.stdout, (res.stdout, res.stderr)
+    assert "value too great" not in res.stderr and "ignored" not in res.stderr

tests/test_issue30_stall_watch.py

@@ -488,12 +488,7 @@ def test_scoped_env_drops_the_pane_inherited_workspace_for_name_lookups_only():
 import test_issue18_liveness_contract as t18  # noqa: E402
 
 
-LIVE_BROKER = ("MQTT_BROKER", "MQTT_PORT", "MQTT_TLS", "MQTT_USERNAME", "MQTT_PASSWORD")
-
-
-def _live_broker() -> dict:
-    """This environment's broker (the #18 harness defaults to a public one)."""
-    return {k: os.environ[k] for k in LIVE_BROKER if k in os.environ}
+from _local_mqtt import local_broker_env  # noqa: E402  (Issue #26: the session's in-process broker)
 
 
 def _watchers(jid: str) -> list:
@@ -507,7 +502,7 @@ def test_s9_the_wrapper_runs_the_watcher_for_the_jobs_life(tmp_path, mock_herdr,
     reg.mkdir()
     t18._headless_agent(Path(mock_herdr), "headless-codex")
     proc, env = t18._submit(tmp_path, reg, "herdr:headless-codex", "codex", ["--ack-timeout", "40"],
-                            env_extra={"MAM_STALL_WATCH": switch, **_live_broker()}, timeout_s=60)
+                            env_extra={"MAM_STALL_WATCH": switch, **local_broker_env()}, timeout_s=60)
     jid = t18._wait_job_id(reg)
     try:
         time.sleep(1)
@@ -533,7 +528,7 @@ def test_s10_remote_and_human_jobs_are_not_watched(tmp_path, agent, session, why
     reg = tmp_path / "jobs"
     reg.mkdir()
     proc, env = t18._submit(tmp_path, reg, session, agent, ["--ack-timeout", "40"], timeout_s=60,
-                            env_extra={"MAM_DISPATCH_SECRET": "s10-secret", **_live_broker()})
+                            env_extra={"MAM_DISPATCH_SECRET": "s10-secret", **local_broker_env()})
     jid = t18._wait_job_id(reg)
     try:
         time.sleep(1)
@@ -620,7 +615,7 @@ def test_r2_the_loop_relays_a_blocked_notice_while_the_job_is_open(mam_sandbox,
                                "SKS_IDLE_TIMEOUT_MS": "500", "SKS_EMPTY_GIVEUP": "1",
                                "MAM_ACK_TIMEOUT_S": "40", "MAM_JOBS_DIR": str(reg),
                                "PYTHONPATH": str(SKILLS), "PYTHONUNBUFFERED": "1"})
-    env.update(_live_broker())
+    env.update(local_broker_env())
     script = '''from lib_py.loop.runner import LoopRunner
 r = LoopRunner(["--creator", "headless-codex", "--task", "review check", "--job-timeout", "60"])
 r.parse_args()

tests/test_issue33_budget.py

@@ -25,6 +25,8 @@ from pathlib import Path
 
 import pytest
 
+from _local_mqtt import local_broker_env
+
 REPO_ROOT = Path(__file__).resolve().parent.parent
 SKILLS = REPO_ROOT / ".agents" / "skills"
 WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
@@ -48,10 +50,12 @@ def _env(home: Path, extra=None, broker: bool = False) -> dict:
     for k in ("MAM_JOBS_DIR", "MAM_REAL_ROOT", "MAM_JOB_ID", "MAM_JOB_TIMEOUT_S", "MAM_CHILD_WAIT_S",
               "MAM_WAIT_GRACE_S"):
         env.pop(k, None)
-    if not broker:
+    for k in ("MQTT_BROKER", "MQTT_HOST", "MQTT_PORT", "MQTT_TLS", "MQTT_USERNAME", "MQTT_PASSWORD"):
+        env.pop(k, None)
+    if broker:
+        env.update(local_broker_env())  # Issue #26: the session's in-process broker
+    else:
         env["MAM_ENV_FILE"] = str(home / ".mam.env.absent")
-        for k in ("MQTT_BROKER", "MQTT_HOST", "MQTT_PORT", "MQTT_TLS", "MQTT_USERNAME", "MQTT_PASSWORD"):
-            env.pop(k, None)
         env.update(UNREACHABLE)
     if extra:
         env.update({k: str(v) for k, v in extra.items()})

tests/test_issue6_fail_closed.py

@@ -16,6 +16,8 @@ from unittest.mock import MagicMock
 
 import pytest
 
+from _local_mqtt import local_broker_env
+
 REPO_ROOT = Path(__file__).resolve().parent.parent
 SKILLS = REPO_ROOT / ".agents" / "skills"
 WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
@@ -25,6 +27,7 @@ PUBLISH_PY = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "publish_even
 
 def _run_wrapper(args, env=None, cwd=None, timeout=60):
     run_env = dict(os.environ)
+    run_env.update(local_broker_env())  # Issue #26: the session's in-process broker
     run_env["DELEGATE_JOB_PYTHON"] = sys.executable
     run_env["AGENT_PYTHON_BIN"] = sys.executable
     if cwd:

tests/test_local_mqtt.py

@@ -0,0 +1,344 @@
+#!/usr/bin/env python3
+"""Issue #26 - the suite's in-process MQTT broker (tests/_local_mqtt.py).
+
+C0  conformance with the clients the repo really runs (mqtt_common.make_client,
+    publish_event.py, job_subscriber.py);
+C2  a static guard that keeps tests off third-party brokers;
+C3  the session fixture points the product's broker config at the local one.
+"""
+from __future__ import annotations
+
+import ast
+import json
+import os
+import queue
+import subprocess
+import sys
+import time
+import uuid
+from pathlib import Path
+
+import pytest
+
+import _local_mqtt
+from _local_mqtt import local_broker_env
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+TESTS = Path(__file__).resolve().parent
+SCRIPTS = REPO_ROOT / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "scripts"
+SKILLS = REPO_ROOT / ".agents" / "skills"
+if str(SKILLS) not in sys.path:
+    sys.path.insert(0, str(SKILLS))
+
+from lib_py.delegation import mqtt_common  # noqa: E402
+
+
+# ---------------------------------------------------------------------------
+# C3 - the session fixture
+# ---------------------------------------------------------------------------
+
+def test_c3_the_product_config_resolves_to_the_local_broker():
+    cfg = mqtt_common.broker_config_from_env()
+    broker = _local_mqtt.current_broker()
+    assert (cfg.host, cfg.port, cfg.tls) == ("127.0.0.1", broker.port, False)
+    assert cfg.username is None and cfg.password is None
+    assert "MQTT_HOST" not in os.environ
+    assert local_broker_env() == {"MQTT_BROKER": "127.0.0.1", "MQTT_PORT": str(broker.port), "MQTT_TLS": "0"}
+
+
+# ---------------------------------------------------------------------------
+# C0 - conformance with the repo's real clients
+# ---------------------------------------------------------------------------
+
+class _Client:
+    """A real mqtt_common.make_client client, connected and looping.
+
+    Every step that changes broker state waits for the broker's ack (CONNACK,
+    SUBACK, UNSUBACK, PUBACK), and the broker records/applies the change
+    before it acks - so a test may read broker state right after a step.
+    paho's connect() only sends CONNECT; it does not wait for the CONNACK."""
+
+    def __init__(self, role):
+        self.msgs = queue.Queue()
+        self.subacks = queue.Queue()
+        self.unsubacks = queue.Queue()
+        self.connacks = queue.Queue()
+        self.c = mqtt_common.make_client(role, mqtt_common.broker_config_from_env())
+        self.client_id = self.c._client_id.decode()
+        self.c.on_connect = lambda _c, _u, _f, rc, _p: self.connacks.put(mqtt_common.reason_code_value(rc))
+        self.c.on_message = lambda _c, _u, m: self.msgs.put((m.topic, m.payload, m.qos, m.retain))
+        self.c.on_subscribe = lambda _c, _u, mid, rcs, _p: self.subacks.put([mqtt_common.reason_code_value(r) for r in rcs])
+        self.c.on_unsubscribe = lambda *_a: self.unsubacks.put(True)
+        cfg = mqtt_common.broker_config_from_env()
+        self.c.connect(cfg.host, cfg.port, cfg.keepalive)
+        self.c.loop_start()
+        assert self.connacks.get(timeout=5) == 0, "CONNACK refused"
+
+    def sub(self, filt, qos=1):
+        self.c.subscribe(filt, qos=qos)
+        return self.subacks.get(timeout=5)
+
+    def unsub(self, filt):
+        self.c.unsubscribe(filt)
+        return self.unsubacks.get(timeout=5)
+
+    def get(self, timeout=5):
+        return self.msgs.get(timeout=timeout)
+
+    def none(self, timeout=1.0):
+        try:
+            self.msgs.get(timeout=timeout)
+        except queue.Empty:
+            return True
+        return False
+
+    def close(self):
+        self.c.loop_stop()
+        self.c.disconnect()
+
+
+def _c0_conformance(broker):
+    root = "mam/t26-" + uuid.uuid4().hex[:8]
+    a, b = _Client("c0-pub"), _Client("c0-sub")
+    try:
+        levels = broker.stats()["levels_by_client"]
+        assert (levels.get(a.client_id), levels.get(b.client_id)) == (4, 4), "paho's default MQTT 3.1.1"
+        assert b.sub(f"{root}/+/events", qos=1) == [1]
+        a.c.publish(f"{root}/j1/events", b"one", qos=1).wait_for_publish(5)
+        assert b.get() == (f"{root}/j1/events", b"one", 1, False)
+        assert b.sub(f"{root}/#", qos=0) == [0]
+        a.c.publish(f"{root}/j1/deep/x", b"two", qos=1).wait_for_publish(5)
+        assert b.get() == (f"{root}/j1/deep/x", b"two", 0, False), "min(pub qos, sub qos)"
+        assert b.unsub(f"{root}/#")
+        a.c.publish(f"{root}/j1/deep/y", b"three", qos=1).wait_for_publish(5)
+        assert b.none(), "unsubscribed"
+        # retained: replayed to a late subscriber, deleted by an empty payload
+        a.c.publish(f"{root}/j2/state", b"retained", qos=1, retain=True).wait_for_publish(5)
+        c = _Client("c0-late")
+        try:
+            assert c.sub(f"{root}/+/state") == [1]
+            assert c.get() == (f"{root}/j2/state", b"retained", 1, True)
+        finally:
+            c.close()
+        a.c.publish(f"{root}/j2/state", b"", qos=1, retain=True).wait_for_publish(5)
+        d = _Client("c0-later")
+        try:
+            d.sub(f"{root}/+/state")
+            assert d.none(), "an empty retained payload deletes the retained message"
+        finally:
+            d.close()
+        assert f"{root}/j2/state" not in broker.retained
+    finally:
+        a.close()
+        b.close()
+
+
+def test_c0_make_client_speaks_3_1_1_wildcards_qos1_and_retained():
+    _c0_conformance(_local_mqtt.current_broker())
+
+
+def test_c0_holds_when_the_broker_is_slow(monkeypatch):
+    """The C0 races, made deterministic. The broker takes its time on CONNECT
+    and UNSUBSCRIBE, and routes each PUBLISH later still (longer than a new
+    client takes to connect and subscribe). A test that reads broker state
+    before the matching ack, or sleeps a fixed time, fails here; so does a
+    broker that acks a PUBLISH before routing it."""
+    broker = _local_mqtt.current_broker()
+    handle, route = broker._handle, broker._route
+    delay = {_local_mqtt.CONNECT: 0.3, _local_mqtt.UNSUBSCRIBE: 1.5}
+
+    def slow_handle(conn, ptype, flags, body):
+        time.sleep(delay.get(ptype, 0))
+        return handle(conn, ptype, flags, body)
+
+    def slow_route(*args):
+        time.sleep(0.8)
+        return route(*args)
+
+    monkeypatch.setattr(broker, "_handle", slow_handle)
+    monkeypatch.setattr(broker, "_route", slow_route)
+    _c0_conformance(broker)
+
+
+def _env(tmp_path: Path) -> dict:
+    env = dict(os.environ)
+    env.update(local_broker_env())
+    env["PYTHONPATH"] = str(SKILLS)
+    env["DELEGATE_JOB_LOGS_DIR"] = str(tmp_path / "logs")
+    env["MAM_ENV_FILE"] = str(tmp_path / ".mam.env.absent")
+    return env
+
+
+@pytest.mark.parametrize("terminal,rc", [("completed", 0), ("error", 1)])
+def test_c0_publish_event_reaches_job_subscriber_over_the_local_broker(tmp_path, terminal, rc):
+    """The real producer -> the real consumer; the subscriber's exit code and
+    console record are the evidence. `error` is the failing twin (RC_ERROR)."""
+    env = _env(tmp_path)
+    reg = tmp_path / "jobs"
+    jid = subprocess.run([sys.executable, str(SCRIPTS / "registry.py"), "--registry-dir", str(reg), "register",
+                          "--prompt", "c0"], capture_output=True, text=True, env=env, cwd=str(tmp_path),
+                         check=True).stdout.strip()
+    out = tmp_path / "sub.out"
+    with open(out, "wb") as fh:
+        sub = subprocess.Popen([sys.executable, str(SCRIPTS / "job_subscriber.py"), "--registry-dir", str(reg),
+                                "--job", jid, "--timeout", "40", "--idle-timeout", "30"],
+                               env=env, cwd=str(tmp_path), stdout=fh, stderr=subprocess.STDOUT)
+    try:
+        deadline = time.time() + 20
+        while time.time() < deadline and "SUBSCRIBED" not in out.read_text(errors="replace"):
+            time.sleep(0.1)
+        record = json.loads((reg / f"{jid}.json").read_text())
+        assert f"SUBSCRIBED {record['topic_prefix']}/events" in out.read_text(), out.read_text()
+        for ev in ("started", terminal):
+            res = subprocess.run([sys.executable, str(SCRIPTS / "publish_event.py"), "--registry-dir", str(reg),
+                                  "--job", jid, "--event", ev, "--detail", f"c0 {ev}",
+                                  "--expect-dispatch-token", record["dispatch_token"]],
+                                 capture_output=True, text=True, env=env, cwd=str(tmp_path), timeout=60)
+            assert res.returncode == 0, res.stderr
+        assert sub.wait(timeout=30) == rc, out.read_text()
+    finally:
+        if sub.poll() is None:
+            sub.kill()
+            sub.wait(timeout=10)
+    text = out.read_text()
+    assert f"job={jid}  seq=1  started" in text, text
+    assert f"job={jid}  seq=2  {terminal}" in text and f"c0 {terminal}" in text, text
+
+
+# ---------------------------------------------------------------------------
+# C2 - static guard over tests/*.py
+# ---------------------------------------------------------------------------
+
+# Tests that pin the product's public default broker by name. None today;
+# add one here (with its reason) rather than weakening the scan.
+PUBLIC_BROKER_ALLOWLIST: dict = {}
+PUBLIC_BROKER = "broker." + "hivemq.com"
+REDIRECT_KEYS = {"MQTT_BROKER", "MQTT_HOST"}
+
+
+def _const(node):
+    return node.value if isinstance(node, ast.Constant) and isinstance(node.value, str) else None
+
+
+def _dict_has_key(node) -> bool:
+    return isinstance(node, ast.Dict) and any(_const(k) in REDIRECT_KEYS for k in node.keys if k is not None)
+
+
+def _module_env_dicts(tree) -> set:
+    """Module-level names bound to a dict literal that names a broker (UNREACHABLE)."""
+    names = set()
+    for node in tree.body:
+        if isinstance(node, ast.Assign) and _dict_has_key(node.value):
+            names.update(t.id for t in node.targets if isinstance(t, ast.Name))
+    return names
+
+
+def _names_broker(node, env_dicts) -> bool:
+    if _dict_has_key(node):
+        return True
+    if isinstance(node, ast.Name) and node.id in env_dicts:
+        return True
+    if isinstance(node, ast.Call):
+        fn = node.func.id if isinstance(node.func, ast.Name) else getattr(node.func, "attr", None)
+        if fn == "local_broker_env":
+            return True
+        if fn == "dict" and any(k.arg in REDIRECT_KEYS for k in node.keywords):
+            return True
+    return False
+
+
+def _deletes_and_resets(func, env_dicts):
+    loop_keys = {}   # loop variable -> iterates over a literal that names a redirect key
+    for node in ast.walk(func):
+        if isinstance(node, ast.For) and isinstance(node.target, ast.Name) and isinstance(node.iter, (ast.Tuple, ast.List)):
+            if any(_const(e) in REDIRECT_KEYS for e in node.iter.elts):
+                loop_keys[node.target.id] = True
+    deletes = resets = False
+    for node in ast.walk(func):
+        if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
+            attr, args = node.func.attr, node.args
+            first = args[0] if args else None
+            names_key = _const(first) in REDIRECT_KEYS or (isinstance(first, ast.Name) and first.id in loop_keys)
+            if attr in ("pop", "delenv") and names_key:
+                deletes = True
+            if attr == "setenv" and _const(first) in REDIRECT_KEYS:
+                resets = True
+            if attr == "update" and (any(_names_broker(a, env_dicts) for a in args)
+                                     or any(k.arg in REDIRECT_KEYS for k in node.keywords)):
+                resets = True
+        elif isinstance(node, ast.Delete):
+            if any(isinstance(t, ast.Subscript) and _const(t.slice) in REDIRECT_KEYS for t in node.targets):
+                deletes = True
+        elif isinstance(node, (ast.Assign, ast.AugAssign)):
+            targets = node.targets if isinstance(node, ast.Assign) else [node.target]
+            if any(isinstance(t, ast.Subscript) and _const(t.slice) in REDIRECT_KEYS for t in targets):
+                resets = True
+        elif isinstance(node, ast.Dict) and any(k is None and _names_broker(v, env_dicts)
+                                                for k, v in zip(node.keys, node.values)):
+            resets = True   # {**os.environ, **UNREACHABLE}
+    return deletes, resets
+
+
+def _scan(path: Path):
+    tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+    public, stripping = [], []
+    for node in ast.walk(tree):
+        if _const(node) is not None and PUBLIC_BROKER in node.value:
+            public.append(node.lineno)
+    env_dicts = _module_env_dicts(tree)
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+            deletes, resets = _deletes_and_resets(node, env_dicts)
+            if deletes and not resets:
+                stripping.append(f"{node.name}:{node.lineno}")
+    return public, stripping
+
+
+def test_c2_no_test_names_the_public_broker_or_strips_the_local_one():
+    """(a) the public default broker's name appears only in allowlisted
+    tests; (b) no helper deletes MQTT_BROKER/MQTT_HOST without naming a
+    broker again in the same function - without MQTT_BROKER/MQTT_HOST the
+    product falls back to the public default."""
+    public, stripping = {}, {}
+    for path in sorted(TESTS.glob("*.py")):
+        p, s = _scan(path)
+        if p and path.name not in PUBLIC_BROKER_ALLOWLIST:
+            public[path.name] = p
+        if s:
+            stripping[path.name] = s
+    assert not public, f"tests naming the public broker: {public}"
+    assert not stripping, f"helpers that drop the broker and fall back to the public default: {stripping}"
+
+
+def test_c2_the_guard_catches_what_it_is_for(tmp_path):
+    """The failing twins: each forbidden shape is reported, and the allowed
+    shapes (pop then UNREACHABLE / local_broker_env / setenv) are not."""
+    bad = tmp_path / "bad.py"
+    bad.write_text(
+        "import os\n"
+        f"URL = '{PUBLIC_BROKER}'\n"
+        "def strip_loop(env):\n"
+        "    for k in ('MQTT_BROKER', 'MQTT_PORT'):\n"
+        "        env.pop(k, None)\n"
+        "    return env\n"
+        "def strip_direct(mp):\n"
+        "    mp.delenv('MQTT_HOST', raising=False)\n"
+        "def strip_del(env):\n"
+        "    del env['MQTT_BROKER']\n")
+    good = tmp_path / "good.py"
+    good.write_text(
+        "UNREACHABLE = {'MQTT_BROKER': '127.0.0.1', 'MQTT_PORT': '9'}\n"
+        "def a(env):\n"
+        "    for k in ('MQTT_BROKER', 'MQTT_HOST'):\n"
+        "        env.pop(k, None)\n"
+        "    env.update(UNREACHABLE)\n"
+        "def b(env):\n"
+        "    env.pop('MQTT_BROKER', None)\n"
+        "    env.update(local_broker_env())\n"
+        "def c(mp):\n"
+        "    mp.delenv('MQTT_HOST')\n"
+        "    mp.setenv('MQTT_BROKER', 'x')\n")
+    public, stripping = _scan(bad)
+    assert public == [2]
+    assert stripping == ["strip_loop:3", "strip_direct:7", "strip_del:9"]
+    assert _scan(good) == ([], [])