fix(issue-5): address phase 1 cross-review findings (R1, R2, F-1)

Three reviewers examined fc0bd96..8def235. Codex and Claude each reproduced a
defect; all three findings were re-verified here before being accepted, and none
were rebutted.

R1 (codex) — the shell helpers ran `python -m lib_py.delegation.cli`. For -m,
sys.path[0] is the working directory, which therefore beats PYTHONPATH: with cwd
inside a tree containing lib_py, the live package won even with the frozen path
listed first. The documented B-13 guarantee for this path did not hold.
Reproduced, then replaced with an explicit bootstrap that inserts SKILL_DIR
itself. cwd is left alone, because the registry and audit-log paths are relative
to it. The file-based shim path was never affected.

R2 (codex) — the persistent connection was reused on host and port alone. Two
jobs sharing an endpoint but carrying different credentials or TLS settings
silently published over the first connection. Reproduced (the client factory ran
once, under the first account, with TLS off). The guard now compares an
immutable snapshot of the whole connection identity and names only the differing
field, never its value.

F-1 (claude) — paho queues a publish issued while disconnected and raises
RuntimeError from wait_for_publish. RuntimeError is deliberately outside the
retry tuple, so the one state a persistent connection exists to survive, the gap
between a drop and paho's reconnect, failed on the first attempt. Confirmed
against paho 2.1.0. That single condition is now translated into a retryable
error; the tuple is unchanged, so genuine programming errors are still not
retried.

Two fixtures were hiding these: FakeInfo had no rc and never raised. It now
mirrors paho's contract, which is what makes the new tests meaningful.

Also folds in the duplication both Claude (O-1) and I had flagged: the CLI path
kept its own copy of connect-publish-disconnect plus its own timeout constants.
_publish_once now delegates to OneShotPublisher.publish_raw, so a transport fix
lands once. Having two copies is exactly why these bugs existed in one of them.

Documentation corrected rather than defended: the benchmark's rationale for
keeping three forks was imprecise (the || true already guarantees rc 0; the
lookup buys quietness and legacy parity), SKILL.md no longer claims PYTHONPATH
decides the frozen path, and the daemon's directory creation is recorded as a
deliberate difference. Trailing whitespace inherited by the relocated file is
stripped, so the commit range is now clean too.

Verified: each regression fails against the pre-fix code and passes after
(R1 1, R2 4, F-1 1). 66 phase-1 tests pass.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd

.agents/skills/lib.sh

@@ -1949,6 +1949,23 @@ _start_watchdog_legacy() {
 # loop run frozen delegation code.
 # ---------------------------------------------------------------------------
 
+# _delegation_py <cli-args...>
+# Run the delegation CLI from *this* tree.
+#
+# NOT `python -m`: for -m, sys.path[0] is the current directory, which therefore
+# beats PYTHONPATH. A frozen loop whose cwd happened to contain a lib_py package
+# would silently execute live code, losing B-13. An explicit insert of SKILL_DIR
+# cannot be shadowed that way. cwd is deliberately left alone, because the
+# registry and audit-log paths are relative to it.
+_delegation_py() {
+  PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -c '
+import sys
+sys.path.insert(0, sys.argv[1])
+from lib_py.delegation.cli import main
+sys.exit(main(sys.argv[2:]))
+' "$SKILL_DIR" "$@"
+}
+
 # delegate_submit_job <prompt> <agent> <agent_session>
 # Register a job. Prints the new JID on stdout and nothing else.
 delegate_submit_job() {
@@ -1960,8 +1977,7 @@ delegate_submit_job() {
     echo "ERROR: multi-agent-mux-delegate-job registry.py not found under .agents/skills/" >&2
     return 1
   fi
-  PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -m lib_py.delegation.cli \
+  _delegation_py submit --prompt "$prompt" --agent "$agent" --agent-session "$session"
-    submit --prompt "$prompt" --agent "$agent" --agent-session "$session"
 }
 
 # delegate_publish_event <job_id> <event> [detail]
@@ -1974,8 +1990,7 @@ delegate_publish_event() {
   local pub
   pub="$(_delegate_script publish_event.py)"
   [ -n "$pub" ] && [ -f "$pub" ] || return 0
-  PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -m lib_py.delegation.cli \
+  _delegation_py publish --job "$job_id" --event "$event" --detail "$detail" || true
-    publish --job "$job_id" --event "$event" --detail "$detail" || true
 }
 
 # ensure_subscriber_daemon <job_id> [workdir]
@@ -1983,8 +1998,7 @@ delegate_publish_event() {
 # Prints the PID(s) on stdout, exactly as start_watchdog always did.
 ensure_subscriber_daemon() {
   if [ "${MAM_DELEGATION_PY:-1}" = "0" ]; then _start_watchdog_legacy "$@"; return $?; fi
-  PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -m lib_py.delegation.cli \
+  _delegation_py ensure-daemon --job "$1" --workdir "${2:-$PWD}"
-    ensure-daemon --job "$1" --workdir "${2:-$PWD}"
 }
 
 # start_watchdog — retained name; create_session.sh and operator docs use it.

.agents/skills/lib_py/delegation/bench.py

@@ -27,10 +27,22 @@ from lib_py.delegation.publisher import OneShotPublisher, PersistentPublisher
 
 
 class FakeInfo(object):
-    def __init__(self, published=True):
+    """Stand-in for paho's MQTTMessageInfo, including its failure contract.
+
+    paho carries an ``rc`` and raises RuntimeError from ``wait_for_publish``
+    when the publish could not be handed to a live connection. The earlier
+    version of this fake had neither, which is precisely why a real defect in
+    that path stayed invisible while every test passed.
+    """
+
+    def __init__(self, published=True, rc=0):
         self._published = published
+        self.rc = rc
 
     def wait_for_publish(self, timeout=None):
+        if self.rc:
+            raise RuntimeError(
+                "Message publish failed: The client is not currently connected.")
         return None
 
     def is_published(self):

.agents/skills/lib_py/delegation/daemon.py

@@ -50,6 +50,12 @@ def _ignore_sighup():  # pragma: no cover - runs in the forked child
 
 
 def spawn(monitor_script: str, workdir: str) -> int:
+    """Start the subscriber detached and return its pid.
+
+    One deliberate difference from the shell original: the log directory is
+    created if missing. The legacy `>>` redirect failed outright when `.mam/`
+    did not exist, so this is strictly more robust, not a contract change.
+    """
     log_file = os.path.join(workdir, LOG_RELPATH)
     os.makedirs(os.path.dirname(log_file), exist_ok=True)
     log_handle = open(log_file, "ab")

.agents/skills/lib_py/delegation/publish_event.py

@@ -27,7 +27,6 @@ import hmac
 import json
 import logging
 import sys
-import time
 from typing import Any, Dict, Optional
 
 from lib_py.delegation import mqtt_common
@@ -38,7 +37,6 @@ from lib_py.delegation.mqtt_common import (
     broker_config_from_job,
     events_topic_for,
     load_job,
-    make_client,
     next_seq,
     with_retry,
 )
@@ -55,8 +53,13 @@ EVENT_TO_STATUS = {
     "cancelled": "cancelled",
 }
 
-CONNECT_ACK_TIMEOUT = 10  # seconds to wait for CONNACK
+# Re-exported from the transport module, which owns the single implementation.
-PUBLISH_ACK_TIMEOUT = 5   # seconds to wait for QoS-1 PUBACK
+# Two copies of these numbers meant tuning one silently diverged from the other.
+from lib_py.delegation.publisher import (  # noqa: F401  (re-exported for callers)
+    CONNECT_ACK_TIMEOUT,
+    PUBLISH_ACK_TIMEOUT,
+    OneShotPublisher,
+)
 
 
 def build_payload(
@@ -91,37 +94,12 @@ def build_payload(
 def _publish_once(config, topic: str, body: bytes, retain: bool) -> None:
     """Connect, publish one QoS-1 message, wait for the broker ACK, disconnect.
 
-    Raises on any failure so ``with_retry`` can re-run the whole sequence (a
+    Delegates to the transport module so the CLI path and the service path share
-    fresh connection per attempt is the robust choice for a PoC)."""
+    one implementation. Keeping a second copy here is what let the persistent
-    client = make_client("publisher", config)
+    transport's handshake and not-connected bugs exist in only one of them.
-    connected = {"rc": None}
+    Raises on any failure so ``with_retry`` can re-run the whole sequence.
-
+    """
-    def on_connect(_c, _u, _flags, reason_code, _props):
+    OneShotPublisher().publish_raw(config, topic, body, retain)
-        connected["rc"] = reason_code
-
-    client.on_connect = on_connect
-    client.connect(config.host, config.port, config.keepalive)
-    client.loop_start()
-    try:
-        # Wait for CONNACK so we fail fast on auth/TLS errors.
-        deadline = time.monotonic() + CONNECT_ACK_TIMEOUT
-        while connected["rc"] is None and time.monotonic() < deadline:
-            time.sleep(0.05)
-        if connected["rc"] is None:
-            raise TimeoutError("no CONNACK from broker")
-        if mqtt_common.reason_code_value(connected["rc"]) != 0:
-            raise ConnectionError(f"broker refused connection: rc={connected['rc']}")
-
-        info = client.publish(topic, payload=body, qos=1, retain=retain)
-        info.wait_for_publish(timeout=PUBLISH_ACK_TIMEOUT)
-        if not info.is_published():
-            raise TimeoutError("publish not acknowledged within timeout")
-    finally:
-        client.loop_stop()
-        try:
-            client.disconnect()
-        except Exception:  # pragma: no cover - disconnect best effort
-            pass
 
 
 class PreparedEvent(object):

.agents/skills/lib_py/delegation/publisher.py

@@ -12,7 +12,9 @@ from __future__ import annotations
 
 import logging
 import time
-from typing import Optional, Tuple
+from typing import Any, Optional, Tuple
+
+import paho.mqtt.client as mqtt
 
 from lib_py.delegation import mqtt_common
 
@@ -22,6 +24,25 @@ CONNECT_ACK_TIMEOUT = 10  # seconds to wait for CONNACK
 PUBLISH_ACK_TIMEOUT = 5   # seconds to wait for QoS-1 PUBACK
 
 
+def _send(client, topic: str, body: bytes, retain: bool) -> None:
+    """Publish one QoS-1 message and wait for the broker ACK.
+
+    paho queues a publish issued while disconnected and returns
+    ``MQTT_ERR_NO_CONN``; ``wait_for_publish`` then raises ``RuntimeError``,
+    which is deliberately NOT in the retry tuple (a RuntimeError normally means a
+    programming error and must not be retried blindly). Left alone, that turned
+    the one state a persistent connection exists to survive -- the gap between a
+    drop and paho's reconnect -- into a single-attempt failure. Detect that one
+    condition explicitly and raise a retryable error instead.
+    """
+    info = client.publish(topic, payload=body, qos=1, retain=retain)
+    if getattr(info, "rc", mqtt.MQTT_ERR_SUCCESS) == mqtt.MQTT_ERR_NO_CONN:
+        raise ConnectionError("client is not connected (paho is reconnecting)")
+    info.wait_for_publish(timeout=PUBLISH_ACK_TIMEOUT)
+    if not info.is_published():
+        raise TimeoutError("publish not acknowledged within timeout")
+
+
 class _BasePublisher(object):
     """Shared retry wrapper so both transports report outcomes identically."""
 
@@ -56,7 +77,15 @@ class OneShotPublisher(_BasePublisher):
         self.connect_count = 0
 
     def publish(self, prepared):
-        client = self.client_factory("publisher", prepared.config)
+        self.publish_raw(prepared.config, prepared.topic, prepared.body, prepared.retain)
+
+    def publish_raw(self, config, topic: str, body: bytes, retain: bool) -> None:
+        """Connect, publish, wait for the ACK, disconnect.
+
+        The single implementation of one-shot publishing: ``publish_event``'s
+        CLI path calls straight into this, so the two entry points cannot drift.
+        """
+        client = self.client_factory("publisher", config)
         self.connect_count += 1
         connected = {"rc": None}
 
@@ -64,7 +93,7 @@ class OneShotPublisher(_BasePublisher):
             connected["rc"] = reason_code
 
         client.on_connect = on_connect
-        client.connect(prepared.config.host, prepared.config.port, prepared.config.keepalive)
+        client.connect(config.host, config.port, config.keepalive)
         client.loop_start()
         try:
             deadline = time.monotonic() + CONNECT_ACK_TIMEOUT
@@ -74,11 +103,7 @@ class OneShotPublisher(_BasePublisher):
                 raise TimeoutError("no CONNACK from broker")
             if mqtt_common.reason_code_value(connected["rc"]) != 0:
                 raise ConnectionError("broker refused connection: rc={}".format(connected["rc"]))
-            info = client.publish(prepared.topic, payload=prepared.body,
+            _send(client, topic, body, retain)
-                                  qos=1, retain=prepared.retain)
-            info.wait_for_publish(timeout=PUBLISH_ACK_TIMEOUT)
-            if not info.is_published():
-                raise TimeoutError("publish not acknowledged within timeout")
         finally:
             client.loop_stop()
             try:
@@ -100,6 +125,7 @@ class PersistentPublisher(_BasePublisher):
         self.connect_count = 0
         self._client = None
         self._config = None
+        self._fingerprint_bound = None
 
     def __enter__(self):
         return self.open()
@@ -111,12 +137,35 @@ class PersistentPublisher(_BasePublisher):
     def open(self):
         return self
 
+    @staticmethod
+    def _fingerprint(config) -> tuple:
+        """An immutable snapshot of everything that decides *which connection*.
+
+        Endpoint alone is not enough: two jobs can share a host and port while
+        carrying different credentials or TLS material, and reusing the first
+        connection would publish under the wrong principal or silently drop a
+        requested TLS setting. Snapshotting also means later mutation of the
+        original config object cannot change what the guard believes is bound.
+        """
+        fields = ("host", "port", "username", "password", "tls", "ca_certs",
+                  "certfile", "keyfile", "keepalive", "client_id_prefix")
+        return tuple((name, getattr(config, name, None)) for name in fields)
+
+    @staticmethod
+    def _describe_mismatch(bound: tuple, incoming: tuple) -> str:
+        """Name the differing fields without ever echoing their values."""
+        differing = [name for (name, old), (_n, new) in zip(bound, incoming) if old != new]
+        return ", ".join(differing) or "connection settings"
+
     def _ensure_client(self, config):
         if self._client is not None:
-            if (config.host, config.port) != (self._config.host, self._config.port):
+            incoming = self._fingerprint(config)
+            if incoming != self._fingerprint_bound:
                 raise ValueError(
-                    "PersistentPublisher is bound to {}:{}; refusing an event for {}:{}".format(
+                    "PersistentPublisher is bound to {}:{}; refusing an event whose "
-                        self._config.host, self._config.port, config.host, config.port)
+                    "{} differ".format(
+                        self._config.host, self._config.port,
+                        self._describe_mismatch(self._fingerprint_bound, incoming))
                 )
             return self._client
 
@@ -151,6 +200,7 @@ class PersistentPublisher(_BasePublisher):
             raise
         self._client = client
         self._config = config
+        self._fingerprint_bound = self._fingerprint(config)
         return client
 
     @staticmethod
@@ -163,12 +213,10 @@ class PersistentPublisher(_BasePublisher):
 
     def publish(self, prepared):
         client = self._ensure_client(prepared.config)
-        info = client.publish(prepared.topic, payload=prepared.body,
+        # Both failure shapes (queued-while-disconnected and no PUBACK) are raised
-                              qos=1, retain=prepared.retain)
+        # as retryable errors by _send, so the caller's with_retry can ride out a
-        info.wait_for_publish(timeout=PUBLISH_ACK_TIMEOUT)
+        # reconnect instead of recording a one-attempt failure.
-        if not info.is_published():
+        _send(client, prepared.topic, prepared.body, prepared.retain)
-            # Surfaced to the caller's with_retry; paho reconnects underneath.
-            raise TimeoutError("publish not acknowledged within timeout")
 
     def close(self):
         if self._client is None:
@@ -178,3 +226,4 @@ class PersistentPublisher(_BasePublisher):
         finally:
             self._client = None
             self._config = None
+            self._fingerprint_bound = None

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

@@ -84,14 +84,25 @@ Which tree wins, by execution context:
 
 | Context | Resolved by | Tree that wins |
 |---|---|---|
-| Loop process and its children (frozen `lib.sh` sourced) | `PYTHONPATH` | the frozen snapshot |
+| Loop process and its children (frozen `lib.sh` sourced) | the helper's explicit bootstrap from `SKILL_DIR` | the frozen snapshot |
 | Agent session running a **frozen** script path while `PYTHONPATH` still points live | the shim, from its own location | the frozen snapshot (B-13 preserved) |
 | Live helpers (create/stop/resume/reconcile) | `PYTHONPATH` or the shim (same tree) | the live tree |
 | Standalone, no `PYTHONPATH` | the shim's fallback append | the shim's own tree |
 
-`lib.sh` calls the package through `python -m lib_py.delegation.cli`. Setting
+`lib.sh` calls the package through `_delegation_py`, which inserts its own
-`MAM_DELEGATION_PY=0` switches the shell helpers back to their preserved legacy
+`SKILL_DIR` at the front of the import path. It deliberately does **not** use
-bodies with no redeploy.
+`python -m`: for `-m` the current directory precedes `PYTHONPATH`, so a frozen
+loop whose working directory happened to contain a `lib_py` package would have
+silently executed live code. The working directory is left untouched, because
+the registry and audit-log paths are relative to it.
+
+Setting `MAM_DELEGATION_PY=0` switches the shell helpers back to their preserved
+legacy bodies with no redeploy.
+
+One in-process caveat: the package binds once per process and is never rebound,
+so do not import a frozen shim after a live `lib_py.delegation` is already bound
+inside the same interpreter. Production never does this — each delegated job is
+a fresh interpreter — but a test harness could.
 
 ---
 

docs/contracts/issue-5/phase1-benchmark.md

@@ -68,7 +68,9 @@ Rev.1 계획 job의 `started` 이벤트는 payload timestamp 05:58:49Z에서 모
 | 레거시 (`_delegate_publish_event_legacy`) | 3 (`$(_delegate_py_bin)`, `$(_delegate_script)`, python) |
 | 신규 (`-m lib_py.delegation.cli`) | 3 (`$(_delegate_py_bin)`, `$(_delegate_script)`, python) |
 
-> 계획서 D8은 신규 경로가 2 fork일 것으로 예상했으나, **실제 구현은 3 fork로 동일하다.** `_delegate_script publish_event.py` 호출을 유지했기 때문이며, 이는 의도적이다: 그 호출이 "스크립트 트리가 없으면 아무 것도 하지 않는다"는 비치명 계약(C-02)을 만들어내는 유일한 지점이다. 이를 제거하면 fork 1회를 아끼는 대신 계약이 바뀐다. fork 절감은 Phase 4에서 계약과 함께 재검토한다.
+> 계획서 D8은 신규 경로가 2 fork일 것으로 예상했으나, **실제 구현은 3 fork로 동일하다.** `_delegate_script publish_event.py` 조회를 유지했기 때문이다.
+>
+> **이 조회가 실제로 보장하는 것 (교차 리뷰 `cc3468cb` 정정)**: 초기 서술은 이 조회가 "비치명 계약(C-02)을 만드는 유일한 지점"이라고 했으나 부정확하다. rc 0 자체는 `delegate_publish_event`의 `|| true`만으로도 보장되고, C-02도 rc만 단정한다. 조회가 보장하는 것은 **위임 스크립트 트리가 없을 때 인터프리터를 아예 띄우지 않아 stderr에 트레이스백을 남기지 않는다는 정숙성과, 레거시 본체와의 동작 동일성**이다(인터프리터 미호출 단정은 C-01에 있다). 유지 결정 자체는 세 리뷰어 전원이 지지했으며, fork 절감은 Phase 4에서 이 계약과 함께 재검토한다.
 
 ---
 

tests/test_issue5_phase1_delegation.py

@@ -1000,3 +1000,209 @@ def test_persistent_publisher_reuses_a_healthy_connection_across_events():
         publisher.close()
     assert calls["n"] == 1, "a healthy connection must be reused"
     assert calls["publish"] == 4
+
+
+# ==============================================================================
+# 8. Cross-review regressions (job 60c980ac)
+#
+# R1 (codex): `python -m` puts the working directory ahead of PYTHONPATH, so a
+#             frozen loop whose cwd contained a lib_py package silently ran live
+#             code. The shell helpers now use an explicit bootstrap instead.
+# R2 (codex): the persistent connection was reused on host/port alone, ignoring
+#             credentials and TLS.
+# F-1 (claude): paho queues a publish issued while disconnected and then raises
+#             RuntimeError, which sat outside the retry tuple -- so the one state
+#             a persistent connection exists to survive failed on first attempt.
+# ==============================================================================
+
+def _decoy_tree(root: Path) -> Path:
+    """A directory that looks like a skills tree and screams if it is imported."""
+    pkg = root / "lib_py" / "delegation"
+    pkg.mkdir(parents=True)
+    (root / "lib_py" / "__init__.py").write_text("", encoding="utf-8")
+    (pkg / "__init__.py").write_text("", encoding="utf-8")
+    (pkg / "cli.py").write_text(
+        "def main(argv=None):\n"
+        "    print('DECOY-TREE-EXECUTED')\n"
+        "    return 0\n",
+        encoding="utf-8",
+    )
+    return root
+
+
+def test_r1_shell_helper_ignores_a_lib_py_in_the_working_directory(mam_sandbox, tmp_path):
+    """R1: the helper must run its own tree even when cwd provides lib_py.
+
+    This is the frozen-loop shape: a decoy package sitting in the working
+    directory must not win over the tree the sourced lib.sh belongs to.
+    """
+    decoy = _decoy_tree(tmp_path / "decoy_ws")
+    (decoy / ".mam" / "jobs").mkdir(parents=True)
+
+    lib_sh = _sandbox_skills(mam_sandbox) / "lib.sh"
+    env = dict(os.environ)
+    env.pop("HERDR_SESSION_NAME", None)
+    env.pop("PYTHONPATH", None)
+    env["MAM_ENV_FILE"] = str(Path(mam_sandbox) / ".mam.env.absent")
+
+    res = subprocess.run(
+        ["bash", "-c",
+         'cd "{}" && source "{}" >/dev/null 2>&1 && '
+         'delegate_submit_job "r1 probe" claude-code herdr:claude'.format(decoy, lib_sh)],
+        capture_output=True, text=True, env=env, timeout=120,
+    )
+    assert "DECOY-TREE-EXECUTED" not in res.stdout, (
+        "the working directory's lib_py shadowed the helper's own tree:\n" + res.stdout
+    )
+    assert res.returncode == 0, res.stderr
+    job_id = res.stdout.strip()
+    assert re.fullmatch(r"[0-9a-f]+", job_id), repr(res.stdout)
+    assert (decoy / ".mam" / "jobs" / (job_id + ".json")).exists(), (
+        "the real cli should have registered the job in the cwd registry"
+    )
+
+
+@pytest.mark.parametrize("field,first,second", [
+    ("username", "account-a", "account-b"),
+    ("password", "secret-a", "secret-b"),
+    ("tls", False, True),
+    ("ca_certs", None, "/tmp/ca.pem"),
+])
+def test_r2_persistent_publisher_refuses_a_different_connection_identity(field, first, second):
+    """R2: same endpoint is not the same connection when auth or TLS differ."""
+    from lib_py.delegation import mqtt_common, publish_event
+    from lib_py.delegation.publisher import PersistentPublisher
+
+    built = []
+
+    def factory(_role, config):
+        built.append(getattr(config, field))
+        from lib_py.delegation.bench import FakeClient
+        client = FakeClient(connect_latency=0.0, counters={})
+        return client
+
+    def prepared(value):
+        config = mqtt_common.broker_config_from_env(
+            {"host": "same.example", "port": 1883, field: value})
+        return publish_event.PreparedEvent(
+            "j", "progress", "d", 1, "t", False, {}, b"{}", config)
+
+    publisher = PersistentPublisher(client_factory=factory)
+    try:
+        ok, _err = publisher.publish_with_retry(prepared(first), attempts=1)
+        assert ok is True
+        ok2, err2 = publisher.publish_with_retry(prepared(second), attempts=1)
+    finally:
+        publisher.close()
+
+    assert ok2 is False, "an event with a different {} must not reuse the connection".format(field)
+    assert field in err2, err2
+    assert str(second) not in err2, "the error must name the field, never its value"
+    assert built == [first], "the second event must not have been published on the first client"
+
+
+def test_r2_identical_configuration_still_reuses_the_connection():
+    """The guard must not become so strict that reuse never happens."""
+    from lib_py.delegation.publisher import PersistentPublisher
+
+    counters = {"n": 0}
+
+    def factory(_role, _config):
+        counters["n"] += 1
+        from lib_py.delegation.bench import FakeClient
+        return FakeClient(connect_latency=0.0, counters={})
+
+    publisher = PersistentPublisher(client_factory=factory)
+    try:
+        for _ in range(3):
+            assert publisher.publish_with_retry(_prepared(), attempts=1) == (True, None)
+    finally:
+        publisher.close()
+    assert counters["n"] == 1
+
+
+class _FlakyConnectionClient(object):
+    """paho-shaped: the first N publishes are issued while disconnected."""
+
+    def __init__(self, no_conn_times, calls):
+        import paho.mqtt.client as mqtt
+        self._mqtt = mqtt
+        self._no_conn_times = no_conn_times
+        self._calls = calls
+        self.on_connect = None
+
+    def connect(self, host, port, keepalive=60):
+        self._calls["connect"] = self._calls.get("connect", 0) + 1
+        self.on_connect(self, None, None, 0, None)
+
+    def loop_start(self):
+        return None
+
+    def loop_stop(self):
+        return None
+
+    def reconnect_delay_set(self, **_kw):
+        return None
+
+    def publish(self, *_a, **_k):
+        from lib_py.delegation.bench import FakeInfo
+        n = self._calls.get("publish", 0)
+        self._calls["publish"] = n + 1
+        if n < self._no_conn_times:
+            return FakeInfo(published=False, rc=int(self._mqtt.MQTT_ERR_NO_CONN))
+        return FakeInfo(published=True, rc=0)
+
+    def disconnect(self):
+        return None
+
+
+def test_f1_publish_issued_while_disconnected_is_retried_until_reconnect():
+    """F-1: the gap between a drop and paho's reconnect must be survivable.
+
+    paho queues such a publish and raises RuntimeError from wait_for_publish,
+    which is not in the retry tuple; it used to end the attempt immediately."""
+    from lib_py.delegation.publisher import PersistentPublisher
+
+    calls = {}
+    publisher = PersistentPublisher(
+        client_factory=lambda _r, _c: _FlakyConnectionClient(2, calls))
+    try:
+        ok, err = publisher.publish_with_retry(_prepared(), attempts=3)
+    finally:
+        publisher.close()
+    assert ok is True, "should have recovered on the third attempt, got: {}".format(err)
+    assert calls["publish"] == 3, "every attempt must reach the transport"
+    assert calls["connect"] == 1, "recovery must reuse the connection, not rebuild it"
+
+
+def test_f1_disconnected_publish_still_fails_when_it_never_recovers():
+    """Retrying must not mask a genuinely dead connection."""
+    from lib_py.delegation.publisher import PersistentPublisher
+
+    calls = {}
+    publisher = PersistentPublisher(
+        client_factory=lambda _r, _c: _FlakyConnectionClient(99, calls))
+    try:
+        ok, err = publisher.publish_with_retry(_prepared(), attempts=2)
+    finally:
+        publisher.close()
+    assert ok is False
+    assert "not connected" in err, err
+    assert calls["publish"] == 2
+
+
+def test_one_shot_publish_has_a_single_implementation():
+    """O-1: the CLI path and the service path must not drift apart again."""
+    from lib_py.delegation import publish_event
+    from lib_py.delegation.publisher import OneShotPublisher
+
+    source = Path(publish_event.__file__).read_text(encoding="utf-8")
+    assert "OneShotPublisher().publish_raw(" in source, (
+        "_publish_once must delegate to the transport module"
+    )
+    for leaked in ("client.loop_start()", "no CONNACK from broker", "wait_for_publish("):
+        assert leaked not in source, (
+            "publish_event still carries its own copy of the transport logic: {}".format(leaked)
+        )
+    assert publish_event.CONNECT_ACK_TIMEOUT is not None
+    assert hasattr(OneShotPublisher, "publish_raw")