diff --git a/ZB.MOM.WW.OtOpcUa.slnx b/ZB.MOM.WW.OtOpcUa.slnx
index 02a8700d..7c5147f0 100644
--- a/ZB.MOM.WW.OtOpcUa.slnx
+++ b/ZB.MOM.WW.OtOpcUa.slnx
@@ -108,6 +108,7 @@
     <Project Path="tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser.Tests/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser.Tests.csproj" />
     <Project Path="tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser.IntegrationTests.csproj" />
     <Project Path="tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests.csproj" />
+    <Project Path="tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests.csproj" />
   </Folder>
   <Folder Name="/tests/Drivers/Driver CLIs/">
     <Project Path="tests/Drivers/Cli/ZB.MOM.WW.OtOpcUa.Driver.Cli.Common.Tests/ZB.MOM.WW.OtOpcUa.Driver.Cli.Common.Tests.csproj" />
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/.gitignore b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/.gitignore
new file mode 100644
index 00000000..b69c261a
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/.gitignore
@@ -0,0 +1,14 @@
+# Fixture credential + TLS material produced by gen-fixture-material.sh.
+#
+# NEVER commit anything in here: it holds the broker password hash, the fixture CA's
+# PRIVATE key and the server's private key. The generator script is the artifact in
+# git; its output is regenerated per machine.
+secrets/
+
+# Belt-and-braces: catch key/cert material written directly into this directory by a
+# hand-run openssl command that forgot the secrets/ prefix.
+*.key
+*.crt
+*.csr
+*.srl
+passwd
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/docker-compose.yml b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/docker-compose.yml
new file mode 100644
index 00000000..a1ecfec7
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/docker-compose.yml
@@ -0,0 +1,77 @@
+# MQTT integration-test fixture — Eclipse Mosquitto (auth + TLS) + a JSON publisher.
+#
+# PREREQUISITE — run this first, in this directory:
+#
+#   MQTT_FIXTURE_PASSWORD='<pick-one>' ./gen-fixture-material.sh
+#
+# It writes ./secrets/ (password file + CA + server certificate/key), which is
+# gitignored and which both services below mount. Without it the broker will not
+# start: there is no anonymous fallback, by design.
+#
+# Bring up (on the shared Docker host — see infra/README.md §1):
+#
+#   ssh dohertj2@10.100.0.35 'cd /opt/otopcua-mqtt \
+#       && MQTT_FIXTURE_USERNAME=otopcua MQTT_FIXTURE_PASSWORD=<same> docker compose up -d'
+#
+# Endpoints: 10.100.0.35:8883 (TLS + auth) · 10.100.0.35:1883 (plaintext + auth, smoke)
+#
+# Unlike the Modbus / S7 fixtures there are no compose profiles: both services are
+# always wanted together (a broker with nothing publishing into it tests nothing), and
+# they bind different ports so nothing contends.
+services:
+  mosquitto:
+    image: eclipse-mosquitto:2.0.22
+    container_name: otopcua-mosquitto
+    restart: unless-stopped
+    # Every lmxopcua fixture container carries this so the shared Docker host stays
+    # discoverable with `docker ps --filter label=project=lmxopcua`.
+    labels:
+      project: lmxopcua
+    ports:
+      - "1883:1883"
+      - "8883:8883"
+    volumes:
+      - ./mosquitto.conf:/mosquitto/config/mosquitto.conf:ro
+      - ./secrets:/mosquitto/secrets:ro
+    healthcheck:
+      # Authenticated round-trip against the broker's own $SYS tree ($$ escapes the
+      # compose variable syntax). Proves the password file loaded and the listener
+      # accepts a real session — a bare port check would go green on a broker that
+      # rejects every CONNECT.
+      test:
+        - CMD-SHELL
+        - >-
+          mosquitto_sub -h 127.0.0.1 -p 1883
+          -u "$$MQTT_FIXTURE_USERNAME" -P "$$MQTT_FIXTURE_PASSWORD"
+          -t '$$SYS/broker/uptime' -C 1 -W 3
+      interval: 5s
+      timeout: 8s
+      retries: 12
+      start_period: 5s
+    environment:
+      # Consumed only by the healthcheck above; the broker itself reads the hashed
+      # password file. Supplied by the operator's environment at `docker compose up` —
+      # never defaulted here, so a missing value fails loudly instead of silently
+      # provisioning a known-password broker.
+      MQTT_FIXTURE_USERNAME: ${MQTT_FIXTURE_USERNAME:?set MQTT_FIXTURE_USERNAME (must match gen-fixture-material.sh)}
+      MQTT_FIXTURE_PASSWORD: ${MQTT_FIXTURE_PASSWORD:?set MQTT_FIXTURE_PASSWORD (must match gen-fixture-material.sh)}
+
+  publisher:
+    image: eclipse-mosquitto:2.0.22
+    container_name: otopcua-mqtt-publisher
+    restart: unless-stopped
+    labels:
+      project: lmxopcua
+    depends_on:
+      mosquitto:
+        condition: service_healthy
+    volumes:
+      - ./publisher/publish.sh:/publish.sh:ro
+    environment:
+      BROKER_HOST: mosquitto
+      BROKER_PORT: "1883"
+      TOPIC_PREFIX: ${MQTT_FIXTURE_TOPIC_PREFIX:-otopcua/fixture}
+      PUBLISH_INTERVAL: ${MQTT_FIXTURE_PUBLISH_INTERVAL:-2}
+      MQTT_FIXTURE_USERNAME: ${MQTT_FIXTURE_USERNAME:?set MQTT_FIXTURE_USERNAME}
+      MQTT_FIXTURE_PASSWORD: ${MQTT_FIXTURE_PASSWORD:?set MQTT_FIXTURE_PASSWORD}
+    entrypoint: ["/bin/sh", "/publish.sh"]
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/gen-fixture-material.sh b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/gen-fixture-material.sh
new file mode 100755
index 00000000..3b72538f
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/gen-fixture-material.sh
@@ -0,0 +1,137 @@
+#!/usr/bin/env sh
+# ---------------------------------------------------------------------------
+# Generates the MQTT fixture's password file + TLS material into ./secrets/.
+#
+# NOTHING this script writes is committed: ./secrets/ is excluded by the sibling
+# .gitignore. The SCRIPT is the artifact in git, the OUTPUT never is. Re-run it on any
+# machine that needs the fixture (or on the Docker host); the material is throwaway by
+# construction — a self-issued CA, a 2048-bit server key, and whatever password the
+# operator supplies in the environment.
+#
+# Usage:
+#   MQTT_FIXTURE_PASSWORD='<pick-one>' ./gen-fixture-material.sh
+#
+# Environment:
+#   MQTT_FIXTURE_PASSWORD   (required) broker password. No default — a committed or
+#                           defaulted password is exactly what the plan's
+#                           "secrets from env, never committed" rule forbids.
+#   MQTT_FIXTURE_USERNAME   broker username                       (default: otopcua)
+#   MQTT_FIXTURE_SAN_IPS    comma-separated IP SANs for the server certificate
+#                           (default: 10.100.0.35,127.0.0.1)
+#   MQTT_FIXTURE_SAN_DNS    comma-separated DNS SANs
+#                           (default: localhost,mosquitto)
+#   MQTT_FIXTURE_CERT_CN    server certificate CN         (default: first IP SAN)
+#   MQTT_FIXTURE_CERT_DAYS  certificate lifetime in days             (default: 825)
+#   MQTT_FIXTURE_SECRETS_DIR  output directory        (default: <script dir>/secrets)
+#   MOSQUITTO_IMAGE         image used to run mosquitto_passwd
+#                                                 (default: eclipse-mosquitto:2.0.22)
+#
+# The SANs matter: MqttConnection pins the TLS target host with
+# `.WithTargetHost(options.Host)`, so hostname verification is enforced and the
+# certificate MUST carry a SAN matching whatever host the tests dial — the shared
+# Docker host's IP literal (10.100.0.35) in the normal case. A cert without the right
+# SAN fails the handshake even with a correct CA pin, and the failure looks like a pin
+# bug.
+#
+# Requires: openssl, docker (only for hashing the password — mosquitto 2.0 will not
+# read a plaintext password file, and mosquitto_passwd is the only supported way to
+# produce its PBKDF2-SHA512 format).
+# ---------------------------------------------------------------------------
+set -eu
+
+SCRIPT_DIR=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
+OUT=${MQTT_FIXTURE_SECRETS_DIR:-"$SCRIPT_DIR/secrets"}
+USERNAME=${MQTT_FIXTURE_USERNAME:-otopcua}
+SAN_IPS=${MQTT_FIXTURE_SAN_IPS:-10.100.0.35,127.0.0.1}
+SAN_DNS=${MQTT_FIXTURE_SAN_DNS:-localhost,mosquitto}
+DAYS=${MQTT_FIXTURE_CERT_DAYS:-825}
+MOSQUITTO_IMAGE=${MOSQUITTO_IMAGE:-eclipse-mosquitto:2.0.22}
+
+if [ -z "${MQTT_FIXTURE_PASSWORD:-}" ]; then
+    echo "error: MQTT_FIXTURE_PASSWORD is not set." >&2
+    echo "       Supply the broker password in the environment; this script never" >&2
+    echo "       defaults or commits one. Example:" >&2
+    echo "         MQTT_FIXTURE_PASSWORD='\$(openssl rand -base64 18)' $0" >&2
+    exit 2
+fi
+
+command -v openssl >/dev/null 2>&1 || { echo "error: openssl not found on PATH." >&2; exit 2; }
+command -v docker  >/dev/null 2>&1 || {
+    echo "error: docker not found on PATH — needed to run mosquitto_passwd." >&2
+    echo "       Run this script on the Docker host (10.100.0.35) instead, or install docker." >&2
+    exit 2
+}
+
+CN=${MQTT_FIXTURE_CERT_CN:-$(printf '%s' "$SAN_IPS" | cut -d, -f1)}
+
+mkdir -p "$OUT"
+chmod 0755 "$OUT"
+
+echo "==> generating fixture material into $OUT"
+echo "    username : $USERNAME"
+echo "    cert CN  : $CN"
+echo "    IP SANs  : $SAN_IPS"
+echo "    DNS SANs : $SAN_DNS"
+
+# --- CA -------------------------------------------------------------------
+openssl req -x509 -newkey rsa:2048 -nodes \
+    -keyout "$OUT/ca.key" -out "$OUT/ca.crt" -days "$DAYS" \
+    -subj "/CN=OtOpcUa MQTT fixture CA/O=OtOpcUa test fixture" \
+    -addext "basicConstraints=critical,CA:TRUE,pathlen:0" \
+    -addext "keyUsage=critical,keyCertSign,cRLSign" \
+    >/dev/null 2>&1
+
+# --- server leaf ----------------------------------------------------------
+# The SAN list is assembled into a temp extfile rather than a process substitution,
+# which is a bashism this /bin/sh script cannot use.
+EXT=$(mktemp)
+trap 'rm -f "$EXT" "$OUT/server.csr" "$OUT/ca.srl"' EXIT
+
+SAN=""
+OLD_IFS=$IFS
+IFS=,
+for ip in $SAN_IPS;  do [ -n "$ip"  ] && SAN="$SAN,IP:$ip";   done
+for dn in $SAN_DNS;  do [ -n "$dn"  ] && SAN="$SAN,DNS:$dn";  done
+IFS=$OLD_IFS
+SAN=${SAN#,}
+
+cat >"$EXT" <<EOF
+subjectAltName = $SAN
+basicConstraints = critical,CA:FALSE
+keyUsage = critical,digitalSignature,keyEncipherment
+extendedKeyUsage = serverAuth
+EOF
+
+openssl req -newkey rsa:2048 -nodes \
+    -keyout "$OUT/server.key" -out "$OUT/server.csr" \
+    -subj "/CN=$CN/O=OtOpcUa test fixture" \
+    >/dev/null 2>&1
+
+openssl x509 -req -in "$OUT/server.csr" \
+    -CA "$OUT/ca.crt" -CAkey "$OUT/ca.key" -CAcreateserial \
+    -out "$OUT/server.crt" -days "$DAYS" -sha256 -extfile "$EXT" \
+    >/dev/null 2>&1
+
+# --- password file --------------------------------------------------------
+# `--user` so the file lands owned by the invoking user rather than by root: the host
+# user must be able to chmod it below, and (unlike a root-owned 0600 file) the broker
+# container's `mosquitto` user must be able to READ it.
+rm -f "$OUT/passwd"
+docker run --rm --user "$(id -u):$(id -g)" -v "$OUT":/out "$MOSQUITTO_IMAGE" \
+    mosquitto_passwd -b -c /out/passwd "$USERNAME" "$MQTT_FIXTURE_PASSWORD" >/dev/null
+
+# mosquitto runs as uid 1883 inside the container and reads these through a read-only
+# bind mount, so they must be world-readable. This is a throwaway fixture key by
+# construction — see the banner.
+chmod 0644 "$OUT/ca.crt" "$OUT/server.crt" "$OUT/server.key" "$OUT/passwd"
+chmod 0600 "$OUT/ca.key"
+
+echo "==> done. Files written:"
+ls -l "$OUT"
+echo
+echo "Next:"
+echo "  1. Export the same creds for the test suite:"
+echo "       export MQTT_FIXTURE_USERNAME='$USERNAME'"
+echo "       export MQTT_FIXTURE_PASSWORD='<the same password>'"
+echo "       export MQTT_FIXTURE_CA_CERT='$OUT/ca.crt'"
+echo "  2. Push this directory to the Docker host and bring the stack up."
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/mosquitto.conf b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/mosquitto.conf
new file mode 100644
index 00000000..ea771e03
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/mosquitto.conf
@@ -0,0 +1,65 @@
+# Eclipse Mosquitto configuration for the OtOpcUa MQTT driver integration fixture.
+#
+# SECURITY POSTURE — deliberate, and the whole reason this fixture exists.
+# `allow_anonymous false` is set GLOBALLY (with per_listener_settings false), so it
+# governs BOTH listeners. An anonymous fixture would let the driver's auth and TLS
+# code paths — including the CA-pin logic in MqttConnection.ConfigureTls /
+# ValidateAgainstPinnedCa — go completely untested, which is exactly backwards: those
+# are the paths whose failure mode is silent (a pin that never rejects anything looks
+# identical to a pin that works).
+#
+# The password file and the TLS material are NOT in git. Generate them first with
+# `./gen-fixture-material.sh` (see that script's banner); it writes ./secrets/, which
+# .gitignore excludes.
+
+# One global security policy rather than per-listener ones: with
+# per_listener_settings true, `allow_anonymous`/`password_file` would have to be
+# repeated under EVERY listener block, and a listener that forgot them would silently
+# fall back to allowing anonymous connections. Global + fail-closed is the safer shape.
+per_listener_settings false
+
+allow_anonymous false
+password_file /mosquitto/secrets/passwd
+
+# No persistence: the fixture is throwaway, and a persisted retained-message store
+# surviving a `docker compose down` would make the retained-seed test's outcome depend
+# on a previous run rather than on this one's publisher.
+persistence false
+
+log_dest stdout
+log_type error
+log_type warning
+log_type notice
+log_type information
+connection_messages true
+
+# ---------------------------------------------------------------------------
+# Plain (non-TLS) listener — smoke only.
+#
+# Still authenticated (allow_anonymous false above applies here too); it is the
+# TRANSPORT that is plaintext, not the access policy. It exists so the subscribe /
+# retained-seed / unauthored-topic legs can be exercised without dragging the TLS
+# handshake into every assertion, and so an operator can `mosquitto_sub` the fixture
+# by hand. Never a model for a deployment.
+# ---------------------------------------------------------------------------
+listener 1883 0.0.0.0
+protocol mqtt
+
+# ---------------------------------------------------------------------------
+# TLS listener — the real one.
+#
+# `require_certificate false`: the driver authenticates with username/password over
+# TLS (the shape MqttDriverOptions models today); mutual TLS is not a driver feature
+# yet, so demanding a client certificate here would make the fixture untestable rather
+# than more secure.
+#
+# tls_version is deliberately NOT pinned: mosquitto's `tls_version` selects exactly
+# one version, so pinning 1.2 would refuse a .NET client that negotiated 1.3. Leaving
+# it unset lets OpenSSL negotiate the best mutually-supported version.
+# ---------------------------------------------------------------------------
+listener 8883 0.0.0.0
+protocol mqtt
+cafile /mosquitto/secrets/ca.crt
+certfile /mosquitto/secrets/server.crt
+keyfile /mosquitto/secrets/server.key
+require_certificate false
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/publisher/publish.sh b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/publisher/publish.sh
new file mode 100755
index 00000000..5e8d9013
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/publisher/publish.sh
@@ -0,0 +1,67 @@
+#!/bin/sh
+# ---------------------------------------------------------------------------
+# The fixture's JSON publisher. Runs inside a second eclipse-mosquitto container (the
+# image already ships mosquitto_pub, so no bespoke image / paho dependency is needed)
+# and feeds the broker over the PLAIN listener from inside the compose network —
+# authenticated like every other client, since the broker is allow_anonymous false.
+#
+# Topics (all under $TOPIC_PREFIX, default otopcua/fixture):
+#
+#   retained/seed        published EXACTLY ONCE, retain=true, never republished.
+#                        This is what makes the retained-seed test meaningful: a
+#                        subscriber that connects minutes later and still receives a
+#                        value on this topic can only have received the BROKER'S
+#                        RETAINED COPY. On a topic that is also republished every few
+#                        seconds, a retained seed is indistinguishable from a live
+#                        publish that happened to arrive quickly.
+#   line1/temperature    JSON object, retain=true, republished every $PUBLISH_INTERVAL
+#   line1/counter        JSON object, retain=true, monotonically increasing
+#   line1/state          bare scalar text (no JSON), retain=true
+#   noise/chatter        JSON, retain=FALSE — deliberately never authored as a tag.
+#                        A broker carrying traffic the deployment did not author is
+#                        the normal case, and the driver must stay silent about it.
+# ---------------------------------------------------------------------------
+set -eu
+
+HOST=${BROKER_HOST:-mosquitto}
+PORT=${BROKER_PORT:-1883}
+PREFIX=${TOPIC_PREFIX:-otopcua/fixture}
+INTERVAL=${PUBLISH_INTERVAL:-2}
+
+: "${MQTT_FIXTURE_USERNAME:?publisher needs MQTT_FIXTURE_USERNAME}"
+: "${MQTT_FIXTURE_PASSWORD:?publisher needs MQTT_FIXTURE_PASSWORD}"
+
+pub_retained() {
+    mosquitto_pub -h "$HOST" -p "$PORT" \
+        -u "$MQTT_FIXTURE_USERNAME" -P "$MQTT_FIXTURE_PASSWORD" \
+        -q 1 -r -t "$1" -m "$2"
+}
+
+pub_volatile() {
+    mosquitto_pub -h "$HOST" -p "$PORT" \
+        -u "$MQTT_FIXTURE_USERNAME" -P "$MQTT_FIXTURE_PASSWORD" \
+        -q 0 -t "$1" -m "$2"
+}
+
+echo "publisher: broker=$HOST:$PORT prefix=$PREFIX interval=${INTERVAL}s"
+
+# The one-shot retained seed. Published before the loop starts and never again.
+pub_retained "$PREFIX/retained/seed" \
+    '{"value":42.5,"unit":"C","note":"published once at fixture start; never republished"}'
+echo "publisher: retained seed published on $PREFIX/retained/seed"
+
+n=0
+while true; do
+    n=$((n + 1))
+    temperature=$(awk "BEGIN { printf \"%.2f\", 20 + ($n % 50) / 10 }")
+
+    pub_retained "$PREFIX/line1/temperature" \
+        "{\"value\":$temperature,\"unit\":\"C\",\"seq\":$n}"
+    pub_retained "$PREFIX/line1/counter" "{\"value\":$n}"
+    pub_retained "$PREFIX/line1/state"   "RUNNING"
+
+    # Unauthored chatter — the driver must never materialise a tag for this.
+    pub_volatile "$PREFIX/noise/chatter" "{\"value\":$n,\"source\":\"chatter\"}"
+
+    sleep "$INTERVAL"
+done
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/MqttFixture.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/MqttFixture.cs
new file mode 100644
index 00000000..7fecc60c
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/MqttFixture.cs
@@ -0,0 +1,205 @@
+using System.Net.Sockets;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests;
+
+/// <summary>
+///     Environment gate + reachability probe for the Mosquitto fixture defined in
+///     <c>Docker/docker-compose.yml</c> (auth + TLS on <c>:8883</c>, authenticated plaintext on
+///     <c>:1883</c>, plus a JSON publisher emitting retained messages).
+///     <para>
+///         <b>Unset env ⇒ skip, never fail.</b> Unlike the Modbus / S7 fixtures — which default to
+///         the shared Docker host and probe it unconditionally — this one has <b>no default
+///         endpoint</b>: without <c>MQTT_FIXTURE_ENDPOINT</c> the whole suite reports
+///         <see cref="NotConfigured"/> and every test calls <c>Assert.Skip</c>. That is deliberate.
+///         The fixture needs credentials that are generated per machine and never committed, so
+///         "probe a well-known host and hope" would produce auth failures rather than a clean skip
+///         on a dev box that never provisioned it. `dotnet test` on macOS with no Docker therefore
+///         stays green.
+///     </para>
+/// </summary>
+/// <remarks>
+///     <para>
+///         Env vars, all read once at fixture construction:
+///     </para>
+///     <list type="bullet">
+///         <item>
+///             <c>MQTT_FIXTURE_ENDPOINT</c> — <c>host:port</c> of the <b>TLS</b> listener, e.g.
+///             <c>10.100.0.35:8883</c>. <b>The gate</b>: absent ⇒ the whole suite skips.
+///         </item>
+///         <item>
+///             <c>MQTT_FIXTURE_PLAIN_ENDPOINT</c> — <c>host:port</c> of the plaintext listener.
+///             Defaults to the TLS host on port 1883.
+///         </item>
+///         <item>
+///             <c>MQTT_FIXTURE_USERNAME</c> / <c>MQTT_FIXTURE_PASSWORD</c> — the broker credentials
+///             <c>gen-fixture-material.sh</c> was run with. The password has no default and is
+///             never committed; without it the suite skips.
+///         </item>
+///         <item>
+///             <c>MQTT_FIXTURE_CA_CERT</c> — path to the fixture CA (<c>secrets/ca.crt</c>).
+///             Optional: supplied, the TLS legs pin to it and the CA-pin test runs; absent, the
+///             connect legs fall back to <c>AllowUntrustedServerCertificate</c> and the pin test
+///             skips.
+///         </item>
+///         <item>
+///             <c>MQTT_FIXTURE_TOPIC_PREFIX</c> — must match the publisher's. Default
+///             <c>otopcua/fixture</c>.
+///         </item>
+///     </list>
+///     <para>
+///         The probe is one-shot and its socket is closed immediately: tests open their own real
+///         <see cref="MqttConnection"/> / <see cref="MqttDriver"/> sessions, and sharing a socket
+///         would serialise them.
+///     </para>
+/// </remarks>
+public sealed class MqttFixture
+{
+    /// <summary>The gate: absent ⇒ the whole suite skips.</summary>
+    public const string EndpointEnvVar = "MQTT_FIXTURE_ENDPOINT";
+
+    /// <summary>Optional override for the plaintext listener; defaults to the TLS host on 1883.</summary>
+    public const string PlainEndpointEnvVar = "MQTT_FIXTURE_PLAIN_ENDPOINT";
+
+    /// <summary>Broker username; defaults to the generator script's own default.</summary>
+    public const string UsernameEnvVar = "MQTT_FIXTURE_USERNAME";
+
+    /// <summary>Broker password. No default — a defaulted fixture password is a committed secret.</summary>
+    public const string PasswordEnvVar = "MQTT_FIXTURE_PASSWORD";
+
+    /// <summary>Path to the fixture CA PEM, for the CA-pin legs.</summary>
+    public const string CaCertEnvVar = "MQTT_FIXTURE_CA_CERT";
+
+    /// <summary>Topic prefix the fixture publisher emits under.</summary>
+    public const string TopicPrefixEnvVar = "MQTT_FIXTURE_TOPIC_PREFIX";
+
+    private const string DefaultUsername = "otopcua";
+    private const string DefaultTopicPrefix = "otopcua/fixture";
+    private const int DefaultPlainPort = 1883;
+
+    private static readonly TimeSpan ProbeTimeout = TimeSpan.FromSeconds(3);
+
+    /// <summary>Initializes a new instance of the <see cref="MqttFixture"/> class.</summary>
+    public MqttFixture()
+    {
+        Username = Environment.GetEnvironmentVariable(UsernameEnvVar) is { Length: > 0 } u ? u : DefaultUsername;
+        Password = Environment.GetEnvironmentVariable(PasswordEnvVar);
+        TopicPrefix = Environment.GetEnvironmentVariable(TopicPrefixEnvVar) is { Length: > 0 } p
+            ? p.TrimEnd('/')
+            : DefaultTopicPrefix;
+
+        var caPath = Environment.GetEnvironmentVariable(CaCertEnvVar);
+        CaCertificatePath = string.IsNullOrWhiteSpace(caPath) ? null : caPath;
+
+        var tlsRaw = Environment.GetEnvironmentVariable(EndpointEnvVar);
+        if (string.IsNullOrWhiteSpace(tlsRaw))
+        {
+            SkipReason =
+                $"Skipped: set {EndpointEnvVar} (e.g. 10.100.0.35:8883) plus {UsernameEnvVar}/{PasswordEnvVar} to run "
+                + "the MQTT fixture live suite. Bring the fixture up first — see "
+                + "tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/docker-compose.yml.";
+            return;
+        }
+
+        (TlsHost, TlsPort) = ParseEndpoint(tlsRaw, defaultPort: 8883);
+
+        var plainRaw = Environment.GetEnvironmentVariable(PlainEndpointEnvVar);
+        (PlainHost, PlainPort) = string.IsNullOrWhiteSpace(plainRaw)
+            ? (TlsHost, DefaultPlainPort)
+            : ParseEndpoint(plainRaw, DefaultPlainPort);
+
+        if (string.IsNullOrEmpty(Password))
+        {
+            // A blank password is NOT a valid "anonymous" configuration for this fixture — the
+            // broker runs allow_anonymous false — so treat it as unconfigured rather than letting
+            // every test fail with an indistinguishable "not authorized".
+            SkipReason =
+                $"Skipped: {EndpointEnvVar} is set but {PasswordEnvVar} is not. The fixture broker runs "
+                + "allow_anonymous=false; supply the password gen-fixture-material.sh was run with.";
+            return;
+        }
+
+        // Probe the TLS listener: it is the endpoint the gate names, and a broker that is up serves
+        // both listeners from one process.
+        SkipReason = Probe(TlsHost, TlsPort);
+    }
+
+    /// <summary>Gets the TLS listener host.</summary>
+    public string TlsHost { get; } = string.Empty;
+
+    /// <summary>Gets the TLS listener port.</summary>
+    public int TlsPort { get; }
+
+    /// <summary>Gets the plaintext listener host.</summary>
+    public string PlainHost { get; } = string.Empty;
+
+    /// <summary>Gets the plaintext listener port.</summary>
+    public int PlainPort { get; }
+
+    /// <summary>Gets the broker username.</summary>
+    public string Username { get; }
+
+    /// <summary>Gets the broker password, or <see langword="null"/> when unconfigured.</summary>
+    public string? Password { get; }
+
+    /// <summary>Gets the fixture CA PEM path, or <see langword="null"/> when not supplied.</summary>
+    public string? CaCertificatePath { get; }
+
+    /// <summary>Gets the topic prefix the fixture publisher emits under (no trailing slash).</summary>
+    public string TopicPrefix { get; }
+
+    /// <summary>
+    ///     Gets the reason every test in the suite must skip, or <see langword="null"/> when the
+    ///     fixture is configured and reachable.
+    /// </summary>
+    public string? SkipReason { get; }
+
+    /// <summary>Gets a value indicating whether the suite must skip.</summary>
+    public bool NotConfigured => SkipReason is not null;
+
+    /// <summary>Composes a fixture topic under <see cref="TopicPrefix"/>.</summary>
+    /// <param name="suffix">The topic suffix, e.g. <c>line1/temperature</c>.</param>
+    /// <returns>The full topic.</returns>
+    public string Topic(string suffix) => $"{TopicPrefix}/{suffix.TrimStart('/')}";
+
+    private static (string Host, int Port) ParseEndpoint(string raw, int defaultPort)
+    {
+        var parts = raw.Trim().Split(':', 2);
+        var port = parts.Length == 2 && int.TryParse(parts[1], out var parsed) ? parsed : defaultPort;
+        return (parts[0], port);
+    }
+
+    private static string? Probe(string host, int port)
+    {
+        try
+        {
+            using var client = new TcpClient();
+            var connect = client.ConnectAsync(host, port);
+            if (!connect.Wait(ProbeTimeout) || !client.Connected)
+            {
+                return Unreachable(host, port, $"no TCP connection within {ProbeTimeout.TotalSeconds:0}s");
+            }
+        }
+        catch (Exception ex)
+        {
+            return Unreachable(host, port, $"{ex.GetType().Name}: {ex.Message}");
+        }
+
+        return null;
+    }
+
+    private static string Unreachable(string host, int port, string detail) =>
+        $"Skipped: the MQTT fixture broker at {host}:{port} is configured but unreachable ({detail}). "
+        + "Bring the stack up on the Docker host: "
+        + "`ssh dohertj2@10.100.0.35 'cd /opt/otopcua-mqtt && MQTT_FIXTURE_USERNAME=... MQTT_FIXTURE_PASSWORD=... docker compose up -d'`.";
+}
+
+/// <summary>
+///     Collection definition so the reachability probe runs once per test session rather than once
+///     per test — probing a firewalled endpoint per test would burn the probe timeout each time.
+/// </summary>
+[Xunit.CollectionDefinition(Name)]
+public sealed class MqttFixtureCollection : Xunit.ICollectionFixture<MqttFixture>
+{
+    /// <summary>The collection name.</summary>
+    public const string Name = "MqttFixture";
+}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/PlainMqttLiveTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/PlainMqttLiveTests.cs
new file mode 100644
index 00000000..6633f7c4
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/PlainMqttLiveTests.cs
@@ -0,0 +1,616 @@
+using System.Security.Cryptography;
+using System.Security.Cryptography.X509Certificates;
+using System.Text;
+using MQTTnet;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests;
+
+/// <summary>
+///     Live validation of the plain-MQTT (P1) driver against the real Mosquitto fixture in
+///     <c>Docker/docker-compose.yml</c> — auth + TLS on <c>:8883</c>, authenticated plaintext on
+///     <c>:1883</c>, and a publisher emitting retained JSON. Every component exercised here is the
+///     production one: <see cref="MqttConnection"/> (broker session, TLS, CA pin),
+///     <see cref="MqttSubscriptionManager"/> (SUBSCRIBE, routing, extraction) and
+///     <see cref="MqttDriver"/> (the <c>ISubscribable</c>/<c>IReadable</c> surface).
+///     <para>
+///         <b>Env-gated + skip-clean.</b> Without <c>MQTT_FIXTURE_ENDPOINT</c> (and the credentials)
+///         every test calls <c>Assert.Skip</c>, so <c>dotnet test</c> is green on a laptop with no
+///         Docker. See <see cref="MqttFixture"/> for the full env-var list.
+///     </para>
+///     <para>
+///         <b>Why these four things.</b> The TLS/auth legs are the first time the Task-3 handshake
+///         code meets a real broker — until now the CA-pin path had only ever seen synthetic certs,
+///         and its failure mode (a pin that accepts everything) is invisible without a negative
+///         control, which is why <see cref="Tls_connect_pinned_to_a_foreign_ca_is_rejected"/> exists
+///         alongside the positive. The subscribe leg asserts the published <b>RawPath</b>, the single
+///         most defect-prone identity in this plan. And the unauthored-topic leg is written so it
+///         cannot pass vacuously — it proves an authored tag was flowing throughout the window in
+///         which the unauthored topic produced nothing.
+///     </para>
+/// </summary>
+[Collection(MqttFixtureCollection.Name)]
+[Trait("Category", "LiveIntegration")]
+public sealed class PlainMqttLiveTests(MqttFixture fixture)
+{
+    /// <summary>OPC UA <c>BadWaitingForInitialData</c> — "no value has ever been observed".</summary>
+    private const uint BadWaitingForInitialData = 0x80320000u;
+
+    /// <summary>OPC UA <c>BadNodeIdUnknown</c> — "that reference is not an authored MQTT tag".</summary>
+    private const uint BadNodeIdUnknown = 0x80340000u;
+
+    /// <summary>OPC UA <c>Good</c>.</summary>
+    private const uint Good = 0x00000000u;
+
+    /// <summary>
+    ///     Bound on every wait for broker traffic. The fixture publisher emits every 2s by default,
+    ///     so 30s is ~15 publish cycles — generous enough to absorb a slow container start, short
+    ///     enough that a genuinely dark subscription fails rather than hangs (the plan's per-op
+    ///     bounded-deadline rule).
+    /// </summary>
+    private static readonly TimeSpan MessageTimeout = TimeSpan.FromSeconds(30);
+
+    private readonly MqttFixture _fx = fixture;
+
+    // -----------------------------------------------------------------------
+    // TLS + auth
+    // -----------------------------------------------------------------------
+
+    /// <summary>
+    ///     Connects over the TLS listener with real credentials and asserts the session establishes.
+    ///     Pins to the fixture CA when <c>MQTT_FIXTURE_CA_CERT</c> is supplied — so on a fully
+    ///     provisioned rig this is simultaneously the CA-pin happy path — and falls back to
+    ///     <c>AllowUntrustedServerCertificate</c> otherwise so the auth leg still runs.
+    /// </summary>
+    [Fact]
+    public async Task Tls_auth_connect_succeeds()
+    {
+        SkipIfUnconfigured();
+
+        var ct = TestContext.Current.CancellationToken;
+        await using var connection = new MqttConnection(TlsOptions(), "mqtt-live-tls");
+
+        await connection.ConnectAsync(ct);
+
+        connection.IsConnected.ShouldBeTrue("a TLS+auth connect against the fixture broker must establish a session");
+        connection.State.ShouldBe(MqttConnectionState.Connected);
+
+        TestContext.Current.SendDiagnosticMessage(
+            $"TLS+auth connect to {_fx.TlsHost}:{_fx.TlsPort} succeeded "
+            + $"(ca pin: {_fx.CaCertificatePath ?? "<none — AllowUntrustedServerCertificate>"}).");
+    }
+
+    /// <summary>
+    ///     The auth negative, and the assertion that gives every other test in this file its meaning:
+    ///     a wrong password must never yield a usable session. The broker runs
+    ///     <c>allow_anonymous false</c>, so if bad credentials produced a session the fixture would be
+    ///     misconfigured and "connected" would prove nothing about auth.
+    ///     <para>
+    ///         <b>LIVE-GATE FINDING (2026-07-24) — the invariant asserted here is deliberately "no
+    ///         session", not "ConnectAsync throws".</b> Against real Mosquitto,
+    ///         <see cref="MqttConnection.ConnectAsync"/> <b>returns normally</b> for a CONNECT the
+    ///         broker rejects as not-authorized: MQTTnet 5 does not throw on an unsuccessful CONNACK,
+    ///         and <c>ConnectCoreAsync</c> then unconditionally publishes
+    ///         <see cref="MqttConnectionState.Connected"/> and starts the reconnect supervisor.
+    ///         Measured at +300 ms with a wrong password: <c>thrown=&lt;none&gt;</c>,
+    ///         <c>IsConnected=False</c>, <c>State=Reconnecting</c> (the broker's socket close had
+    ///         fired <c>DisconnectedAsync</c> by then). Broker side:
+    ///         <c>Client … disconnected, not authorised.</c>
+    ///     </para>
+    ///     <para>
+    ///         That is a driver-side defect, not a fixture one — <c>InitializeAsync</c> reports
+    ///         <c>DriverState.Healthy</c> / <c>HostState.Running</c> off a connect that never
+    ///         authenticated, so a deployment carrying the wrong broker password seals green and the
+    ///         AdminUI "Test connect" probe is expected to agree. Fixing it belongs in the driver
+    ///         (<c>MqttConnection</c> must inspect the connect result / opt into MQTTnet's
+    ///         throw-on-unsuccessful-CONNACK), which is outside this task's file scope; it is reported
+    ///         as a live-gate finding.
+    ///     </para>
+    ///     <para>
+    ///         The test is written to hold either way: it records whether the call threw (as a
+    ///         diagnostic, not an assertion) and asserts only the invariant that must be true under
+    ///         both behaviours — no session is ever established. A driver fixed to throw still passes.
+    ///     </para>
+    /// </summary>
+    [Fact]
+    public async Task Tls_connect_with_wrong_password_never_establishes_a_session()
+    {
+        SkipIfUnconfigured();
+
+        var ct = TestContext.Current.CancellationToken;
+        var options = TlsOptions() with { Password = _fx.Password + "-definitely-wrong" };
+        await using var connection = new MqttConnection(options, "mqtt-live-badpass");
+
+        Exception? thrown = null;
+        try
+        {
+            await connection.ConnectAsync(ct);
+        }
+        catch (Exception ex)
+        {
+            thrown = ex;
+        }
+
+        // Poll rather than sample once: the rejection arrives as a broker-initiated socket close a
+        // few milliseconds after ConnectAsync returns, so a single immediate read could catch the
+        // optimistic Connected state and pass for the wrong reason. Every sample must show no
+        // session — a correct password would make IsConnected true within the first sample.
+        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5);
+        while (DateTime.UtcNow < deadline)
+        {
+            connection.IsConnected.ShouldBeFalse(
+                "a CONNECT the broker rejected as not-authorized must never yield a usable session");
+            await Task.Delay(100, ct);
+        }
+
+        connection.State.ShouldNotBe(
+            MqttConnectionState.Connected,
+            "the settled state after a rejected CONNECT must not advertise a healthy session");
+
+        TestContext.Current.SendDiagnosticMessage(
+            $"wrong-password connect: threw={thrown?.GetType().Name ?? "<nothing — see the LIVE-GATE FINDING in this test's docs>"}, "
+            + $"IsConnected={connection.IsConnected}, State={connection.State}.");
+    }
+
+    /// <summary>
+    ///     CA-pin happy path: <c>CaCertificatePath</c> set to the fixture's own CA, with the
+    ///     untrusted-certificate escape hatch explicitly OFF, so the handshake can only succeed by
+    ///     chaining the broker's leaf up to the pinned root through
+    ///     <c>MqttConnection.ValidateAgainstPinnedCa</c>.
+    /// </summary>
+    [Fact]
+    public async Task Tls_connect_pinned_to_fixture_ca_succeeds()
+    {
+        SkipIfUnconfigured();
+        if (_fx.CaCertificatePath is null)
+        {
+            Assert.Skip(
+                $"Skipped: set {MqttFixture.CaCertEnvVar} to the fixture CA (Docker/secrets/ca.crt) to run the CA-pin leg.");
+        }
+
+        File.Exists(_fx.CaCertificatePath).ShouldBeTrue(
+            $"{MqttFixture.CaCertEnvVar}='{_fx.CaCertificatePath}' must point at the fixture CA PEM");
+
+        var ct = TestContext.Current.CancellationToken;
+        var options = TlsOptions() with
+        {
+            CaCertificatePath = _fx.CaCertificatePath,
+            AllowUntrustedServerCertificate = false,
+        };
+        await using var connection = new MqttConnection(options, "mqtt-live-capin");
+
+        await connection.ConnectAsync(ct);
+
+        connection.IsConnected.ShouldBeTrue("the broker leaf must chain up to the pinned fixture CA");
+    }
+
+    /// <summary>
+    ///     The pin's falsifiability control, and the reason the positive above means anything: pin to
+    ///     a CA the broker's certificate was <b>not</b> issued by and assert the handshake fails. A
+    ///     validator that returned <c>true</c> unconditionally — the silent failure mode for this
+    ///     kind of code — would pass the positive test and fail here.
+    ///     <para>
+    ///         The foreign CA is generated in-process rather than taken from the fixture, so this leg
+    ///         needs no extra rig provisioning and cannot accidentally be handed the real CA.
+    ///     </para>
+    /// </summary>
+    [Fact]
+    public async Task Tls_connect_pinned_to_a_foreign_ca_is_rejected()
+    {
+        SkipIfUnconfigured();
+
+        var ct = TestContext.Current.CancellationToken;
+        var foreignCaPath = Path.Combine(Path.GetTempPath(), $"otopcua-mqtt-foreign-ca-{Guid.NewGuid():N}.pem");
+        await File.WriteAllTextAsync(foreignCaPath, CreateForeignCaPem(), ct);
+
+        try
+        {
+            var options = TlsOptions() with
+            {
+                CaCertificatePath = foreignCaPath,
+                AllowUntrustedServerCertificate = false,
+                // The handshake fails fast; no need to sit on the default 15s budget.
+                ConnectTimeoutSeconds = 10,
+            };
+            await using var connection = new MqttConnection(options, "mqtt-live-foreignca");
+
+            await Should.ThrowAsync<Exception>(async () => await connection.ConnectAsync(ct));
+
+            connection.IsConnected.ShouldBeFalse(
+                "a broker certificate that does not chain to the pinned CA must be refused — fail closed");
+        }
+        finally
+        {
+            File.Delete(foreignCaPath);
+        }
+    }
+
+    // -----------------------------------------------------------------------
+    // Plain subscribe / read
+    // -----------------------------------------------------------------------
+
+    /// <summary>
+    ///     The identity assertion. Subscribes an authored tag whose RawPath is deliberately unrelated
+    ///     to its MQTT topic, and asserts the driver publishes the change under the <b>RawPath</b> —
+    ///     the v3 driver wire reference <c>DriverHostActor</c> fans out to the raw + UNS NodeIds. A
+    ///     driver that keyed notifications by topic (or by the TagConfig blob, the retired pre-v3
+    ///     shape) would deliver a value that no node in the address space is listening for, and every
+    ///     "did a value arrive" assertion would still pass.
+    /// </summary>
+    [Fact]
+    public async Task Plain_subscribe_delivers_value_under_rawpath()
+    {
+        SkipIfUnconfigured();
+
+        var ct = TestContext.Current.CancellationToken;
+        const string rawPath = "Fixture/Mqtt/Line1/Temperature";
+        var tag = JsonTag(rawPath, _fx.Topic("line1/temperature"), "$.value", DriverDataType.Float64);
+
+        await using var driver = new MqttDriver(PlainOptions(tag), "mqtt-live-subscribe");
+        var recorder = new DataChangeRecorder(driver);
+
+        await driver.InitializeAsync(string.Empty, ct);
+        await driver.SubscribeAsync([rawPath], TimeSpan.FromSeconds(1), ct);
+
+        var change = await recorder.WaitForAsync(rawPath, MessageTimeout, ct);
+
+        change.FullReference.ShouldBe(
+            rawPath,
+            "the driver must publish under the tag's RawPath, never its topic or its TagConfig blob");
+        change.Snapshot.StatusCode.ShouldBe(Good);
+        change.Snapshot.Value.ShouldBeOfType<double>().ShouldBeInRange(-100d, 1000d);
+
+        // ...and the read path serves the same last value under the same key.
+        var read = await driver.ReadAsync([rawPath], ct);
+        read.Count.ShouldBe(1);
+        read[0].StatusCode.ShouldBe(Good);
+        read[0].Value.ShouldNotBeNull();
+    }
+
+    /// <summary>
+    ///     Retained-message seeding. The fixture publisher emits <c>retained/seed</c> exactly once,
+    ///     at container start, and never republishes it — so a subscriber connecting later can only
+    ///     obtain a value from the broker's retained copy. Asserting the value on a topic that is
+    ///     <i>also</i> republished every couple of seconds would prove nothing: a live publish
+    ///     arriving quickly looks identical to a retained seed.
+    /// </summary>
+    [Fact]
+    public async Task Retained_message_seeds_value_on_subscribe()
+    {
+        SkipIfUnconfigured();
+
+        var ct = TestContext.Current.CancellationToken;
+        const string rawPath = "Fixture/Mqtt/Retained/Seed";
+        var tag = JsonTag(rawPath, _fx.Topic("retained/seed"), "$.value", DriverDataType.Float64);
+
+        await using var driver = new MqttDriver(PlainOptions(tag), "mqtt-live-retained");
+        var recorder = new DataChangeRecorder(driver);
+
+        await driver.InitializeAsync(string.Empty, ct);
+        await driver.SubscribeAsync([rawPath], TimeSpan.FromSeconds(1), ct);
+
+        var change = await recorder.WaitForAsync(rawPath, MessageTimeout, ct);
+
+        change.FullReference.ShouldBe(rawPath);
+        change.Snapshot.StatusCode.ShouldBe(Good);
+        change.Snapshot.Value.ShouldBeOfType<double>().ShouldBe(
+            42.5d,
+            "the retained seed carries the publisher's one-shot value; anything else means the "
+            + "subscription was seeded from somewhere other than the broker's retained store");
+    }
+
+    /// <summary>
+    ///     A chatty broker must not auto-provision. Publishes on a topic no tag was authored for and
+    ///     asserts the driver never raises a notification for it.
+    ///     <para>
+    ///         <b>Deliberately falsifiable.</b> A version of this test that authored nothing would
+    ///         pass vacuously — silence proves nothing if the pipeline is dead. So a real tag is
+    ///         authored and subscribed first, and the test asserts that authored tag kept delivering
+    ///         <i>throughout</i> the window in which the unauthored topic was being published. The
+    ///         only way both assertions hold is if the driver was live and chose to ignore the
+    ///         unauthored traffic.
+    ///     </para>
+    /// </summary>
+    [Fact]
+    public async Task Unauthored_topic_is_silent_while_an_authored_tag_keeps_flowing()
+    {
+        SkipIfUnconfigured();
+
+        var ct = TestContext.Current.CancellationToken;
+        const string authoredRawPath = "Fixture/Mqtt/Line1/Counter";
+        const string unauthoredRawPath = "Fixture/Mqtt/Line1/NotAuthored";
+
+        // Unique per run so a previous run's retained message can never seed it.
+        var unauthoredTopic = _fx.Topic($"unauthored/{Guid.NewGuid():N}");
+
+        var tag = JsonTag(authoredRawPath, _fx.Topic("line1/counter"), "$.value", DriverDataType.Int64);
+
+        await using var driver = new MqttDriver(PlainOptions(tag), "mqtt-live-unauthored");
+        var recorder = new DataChangeRecorder(driver);
+
+        await driver.InitializeAsync(string.Empty, ct);
+
+        // Subscribing an unauthored reference alongside the authored one: the driver must say
+        // BadNodeIdUnknown rather than leaving it forever "waiting for initial data" — and must
+        // never later turn it Good off broker traffic.
+        await driver.SubscribeAsync([authoredRawPath, unauthoredRawPath], TimeSpan.FromSeconds(1), ct);
+
+        // Proof the pipeline is live BEFORE the unauthored traffic starts.
+        await recorder.WaitForAsync(authoredRawPath, MessageTimeout, ct);
+        var beforeCount = recorder.CountFor(authoredRawPath);
+
+        // Emit unauthored traffic with our own client — the fixture publisher's own noise topic is
+        // not enough, because a test must be able to point at the exact messages it caused.
+        await PublishUnauthoredAsync(unauthoredTopic, messages: 5, ct);
+
+        // Let the authored tag tick at least once more while the unauthored messages sit delivered.
+        await recorder.WaitForCountAsync(authoredRawPath, beforeCount + 1, MessageTimeout, ct);
+
+        var seen = recorder.Snapshot();
+        seen.ShouldAllBe(
+            e => e.FullReference == authoredRawPath,
+            "the driver must raise notifications ONLY for authored tags; a broker's other topics are none of its business. "
+            + $"Saw: {string.Join(", ", seen.Select(e => e.FullReference).Distinct())}");
+
+        var read = await driver.ReadAsync([authoredRawPath, unauthoredRawPath], ct);
+        read[0].StatusCode.ShouldBe(Good, "the authored tag was flowing throughout — this is the falsifiability control");
+        read[1].StatusCode.ShouldBeOneOf(
+            BadNodeIdUnknown,
+            BadWaitingForInitialData);
+        read[1].Value.ShouldBeNull("an unauthored reference must never acquire a value from broker traffic");
+    }
+
+    // -----------------------------------------------------------------------
+    // helpers
+    // -----------------------------------------------------------------------
+
+    private void SkipIfUnconfigured()
+    {
+        if (_fx.NotConfigured)
+        {
+            Assert.Skip(_fx.SkipReason!);
+        }
+    }
+
+    /// <summary>Driver options for the TLS listener, pinning the fixture CA when one was supplied.</summary>
+    private MqttDriverOptions TlsOptions(params RawTagEntry[] rawTags) => new()
+    {
+        Host = _fx.TlsHost,
+        Port = _fx.TlsPort,
+        UseTls = true,
+        // With no CA to pin, the fixture's self-issued chain cannot validate against the OS trust
+        // store, so the documented dev escape hatch is the only way the auth leg can run at all.
+        CaCertificatePath = _fx.CaCertificatePath,
+        AllowUntrustedServerCertificate = _fx.CaCertificatePath is null,
+        Username = _fx.Username,
+        Password = _fx.Password,
+        ClientId = $"otopcua-live-{Guid.NewGuid():N}",
+        ConnectTimeoutSeconds = 15,
+        Mode = MqttMode.Plain,
+        Plain = new MqttPlainOptions { DefaultQos = 1 },
+        RawTags = rawTags,
+    };
+
+    /// <summary>Driver options for the plaintext listener — still authenticated.</summary>
+    private MqttDriverOptions PlainOptions(params RawTagEntry[] rawTags) => new()
+    {
+        Host = _fx.PlainHost,
+        Port = _fx.PlainPort,
+        UseTls = false,
+        Username = _fx.Username,
+        Password = _fx.Password,
+        ClientId = $"otopcua-live-{Guid.NewGuid():N}",
+        ConnectTimeoutSeconds = 15,
+        Mode = MqttMode.Plain,
+        Plain = new MqttPlainOptions { DefaultQos = 1 },
+        RawTags = rawTags,
+    };
+
+    private static RawTagEntry JsonTag(string rawPath, string topic, string jsonPath, DriverDataType dataType) =>
+        new(
+            rawPath,
+            $$"""
+              {"topic":"{{topic}}","payloadFormat":"Json","jsonPath":"{{jsonPath}}","dataType":"{{dataType}}","qos":1,"retainSeed":true}
+              """,
+            WriteIdempotent: false);
+
+    /// <summary>
+    ///     Publishes on a topic no tag is authored for, with a raw MQTTnet client rather than through
+    ///     the driver (the driver has no write path in P1, and the point is traffic the driver did
+    ///     not originate).
+    /// </summary>
+    private async Task PublishUnauthoredAsync(string topic, int messages, CancellationToken ct)
+    {
+        using var client = new MqttClientFactory().CreateMqttClient();
+        var options = new MqttClientOptionsBuilder()
+            .WithTcpServer(_fx.PlainHost, _fx.PlainPort)
+            .WithCredentials(_fx.Username, _fx.Password)
+            .WithClientId($"otopcua-live-noise-{Guid.NewGuid():N}")
+            .WithTimeout(TimeSpan.FromSeconds(10))
+            .Build();
+
+        await client.ConnectAsync(options, ct);
+        try
+        {
+            for (var i = 0; i < messages; i++)
+            {
+                await client.PublishStringAsync(
+                    topic,
+                    $$"""{"value":{{i}},"source":"unauthored-live-test"}""",
+                    MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce,
+                    retain: false,
+                    ct);
+            }
+        }
+        finally
+        {
+            await client.DisconnectAsync(cancellationToken: ct);
+        }
+    }
+
+    /// <summary>
+    ///     A self-signed CA the fixture broker's certificate was definitively NOT issued by, as PEM.
+    ///     Generated per call so nothing about it can be shared with — or mistaken for — the real
+    ///     fixture CA.
+    /// </summary>
+    private static string CreateForeignCaPem()
+    {
+        using var key = RSA.Create(2048);
+        var request = new CertificateRequest(
+            "CN=OtOpcUa MQTT foreign CA (negative control)",
+            key,
+            HashAlgorithmName.SHA256,
+            RSASignaturePadding.Pkcs1);
+
+        request.CertificateExtensions.Add(new X509BasicConstraintsExtension(true, false, 0, true));
+        request.CertificateExtensions.Add(
+            new X509KeyUsageExtension(X509KeyUsageFlags.KeyCertSign | X509KeyUsageFlags.CrlSign, true));
+
+        using var certificate = request.CreateSelfSigned(
+            DateTimeOffset.UtcNow.AddDays(-1),
+            DateTimeOffset.UtcNow.AddDays(1));
+
+        var builder = new StringBuilder();
+        builder.AppendLine("-----BEGIN CERTIFICATE-----");
+        builder.AppendLine(Convert.ToBase64String(certificate.RawData, Base64FormattingOptions.InsertLineBreaks));
+        builder.AppendLine("-----END CERTIFICATE-----");
+        return builder.ToString();
+    }
+
+    /// <summary>
+    ///     Records every <c>OnDataChange</c> the driver raises and lets a test await a specific
+    ///     RawPath. Attached in the constructor — <b>before</b> the MQTT subscribe — because the
+    ///     retained seed can arrive inside the SUBSCRIBE round-trip, and a handler attached
+    ///     afterwards would miss exactly the message the retained-seed test is about.
+    /// </summary>
+    private sealed class DataChangeRecorder
+    {
+        private readonly List<DataChangeEventArgs> _events = [];
+        private readonly Lock _gate = new();
+        private readonly List<Waiter> _waiters = [];
+
+        public DataChangeRecorder(MqttDriver driver) => driver.OnDataChange += OnDataChange;
+
+        /// <summary>A point-in-time copy of everything recorded so far.</summary>
+        /// <returns>The recorded events.</returns>
+        public IReadOnlyList<DataChangeEventArgs> Snapshot()
+        {
+            lock (_gate)
+            {
+                return [.. _events];
+            }
+        }
+
+        /// <summary>How many notifications have been recorded for one RawPath.</summary>
+        /// <param name="rawPath">The RawPath to count.</param>
+        /// <returns>The count.</returns>
+        public int CountFor(string rawPath)
+        {
+            lock (_gate)
+            {
+                return _events.Count(e => e.FullReference == rawPath);
+            }
+        }
+
+        /// <summary>Awaits the first notification for <paramref name="rawPath"/>.</summary>
+        /// <param name="rawPath">The RawPath to await.</param>
+        /// <param name="timeout">Bound on the wait.</param>
+        /// <param name="cancellationToken">Caller cancellation.</param>
+        /// <returns>The first matching notification.</returns>
+        public Task<DataChangeEventArgs> WaitForAsync(
+            string rawPath,
+            TimeSpan timeout,
+            CancellationToken cancellationToken)
+            => WaitForCountAsync(rawPath, count: 1, timeout, cancellationToken);
+
+        /// <summary>
+        ///     Awaits the <paramref name="count"/>-th notification for <paramref name="rawPath"/>.
+        ///     Pre-scans what has already been recorded, so a caller that registers the wait after
+        ///     the message arrived is served immediately rather than timing out.
+        /// </summary>
+        /// <param name="rawPath">The RawPath to await.</param>
+        /// <param name="count">How many notifications must have been seen.</param>
+        /// <param name="timeout">Bound on the wait.</param>
+        /// <param name="cancellationToken">Caller cancellation.</param>
+        /// <returns>The <paramref name="count"/>-th matching notification.</returns>
+        public async Task<DataChangeEventArgs> WaitForCountAsync(
+            string rawPath,
+            int count,
+            TimeSpan timeout,
+            CancellationToken cancellationToken)
+        {
+            Waiter waiter;
+            lock (_gate)
+            {
+                var matches = _events.Where(e => e.FullReference == rawPath).ToList();
+                if (matches.Count >= count)
+                {
+                    return matches[count - 1];
+                }
+
+                waiter = new Waiter(rawPath, count);
+                _waiters.Add(waiter);
+            }
+
+            try
+            {
+                return await waiter.Completion.Task.WaitAsync(timeout, cancellationToken);
+            }
+            catch (TimeoutException)
+            {
+                throw new TimeoutException(
+                    $"No notification #{count} for RawPath '{rawPath}' within {timeout.TotalSeconds:0}s. "
+                    + $"Recorded so far: {DescribeRecorded()}. Is the fixture publisher container running "
+                    + "(`docker logs otopcua-mqtt-publisher`), and does MQTT_FIXTURE_TOPIC_PREFIX match it?");
+            }
+            finally
+            {
+                lock (_gate)
+                {
+                    _waiters.Remove(waiter);
+                }
+            }
+        }
+
+        private string DescribeRecorded()
+        {
+            var snapshot = Snapshot();
+            return snapshot.Count == 0
+                ? "(nothing)"
+                : string.Join(", ", snapshot.GroupBy(e => e.FullReference).Select(g => $"{g.Key}×{g.Count()}"));
+        }
+
+        private void OnDataChange(object? sender, DataChangeEventArgs args)
+        {
+            List<Waiter>? satisfied = null;
+            lock (_gate)
+            {
+                _events.Add(args);
+                foreach (var waiter in _waiters)
+                {
+                    if (waiter.RawPath != args.FullReference) continue;
+                    if (_events.Count(e => e.FullReference == waiter.RawPath) < waiter.Count) continue;
+
+                    (satisfied ??= []).Add(waiter);
+                }
+            }
+
+            // Completed OUTSIDE the lock, and with RunContinuationsAsynchronously below, so a test
+            // continuation never runs on MQTTnet's dispatcher thread while holding this lock —
+            // that would stall delivery for every other subscription.
+            if (satisfied is null) return;
+            foreach (var waiter in satisfied)
+            {
+                waiter.Completion.TrySetResult(args);
+            }
+        }
+
+        private sealed record Waiter(string RawPath, int Count)
+        {
+            public TaskCompletionSource<DataChangeEventArgs> Completion { get; } =
+                new(TaskCreationOptions.RunContinuationsAsynchronously);
+        }
+    }
+}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests.csproj b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests.csproj
new file mode 100644
index 00000000..f7b27445
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests.csproj
@@ -0,0 +1,49 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <!-- OTOPCUA0001 (arch-review 07/C-1): this suite invokes driver capability methods DIRECTLY to
+         exercise wire-level behavior against a real broker — the analyzer's documented intentional
+         case ("move the suppression into a NoWarn for the test project"). Not a resilience-dispatch
+         gap. Same suppression as the S7 / AB CIP integration suites. -->
+    <PropertyGroup>
+        <NoWarn>$(NoWarn);OTOPCUA0001</NoWarn>
+    </PropertyGroup>
+
+    <PropertyGroup>
+        <TargetFramework>net10.0</TargetFramework>
+        <Nullable>enable</Nullable>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <IsPackable>false</IsPackable>
+        <IsTestProject>true</IsTestProject>
+        <RootNamespace>ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests</RootNamespace>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <PackageReference Include="xunit.v3"/>
+        <PackageReference Include="Shouldly"/>
+        <PackageReference Include="Microsoft.NET.Test.Sdk"/>
+        <PackageReference Include="xunit.runner.visualstudio">
+            <PrivateAssets>all</PrivateAssets>
+            <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+        </PackageReference>
+        <!--
+            Direct reference (not merely transitive through the driver) because the tests
+            publish into the fixture broker with their own raw MQTTnet client — the
+            unauthored-topic test has to emit traffic the DRIVER never emits.
+        -->
+        <PackageReference Include="MQTTnet"/>
+    </ItemGroup>
+
+    <ItemGroup>
+        <ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts.csproj"/>
+        <ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Mqtt\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj"/>
+    </ItemGroup>
+
+    <!--
+        Docker/ is DELIBERATELY not copied to the output directory (the sibling Modbus /
+        S7 suites copy theirs because their tests read simulator profile JSONs at
+        runtime; nothing here does). Copying it would also drag Docker/secrets/ — the
+        generated broker password hash and the CA private key — into bin/, i.e. out of
+        the one directory .gitignore protects.
+    -->
+
+</Project>