test(issue-26): no test depends on a third-party MQTT broker

- tests/_local_mqtt.py: in-process MQTT 3.1.1 broker on 127.0.0.1 (what
  the repo uses: protocol level 4, clean session, QoS 0/1, retained).
- conftest: one session broker named by MQTT_BROKER/MQTT_PORT/MQTT_TLS;
  other redirecting/auth MQTT_* dropped and restored afterwards.
- Harnesses of the 22 broker-dependent tests (#18 acceptance1/2/3/5,
  #6 t1_1/2/7/8/14/15/16/20, #19 MQTT loop x2, #33 b3/r3/b7, #30
  s9/s10/r2) use it; the "broker unreachable" skips are gone (a local
  broker that does not deliver is a failure). acceptance1/5 and #19 also
  assert on the subscriber, runner and audit logs the real producers write.
- tests/test_local_mqtt.py: conformance with the real clients and the real
  publish_event -> job_subscriber path (C0), an AST guard against public
  broker literals and MQTT_* deleted without a replacement (C2), and the
  session's broker config (C3).

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

tests/_local_mqtt.py

@@ -0,0 +1,272 @@
+"""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),
+                    "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
+            if qos == 1:
+                pid = body[i:i + 2]
+                i += 2
+                conn.send(_packet(PUBACK << 4, pid))
+            self._route(topic.decode("utf-8"), body[i:], qos, retain)
+            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_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_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:
-        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)
+    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")
         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,304 @@
+#!/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."""
+
+    def __init__(self, role):
+        self.msgs = queue.Queue()
+        self.subacks = queue.Queue()
+        self.c = mqtt_common.make_client(role, mqtt_common.broker_config_from_env())
+        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])
+        cfg = mqtt_common.broker_config_from_env()
+        self.c.connect(cfg.host, cfg.port, cfg.keepalive)
+        self.c.loop_start()
+
+    def sub(self, filt, qos=1):
+        self.c.subscribe(filt, qos=qos)
+        return self.subacks.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 test_c0_make_client_speaks_3_1_1_wildcards_qos1_and_retained():
+    broker = _local_mqtt.current_broker()
+    before = broker.stats()["connections"]
+    root = "mam/t26-" + uuid.uuid4().hex[:8]
+    a, b = _Client("c0-pub"), _Client("c0-sub")
+    try:
+        assert broker.stats()["protocol_levels"][before:] == [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)"
+        b.c.unsubscribe(f"{root}/#")
+        time.sleep(0.3)
+        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 _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) == ([], [])