fix(issue-5): scope the persistent publisher's CONNACK result to its own connection

Found while self-reviewing phase 1. PersistentPublisher held the connect
handshake result on the instance, so a refused attempt left its reason code
behind for the next one. On retry the wait loop saw a non-None value straight
away and never waited for the new connection's CONNACK, which is wrong in both
directions:

  * a healthy second connection was rejected using the first attempt's rc, and
  * once a late callback had written a success rc, a subsequent connection that
    never acknowledged could be accepted on that stale value, letting a publish
    proceed on a socket that was not ready.

Both need an asynchronous CONNACK to reproduce, which is exactly how paho
delivers it; the earlier fake called the callback inline and hid it. The
handshake state is now a local the callback closes over, so it can only ever
describe the connection that produced it, and a client that fails the handshake
is torn down instead of leaked.

No production path is affected today: the CLI and shell helpers publish once
per process and use OneShotPublisher. The defect mattered because the phase 2a
engine is meant to build on this prototype, which is where a transient refusal
would have become a stuck publisher.

Also tightens PV-1 to check the bound tree against pytest's real basetemp
rather than sniffing for "pytest-" in the path.

Verified: the three new regressions fail against the previous implementation
(2 of 3, the third covers the healthy path) and pass against this one.

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

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

@@ -100,7 +100,6 @@ class PersistentPublisher(_BasePublisher):
         self.connect_count = 0
         self._client = None
         self._config = None
-        self._connected = {"rc": None}
 
     def __enter__(self):
         return self.open()
@@ -124,8 +123,14 @@ class PersistentPublisher(_BasePublisher):
         client = self.client_factory("publisher", config)
         self.connect_count += 1
 
+        # The handshake result belongs to THIS client, so it lives in a local the
+        # callback closes over. Holding it on self would let a refused attempt's
+        # reason code be read by the next attempt: the retry would either reject a
+        # healthy connection or, worse, accept one whose CONNACK never arrived.
+        connected = {"rc": None}
+
         def on_connect(_c, _u, _flags, reason_code, _props):
-            self._connected["rc"] = reason_code
+            connected["rc"] = reason_code
 
         client.on_connect = on_connect
         client.connect(config.host, config.port, config.keepalive)
@@ -133,19 +138,29 @@ class PersistentPublisher(_BasePublisher):
         reconnect_delay_set = getattr(client, "reconnect_delay_set", None)
         if reconnect_delay_set is not None:
             reconnect_delay_set(min_delay=1, max_delay=16)
-        deadline = time.monotonic() + CONNECT_ACK_TIMEOUT
-        while self._connected["rc"] is None and time.monotonic() < deadline:
-            time.sleep(0.05)
-        if self._connected["rc"] is None:
-            client.loop_stop()
-            raise TimeoutError("no CONNACK from broker")
-        if mqtt_common.reason_code_value(self._connected["rc"]) != 0:
-            client.loop_stop()
-            raise ConnectionError("broker refused connection: rc={}".format(self._connected["rc"]))
+        try:
+            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("broker refused connection: rc={}".format(connected["rc"]))
+        except Exception:
+            self._discard(client)   # never leak the half-open client on the way out
+            raise
         self._client = client
         self._config = config
         return client
 
+    @staticmethod
+    def _discard(client):
+        try:
+            client.loop_stop()
+            client.disconnect()
+        except Exception:  # pragma: no cover - best effort
+            pass
+
     def publish(self, prepared):
         client = self._ensure_client(prepared.config)
         info = client.publish(prepared.topic, payload=prepared.body,
@@ -159,11 +174,7 @@ class PersistentPublisher(_BasePublisher):
         if self._client is None:
             return
         try:
-            self._client.loop_stop()
-            self._client.disconnect()
-        except Exception:  # pragma: no cover - best effort
-            pass
+            self._discard(self._client)
         finally:
             self._client = None
             self._config = None
-            self._connected = {"rc": None}

tests/test_issue5_phase1_delegation.py

@@ -415,7 +415,7 @@ def test_pv5_all_four_shims_share_one_bootstrap_block():
     assert "sys.path.append(_SKILLS)" in base, "standalone fallback must remain"
 
 
-def test_pv1_identity_and_provenance_of_every_alias(mam_sandbox):
+def test_pv1_identity_and_provenance_of_every_alias(mam_sandbox, tmp_path_factory):
     """PV-1: each alias IS the package module, and all four share one tree."""
     import lib_py.delegation
 
@@ -432,9 +432,19 @@ def test_pv1_identity_and_provenance_of_every_alias(mam_sandbox):
     assert len(paths) == 1, "delegation modules span multiple trees: {}".format(paths)
     bound = Path(paths.pop())
     real = PKG_DIR.resolve()
-    basetemp_ok = "pytest-" in str(bound)
-    assert bound == real or basetemp_ok, (
-        "delegation bound to an unexpected tree: {}".format(bound)
+    # Provenance, not just identity: the bound tree is either the real package or a
+    # copy under pytest's own basetemp. Checked against the actual basetemp rather
+    # than sniffing for "pytest-" in the path, so an unrelated directory that merely
+    # looks temporary cannot satisfy it.
+    basetemp = Path(tmp_path_factory.getbasetemp()).resolve()
+    try:
+        bound.relative_to(basetemp)
+        under_basetemp = True
+    except ValueError:
+        under_basetemp = False
+    assert bound == real or under_basetemp, (
+        "delegation bound to an unexpected tree: {} (real={}, basetemp={})".format(
+            bound, real, basetemp)
     )
     # the parent package must never be dragged along with it
     assert Path(lib_py.__file__).resolve().parent == (SKILLS / "lib_py").resolve()
@@ -880,3 +890,113 @@ def test_bench_broker_transport_requires_a_job():
 
     with pytest.raises(ValueError):
         bench.run_benchmark(n=1, transport="broker")
+
+
+# ==============================================================================
+# 7. Self-review regressions — the connect handshake is per-connection
+#
+# Found reviewing phase 1: PersistentPublisher held the CONNACK result on the
+# instance, so a refused attempt left its reason code behind for the next one.
+# The retry then either rejected a healthy connection or, worse, accepted one
+# whose CONNACK had never arrived. Both need an asynchronous callback to
+# reproduce, which is how real paho delivers it.
+# ==============================================================================
+
+class _AsyncAckClient(object):
+    """paho-shaped stub: connect() returns at once, CONNACK lands on a timer."""
+
+    def __init__(self, rc_sequence, calls, ack_delay=0.05):
+        self._rc_sequence = rc_sequence
+        self._calls = calls
+        self._ack_delay = ack_delay
+        self.on_connect = None
+
+    def connect(self, host, port, keepalive=60):
+        import threading
+        idx = self._calls["n"]
+        self._calls["n"] += 1
+        rc = self._rc_sequence[min(idx, len(self._rc_sequence) - 1)]
+        if rc is None:
+            return  # models a broker that never acknowledges
+        timer = threading.Timer(self._ack_delay,
+                                lambda: self.on_connect(self, None, None, rc, None))
+        timer.daemon = True
+        timer.start()
+
+    def loop_start(self):
+        return None
+
+    def loop_stop(self):
+        self._calls["loop_stop"] = self._calls.get("loop_stop", 0) + 1
+
+    def reconnect_delay_set(self, **_kw):
+        return None
+
+    def publish(self, *_a, **_k):
+        from lib_py.delegation.bench import FakeInfo
+        self._calls["publish"] = self._calls.get("publish", 0) + 1
+        return FakeInfo()
+
+    def disconnect(self):
+        self._calls["disconnect"] = self._calls.get("disconnect", 0) + 1
+
+
+def _async_factory(rc_sequence, calls):
+    def factory(_role, _config):
+        return _AsyncAckClient(rc_sequence, calls)
+    return factory
+
+
+def test_persistent_publisher_recovers_on_the_attempt_after_a_refusal():
+    """A refused connection must not decide the next attempt's outcome."""
+    from lib_py.delegation.publisher import PersistentPublisher
+
+    calls = {"n": 0}
+    publisher = PersistentPublisher(client_factory=_async_factory([5, 0], calls))
+    try:
+        ok, err = publisher.publish_with_retry(_prepared(), attempts=3)
+    finally:
+        publisher.close()
+    assert ok is True, err
+    assert calls["n"] == 2, (
+        "attempt 2 had a healthy connection and should have been the last one; "
+        "got {} connect() calls".format(calls["n"])
+    )
+    assert calls.get("disconnect", 0) >= 1, "the refused client must be torn down, not leaked"
+
+
+def test_persistent_publisher_never_accepts_a_connection_that_did_not_ack():
+    """A silent broker must time out even when an earlier attempt acked."""
+    from lib_py.delegation.publisher import PersistentPublisher
+    import lib_py.delegation.publisher as publisher_mod
+
+    calls = {"n": 0}
+    # first connection is refused (leaves a reason code behind), second never acks
+    publisher = PersistentPublisher(client_factory=_async_factory([5, None], calls))
+    original = publisher_mod.CONNECT_ACK_TIMEOUT
+    publisher_mod.CONNECT_ACK_TIMEOUT = 0.3   # keep the test quick
+    try:
+        ok, err = publisher.publish_with_retry(_prepared(), attempts=2)
+    finally:
+        publisher_mod.CONNECT_ACK_TIMEOUT = original
+        publisher.close()
+    assert ok is False
+    assert "no CONNACK" in err, (
+        "the silent second connection must time out on its own evidence, not "
+        "inherit the first attempt's reason code; got: {}".format(err)
+    )
+
+
+def test_persistent_publisher_reuses_a_healthy_connection_across_events():
+    """The successful path still connects exactly once."""
+    from lib_py.delegation.publisher import PersistentPublisher
+
+    calls = {"n": 0}
+    publisher = PersistentPublisher(client_factory=_async_factory([0], calls))
+    try:
+        for _ in range(4):
+            assert publisher.publish_with_retry(_prepared(), attempts=1) == (True, None)
+    finally:
+        publisher.close()
+    assert calls["n"] == 1, "a healthy connection must be reused"
+    assert calls["publish"] == 4