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>

tests/_local_mqtt.py

@@ -125,6 +125,7 @@ class LocalBroker:
     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 -------------------------------------------------------------
@@ -188,11 +189,15 @@ class LocalBroker:
             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
-                conn.send(_packet(PUBACK << 4, pid))
+            # 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(), []