diff --git a/include/mqtt_client.h b/include/mqtt_client.h
index 7e7b38a..2db00cd 100644
--- a/include/mqtt_client.h
+++ b/include/mqtt_client.h
@@ -568,8 +568,12 @@ int esp_mqtt_client_unsubscribe(esp_mqtt_client_handle_t client,
  * (10s) or if publishing payloads longer than internal buffer (due to message
  *   fragmentation)
  * - Client doesn't have to be connected for this API to work, enqueueing the
- * messages with qos>1 (returning -1 for all the qos=0 messages if
- * disconnected). If MQTT_SKIP_PUBLISH_IF_DISCONNECTED is enabled, this API will
+ * messages with qos>0 (returning -1 for all the qos=0 messages if
+ * disconnected).
+ * - In case of MQTT v5, if the server quota for inflight messages is exceeded,
+ *   message will be enqueued and sent later when quota is available.
+ * - QoS 0 messages are sent immediately in the calling task, not via the outbox.
+ * - If MQTT_SKIP_PUBLISH_IF_DISCONNECTED is enabled, this API will
  * not attempt to publish when the client is not connected and will always
  * return -1.
  * - It is thread safe, please refer to `esp_mqtt_client_subscribe` for details
@@ -597,6 +601,8 @@ int esp_mqtt_client_publish(esp_mqtt_client_handle_t client, const char *topic,
  * (in contrast to the esp_mqtt_client_publish() which sends the publish message
  * immediately in the user task's context). Thus, it could be used as a non
  * blocking version of esp_mqtt_client_publish().
+ * - When MQTT v5 inflight quota is exceeded, queued QoS 1/2 messages are held
+ *   in the outbox. QoS 0 messages enqueued with store=true are not affected.
  *
  * @param client    *MQTT* client handle
  * @param topic     topic string
diff --git a/lib/include/mqtt5_client_priv.h b/lib/include/mqtt5_client_priv.h
index b3d355b..8f54478 100644
--- a/lib/include/mqtt5_client_priv.h
+++ b/lib/include/mqtt5_client_priv.h
@@ -8,7 +8,6 @@
 #define _MQTT5_CLIENT_PRIV_H_
 
 #include "mqtt5_client.h"
-#include "mqtt_client_priv.h"
 #include "mqtt5_msg.h"
 
 #ifdef __cplusplus
@@ -45,6 +44,7 @@ void esp_mqtt5_parse_suback(esp_mqtt5_client_handle_t client);
 void esp_mqtt5_parse_disconnect(esp_mqtt5_client_handle_t client, int *disconnect_rsp_code);
 esp_err_t esp_mqtt5_parse_connack(esp_mqtt5_client_handle_t client, int *connect_rsp_code);
 void esp_mqtt5_client_destory(esp_mqtt5_client_handle_t client);
+esp_err_t esp_mqtt5_client_check_inflight_maximum(esp_mqtt5_client_handle_t client);
 esp_err_t esp_mqtt5_client_publish_check(esp_mqtt5_client_handle_t client, int qos, int retain);
 esp_err_t esp_mqtt5_client_subscribe_check(esp_mqtt5_client_handle_t client, int qos);
 esp_err_t esp_mqtt5_create_default_config(esp_mqtt5_client_handle_t client);
diff --git a/mqtt5_client.c b/mqtt5_client.c
index c687b3a..0104f58 100644
--- a/mqtt5_client.c
+++ b/mqtt5_client.c
@@ -10,6 +10,9 @@
 
 static const char *TAG = "mqtt5_client";
 
+// Receive Maximum is optional in CONNACK; when absent the limit is 65535
+#define MQTT5_DEFAULT_RECEIVE_MAXIMUM 65535
+
 static void esp_mqtt5_print_error_code(esp_mqtt5_client_handle_t client, int code);
 static esp_err_t esp_mqtt5_client_update_topic_alias(mqtt5_topic_alias_handle_t topic_alias_handle,
                                                      uint16_t topic_alias, char *topic, size_t topic_len);
@@ -21,12 +24,8 @@ static esp_err_t esp_mqtt5_user_property_copy(mqtt5_user_property_handle_t user_
 
 void esp_mqtt5_increment_packet_counter(esp_mqtt5_client_handle_t client)
 {
-    bool msg_dup = mqtt5_get_dup(client->mqtt_state.connection.outbound_message.data);
-
-    if (msg_dup == false) {
-        client->send_publish_packet_count ++;
-        ESP_LOGD(TAG, "Sent (%d) qos > 0 publish packet without ack", client->send_publish_packet_count);
-    }
+    client->send_publish_packet_count ++;
+    ESP_LOGD(TAG, "Sent (%d) qos > 0 publish packet without ack", client->send_publish_packet_count);
 }
 
 void esp_mqtt5_decrement_packet_counter(esp_mqtt5_client_handle_t client)
@@ -104,6 +103,7 @@ esp_err_t esp_mqtt5_parse_connack(esp_mqtt5_client_handle_t client, int *connect
     size_t len = client->mqtt_state.in_buffer_read_len;
     client->mqtt_state.in_buffer_read_len = 0;
     uint8_t ack_flag = 0;
+    client->mqtt5_config->server_resp_property_info.receive_maximum = MQTT5_DEFAULT_RECEIVE_MAXIMUM;
 
     if (mqtt5_msg_parse_connack_property(client->mqtt_state.in_buffer, len, &client->mqtt_state.
                                          connection.information, &client->mqtt5_config->connect_property_info, &client->mqtt5_config->server_resp_property_info,
@@ -195,7 +195,7 @@ esp_err_t esp_mqtt5_create_default_config(esp_mqtt5_client_handle_t client)
         client->mqtt5_config->server_resp_property_info.wildcard_subscribe_available = true;
         client->mqtt5_config->server_resp_property_info.subscribe_identifiers_available = true;
         client->mqtt5_config->server_resp_property_info.shared_subscribe_available = true;
-        client->mqtt5_config->server_resp_property_info.receive_maximum = 65535;
+        client->mqtt5_config->server_resp_property_info.receive_maximum = MQTT5_DEFAULT_RECEIVE_MAXIMUM;
     }
 
     return ESP_OK;
@@ -375,9 +375,13 @@ esp_err_t esp_mqtt5_client_publish_check(esp_mqtt5_client_handle_t client, int q
         return ESP_FAIL;
     }
 
-    /* Flow control to check PUBLISH(No PUBACK or PUBCOMP received) packet sent count(Only record QoS1 and QoS2)*/
-    if (client->send_publish_packet_count > client->mqtt5_config->server_resp_property_info.receive_maximum) {
-        ESP_LOGE(TAG, "Client send more than %d QoS1 and QoS2 PUBLISH packet without no ack",
+    return ESP_OK;
+}
+
+esp_err_t esp_mqtt5_client_check_inflight_maximum(esp_mqtt5_client_handle_t client)
+{
+    if (client->send_publish_packet_count >= client->mqtt5_config->server_resp_property_info.receive_maximum) {
+        ESP_LOGD(TAG, "Broker quota for QoS > 0 exceeded. Quota is %d messages",
                  client->mqtt5_config->server_resp_property_info.receive_maximum);
         return ESP_FAIL;
     }
diff --git a/mqtt_client.c b/mqtt_client.c
index 252e8dc..da35277 100644
--- a/mqtt_client.c
+++ b/mqtt_client.c
@@ -843,6 +843,43 @@ static inline esp_err_t esp_mqtt_write(esp_mqtt_client_handle_t client)
     return ESP_OK;
 }
 
+#ifdef MQTT_PROTOCOL_5
+static void mqtt_requeue_transmitted_messages(esp_mqtt_client_handle_t client)
+{
+    outbox_item_handle_t item;
+
+    // Receive Maximum is scoped to the network connection. Requeue previous
+    // inflight packets so the new connection admits and counts them once.
+    //
+    // [MQTT-4.4.0-1] only sanctions resending unacknowledged QoS>0 PUBLISH and
+    // PUBREL packets, so requeuing subscribe and unsubscribe is not correct.
+    // It is kept because it is what the client has always done: the periodic
+    // retransmit path resends any TRANSMITTED packet regardless of type, so
+    // dropping them here would silently break subscriptions that work today.
+    while ((item = outbox_dequeue(client->outbox, TRANSMITTED, NULL)) != NULL) {
+        size_t len;
+        uint16_t msg_id;
+        int msg_type;
+        int msg_qos;
+        uint8_t *data = outbox_item_get_data(item, &len, &msg_id, &msg_type, &msg_qos);
+
+        if (data == NULL) {
+            ESP_LOGE(TAG, "Failed to read transmitted outbox item");
+            break;
+        }
+
+        if (msg_type == MQTT_MSG_TYPE_PUBLISH && msg_qos > 0) {
+            mqtt_set_dup(data);
+        }
+
+        if (outbox_set_pending(client->outbox, msg_id, QUEUED) != ESP_OK) {
+            ESP_LOGE(TAG, "Failed to requeue transmitted message id=%d", msg_id);
+            break;
+        }
+    }
+}
+#endif
+
 static esp_err_t esp_mqtt_connect(esp_mqtt_client_handle_t client, int timeout_ms)
 {
     int read_len, connect_rsp_code = 0;
@@ -909,6 +946,7 @@ static esp_err_t esp_mqtt_connect(esp_mqtt_client_handle_t client, int timeout_m
 
         if (esp_mqtt5_parse_connack(client, &connect_rsp_code) == ESP_OK) {
             client->send_publish_packet_count = 0;
+            mqtt_requeue_transmitted_messages(client);
             return ESP_OK;
         }
 
@@ -1690,14 +1728,14 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
         break;
 
     case MQTT_MSG_TYPE_PUBACK:
+        if (remove_initiator_message(client, MQTT_MSG_TYPE_PUBLISH, msg_id)) {
 #ifdef MQTT_PROTOCOL_5
-        if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
-            esp_mqtt5_decrement_packet_counter(client);
-        }
+
+            if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
+                esp_mqtt5_decrement_packet_counter(client);
+            }
 
 #endif
-
-        if (remove_initiator_message(client, MQTT_MSG_TYPE_PUBLISH, msg_id)) {
             ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBACK, finish QoS1 publish");
 #ifdef MQTT_PROTOCOL_5
             esp_mqtt5_parse_puback(client);
@@ -1753,15 +1791,15 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
 
     case MQTT_MSG_TYPE_PUBCOMP:
         ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBCOMP");
-#ifdef MQTT_PROTOCOL_5
-
-        if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
-            esp_mqtt5_decrement_packet_counter(client);
-        }
-
-#endif
 
         if (remove_initiator_message(client, MQTT_MSG_TYPE_PUBLISH, msg_id)) {
+#ifdef MQTT_PROTOCOL_5
+
+            if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
+                esp_mqtt5_decrement_packet_counter(client);
+            }
+
+#endif
             ESP_LOGD(TAG, "Receive MQTT_MSG_TYPE_PUBCOMP, finish QoS2 publish");
 #ifdef MQTT_PROTOCOL_5
             esp_mqtt5_parse_pubcomp(client);
@@ -1832,6 +1870,25 @@ static esp_err_t mqtt_resend_queued(esp_mqtt_client_handle_t client, outbox_item
     return ESP_OK;
 }
 
+#ifdef MQTT_PROTOCOL_5
+static outbox_item_handle_t mqtt_get_queued_qos0(outbox_handle_t outbox)
+{
+    outbox_item_handle_t item = outbox_get(outbox, 0);
+    size_t len;
+    uint16_t msg_id;
+    int msg_type;
+    int msg_qos;
+
+    if (item && outbox_item_get_pending(item) == QUEUED &&
+            outbox_item_get_data(item, &len, &msg_id, &msg_type, &msg_qos) != NULL &&
+            msg_id == 0 && msg_type == MQTT_MSG_TYPE_PUBLISH && msg_qos == 0) {
+        return item;
+    }
+
+    return NULL;
+}
+#endif
+
 static esp_err_t mqtt_resend_pubrel(esp_mqtt_client_handle_t client, outbox_item_handle_t item)
 {
     client->mqtt_state.connection.outbound_message.data = outbox_item_get_data(item,
@@ -2021,6 +2078,23 @@ static void esp_mqtt_task(void *pv)
 
             // resend all non-transmitted messages first
             outbox_item_handle_t item = outbox_dequeue(client->outbox, QUEUED, NULL);
+#ifdef MQTT_PROTOCOL_5
+
+            if (item && client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5 &&
+                    esp_mqtt5_client_check_inflight_maximum(client) != ESP_OK) {
+                size_t len;
+                uint16_t msg_id;
+                int msg_type = 0;
+                int msg_qos = 0;
+
+                if (outbox_item_get_data(item, &len, &msg_id, &msg_type, &msg_qos) != NULL &&
+                        msg_type == MQTT_MSG_TYPE_PUBLISH && msg_qos > 0) {
+                    // Receive Maximum applies only to QoS 1 and QoS 2.
+                    item = mqtt_get_queued_qos0(client->outbox);
+                }
+            }
+
+#endif
 
             if (item) {
                 if (mqtt_resend_queued(client, item) == ESP_OK) {
@@ -2029,14 +2103,13 @@ static void esp_mqtt_task(void *pv)
                         if (outbox_delete_item(client->outbox, item) != ESP_OK) {
                             ESP_LOGE(TAG, "Failed to remove queued qos0 message from the outbox");
                         }
-                    }
-
-                    if (client->mqtt_state.pending_publish_qos > 0 &&
-                            mqtt_get_type(client->mqtt_state.connection.outbound_message.data) == MQTT_MSG_TYPE_PUBLISH) {
+                    } else {
+                        outbox_set_tick(client->outbox, client->mqtt_state.pending_msg_id, platform_tick_get_ms());
                         outbox_set_pending(client->outbox, client->mqtt_state.pending_msg_id, TRANSMITTED);
 #ifdef MQTT_PROTOCOL_5
 
-                        if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
+                        if (client->mqtt_state.pending_msg_type == MQTT_MSG_TYPE_PUBLISH &&
+                                client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
                             esp_mqtt5_increment_packet_counter(client);
                         }
 
@@ -2050,36 +2123,13 @@ static void esp_mqtt_task(void *pv)
                 item = outbox_dequeue(client->outbox, TRANSMITTED, &msg_tick);
 
                 if (item && (last_retransmit - msg_tick > client->config->message_retransmit_timeout))  {
-                    if (mqtt_resend_queued(client, item) == ESP_OK) {
-#ifdef MQTT_PROTOCOL_5
-
-                        if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5 &&
-                                client->mqtt_state.pending_publish_qos > 0 &&
-                                mqtt_get_type(client->mqtt_state.connection.outbound_message.data) == MQTT_MSG_TYPE_PUBLISH) {
-                            esp_mqtt5_increment_packet_counter(client);
-                        }
-
-#endif
-                    }
+                    mqtt_resend_queued(client, item);
                 }
 
                 item = outbox_dequeue(client->outbox, ACKNOWLEDGED, &msg_tick);
 
                 if (item && (last_retransmit - msg_tick > client->config->message_retransmit_timeout))  {
-                    if (mqtt_resend_pubrel(client, item) == ESP_OK) {
-#ifdef MQTT_PROTOCOL_5
-
-                        // Do not count PUBREL as a new inflight PUBLISH
-                        // Only PUBLISH QoS>0 contributes to inflight limitation
-                        // (outbound_message here is PUBREL, so this condition will be false)
-                        if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5 &&
-                                client->mqtt_state.pending_publish_qos > 0 &&
-                                mqtt_get_type(client->mqtt_state.connection.outbound_message.data) == MQTT_MSG_TYPE_PUBLISH) {
-                            esp_mqtt5_increment_packet_counter(client);
-                        }
-
-#endif
-                    }
+                    mqtt_resend_pubrel(client, item);
                 }
             }
 
@@ -2311,6 +2361,8 @@ int esp_mqtt_client_subscribe_multiple(esp_mqtt_client_handle_t client,
     }
 
     MQTT_API_LOCK(client);
+    // Reset pending state to avoid inheriting previous PUBLISH QoS or type
+    mqtt_reset_pending_message(client);
 
     if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
 #ifdef MQTT_PROTOCOL_5
@@ -2572,6 +2624,17 @@ int esp_mqtt_client_publish(esp_mqtt_client_handle_t client, const char *topic,
         goto cannot_publish;
     }
 
+#ifdef MQTT_PROTOCOL_5
+
+    if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5 && qos > 0) {
+        if (esp_mqtt5_client_check_inflight_maximum(client) != ESP_OK) {
+            ESP_LOGW(TAG, "Unable to publish now: maximum inflight messages reached");
+            MQTT_API_UNLOCK(client);
+            return pending_msg_id;
+        }
+    }
+
+#endif
     /* Provide support for sending fragmented message if it doesn't fit buffer */
     int remaining_len = len;
     const char *current_data = data;
diff --git a/test/apps/mqtt_conformance/README.md b/test/apps/mqtt_conformance/README.md
index 095a557..e270272 100644
--- a/test/apps/mqtt_conformance/README.md
+++ b/test/apps/mqtt_conformance/README.md
@@ -17,7 +17,7 @@ This app exposes a console API for pytest-embedded HIL tests that target MQTT co
 
 ## JSON config keys
 
-All configuration is passed as a base64-encoded JSON object with exactly one top-level key, naming which config category the blob targets. The JSON shape mirrors the real esp_mqtt C struct layout, so field names/paths match `mqtt_client.h` / `mqtt5_client.h` directly.
+All configuration is passed as a base64-encoded JSON object with a recognized top-level key naming the config category the blob targets. The JSON shape mirrors the real esp_mqtt C struct layout, so field names/paths match `mqtt_client.h` / `mqtt5_client.h` directly.
 
 ### `mqtt_config` (used with `init`)
 
@@ -25,63 +25,67 @@ Mirrors `esp_mqtt_client_config_t`'s nesting:
 
 ```json
 {
-  "mqtt_config": {
-    "broker": { "address": { "uri": "mqtt://192.168.1.1:1883" } },
-    "credentials": { "client_id": "my-client" },
-    "session": { "keepalive": 30, "disable_clean_session": false, "protocol_ver": 3 },
-    "network": { "disable_auto_reconnect": true }
-  }
+    "mqtt_config": {
+        "broker": { "address": { "uri": "mqtt://192.168.1.1:1883" } },
+        "credentials": { "client_id": "my-client" },
+        "session": {
+            "keepalive": 30,
+            "disable_clean_session": false,
+            "protocol_ver": 3
+        },
+        "network": { "disable_auto_reconnect": true }
+    }
 }
 ```
 
-| Path | Type | Description |
-|------|------|-------------|
-| `broker.address.uri` | string | Broker URI (e.g. `mqtt://192.168.1.1:1883`) |
-| `credentials.client_id` | string | Client identifier |
-| `session.keepalive` | int | Keepalive interval (seconds) |
-| `session.disable_clean_session` | bool | `true` = persistent session (clean start = false) |
-| `session.protocol_ver` | int | Raw `esp_mqtt_protocol_ver_t` ordinal: `0`=UNDEFINED, `1`=MQTT 3.1, `2`=MQTT 3.1.1, `3`=MQTT 5.0 |
-| `network.disable_auto_reconnect` | bool | Disable MQTT client automatic reconnect |
+| Path                             | Type   | Description                                                                                      |
+| -------------------------------- | ------ | ------------------------------------------------------------------------------------------------ |
+| `broker.address.uri`             | string | Broker URI (e.g. `mqtt://192.168.1.1:1883`)                                                      |
+| `credentials.client_id`          | string | Client identifier                                                                                |
+| `session.keepalive`              | int    | Keepalive interval (seconds)                                                                     |
+| `session.disable_clean_session`  | bool   | `true` = persistent session (clean start = false)                                                |
+| `session.protocol_ver`           | int    | Raw `esp_mqtt_protocol_ver_t` ordinal: `0`=UNDEFINED, `1`=MQTT 3.1, `2`=MQTT 3.1.1, `3`=MQTT 5.0 |
+| `network.disable_auto_reconnect` | bool   | Disable MQTT client automatic reconnect                                                          |
 
 ### `connect_property` (MQTT5 connect properties)
 
 Already flat in C, so the JSON object is flat too:
 
-| Key | Type | Description |
-|-----|------|-------------|
-| `session_expiry_interval` | int | Session expiry (seconds) |
-| `receive_maximum` | int | Receive maximum |
-| `topic_alias_maximum` | int | Topic alias maximum |
-| `maximum_packet_size` | int | Maximum packet size |
-| `will_delay_interval` | int | Will delay interval (seconds) |
+| Key                       | Type | Description                   |
+| ------------------------- | ---- | ----------------------------- |
+| `session_expiry_interval` | int  | Session expiry (seconds)      |
+| `receive_maximum`         | int  | Receive maximum               |
+| `topic_alias_maximum`     | int  | Topic alias maximum           |
+| `maximum_packet_size`     | int  | Maximum packet size           |
+| `will_delay_interval`     | int  | Will delay interval (seconds) |
 
 ### `publish_property` (MQTT5 publish properties)
 
-| Key | Type | Description |
-|-----|------|-------------|
-| `message_expiry_interval` | int | Message expiry (seconds) |
-| `payload_format_indicator` | bool | `true` = UTF-8 encoded payload |
-| `topic_alias` | int | Topic alias |
-| `content_type` | string | Content type |
-| `response_topic` | string | Response topic |
+| Key                        | Type   | Description                    |
+| -------------------------- | ------ | ------------------------------ |
+| `message_expiry_interval`  | int    | Message expiry (seconds)       |
+| `payload_format_indicator` | bool   | `true` = UTF-8 encoded payload |
+| `topic_alias`              | int    | Topic alias                    |
+| `content_type`             | string | Content type                   |
+| `response_topic`           | string | Response topic                 |
 
 ### `subscribe_property` (MQTT5 subscribe properties)
 
-| Key | Type | Description |
-|-----|------|-------------|
-| `subscribe_id` | int | Subscription identifier |
-| `no_local_flag` | bool | No local flag |
-| `retain_as_published_flag` | bool | Retain as published flag |
-| `retain_handle` | int | Retain handling option (0/1/2) |
-| `is_share_subscribe` | bool | Shared subscription flag |
-| `share_name` | string | Shared subscription group name |
+| Key                        | Type   | Description                    |
+| -------------------------- | ------ | ------------------------------ |
+| `subscribe_id`             | int    | Subscription identifier        |
+| `no_local_flag`            | bool   | No local flag                  |
+| `retain_as_published_flag` | bool   | Retain as published flag       |
+| `retain_handle`            | int    | Retain handling option (0/1/2) |
+| `is_share_subscribe`       | bool   | Shared subscription flag       |
+| `share_name`               | string | Shared subscription group name |
 
 ### `disconnect_property` (MQTT5 disconnect properties)
 
-| Key | Type | Description |
-|-----|------|-------------|
-| `session_expiry_interval` | int | Session expiry override on disconnect |
-| `disconnect_reason` | int | Disconnect reason code |
+| Key                       | Type | Description                           |
+| ------------------------- | ---- | ------------------------------------- |
+| `session_expiry_interval` | int  | Session expiry override on disconnect |
+| `disconnect_reason`       | int  | Disconnect reason code                |
 
 ## Conformance mapping
 
@@ -98,12 +102,37 @@ From the repository root (or the mqtt worktree root if using worktrees):
 1. Ensure the environment is active (e.g. `direnv allow` at repo root so IDF and pytest-embedded are available).
 
 2. Initialize the paho.mqtt.testing submodule:
-   ```bash
-   git submodule update --init --recursive test/tools/paho.mqtt.testing
-   ```
+
+    ```bash
+    git submodule update --init --recursive test/tools/paho.mqtt.testing
+    ```
 
 3. Run the conformance tests (connect a board with Ethernet, or use the same target/port as in CI):
-   ```bash
-   pytest test/apps/mqtt_conformance/ -v
-   ```
-   To run a single test or filter by keyword, add e.g. `-k test_mqtt_v311` or the test path.
+
+    ```bash
+    pytest test/apps/mqtt_conformance/ -v
+    ```
+
+    To run a single test or filter by keyword, add e.g. `-k receive_maximum` or the test path.
+
+## Optional environment variables
+
+Each test starts its own fresh in-process paho broker on an OS-assigned ephemeral
+port and tears it down at the end of that test, so brokers never carry state
+between tests and there's no port to configure/coordinate.
+
+- `MQTT_CONFORMANCE_PAHO_BROKER_LOG_LEVEL` — log level for the in-process paho broker's own
+  logger (default: `WARNING`).
+- `MQTT_CONFORMANCE_HOST_IP` — host IPv4 address the DUT should use to reach the in-process
+  broker (default: auto-detected via a UDP socket connect to `8.8.8.8`).
+- `MQTT_CONFORMANCE_CONNECT_RETRIES` — number of `start`/connect attempts before failing
+  (default: 3).
+- `MQTT_CONFORMANCE_RETRY_BACKOFF_SEC` — backoff between connect retries, in seconds
+  (default: 2).
+
+### Timeouts
+
+Tests use **operation-based timeouts** (not a flat 60 s wait): the budget is computed
+from the number of connect, subscribe, publish, and event-wait operations. Whole-test
+ceilings use `@pytest.mark.timeout(...)`. Inflight tests do not rely on timing windows:
+the broker explicitly holds and releases PUBACK or PUBCOMP packets around assertions.
diff --git a/test/apps/mqtt_conformance/main/mqtt_conformance_console.cpp b/test/apps/mqtt_conformance/main/mqtt_conformance_console.cpp
index 7abcd65..1eeebfd 100644
--- a/test/apps/mqtt_conformance/main/mqtt_conformance_console.cpp
+++ b/test/apps/mqtt_conformance/main/mqtt_conformance_console.cpp
@@ -410,7 +410,7 @@ void register_commands()
 
 extern "C" void app_main(void)
 {
-    constexpr size_t max_line = 512;
+    constexpr size_t max_line = 2048;
     ESP_LOGI(TAG, "[APP] Free memory: %" PRIu32 " bytes", esp_get_free_heap_size());
     ESP_LOGI(TAG, "[APP] IDF version: %s", esp_get_idf_version());
     esp_log_level_set("*", ESP_LOG_INFO);
diff --git a/test/apps/mqtt_conformance/pytest_mqtt_conformance.py b/test/apps/mqtt_conformance/pytest_mqtt_conformance.py
index 65a696c..ecc6904 100644
--- a/test/apps/mqtt_conformance/pytest_mqtt_conformance.py
+++ b/test/apps/mqtt_conformance/pytest_mqtt_conformance.py
@@ -4,7 +4,11 @@ from __future__ import annotations
 
 import base64
 import contextlib
+import copy
+import enum
+import importlib
 import json
+import logging
 import os
 import random
 import re
@@ -22,13 +26,23 @@ from pytest_embedded import Dut
 from pytest_embedded_idf.utils import idf_parametrize
 
 TOPIC_SIZE = 16
-DUT_READY_TIMEOUT = 30
-DUT_CONNECT_TIMEOUT = 30
-DUT_SUBSCRIBE_TIMEOUT = 60
-DUT_TEST_TIMEOUT = 60
-PAHO_BROKER_PORT = int(os.getenv("MQTT_CONFORMANCE_PAHO_BROKER_PORT", "18883"))
+DUT_READY_TIMEOUT = 20
+DUT_CONNECT_TIMEOUT = 20
+DUT_SUBSCRIBE_TIMEOUT = 15
+DUT_CMD_TIMEOUT = 10
+DUT_EVENT_TIMEOUT = 20
+PAHO_BROKER_LOG_LEVEL = os.getenv("MQTT_CONFORMANCE_PAHO_BROKER_LOG_LEVEL", "WARNING").upper()
 CONNECT_RETRIES = int(os.getenv("MQTT_CONFORMANCE_CONNECT_RETRIES", "3"))
 RETRY_BACKOFF_SEC = float(os.getenv("MQTT_CONFORMANCE_RETRY_BACKOFF_SEC", "2"))
+DEFAULT_BROKER_RECEIVE_MAXIMUM = 2
+TEST_TIMEOUT_MARGIN_SEC = 2
+
+QUOTA_REJECTION_FORBIDDEN = (
+    b"MQTT5 publish check fail",
+    b"Publish failed, msg_id=-1",
+    b"MQTT_EVENT_ERROR",
+    b"MQTT_EVENT_DISCONNECTED",
+)
 
 PAHO_SPEC_FILE = (
     Path(__file__).resolve().parents[3]
@@ -52,7 +66,7 @@ def build_topic() -> str:
 
 
 # Raw esp_mqtt_protocol_ver_t ordinals (see mqtt_client.h): the wire format
-# mirrors the real enum value rather than a human-friendly "3/4/5" mapping.
+# mirrors the real enum value.
 MQTT_PROTOCOL_V_3_1_1 = 2
 MQTT_PROTOCOL_V_5 = 3
 
@@ -88,6 +102,12 @@ def esp_mqtt_config(
     return base64.b64encode(json.dumps({"mqtt_config": mqtt_config}).encode()).decode()
 
 
+def configure_paho_broker_logging() -> None:
+    logger = logging.getLogger("MQTT broker")
+    level = getattr(logging, PAHO_BROKER_LOG_LEVEL, logging.WARNING)
+    logger.setLevel(level)
+
+
 def require_paho_testing_checked_out() -> None:
     """Hard requirement: fail the test if the paho.mqtt.testing submodule is not available."""
     if not PAHO_SPEC_FILE.exists():
@@ -104,16 +124,68 @@ def get_host_ip4_by_dest_ip(dest_ip: str = "8.8.8.8") -> str:
         return sock.getsockname()[0]
 
 
-class _BrokerHandle(Protocol):
+class MqttPacketType(enum.IntEnum):
+    CONNECT = 1
+    CONNACK = 2
+    PUBLISH = 3
+    PUBACK = 4
+    PUBREC = 5
+    PUBREL = 6
+    PUBCOMP = 7
+    SUBSCRIBE = 8
+    SUBACK = 9
+    UNSUBSCRIBE = 10
+    UNSUBACK = 11
+    PINGREQ = 12
+    PINGRESP = 13
+    DISCONNECT = 14
+    AUTH = 15
+
+
+class BrokerInterface(Protocol):
     uri: str
 
+    def wait_for_held_packets(self, packet_type: MqttPacketType, count: int, timeout: float) -> None: ...
+
+    def held_packet_count(self, packet_type: MqttPacketType) -> int: ...
+
+    def release_held_packets(
+        self,
+        packet_type: MqttPacketType,
+        count: int | None = None,
+        *,
+        keep_holding: bool = False,
+    ) -> None: ...
+
+    def release_all_held_packets(self) -> None: ...
+
+    def held_packet_identifiers(self, packet_type: MqttPacketType) -> list[int]: ...
+
+    def inject_ack(self, packet_type: MqttPacketType, packet_identifier: int) -> None: ...
+
+    def discard_held_packets(self) -> None: ...
+
+    def set_receive_maximum(self, receive_maximum: int) -> None: ...
+
+    def disconnect_clients(self) -> None: ...
+
     def shutdown(self) -> None: ...
 
 
-def _start_paho_broker(port: int, host_ip: str) -> _BrokerHandle:
-    """Start paho V311+V5 broker in-process; return object with .uri and .shutdown().
+def _start_paho_broker(
+    host_ip: str,
+    port: int = 0,
+    receive_maximum: int = DEFAULT_BROKER_RECEIVE_MAXIMUM,
+    hold_packet_types: tuple[MqttPacketType, ...] = (),
+) -> BrokerInterface:
+    """Start paho V311+V5 broker in-process and return its control handle.
+
+    ``port=0`` (the default) binds an OS-assigned ephemeral port.
+
     Imports deferred so idf-ci collection-time mocking does not replace paho.
     """
+    configure_paho_broker_logging()
+
     from mqtt.brokers.V311 import MQTTBrokers as MQTTV3Brokers
     from mqtt.brokers.V5 import MQTTBrokers as MQTTV5Brokers
     from mqtt.brokers.listeners import TCPListeners
@@ -129,7 +201,7 @@ def _start_paho_broker(port: int, host_ip: str) -> _BrokerHandle:
         "publish_on_pubrel": False,
         "topicAliasMaximum": 2,
         "maximumPacketSize": 16384,
-        "receiveMaximum": 2,
+        "receiveMaximum": receive_maximum,
         "serverKeepAlive": 60,
         "maximum_qos": 2,
         "retain_available": True,
@@ -143,15 +215,135 @@ def _start_paho_broker(port: int, host_ip: str) -> _BrokerHandle:
     broker5.setBroker3(broker3)
     TCPListeners.setBrokers(broker3, broker5)
     server = TCPListeners.create(port=port, host="", serve_forever=False)
+    bound_port = server.socket.getsockname()[1]
+
+    original_respond = None
+    v5_brokers_mod = None
+    held_packets: dict[MqttPacketType, list[tuple[object, object, int]]] = {}
+    held_packets_condition = threading.Condition()
+    active_hold_packet_types = set(hold_packet_types)
+    if hold_packet_types:
+        v5_brokers_mod = importlib.import_module("mqtt.brokers.V5.MQTTBrokers")
+        original_respond = v5_brokers_mod.respond
+
+        def controlled_respond(sock, packet, maximumPacketSize=500):
+            packet_type = MqttPacketType(packet.fh.PacketType)
+            with held_packets_condition:
+                if packet_type in active_hold_packet_types:
+                    # paho mutates some packet objects after respond() returns
+                    # (notably setting DUP on PUBLISH), so retain the wire-state
+                    # snapshot that was presented to this interception point.
+                    held_packet = (sock, copy.deepcopy(packet), maximumPacketSize)
+                    held_packets.setdefault(packet_type, []).append(held_packet)
+                    held_packets_condition.notify_all()
+                    return
+            original_respond(sock, packet, maximumPacketSize)
+
+        v5_brokers_mod.respond = controlled_respond  # type: ignore[attr-defined]
 
     class _Broker:
         def __init__(self) -> None:
-            self.uri = f"mqtt://{host_ip}:{port}"
+            self.uri = f"mqtt://{host_ip}:{bound_port}"
             self._broker3 = broker3
             self._broker5 = broker5
             self._server = server
+            self._v5_brokers_mod = v5_brokers_mod
+            self._original_respond = original_respond
+
+        def wait_for_held_packets(self, packet_type: MqttPacketType, count: int, timeout: float) -> None:
+            deadline = time.monotonic() + timeout
+            with held_packets_condition:
+                while len(held_packets.get(packet_type, ())) < count:
+                    remaining = deadline - time.monotonic()
+                    if remaining <= 0:
+                        held_count = len(held_packets.get(packet_type, ()))
+                        raise TimeoutError(
+                            f"Timed out waiting for {count} held {packet_type.name} packets; got {held_count}"
+                        )
+                    held_packets_condition.wait(remaining)
+
+        def held_packet_count(self, packet_type: MqttPacketType) -> int:
+            with held_packets_condition:
+                return len(held_packets.get(packet_type, ()))
+
+        def held_packet_identifiers(self, packet_type: MqttPacketType) -> list[int]:
+            with held_packets_condition:
+                return [int(getattr(packet, "packetIdentifier")) for _, packet, _ in held_packets.get(packet_type, ())]
+
+        def inject_ack(self, packet_type: MqttPacketType, packet_identifier: int) -> None:
+            if packet_type not in (MqttPacketType.PUBACK, MqttPacketType.PUBCOMP):
+                raise ValueError(f"Cannot inject {packet_type.name}; expected PUBACK or PUBCOMP")
+            respond = self._original_respond
+            if respond is None:
+                raise RuntimeError("Cannot inject acknowledgements unless packet holding is enabled")
+
+            from mqtt.formats import MQTTV5
+
+            packet_class = MQTTV5.Pubacks if packet_type == MqttPacketType.PUBACK else MQTTV5.Pubcomps
+            packet = packet_class()
+            packet.packetIdentifier = packet_identifier
+            with lock:
+                sockets = list(self._broker5.clients)
+                if len(sockets) != 1:
+                    raise RuntimeError(f"Expected one connected MQTT5 client, got {len(sockets)}")
+                respond(sockets[0], packet)
+
+        def discard_held_packets(self) -> None:
+            with held_packets_condition:
+                held_packets.clear()
+
+        def set_receive_maximum(self, receive_maximum: int) -> None:
+            if not 1 <= receive_maximum <= 0xFFFF:
+                raise ValueError("Receive Maximum must be in the range 1..65535")
+            with lock:
+                self._broker5.options["receiveMaximum"] = receive_maximum
+
+        def disconnect_clients(self) -> None:
+            with lock:
+                self._broker5.disconnectAll()
+
+        def _send_held_packets(self, pending_packets: list[tuple[object, object, int]]) -> None:
+            respond = self._original_respond
+            if pending_packets and respond is None:
+                raise RuntimeError("Cannot release held packets without the broker response callback")
+            if respond is None:
+                return
+            for sock, packet, maximum_packet_size in pending_packets:
+                respond(sock, packet, maximum_packet_size)
+
+        def release_held_packets(
+            self,
+            packet_type: MqttPacketType,
+            count: int | None = None,
+            *,
+            keep_holding: bool = False,
+        ) -> None:
+            with held_packets_condition:
+                if not keep_holding:
+                    active_hold_packet_types.discard(packet_type)
+                packet_queue = held_packets.get(packet_type, [])
+                release_count = len(packet_queue) if count is None else count
+                if release_count < 0 or release_count > len(packet_queue):
+                    raise ValueError(
+                        f"Cannot release {release_count} held {packet_type.name} packets; {len(packet_queue)} available"
+                    )
+                pending_packets = packet_queue[:release_count]
+                del packet_queue[:release_count]
+                if not packet_queue:
+                    held_packets.pop(packet_type, None)
+            self._send_held_packets(pending_packets)
+
+        def release_all_held_packets(self) -> None:
+            with held_packets_condition:
+                active_hold_packet_types.clear()
+                pending_packets = [packet for packets in held_packets.values() for packet in packets]
+                held_packets.clear()
+            self._send_held_packets(pending_packets)
 
         def shutdown(self) -> None:
+            self.release_all_held_packets()
+            if self._original_respond is not None and self._v5_brokers_mod is not None:
+                self._v5_brokers_mod.respond = self._original_respond  # type: ignore[attr-defined]
             self._broker3.shutdown()
             self._broker5.shutdown()
             if self._server:
@@ -160,28 +352,38 @@ def _start_paho_broker(port: int, host_ip: str) -> _BrokerHandle:
     return _Broker()
 
 
-@pytest.fixture(scope="module")
-def broker() -> Generator[_BrokerHandle, None, None]:
-    """Start paho MQTT broker in-process for the smoke test. No subclass, just V311+V5 + TCP listener."""
+@contextlib.contextmanager
+def broker_started(
+    port: int = 0,
+    *,
+    receive_maximum: int = DEFAULT_BROKER_RECEIVE_MAXIMUM,
+    hold_packet_types: tuple[MqttPacketType, ...] = (),
+) -> Generator[BrokerInterface, None, None]:
+    """Start an in-process paho broker and guarantee shutdown on exit."""
     require_paho_testing_checked_out()
     host_ip = os.getenv("MQTT_CONFORMANCE_HOST_IP", "").strip() or get_host_ip4_by_dest_ip()
-    b = _start_paho_broker(port=PAHO_BROKER_PORT, host_ip=host_ip)
-    yield b
-    b.shutdown()
+    paho_broker = _start_paho_broker(
+        host_ip=host_ip,
+        port=port,
+        receive_maximum=receive_maximum,
+        hold_packet_types=hold_packet_types,
+    )
+    try:
+        yield paho_broker
+    finally:
+        paho_broker.shutdown()
 
 
-@pytest.fixture(scope="module")
-def broker_uri(broker: _BrokerHandle) -> str:
-    return broker.uri
-
-
-@pytest.fixture
-def mqtt_client(dut: Dut, broker_uri: str):
+@contextlib.contextmanager
+def initialized_mqtt_client(dut: Dut, uri: str, *, protocol_ver: int = MQTT_PROTOCOL_V_5) -> Generator[Dut, None, None]:
+    """Init the MQTT client against ``uri`` and guarantee ``destroy`` on exit."""
     require_paho_testing_checked_out()
     dut.expect(re.compile(rb"mqtt>"), timeout=DUT_READY_TIMEOUT)
-    dut.write(f"init {esp_mqtt_config(protocol_ver=MQTT_PROTOCOL_V_3_1_1, uri=broker_uri)}")
-    yield dut
-    dut.write("destroy")
+    dut.write(f"init {esp_mqtt_config(protocol_ver=protocol_ver, uri=uri)}")
+    try:
+        yield dut
+    finally:
+        dut.write("destroy")
 
 
 def start_client(dut: Dut) -> None:
@@ -201,32 +403,693 @@ def stop_client(dut: Dut) -> None:
     dut.write("stop")
 
 
-@pytest.mark.eth_ip101
-@idf_parametrize("target", ["esp32"], indirect=["target"])
-def test_mqtt_v311_subscribe_and_qos1_publish__sec_3_8_4_and_4_3(mqtt_client: Dut) -> None:
-    """
-    MQTT v3.1.1 conformance smoke case:
-    - section 3.8.4: SUBSCRIBE/SUBACK interaction
-    - section 4.3: QoS 1 publish flow (at least once semantics)
+@contextlib.contextmanager
+def started_client(dut: Dut) -> Generator[Dut, None, None]:
+    """Start the MQTT client and guarantee ``stop`` is issued on exit, even on failure."""
+    try:
+        start_client(dut)
+        yield dut
+    finally:
+        stop_client(dut)
 
-    Reference suite integrated from:
-    test/tools/paho.mqtt.testing/interoperability/specifications/MQTTV311.py
+
+def case_timeout(
+    *,
+    connect_operations: int = 0,
+    subscribe_operations: int = 0,
+    publish_operations: int = 0,
+    event_wait_operations: int = 0,
+    timeout_margin: int = TEST_TIMEOUT_MARGIN_SEC,
+) -> int:
+    """Compound a timeout from the operations performed by a test or test phase."""
+    timeout = (
+        connect_operations * DUT_CONNECT_TIMEOUT
+        + subscribe_operations * DUT_SUBSCRIBE_TIMEOUT
+        + event_wait_operations * DUT_EVENT_TIMEOUT
+    )
+    if publish_operations:
+        timeout += max(6, int(publish_operations * 1.5 + timeout_margin))
+    elif timeout:
+        timeout += timeout_margin
+    return timeout
+
+
+def subscribed_to(dut: Dut, topic: str, qos: int, timeout: int = DUT_SUBSCRIBE_TIMEOUT) -> None:
+    """Issue ``subscribe`` and wait for the resulting MQTT_EVENT_SUBSCRIBED."""
+    dut.write(f"subscribe {topic} {qos}")
+    dut.expect(re.compile(rb"MQTT_EVENT_SUBSCRIBED"), timeout=timeout)
+
+
+def publish_from_dut(
+    dut: Dut,
+    topic: str,
+    qos: int,
+    *,
+    payload_prefix: str,
+    message_count: int,
+    enqueue: int = 0,
+    retain: int = 0,
+    pattern_repetitions: int = 1,
+) -> None:
+    """Write ``message_count`` individual ``publish`` commands (default: publish path)."""
+    for i in range(message_count):
+        publish_payload = f"{payload_prefix}{i}" if message_count > 1 else payload_prefix
+        dut.write(f"publish {topic} {publish_payload} {pattern_repetitions} {qos} {retain} {enqueue}")
+
+
+def data_payload_patterns(prefix: str, n_messages: int) -> dict[bytes, int]:
+    """Return one expected DATA payload marker for each uniquely suffixed message."""
+    return {f"MQTT_EVENT_DATA_PAYLOAD {prefix}{i}".encode(): 1 for i in range(n_messages)}
+
+
+def _check_forbidden_log_line(line: bytes, forbidden: tuple[bytes, ...]) -> None:
+    for bad in forbidden:
+        if bad in line:
+            pytest.fail(f"Forbidden log line containing {bad!r}: {line!r}")
+
+
+def expect_n(
+    dut: Dut,
+    patterns: "dict[bytes, int]",
+    timeout: int | None = None,
+    forbidden: tuple[bytes, ...] = QUOTA_REJECTION_FORBIDDEN,
+) -> "dict[bytes, int]":
+    """Wait until each pattern appears at least the required number of times.
+
+    Searches a combined regex against DUT output so all patterns are matched
+    in a single sequential pass — patterns that appear early in the stream
+    (e.g. log lines emitted during write() calls) are not missed.
+    """
+    keys = list(patterns.keys())
+    combined = re.compile(b"|".join(re.escape(k) for k in keys + list(forbidden)))
+    seen: dict[bytes, int] = {k: 0 for k in keys}
+    if timeout is None:
+        timeout = max(patterns.values()) * 3 + TEST_TIMEOUT_MARGIN_SEC
+    deadline = time.monotonic() + timeout
+    while any(seen[k] < patterns[k] for k in keys):
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise pexpect.TIMEOUT(f"Timed out. seen={seen}, expected={patterns}")
+        m = dut.expect(combined, timeout=remaining)
+        line: bytes = m.group(0)
+        _check_forbidden_log_line(line, forbidden)
+        for key in keys:
+            if key in line:
+                seen[key] += 1
+                break
+    return seen
+
+
+@pytest.mark.eth_ip101
+@pytest.mark.timeout(
+    case_timeout(
+        connect_operations=1,
+        subscribe_operations=1,
+        publish_operations=4,
+    )
+)
+@idf_parametrize("target", ["esp32"], indirect=["target"])
+@pytest.mark.parametrize("enqueue", [0, 1], ids=["publish", "enqueue"])
+@pytest.mark.parametrize(
+    "qos,completion_packet",
+    [(1, MqttPacketType.PUBACK), (2, MqttPacketType.PUBCOMP)],
+    ids=["qos1", "qos2"],
+)
+def test_mqtt5_receive_maximum_defers_publish(
+    dut: Dut,
+    enqueue: int,
+    qos: int,
+    completion_packet: MqttPacketType,
+) -> None:
+    """Receive Maximum defers, rather than drops, QoS 1/2 messages from both APIs."""
+    topic = build_topic()
+    publish_quota = DEFAULT_BROKER_RECEIVE_MAXIMUM
+    initial_prefix = "initial"
+    deferred_payload = "deferred"
+    sentinel_payload = "qos0sentinel"
+
+    with (
+        broker_started(hold_packet_types=(completion_packet,)) as broker,
+        initialized_mqtt_client(dut, broker.uri) as client,
+        started_client(client),
+    ):
+        subscribed_to(client, topic, qos)
+        try:
+            publish_from_dut(
+                client,
+                topic,
+                qos,
+                payload_prefix=initial_prefix,
+                message_count=publish_quota,
+                enqueue=enqueue,
+            )
+            expect_n(
+                client,
+                {b"Publish requested, msg_id=": publish_quota} | data_payload_patterns(initial_prefix, publish_quota),
+                timeout=DUT_EVENT_TIMEOUT,
+            )
+            broker.wait_for_held_packets(completion_packet, publish_quota, timeout=DUT_EVENT_TIMEOUT)
+
+            publish_from_dut(client, topic, qos, payload_prefix=deferred_payload, message_count=1, enqueue=enqueue)
+            deferred_patterns = {b"Publish requested, msg_id=": 1}
+            if enqueue == 0:
+                deferred_patterns[b"Unable to publish now: maximum inflight messages reached"] = 1
+            expect_n(client, deferred_patterns, timeout=DUT_CMD_TIMEOUT)
+
+            # A direct QoS 0 round trip provides a broker/DUT synchronization
+            # point while the publish completion packets remain held.
+            publish_from_dut(client, topic, 0, payload_prefix=sentinel_payload, message_count=1)
+            expect_n(
+                client,
+                {
+                    b"Publish requested, msg_id=": 1,
+                    f"MQTT_EVENT_DATA_PAYLOAD {sentinel_payload}".encode(): 1,
+                },
+                timeout=DUT_EVENT_TIMEOUT,
+            )
+            # Retransmissions may produce duplicate completion packets while
+            # the original quota remains held.
+            assert broker.held_packet_count(completion_packet) >= publish_quota
+
+            broker.release_held_packets(completion_packet, count=1, keep_holding=True)
+            broker.wait_for_held_packets(completion_packet, publish_quota, timeout=DUT_EVENT_TIMEOUT)
+            expect_n(
+                client,
+                {
+                    b"MQTT_EVENT_PUBLISHED": 1,
+                    f"MQTT_EVENT_DATA_PAYLOAD {deferred_payload}".encode(): 1,
+                },
+                timeout=DUT_EVENT_TIMEOUT,
+            )
+        finally:
+            broker.release_held_packets(completion_packet)
+
+        expect_n(client, {b"MQTT_EVENT_PUBLISHED": publish_quota}, timeout=DUT_EVENT_TIMEOUT)
+
+
+@pytest.mark.eth_ip101
+@pytest.mark.timeout(
+    case_timeout(
+        connect_operations=1,
+        subscribe_operations=1,
+        publish_operations=4,
+    )
+)
+@idf_parametrize("target", ["esp32"], indirect=["target"])
+def test_mqtt5_server_receive_maximum_mixed_qos(dut: Dut) -> None:
+    """QoS 1 and QoS 2 PUBLISH packets consume one shared inflight quota."""
+    topic = build_topic()
+    deferred_payload = "mixed_deferred"
+    sentinel_payload = "mixed_sentinel"
+
+    with (
+        broker_started(hold_packet_types=(MqttPacketType.PUBACK, MqttPacketType.PUBCOMP)) as broker,
+        initialized_mqtt_client(dut, broker.uri) as client,
+        started_client(client),
+    ):
+        subscribed_to(client, topic, 2)
+        try:
+            publish_from_dut(client, topic, 1, payload_prefix="mixed_q1", message_count=1)
+            publish_from_dut(client, topic, 2, payload_prefix="mixed_q2", message_count=1)
+            expect_n(
+                client,
+                {
+                    b"Publish requested, msg_id=": 2,
+                    b"MQTT_EVENT_DATA_PAYLOAD mixed_q1": 1,
+                    b"MQTT_EVENT_DATA_PAYLOAD mixed_q2": 1,
+                },
+                timeout=DUT_EVENT_TIMEOUT,
+            )
+            broker.wait_for_held_packets(MqttPacketType.PUBACK, 1, timeout=DUT_EVENT_TIMEOUT)
+            broker.wait_for_held_packets(MqttPacketType.PUBCOMP, 1, timeout=DUT_EVENT_TIMEOUT)
+
+            publish_from_dut(client, topic, 1, payload_prefix=deferred_payload, message_count=1)
+            expect_n(
+                client,
+                {
+                    b"Publish requested, msg_id=": 1,
+                    b"Unable to publish now: maximum inflight messages reached": 1,
+                },
+                timeout=DUT_CMD_TIMEOUT,
+            )
+            publish_from_dut(client, topic, 0, payload_prefix=sentinel_payload, message_count=1)
+            expect_n(
+                client,
+                {
+                    b"Publish requested, msg_id=": 1,
+                    f"MQTT_EVENT_DATA_PAYLOAD {sentinel_payload}".encode(): 1,
+                },
+                timeout=DUT_EVENT_TIMEOUT,
+            )
+            # Retransmissions may add duplicate completion packets while both
+            # original inflight messages remain held.
+            assert broker.held_packet_count(MqttPacketType.PUBACK) >= 1
+            assert broker.held_packet_count(MqttPacketType.PUBCOMP) >= 1
+
+            broker.release_held_packets(MqttPacketType.PUBCOMP, count=1, keep_holding=True)
+            broker.wait_for_held_packets(MqttPacketType.PUBACK, 2, timeout=DUT_EVENT_TIMEOUT)
+            expect_n(
+                client,
+                {
+                    b"MQTT_EVENT_PUBLISHED": 1,
+                    f"MQTT_EVENT_DATA_PAYLOAD {deferred_payload}".encode(): 1,
+                },
+                timeout=DUT_EVENT_TIMEOUT,
+            )
+        finally:
+            broker.release_held_packets(MqttPacketType.PUBACK)
+            broker.release_held_packets(MqttPacketType.PUBCOMP)
+
+        expect_n(client, {b"MQTT_EVENT_PUBLISHED": 2}, timeout=DUT_EVENT_TIMEOUT)
+
+
+@pytest.mark.eth_ip101
+@pytest.mark.timeout(
+    case_timeout(
+        connect_operations=1,
+        subscribe_operations=1,
+        publish_operations=4,
+        event_wait_operations=1,
+    )
+)
+@idf_parametrize("target", ["esp32"], indirect=["target"])
+@pytest.mark.parametrize(
+    "qos,completion_packet",
+    [(1, MqttPacketType.PUBACK), (2, MqttPacketType.PUBCOMP)],
+    ids=["qos1", "qos2"],
+)
+@pytest.mark.parametrize("ack_kind", ["unsolicited", "duplicate"])
+def test_mqtt5_unmatched_completion_does_not_release_receive_maximum(
+    dut: Dut,
+    qos: int,
+    completion_packet: MqttPacketType,
+    ack_kind: str,
+) -> None:
+    """Only an acknowledgement matching a live inflight PUBLISH releases quota."""
+    topic = build_topic()
+    blocked_payload = f"{ack_kind}_blocked"
+    probe_payload = f"{ack_kind}_probe"
+    sentinel_payload = f"{ack_kind}_sentinel"
+
+    with (
+        broker_started(receive_maximum=1, hold_packet_types=(completion_packet,)) as broker,
+        initialized_mqtt_client(dut, broker.uri) as client,
+        started_client(client),
+    ):
+        subscribed_to(client, topic, qos)
+        try:
+            publish_from_dut(client, topic, qos, payload_prefix="active", message_count=1)
+            expect_n(
+                client,
+                {b"Publish requested, msg_id=": 1, b"MQTT_EVENT_DATA_PAYLOAD active": 1},
+                timeout=DUT_EVENT_TIMEOUT,
+            )
+            broker.wait_for_held_packets(completion_packet, 1, timeout=DUT_EVENT_TIMEOUT)
+            active_id = broker.held_packet_identifiers(completion_packet)[0]
+
+            publish_from_dut(client, topic, qos, payload_prefix=blocked_payload, message_count=1, enqueue=1)
+            expect_n(client, {b"Publish requested, msg_id=": 1}, timeout=DUT_CMD_TIMEOUT)
+
+            if ack_kind == "unsolicited":
+                unused_id = 0xFFFF if active_id != 0xFFFF else 0xFFFE
+                broker.inject_ack(completion_packet, unused_id)
+                forbidden_payload = blocked_payload
+            else:
+                broker.release_held_packets(completion_packet, count=1, keep_holding=True)
+                broker.wait_for_held_packets(completion_packet, 1, timeout=DUT_EVENT_TIMEOUT)
+                expect_n(
+                    client,
+                    {
+                        b"MQTT_EVENT_PUBLISHED": 1,
+                        f"MQTT_EVENT_DATA_PAYLOAD {blocked_payload}".encode(): 1,
+                    },
+                    timeout=DUT_EVENT_TIMEOUT,
+                )
+                broker.inject_ack(completion_packet, active_id)
+                publish_from_dut(client, topic, qos, payload_prefix=probe_payload, message_count=1, enqueue=1)
+                expect_n(client, {b"Publish requested, msg_id=": 1}, timeout=DUT_CMD_TIMEOUT)
+                forbidden_payload = probe_payload
+
+            publish_from_dut(client, topic, 0, payload_prefix=sentinel_payload, message_count=1)
+            expect_n(
+                client,
+                {
+                    b"Publish requested, msg_id=": 1,
+                    f"MQTT_EVENT_DATA_PAYLOAD {sentinel_payload}".encode(): 1,
+                },
+                timeout=DUT_EVENT_TIMEOUT,
+                forbidden=QUOTA_REJECTION_FORBIDDEN + (f"MQTT_EVENT_DATA_PAYLOAD {forbidden_payload}".encode(),),
+            )
+        finally:
+            broker.release_all_held_packets()
+
+
+@pytest.mark.eth_ip101
+@pytest.mark.timeout(
+    case_timeout(
+        connect_operations=2,
+        subscribe_operations=2,
+        publish_operations=5,
+        event_wait_operations=2,
+    )
+)
+@idf_parametrize("target", ["esp32"], indirect=["target"])
+def test_mqtt5_reconnect_applies_receive_maximum_to_retransmits(dut: Dut) -> None:
+    """Retransmitted PUBLISH packets consume the newly negotiated connection quota."""
+    topic = build_topic()
+    probe_payload = "reconnect_probe"
+    sentinel_payload = "reconnect_sentinel"
+
+    with (
+        broker_started(receive_maximum=2, hold_packet_types=(MqttPacketType.PUBACK,)) as broker,
+        initialized_mqtt_client(dut, broker.uri) as client,
+        started_client(client),
+    ):
+        subscribed_to(client, topic, 1)
+        try:
+            publish_from_dut(client, topic, 1, payload_prefix="reconnect_active", message_count=2)
+            expect_n(
+                client,
+                {b"Publish requested, msg_id=": 2} | data_payload_patterns("reconnect_active", 2),
+                timeout=DUT_EVENT_TIMEOUT,
+            )
+            broker.wait_for_held_packets(MqttPacketType.PUBACK, 2, timeout=DUT_EVENT_TIMEOUT)
+
+            broker.discard_held_packets()
+            broker.set_receive_maximum(1)
+            broker.disconnect_clients()
+            client.expect(re.compile(rb"MQTT_EVENT_DISCONNECTED"), timeout=DUT_EVENT_TIMEOUT)
+            client.write("reconnect")
+            client.expect(re.compile(rb"MQTT_EVENT_CONNECTED"), timeout=DUT_CONNECT_TIMEOUT)
+            subscribed_to(client, topic, 1)
+
+            broker.wait_for_held_packets(MqttPacketType.PUBACK, 1, timeout=DUT_EVENT_TIMEOUT)
+            publish_from_dut(client, topic, 1, payload_prefix=probe_payload, message_count=1, enqueue=1)
+            expect_n(client, {b"Publish requested, msg_id=": 1}, timeout=DUT_CMD_TIMEOUT)
+
+            publish_from_dut(client, topic, 0, payload_prefix=sentinel_payload, message_count=1)
+            expect_n(
+                client,
+                {
+                    b"Publish requested, msg_id=": 1,
+                    f"MQTT_EVENT_DATA_PAYLOAD {sentinel_payload}".encode(): 1,
+                },
+                timeout=DUT_EVENT_TIMEOUT,
+                forbidden=QUOTA_REJECTION_FORBIDDEN + (f"MQTT_EVENT_DATA_PAYLOAD {probe_payload}".encode(),),
+            )
+            assert broker.held_packet_count(MqttPacketType.PUBACK) == 1
+        finally:
+            broker.release_all_held_packets()
+
+
+@pytest.mark.eth_ip101
+@pytest.mark.timeout(
+    case_timeout(
+        connect_operations=2,
+        subscribe_operations=1,
+        publish_operations=2,
+        event_wait_operations=2,
+    )
+)
+@idf_parametrize("target", ["esp32"], indirect=["target"])
+def test_mqtt5_reconnect_resends_requeued_packets_once(dut: Dut) -> None:
+    """A packet requeued on reconnect is sent once and does not block the outbox.
+
+    This pins current behavior, which knowingly resends more than [MQTT-4.4.0-1]
+    allows; see test_mqtt5_reconnect_resends_only_inflight_publishes__sec_4_4 for
+    the conformant end state. What must hold either way is that requeuing does
+    not break the send path:
+
+    - the requeued SUBSCRIBE leaves QUEUED after being sent, instead of being
+      redelivered on every task loop pass and starving the packets behind it,
+    - its outbox tick follows the send, so the retransmit path does not
+      immediately send it a second time,
+    - the requeued QoS 1 PUBLISH behind it still reaches the broker.
+
+    The broker answers every SUBSCRIBE it receives, so held SUBACKs count the
+    resends. One retransmit timeout (1s by default) after the reconnect the
+    client legitimately retransmits again, which is why the count is checked as
+    soon as the PUBLISH arrives.
     """
     topic = build_topic()
 
-    start_client(mqtt_client)
-    mqtt_client.write(f"subscribe {topic} 1")
-    mqtt_client.expect(re.compile(rb"MQTT_EVENT_SUBSCRIBED"), timeout=DUT_SUBSCRIBE_TIMEOUT)
+    with (
+        broker_started(hold_packet_types=(MqttPacketType.SUBACK, MqttPacketType.PUBACK)) as broker,
+        initialized_mqtt_client(dut, broker.uri) as client,
+        started_client(client),
+    ):
+        try:
+            client.write(f"subscribe {topic} 1")
+            expect_n(client, {b"Subscribe requested, msg_id=": 1}, timeout=DUT_CMD_TIMEOUT)
+            broker.wait_for_held_packets(MqttPacketType.SUBACK, 1, timeout=DUT_EVENT_TIMEOUT)
 
-    mqtt_client.write(f"publish {topic} qos1 4 1 0 1")
-    # DUT may emit DATA_COMPLETE (incoming) before PUBLISHED (outgoing ack); accept either order.
-    mqtt_client.expect(
-        [re.compile(rb"MQTT_EVENT_PUBLISHED"), re.compile(rb"MQTT_EVENT_DATA_COMPLETE")],
-        timeout=DUT_TEST_TIMEOUT,
-    )
-    mqtt_client.expect(
-        [re.compile(rb"MQTT_EVENT_PUBLISHED"), re.compile(rb"MQTT_EVENT_DATA_COMPLETE")],
-        timeout=DUT_TEST_TIMEOUT,
-    )
+            publish_from_dut(client, topic, 1, payload_prefix="requeued", message_count=1)
+            expect_n(client, {b"Publish requested, msg_id=": 1}, timeout=DUT_CMD_TIMEOUT)
+            broker.wait_for_held_packets(MqttPacketType.PUBACK, 1, timeout=DUT_EVENT_TIMEOUT)
 
-    stop_client(mqtt_client)
+            # packets held for the old socket are useless once it is gone
+            broker.discard_held_packets()
+            broker.disconnect_clients()
+            client.expect(re.compile(rb"MQTT_EVENT_DISCONNECTED"), timeout=DUT_EVENT_TIMEOUT)
+            client.write("reconnect")
+            client.expect(re.compile(rb"MQTT_EVENT_CONNECTED"), timeout=DUT_CONNECT_TIMEOUT)
+
+            # The SUBSCRIBE was enqueued first, so it is requeued and sent first.
+            # Reaching the PUBLISH at all proves it was not starved behind it.
+            broker.wait_for_held_packets(MqttPacketType.PUBACK, 1, timeout=DUT_EVENT_TIMEOUT)
+            assert broker.held_packet_count(MqttPacketType.SUBACK) == 1
+        finally:
+            broker.release_all_held_packets()
+
+
+@pytest.mark.eth_ip101
+@pytest.mark.xfail(
+    reason="Known failure: the client requeues every unacknowledged packet on reconnect, "
+    "not just QoS>0 PUBLISH, and the periodic retransmit resends any TRANSMITTED packet "
+    "regardless of type. Fixing it changes long-standing behavior, so it is handled separately.",
+    strict=True,
+)
+@pytest.mark.timeout(
+    case_timeout(
+        connect_operations=2,
+        subscribe_operations=1,
+        publish_operations=2,
+        event_wait_operations=2,
+    )
+)
+@idf_parametrize("target", ["esp32"], indirect=["target"])
+def test_mqtt5_reconnect_resends_only_inflight_publishes__sec_4_4(dut: Dut) -> None:
+    """Only unacknowledged QoS>0 PUBLISH packets are resent on a new connection.
+
+    MQTT5 4.4: resending unacknowledged PUBLISH (QoS > 0) and PUBREL packets is
+    "the only circumstance where a Client or Server is REQUIRED to resend
+    messages. Clients and Servers MUST NOT resend messages at any other time"
+    [MQTT-4.4.0-1].
+
+    Holding the SUBACK leaves a SUBSCRIBE unacknowledged across the reconnect.
+    The broker answers every SUBSCRIBE it receives, so a resent one would show
+    up as a second held SUBACK.
+    """
+    topic = build_topic()
+
+    with (
+        broker_started(hold_packet_types=(MqttPacketType.SUBACK, MqttPacketType.PUBACK)) as broker,
+        initialized_mqtt_client(dut, broker.uri) as client,
+        started_client(client),
+    ):
+        try:
+            client.write(f"subscribe {topic} 1")
+            expect_n(client, {b"Subscribe requested, msg_id=": 1}, timeout=DUT_CMD_TIMEOUT)
+            broker.wait_for_held_packets(MqttPacketType.SUBACK, 1, timeout=DUT_EVENT_TIMEOUT)
+
+            publish_from_dut(client, topic, 1, payload_prefix="inflight", message_count=1)
+            expect_n(client, {b"Publish requested, msg_id=": 1}, timeout=DUT_CMD_TIMEOUT)
+            broker.wait_for_held_packets(MqttPacketType.PUBACK, 1, timeout=DUT_EVENT_TIMEOUT)
+
+            # packets held for the old socket are useless once it is gone
+            broker.discard_held_packets()
+            broker.disconnect_clients()
+            client.expect(re.compile(rb"MQTT_EVENT_DISCONNECTED"), timeout=DUT_EVENT_TIMEOUT)
+            client.write("reconnect")
+            client.expect(re.compile(rb"MQTT_EVENT_CONNECTED"), timeout=DUT_CONNECT_TIMEOUT)
+
+            # The SUBSCRIBE precedes the PUBLISH in the outbox, so once the
+            # retransmitted PUBLISH arrives any resent SUBSCRIBE would already
+            # have been answered.
+            broker.wait_for_held_packets(MqttPacketType.PUBACK, 1, timeout=DUT_EVENT_TIMEOUT)
+            assert broker.held_packet_count(MqttPacketType.SUBACK) == 0
+        finally:
+            broker.release_all_held_packets()
+
+
+@pytest.mark.eth_ip101
+@pytest.mark.timeout(
+    case_timeout(
+        connect_operations=1,
+        subscribe_operations=2,
+        publish_operations=DEFAULT_BROKER_RECEIVE_MAXIMUM,
+    )
+)
+@idf_parametrize("target", ["esp32"], indirect=["target"])
+def test_mqtt5_subscribe_not_delayed_by_receive_maximum(dut: Dut) -> None:
+    """SUBSCRIBE must not be delayed by inflight publish quota.
+
+    MQTT5 §4.9: "The Client MUST NOT delay the sending of any packets
+    other than PUBLISH packets due to having sent Receive Maximum publish
+    packets without receiving acknowledgements for them."
+
+    The SUBACK must arrive while the broker is still holding all QoS 1 PUBACKs.
+    """
+    topic_pub = build_topic()
+    topic_sub = build_topic()
+    publish_quota = DEFAULT_BROKER_RECEIVE_MAXIMUM
+
+    with (
+        broker_started(hold_packet_types=(MqttPacketType.PUBACK,)) as broker,
+        initialized_mqtt_client(dut, broker.uri) as client,
+        started_client(client),
+    ):
+        subscribed_to(client, topic_pub, 1)
+        try:
+            publish_from_dut(client, topic_pub, 1, payload_prefix="subscribe_sat", message_count=publish_quota)
+            expect_n(
+                client,
+                {b"Publish requested, msg_id=": publish_quota} | data_payload_patterns("subscribe_sat", publish_quota),
+                timeout=DUT_EVENT_TIMEOUT,
+            )
+            broker.wait_for_held_packets(MqttPacketType.PUBACK, publish_quota, timeout=DUT_EVENT_TIMEOUT)
+
+            subscribed_to(client, topic_sub, 1)
+            # Retransmissions can produce additional held PUBACKs for the same
+            # inflight packets; none of the original quota has been released.
+            assert broker.held_packet_count(MqttPacketType.PUBACK) >= publish_quota
+        finally:
+            broker.release_held_packets(MqttPacketType.PUBACK)
+
+        expect_n(client, {b"MQTT_EVENT_PUBLISHED": publish_quota}, timeout=DUT_EVENT_TIMEOUT)
+
+
+@pytest.mark.eth_ip101
+@pytest.mark.timeout(
+    case_timeout(
+        connect_operations=1,
+        subscribe_operations=2,
+        publish_operations=DEFAULT_BROKER_RECEIVE_MAXIMUM + 2,
+    )
+)
+@idf_parametrize("target", ["esp32"], indirect=["target"])
+@pytest.mark.parametrize(
+    "qos0_enqueue",
+    [
+        pytest.param(False, id="publish"),
+        pytest.param(True, id="enqueue"),
+    ],
+)
+def test_mqtt5_qos0_not_blocked_by_quota(dut: Dut, qos0_enqueue: bool) -> None:
+    """QoS 0 must bypass a quota-blocked QoS 1 outbox head through either API."""
+    topic_q1 = build_topic()
+    topic_q0 = build_topic()
+    publish_quota = DEFAULT_BROKER_RECEIVE_MAXIMUM  # exactly fills the broker quota
+    blocked_payload = "blocked_qos1"
+    qos0_payload = "qos0_bypass"
+
+    with (
+        broker_started(hold_packet_types=(MqttPacketType.PUBACK,)) as broker,
+        initialized_mqtt_client(dut, broker.uri) as client,
+        started_client(client),
+    ):
+        subscribed_to(client, topic_q1, 1)
+        subscribed_to(client, topic_q0, 0)
+
+        try:
+            publish_from_dut(
+                client,
+                topic_q1,
+                1,
+                payload_prefix="qos0_sat",
+                message_count=publish_quota,
+                enqueue=1,
+            )
+            expect_n(
+                client,
+                {b"Publish requested, msg_id=": publish_quota} | data_payload_patterns("qos0_sat", publish_quota),
+                timeout=DUT_EVENT_TIMEOUT,
+            )
+            broker.wait_for_held_packets(MqttPacketType.PUBACK, publish_quota, timeout=DUT_EVENT_TIMEOUT)
+
+            publish_from_dut(client, topic_q1, 1, payload_prefix=blocked_payload, message_count=1, enqueue=1)
+            expect_n(client, {b"Publish requested, msg_id=": 1}, timeout=DUT_CMD_TIMEOUT)
+
+            publish_from_dut(
+                client,
+                topic_q0,
+                0,
+                payload_prefix=qos0_payload,
+                message_count=1,
+                enqueue=int(qos0_enqueue),
+            )
+            expect_n(
+                client,
+                {
+                    b"Publish requested, msg_id=": 1,
+                    f"MQTT_EVENT_DATA_PAYLOAD {qos0_payload}".encode(): 1,
+                },
+                timeout=DUT_CMD_TIMEOUT,
+            )
+            # Retransmissions can produce additional held PUBACKs for the same
+            # inflight packets; none of the original quota has been released.
+            assert broker.held_packet_count(MqttPacketType.PUBACK) >= publish_quota
+        finally:
+            broker.release_held_packets(MqttPacketType.PUBACK)
+
+        expect_n(client, {b"MQTT_EVENT_PUBLISHED": publish_quota + 1}, timeout=DUT_EVENT_TIMEOUT)
+
+
+@pytest.mark.eth_ip101
+@pytest.mark.timeout(
+    case_timeout(
+        connect_operations=1,
+        subscribe_operations=1,
+        event_wait_operations=2,
+    )
+)
+@idf_parametrize("target", ["esp32"], indirect=["target"])
+@pytest.mark.parametrize(
+    "protocol_ver",
+    [MQTT_PROTOCOL_V_3_1_1, MQTT_PROTOCOL_V_5],
+    ids=["v311", "v5"],
+)
+def test_subscribe_and_qos1_publish__sec_3_8_4_and_4_3(dut: Dut, protocol_ver: int) -> None:
+    """
+    Base subscribe/QoS 1 publish conformance case, run against both protocol versions.
+    Section numbers are identical in both specs:
+    - section 3.8.4: SUBSCRIBE Actions (SUBACK interaction)
+    - section 4.3: Quality of Service levels and protocol flows (QoS 1: at least once semantics)
+
+    """
+    topic = build_topic()
+
+    with (
+        broker_started() as broker,
+        initialized_mqtt_client(dut, broker.uri, protocol_ver=protocol_ver) as client,
+        started_client(client),
+    ):
+        subscribed_to(client, topic, 1)
+
+        publish_from_dut(
+            client,
+            topic,
+            1,
+            payload_prefix="qos1",
+            message_count=1,
+            enqueue=1,
+            pattern_repetitions=4,
+        )
+        expect_n(
+            client,
+            {
+                b"MQTT_EVENT_PUBLISHED": 1,
+                b"MQTT_EVENT_DATA_COMPLETE": 1,
+            },
+            timeout=DUT_EVENT_TIMEOUT,
+        )
diff --git a/test/host/main/CMakeLists.txt b/test/host/main/CMakeLists.txt
index be8879b..d588967 100644
--- a/test/host/main/CMakeLists.txt
+++ b/test/host/main/CMakeLists.txt
@@ -1,4 +1,4 @@
-idf_component_register(SRCS  "test_mqtt_client.cpp" "test_log_intercept.cpp" "test_log_matchers.cpp" "test_log_parser.cpp"
+idf_component_register(SRCS  "test_mqtt_client.cpp" "test_mqtt5_client.cpp" "mqtt5_client_test_adapter.c" "test_log_intercept.cpp" "test_log_matchers.cpp" "test_log_parser.cpp"
                        REQUIRES cmock mqtt esp_timer esp_hw_support http_parser log
                        WHOLE_ARCHIVE)
 
diff --git a/test/host/main/mqtt5_client_test_adapter.c b/test/host/main/mqtt5_client_test_adapter.c
new file mode 100644
index 0000000..20ffea5
--- /dev/null
+++ b/test/host/main/mqtt5_client_test_adapter.c
@@ -0,0 +1,27 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+#include <stdint.h>
+
+#include "mqtt_client_priv.h"
+
+esp_err_t test_mqtt5_check_inflight_maximum(uint16_t send_count, uint16_t receive_maximum)
+{
+    struct esp_mqtt_client client = {0};
+    mqtt5_config_storage_t mqtt5_config = {0};
+    client.mqtt5_config = &mqtt5_config;
+    client.mqtt5_config->server_resp_property_info.receive_maximum = receive_maximum;
+    client.send_publish_packet_count = send_count;
+    return esp_mqtt5_client_check_inflight_maximum(&client);
+}
+
+int test_mqtt5_increment_packet_counter_with_dup(void)
+{
+    struct esp_mqtt_client client = {0};
+    uint8_t publish_header[] = {0x3a}; // PUBLISH, DUP=1, QoS=1
+    client.mqtt_state.connection.outbound_message.data = publish_header;
+    esp_mqtt5_increment_packet_counter(&client);
+    return client.send_publish_packet_count;
+}
diff --git a/test/host/main/test_mqtt5_client.cpp b/test/host/main/test_mqtt5_client.cpp
new file mode 100644
index 0000000..47b5262
--- /dev/null
+++ b/test/host/main/test_mqtt5_client.cpp
@@ -0,0 +1,25 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+#include <catch2/catch_test_macros.hpp>
+#include <cstdint>
+
+#include "esp_err.h"
+
+extern "C" {
+    esp_err_t test_mqtt5_check_inflight_maximum(uint16_t send_count, uint16_t receive_maximum);
+    int test_mqtt5_increment_packet_counter_with_dup(void);
+}
+
+TEST_CASE("MQTT5 inflight quota uses an exact upper bound")
+{
+    REQUIRE(test_mqtt5_check_inflight_maximum(1, 2) == ESP_OK);
+    REQUIRE(test_mqtt5_check_inflight_maximum(2, 2) == ESP_FAIL);
+}
+
+TEST_CASE("MQTT5 first send on a connection counts even when PUBLISH has DUP set")
+{
+    REQUIRE(test_mqtt5_increment_packet_counter_with_dup() == 1);
+}
diff --git a/test/host/sdkconfig.defaults b/test/host/sdkconfig.defaults
index cf0a16d..559f27f 100644
--- a/test/host/sdkconfig.defaults
+++ b/test/host/sdkconfig.defaults
@@ -1,4 +1,5 @@
 CONFIG_IDF_TARGET="linux"
+CONFIG_MQTT_PROTOCOL_5=y
 CONFIG_LOG_DEFAULT_LEVEL_DEBUG=y
 CONFIG_COMPILER_CXX_EXCEPTIONS=y
 CONFIG_COMPILER_CXX_RTTI=y
diff --git a/test/mqtt_outbox_host_test/main/test_outbox.cpp b/test/mqtt_outbox_host_test/main/test_outbox.cpp
index f1bc1ff..579f935 100644
--- a/test/mqtt_outbox_host_test/main/test_outbox.cpp
+++ b/test/mqtt_outbox_host_test/main/test_outbox.cpp
@@ -204,6 +204,31 @@ TEST_CASE("Outbox lookup by msg_id")
         outbox_enqueue(outbox.handle, &message, 0);
         REQUIRE(outbox_get(outbox.handle, 999) == nullptr);
     }
+    SECTION("msg_id zero finds queued QoS 0 behind a QoS 1 head") {
+        auto qos1 = make_msg(1, 1, 3, "qos1", 4);
+        auto qos0_first = make_msg(0, 0, 3, "first", 5);
+        auto qos0_second = make_msg(0, 0, 3, "second", 6);
+        outbox_enqueue(outbox.handle, &qos1, 0);
+        outbox_enqueue(outbox.handle, &qos0_first, 0);
+        outbox_enqueue(outbox.handle, &qos0_second, 0);
+        REQUIRE(outbox_dequeue(outbox.handle, QUEUED, nullptr) == outbox_get(outbox.handle, 1));
+        outbox_item_handle_t item = outbox_get(outbox.handle, 0);
+        REQUIRE(item != nullptr);
+        REQUIRE(outbox_item_get_pending(item) == QUEUED);
+        uint16_t id;
+        int type, qos;
+        size_t len;
+        auto *data = outbox_item_get_data(item, &len, &id, &type, &qos);
+        REQUIRE(id == 0);
+        REQUIRE(type == 3);
+        REQUIRE(qos == 0);
+        REQUIRE(std::string(reinterpret_cast<char *>(data), len) == "first");
+        REQUIRE(outbox_delete_item(outbox.handle, item) == ESP_OK);
+        item = outbox_get(outbox.handle, 0);
+        REQUIRE(item != nullptr);
+        data = outbox_item_get_data(item, &len, &id, &type, &qos);
+        REQUIRE(std::string(reinterpret_cast<char *>(data), len) == "second");
+    }
 }
 
 TEST_CASE("Outbox delete by msg_id and type")