diff --git a/frontend/src/services/mqtt.service.ts b/frontend/src/services/mqtt.service.ts
index 27d56c4824..2bb58b812e 100644
--- a/frontend/src/services/mqtt.service.ts
+++ b/frontend/src/services/mqtt.service.ts
@@ -15,10 +15,13 @@ import { Maybe } from '@/types/common/types'
 import type {
     BinaryPayload,
     ManagedMqttClient,
+    ManagedMqttObserver,
+    MqttConnectionHandlers,
     MqttConnectionOptions,
     MqttConnectionWaiter,
     MqttCredentialProvider,
     MqttCredentials,
+    MqttObserverHandle,
     MqttPublishRequest,
     MqttReconnectOptions,
     MqttReconnectPolicy,
@@ -48,6 +51,8 @@ class MqttService extends BaseService implements MqttServiceI {
 
     protected $destroyed: boolean
 
+    protected $observerSeq: number
+
     constructor ({ app, router, services }: CreateServiceOptions) {
         super({
             name: 'mqtt',
@@ -60,6 +65,7 @@ class MqttService extends BaseService implements MqttServiceI {
         this.$clients = new Map()
         this.$clientOperations = new Map()
         this.$destroyed = false
+        this.$observerSeq = 0
 
         this.init()
     }
@@ -68,111 +74,171 @@ class MqttService extends BaseService implements MqttServiceI {
         this.$mqtt = Mqtt
     }
 
-    getManagedClient (key: string): ManagedMqttClient | null {
-        return this.$clients.get(key) ?? null
+    getManagedClient (clientKey: string): ManagedMqttClient | null {
+        return this.$clients.get(clientKey) ?? null
     }
 
-    hasClient (key: string): boolean {
-        return this.$clients.has(key)
+    hasClient (clientKey: string): boolean {
+        return this.$clients.has(clientKey)
     }
 
-    async runClientOperation<T> (key: string, operation: () => Promise<T> | T): Promise<T> {
-        const previous = this.$clientOperations.get(key) ?? Promise.resolve()
+    async runClientOperation<T> (clientKey: string, operation: () => Promise<T> | T): Promise<T> {
+        const previous = this.$clientOperations.get(clientKey) ?? Promise.resolve()
         const current = previous
             .catch(() => {})
             .then(operation)
 
         const tracked = current.finally(() => {
-            if (this.$clientOperations.get(key) === tracked) {
-                this.$clientOperations.delete(key)
+            if (this.$clientOperations.get(clientKey) === tracked) {
+                this.$clientOperations.delete(clientKey)
             }
         })
-        this.$clientOperations.set(key, tracked)
+        this.$clientOperations.set(clientKey, tracked)
 
         return tracked
     }
 
-    async destroyClient (key: string): Promise<void> {
-        await this.runClientOperation(key, async () => {
-            await this._destroyClientUnlocked(key)
+    async destroyClient (clientKey: string): Promise<void> {
+        await this.runClientOperation(clientKey, async () => {
+            await this._destroyClientUnlocked(clientKey)
         })
     }
 
-    async createClient (key: string, options: Partial<MqttConnectionOptions> = {}): Promise<MqttClient> {
-        const {
-            reconnectPeriod = 0,
-            getCredentials,
-            reconnect,
-            onConnect,
-            onClose,
-            onOffline,
-            onError,
-            onMessage
-        } = options
-
-        if (!key || typeof key !== 'string') {
+    async createClient (clientKey: string, options: Partial<MqttConnectionOptions> = {}): Promise<MqttClient> {
+        if (!clientKey || typeof clientKey !== 'string') {
             throw new Error('MQTT connection key is required')
         }
 
-        if (typeof getCredentials !== 'function') {
-            throw new Error(`MQTT connection "${key}" requires a getCredentials callback`)
+        if (typeof options.getCredentials !== 'function') {
+            throw new Error(`MQTT connection "${clientKey}" requires a getCredentials callback`)
         }
 
-        return this.runClientOperation(key, async () => {
-            if (this.$destroyed) {
-                throw new Error('MqttService has been destroyed')
-            }
+        const observer = this._buildObserver(options)
 
-            if (this.hasClient(key)) {
-                await this._destroyClientUnlocked(key)
+        return this.runClientOperation(clientKey, async () => {
+            if (this.hasClient(clientKey)) {
+                await this._destroyClientUnlocked(clientKey)
             }
+            return this._createAndConnect(clientKey, options, observer)
+        })
+    }
 
-            const managed: ManagedMqttClient = {
-                key,
-                client: null,
-                listeners: new Set(),
-                destroyed: false,
-                intentionalDisconnect: false,
-                status: 'idle',
-                getCredentials: this._buildCredentialsProvider(key, getCredentials),
-                reconnectPolicy: this._normalizeReconnectPolicy({
-                    reconnect,
-                    reconnectPeriod,
-                    hasDynamicCredentials: true
-                }),
-                reconnectAttempt: 0,
-                reconnectGeneration: 0,
-                reconnectTimer: null,
-                subscriptions: new Map(),
-                connectionWaiters: new Set(),
-                terminalFailure: false,
-                lastError: null,
-                connectHandlerPromise: null,
-                handlers: {
-                    onConnect,
-                    onClose,
-                    onOffline,
-                    onError,
-                    onMessage
+    async attachClientObserver (clientKey: string, options: Partial<MqttConnectionOptions> = {}): Promise<MqttObserverHandle> {
+        if (!clientKey || typeof clientKey !== 'string') {
+            throw new Error('MQTT connection key is required')
+        }
+
+        const observer = this._buildObserver(options)
+
+        await this.runClientOperation(clientKey, async () => {
+            const existing = this.$clients.get(clientKey)
+            if (existing && !existing.destroyed) {
+                existing.observers.add(observer)
+                if (existing.status === 'connected' && existing.client?.connected) {
+                    observer.handlers.onConnect?.({} as IConnackPacket, existing.client)
                 }
+                return
             }
 
-            this.$clients.set(key, managed)
+            if (typeof options.getCredentials !== 'function') {
+                throw new Error(`MQTT connection "${clientKey}" requires a getCredentials callback`)
+            }
+            await this._createAndConnect(clientKey, options, observer)
+        })
 
-            try {
-                return await this.connectManagedClient(managed, false)
-            } catch (error) {
-                this.$clients.delete(key)
-                throw error
+        return { key: clientKey, id: observer.id }
+    }
+
+    async detachClientObserver (handle: MqttObserverHandle): Promise<void> {
+        const { key: clientKey, id } = handle
+        await this.runClientOperation(clientKey, async () => {
+            const managed = this.$clients.get(clientKey)
+            if (!managed) return
+
+            for (const observer of managed.observers) {
+                if (observer.id === id) {
+                    managed.observers.delete(observer)
+                    break
+                }
+            }
+
+            if (managed.observers.size === 0) {
+                await this._destroyClientUnlocked(clientKey)
             }
         })
     }
 
-    async publishMessage (key: string, options?: MqttPublishRequest): Promise<void> {
-        const managed = this.getManagedClient(key)
+    private _buildObserver (options: Partial<MqttConnectionOptions>): ManagedMqttObserver {
+        const { onConnect, onClose, onOffline, onError, onMessage } = options
+        return {
+            id: this.$observerSeq++,
+            handlers: { onConnect, onClose, onOffline, onError, onMessage }
+        }
+    }
+
+    private _dispatch<K extends keyof MqttConnectionHandlers> (
+        managed: ManagedMqttClient,
+        event: K,
+        ...args: Parameters<NonNullable<MqttConnectionHandlers[K]>>
+    ): void {
+        for (const observer of managed.observers) {
+            const handler = observer.handlers[event]
+            if (handler) {
+                ;(handler as (...a: unknown[]) => void)(...args)
+            }
+        }
+    }
+
+    private async _createAndConnect (
+        clientKey: string,
+        options: Partial<MqttConnectionOptions>,
+        observer: ManagedMqttObserver
+    ): Promise<MqttClient> {
+        if (this.$destroyed) {
+            throw new Error('MqttService has been destroyed')
+        }
+
+        const { reconnectPeriod = 0, getCredentials, reconnect } = options
+
+        const managed: ManagedMqttClient = {
+            key: clientKey,
+            client: null,
+            listeners: new Set(),
+            destroyed: false,
+            intentionalDisconnect: false,
+            status: 'idle',
+            getCredentials: this._buildCredentialsProvider(clientKey, getCredentials as MqttCredentialProvider),
+            reconnectPolicy: this._normalizeReconnectPolicy({
+                reconnect,
+                reconnectPeriod,
+                hasDynamicCredentials: true
+            }),
+            reconnectAttempt: 0,
+            reconnectGeneration: 0,
+            reconnectTimer: null,
+            subscriptions: new Map(),
+            observers: new Set([observer]),
+            connectionWaiters: new Set(),
+            terminalFailure: false,
+            lastError: null,
+            connectHandlerPromise: null
+        }
+
+        this.$clients.set(clientKey, managed)
+
+        try {
+            return await this.connectManagedClient(managed, false)
+        } catch (error) {
+            this.$clients.delete(clientKey)
+            throw error
+        }
+    }
+
+    async publishMessage (clientKey: string, options?: MqttPublishRequest): Promise<void> {
+        const managed = this.getManagedClient(clientKey)
 
         if (!managed || managed.destroyed) {
-            return Promise.reject(new Error(`MQTT connection "${key}" does not exist`))
+            return Promise.reject(new Error(`MQTT connection "${clientKey}" does not exist`))
         }
 
         const {
@@ -243,7 +309,7 @@ class MqttService extends BaseService implements MqttServiceI {
 
         if (waitForConnection) {
             try {
-                await this.waitForConnection(key, { timeout: connectionTimeout })
+                await this.waitForConnection(clientKey, { timeout: connectionTimeout })
             } catch (error) {
                 if (error instanceof Error && typeof onError === 'function') onError(error)
                 return Promise.reject(error)
@@ -253,7 +319,7 @@ class MqttService extends BaseService implements MqttServiceI {
         const { client } = managed
 
         if (managed.status === 'reconnecting' || !client?.connected) {
-            return Promise.reject(new Error(`MQTT connection "${key}" is not connected`))
+            return Promise.reject(new Error(`MQTT connection "${clientKey}" is not connected`))
         }
 
         return new Promise<void>((resolve, reject) => {
@@ -268,14 +334,14 @@ class MqttService extends BaseService implements MqttServiceI {
         })
     }
 
-    subscribe (key: string, topic: string | string[], options: MqttSubscribeOptions = {}): Promise<void> {
-        const managed = this.$clients.get(key)
+    subscribe (clientKey: string, topic: string | string[], options: MqttSubscribeOptions = {}): Promise<void> {
+        const managed = this.$clients.get(clientKey)
         if (!managed || managed.destroyed) {
-            return Promise.reject(new Error(`MQTT connection "${key}" does not exist`))
+            return Promise.reject(new Error(`MQTT connection "${clientKey}" does not exist`))
         }
 
         if (!managed.client?.connected) {
-            return Promise.reject(new Error(`MQTT connection "${key}" is not connected`))
+            return Promise.reject(new Error(`MQTT connection "${clientKey}" is not connected`))
         }
 
         return new Promise<void>((resolve, reject) => {
@@ -290,14 +356,14 @@ class MqttService extends BaseService implements MqttServiceI {
         })
     }
 
-    unsubscribe (key: string, topic: string | string[]): Promise<void> {
-        const managed = this.getManagedClient(key)
+    unsubscribe (clientKey: string, topic: string | string[]): Promise<void> {
+        const managed = this.getManagedClient(clientKey)
         if (!managed || managed.destroyed) {
             return Promise.resolve()
         }
 
         if (!managed.client?.connected) {
-            return Promise.reject(new Error(`MQTT connection "${key}" is not connected`))
+            return Promise.reject(new Error(`MQTT connection "${clientKey}" is not connected`))
         }
 
         return new Promise<void>((resolve, reject) => {
@@ -312,15 +378,15 @@ class MqttService extends BaseService implements MqttServiceI {
         })
     }
 
-    async endConnection (key: string): Promise<void> {
-        await this.destroyClient(key)
+    async endConnection (clientKey: string): Promise<void> {
+        await this.destroyClient(clientKey)
     }
 
     async destroy (): Promise<void> {
         this.$destroyed = true
 
         const keys = [...this.$clients.keys()]
-        await Promise.all(keys.map(async (key) => this.destroyClient(key)))
+        await Promise.all(keys.map(async (clientKey) => this.destroyClient(clientKey)))
 
         this.$clients.clear()
         this.$mqtt = null
@@ -392,10 +458,10 @@ class MqttService extends BaseService implements MqttServiceI {
         managed.connectionWaiters.clear()
     }
 
-    waitForConnection (key: string, { timeout = 5000 }: MqttWaitForConnectionOptions = {}): Promise<void> {
-        const managed = this.getManagedClient(key)
+    waitForConnection (clientKey: string, { timeout = 5000 }: MqttWaitForConnectionOptions = {}): Promise<void> {
+        const managed = this.getManagedClient(clientKey)
         if (!managed || managed.destroyed) {
-            return Promise.reject(new Error(`MQTT connection "${key}" does not exist`))
+            return Promise.reject(new Error(`MQTT connection "${clientKey}" does not exist`))
         }
 
         if (!Number.isInteger(timeout) || timeout < 0) {
@@ -424,7 +490,7 @@ class MqttService extends BaseService implements MqttServiceI {
             }
 
             waiter.timer = setTimeout(() => {
-                waiter.reject(new Error(`MQTT connection "${key}" did not connect before the timeout elapsed`))
+                waiter.reject(new Error(`MQTT connection "${clientKey}" did not connect before the timeout elapsed`))
             }, timeout)
 
             managed.connectionWaiters.add(waiter)
@@ -505,11 +571,11 @@ class MqttService extends BaseService implements MqttServiceI {
                     await this.replaySubscriptions(managed, client)
                 } catch (error) {
                     if (error instanceof Error) {
-                        managed.handlers.onError?.(error, client)
+                        this._dispatch(managed, 'onError', error, client)
                     }
                 }
 
-                managed.handlers.onConnect?.(connack, client)
+                this._dispatch(managed, 'onConnect', connack, client)
             }
 
             managed.connectHandlerPromise = handleAsync().finally(() => {
@@ -519,46 +585,46 @@ class MqttService extends BaseService implements MqttServiceI {
 
         register('close', () => {
             managed.status = 'disconnected'
-            managed.handlers.onClose?.(client)
+            this._dispatch(managed, 'onClose', client)
             this.scheduleReconnect(managed)
         })
 
         register('disconnect', (packet: IDisconnectPacket) => {
-            managed.handlers.onDisconnect?.(packet, client)
+            this._dispatch(managed, 'onDisconnect', packet, client)
         })
 
         register('offline', () => {
             managed.status = 'disconnected'
-            managed.handlers.onOffline?.(client)
+            this._dispatch(managed, 'onOffline', client)
             this.scheduleReconnect(managed)
         })
 
         register('end', () => {
-            managed.handlers.onEnd?.(client)
+            this._dispatch(managed, 'onEnd', client)
         })
 
         register('reconnect', () => {
-            managed.handlers.onReconnect?.(client)
+            this._dispatch(managed, 'onReconnect', client)
         })
 
         register('error', (error: Error | ErrorWithReasonCode) => {
-            managed.handlers.onError?.(error, client)
+            this._dispatch(managed, 'onError', error, client)
         })
 
         register('message', (topic: string, message: Buffer, packet: IPublishPacket) => {
-            managed.handlers.onMessage?.(topic, message, packet, client)
+            this._dispatch(managed, 'onMessage', topic, message, packet, client)
         })
 
         register('packetsend', (packet: Packet) => {
-            managed.handlers.onPacketSend?.(packet, client)
+            this._dispatch(managed, 'onPacketSend', packet, client)
         })
 
         register('packetreceive', (packet: Packet) => {
-            managed.handlers.onPacketReceive?.(packet, client)
+            this._dispatch(managed, 'onPacketReceive', packet, client)
         })
 
         register('outgoingEmpty', () => {
-            managed.handlers.onOutgoingEmpty?.(client)
+            this._dispatch(managed, 'onOutgoingEmpty', client)
         })
     }
 
@@ -593,9 +659,9 @@ class MqttService extends BaseService implements MqttServiceI {
         }, delay)
     }
 
-    async reconnectClient (key: string): Promise<void> {
-        await this.runClientOperation(key, async () => {
-            const managed = this.$clients.get(key)
+    async reconnectClient (clientKey: string): Promise<void> {
+        await this.runClientOperation(clientKey, async () => {
+            const managed = this.$clients.get(clientKey)
             if (!managed || managed.destroyed || managed.intentionalDisconnect || this.$destroyed) {
                 return
             }
@@ -619,7 +685,7 @@ class MqttService extends BaseService implements MqttServiceI {
                     this.clearReconnectTimer(managed)
                     this.rejectConnectionWaiters(managed, normalizedError)
                 }
-                managed.handlers.onError?.(normalizedError, managed.client)
+                this._dispatch(managed, 'onError', normalizedError, managed.client)
                 if (!managed.terminalFailure) {
                     this.scheduleReconnect(managed)
                 }
@@ -663,7 +729,7 @@ class MqttService extends BaseService implements MqttServiceI {
         }))
     }
 
-    private _buildCredentialsProvider (key: string, getCredentials: MqttCredentialProvider): MqttCredentialProvider {
+    private _buildCredentialsProvider (clientKey: string, getCredentials: MqttCredentialProvider): MqttCredentialProvider {
         return async () => {
             let credentials: MqttCredentials
             try {
@@ -671,28 +737,28 @@ class MqttService extends BaseService implements MqttServiceI {
             } catch (error) {
                 throw this._asTerminalConnectionError(
                     error,
-                    `MQTT credential provider for connection "${key}" failed`
+                    `MQTT credential provider for connection "${clientKey}" failed`
                 )
             }
 
             if (!credentials || typeof credentials !== 'object') {
-                throw this._createTerminalConnectionError(`MQTT credential provider for connection "${key}" must resolve to an object`)
+                throw this._createTerminalConnectionError(`MQTT credential provider for connection "${clientKey}" must resolve to an object`)
             }
 
             if (!this._isValidUrl(credentials.url)) {
-                throw this._createTerminalConnectionError(`MQTT credential provider for connection "${key}" must return a valid url`)
+                throw this._createTerminalConnectionError(`MQTT credential provider for connection "${clientKey}" must return a valid url`)
             }
 
             if (typeof credentials.username !== 'string' || !credentials.username.trim()) {
-                throw this._createTerminalConnectionError(`MQTT credential provider for connection "${key}" must return a non-empty username`)
+                throw this._createTerminalConnectionError(`MQTT credential provider for connection "${clientKey}" must return a non-empty username`)
             }
 
             if (typeof credentials.password !== 'string' || !credentials.password.trim()) {
-                throw this._createTerminalConnectionError(`MQTT credential provider for connection "${key}" must return a non-empty password`)
+                throw this._createTerminalConnectionError(`MQTT credential provider for connection "${clientKey}" must return a non-empty password`)
             }
 
             if (credentials.clientId !== undefined && (typeof credentials.clientId !== 'string' || !credentials.clientId.trim())) {
-                throw this._createTerminalConnectionError(`MQTT credential provider for connection "${key}" must return a non-empty clientId when provided`)
+                throw this._createTerminalConnectionError(`MQTT credential provider for connection "${clientKey}" must return a non-empty clientId when provided`)
             }
 
             if (!credentials.clientId) {
@@ -719,8 +785,8 @@ class MqttService extends BaseService implements MqttServiceI {
         return error instanceof Error && (error as TerminalConnectionError).code === 'MQTT_TERMINAL_CONNECTION_ERROR'
     }
 
-    private async _destroyClientUnlocked (key: string): Promise<void> {
-        const managed = this.$clients.get(key)
+    private async _destroyClientUnlocked (clientKey: string): Promise<void> {
+        const managed = this.$clients.get(clientKey)
         if (!managed) return
 
         managed.destroyed = true
@@ -728,7 +794,7 @@ class MqttService extends BaseService implements MqttServiceI {
         managed.status = 'disconnected'
 
         this.clearReconnectTimer(managed)
-        this.rejectConnectionWaiters(managed, new Error(`MQTT connection "${key}" has been closed`))
+        this.rejectConnectionWaiters(managed, new Error(`MQTT connection "${clientKey}" has been closed`))
 
         for (const off of managed.listeners) {
             try {
@@ -750,7 +816,7 @@ class MqttService extends BaseService implements MqttServiceI {
             this._killStaleClient(client)
         }
 
-        this.$clients.delete(key)
+        this.$clients.delete(clientKey)
     }
 
     private _killStaleClient (client: MqttClient) {
@@ -909,22 +975,22 @@ class MqttService extends BaseService implements MqttServiceI {
         }
 
         const normalized: Record<string, string | string[]> = {}
-        for (const [key, value] of Object.entries(userProperties)) {
-            if (!key.trim()) {
+        for (const [clientKey, value] of Object.entries(userProperties)) {
+            if (!clientKey.trim()) {
                 throw new TypeError('MQTT publish userProperties keys must be non-empty strings')
             }
 
             if (typeof value === 'string') {
-                normalized[key] = value
+                normalized[clientKey] = value
                 continue
             }
 
             if (Array.isArray(value) && value.every(item => typeof item === 'string')) {
-                normalized[key] = value
+                normalized[clientKey] = value
                 continue
             }
 
-            throw new TypeError(`MQTT publish userProperties["${key}"] must be a string or string[]`)
+            throw new TypeError(`MQTT publish userProperties["${clientKey}"] must be a string or string[]`)
         }
 
         return Object.keys(normalized).length ? normalized : undefined
diff --git a/frontend/src/types/services/mqtt.types.ts b/frontend/src/types/services/mqtt.types.ts
index e14dfbb082..2e74a0672f 100644
--- a/frontend/src/types/services/mqtt.types.ts
+++ b/frontend/src/types/services/mqtt.types.ts
@@ -87,6 +87,16 @@ export interface MqttConnectionWaiter {
     timer: ReturnType<typeof setTimeout> | null
 }
 
+export interface ManagedMqttObserver {
+    id: number
+    handlers: MqttConnectionHandlers
+}
+
+export interface MqttObserverHandle {
+    key: string
+    id: number
+}
+
 export interface ManagedMqttClient {
     key: string
     client: MqttClient | null
@@ -100,7 +110,7 @@ export interface ManagedMqttClient {
     reconnectGeneration: number
     reconnectTimer: ReturnType<typeof setTimeout> | null
     subscriptions: Map<string, ManagedMqttSubscription>
-    handlers: MqttConnectionHandlers
+    observers: Set<ManagedMqttObserver>
     connectionWaiters: Set<MqttConnectionWaiter>
     terminalFailure: boolean
     lastError: Error | null
@@ -108,14 +118,16 @@ export interface ManagedMqttClient {
 }
 
 export interface MqttServiceI extends AppService {
-    createClient(key: string, options?: Partial<MqttConnectionOptions>): Promise<MqttClient>
-    destroyClient(key: string): Promise<void>
-    getManagedClient(key: string): ManagedMqttClient | null
-    hasClient(key: string): boolean
-    publishMessage(key: string, options: MqttPublishRequest): Promise<void>
-    subscribe(key: string, topic: string | string[], options?: MqttSubscribeOptions): Promise<void>
-    unsubscribe(key: string, topic: string | string[]): Promise<void>
-    endConnection(key: string): Promise<void>
-    waitForConnection(key: string, options?: MqttWaitForConnectionOptions): Promise<void>
+    createClient(clientKey: string, options?: Partial<MqttConnectionOptions>): Promise<MqttClient>
+    destroyClient(clientKey: string): Promise<void>
+    attachClientObserver(clientKey: string, options?: Partial<MqttConnectionOptions>): Promise<MqttObserverHandle>
+    detachClientObserver(handle: MqttObserverHandle): Promise<void>
+    getManagedClient(clientKey: string): ManagedMqttClient | null
+    hasClient(clientKey: string): boolean
+    publishMessage(clientKey: string, options: MqttPublishRequest): Promise<void>
+    subscribe(clientKey: string, topic: string | string[], options?: MqttSubscribeOptions): Promise<void>
+    unsubscribe(clientKey: string, topic: string | string[]): Promise<void>
+    endConnection(clientKey: string): Promise<void>
+    waitForConnection(clientKey: string, options?: MqttWaitForConnectionOptions): Promise<void>
     reset(): Promise<void>
 }
diff --git a/test/unit/frontend/services/mqtt.service.spec.js b/test/unit/frontend/services/mqtt.service.spec.js
index 4465344d09..a77cbcf264 100644
--- a/test/unit/frontend/services/mqtt.service.spec.js
+++ b/test/unit/frontend/services/mqtt.service.spec.js
@@ -944,4 +944,80 @@ describe('MqttService', async () => {
         await destroyMqttService()
         await destroyMqttService()
     })
+
+    describe('attachClientObserver / detachClientObserver (shared refcounted connections)', () => {
+        const creds = async () => ({ url: 'mqtt://example.com', username: 'user', password: 'pass' })
+
+        test('a second attach on the same key reuses the existing client', async () => {
+            const service = createMqttService({ app: {}, services: {}, router: {} })
+            const client = createMockClient()
+            mockConnect.mockReturnValue(client)
+
+            const first = await service.attachClientObserver('team:team-1', { getCredentials: creds })
+            const second = await service.attachClientObserver('team:team-1', { getCredentials: creds })
+
+            expect(mockConnect).toHaveBeenCalledTimes(1)
+            expect(first.id).not.toBe(second.id)
+            expect(service.getManagedClient('team:team-1').observers.size).toBe(2)
+        })
+
+        test('detaching a non-last attachment keeps the connection alive', async () => {
+            const service = createMqttService({ app: {}, services: {}, router: {} })
+            mockConnect.mockReturnValue(createMockClient())
+
+            const first = await service.attachClientObserver('team:team-1', { getCredentials: creds })
+            const second = await service.attachClientObserver('team:team-1', { getCredentials: creds })
+
+            await service.detachClientObserver(first)
+            expect(service.hasClient('team:team-1')).toBe(true)
+            expect(service.getManagedClient('team:team-1').observers.size).toBe(1)
+
+            await service.detachClientObserver(second)
+            expect(service.hasClient('team:team-1')).toBe(false)
+        })
+
+        test('detaching an unknown handle is a no-op', async () => {
+            const service = createMqttService({ app: {}, services: {}, router: {} })
+            mockConnect.mockReturnValue(createMockClient())
+            await service.attachClientObserver('team:team-1', { getCredentials: creds })
+
+            await service.detachClientObserver({ key: 'team:team-1', id: 999 })
+            expect(service.hasClient('team:team-1')).toBe(true)
+            expect(service.getManagedClient('team:team-1').observers.size).toBe(1)
+        })
+
+        test('a late attach onto an already-connected client fires its onConnect immediately', async () => {
+            const service = createMqttService({ app: {}, services: {}, router: {} })
+            const client = createMockClient()
+            mockConnect.mockReturnValue(client)
+
+            await service.attachClientObserver('team:team-1', { getCredentials: creds })
+            client.emit('connect', { cmd: 'connack' })
+
+            const onConnect = vi.fn()
+            await service.attachClientObserver('team:team-1', { getCredentials: creds, onConnect })
+            expect(onConnect).toHaveBeenCalledTimes(1)
+        })
+
+        test('messages fan out to every observer', async () => {
+            const service = createMqttService({ app: {}, services: {}, router: {} })
+            const client = createMockClient()
+            mockConnect.mockReturnValue(client)
+
+            const onMessageA = vi.fn()
+            const onMessageB = vi.fn()
+            await service.attachClientObserver('team:team-1', { getCredentials: creds, onMessage: onMessageA })
+            await service.attachClientObserver('team:team-1', { getCredentials: creds, onMessage: onMessageB })
+
+            client.emit('message', 'ff/v1/team-1/t/updated', Buffer.from('{}'), { cmd: 'publish' })
+
+            expect(onMessageA).toHaveBeenCalledTimes(1)
+            expect(onMessageB).toHaveBeenCalledTimes(1)
+        })
+
+        test('requires a credential provider when creating a fresh connection', async () => {
+            const service = createMqttService({ app: {}, services: {}, router: {} })
+            await expect(service.attachClientObserver('team:team-1', {})).rejects.toThrow('requires a getCredentials callback')
+        })
+    })
 })