diff --git a/apple/Plugins/KafkaDriverPlugin/KafkaPluginDriver+LiveMessages.swift b/apple/Plugins/KafkaDriverPlugin/KafkaPluginDriver+LiveMessages.swift
new file mode 100644
index 0000000000..c71a3f8b62
--- /dev/null
+++ b/apple/Plugins/KafkaDriverPlugin/KafkaPluginDriver+LiveMessages.swift
@@ -0,0 +1,140 @@
+import Foundation
+import os
+import TableProPluginKit
+
+// Follows topics from each partition's end offset with plain Fetch requests and no consumer group, so it never
+// joins a group or moves a group's committed offsets.
+extension KafkaPluginDriver: PluginLiveMessaging {
+    var liveMessageKind: PluginLiveMessageKind? { .topics }
+
+    func subscribe(
+        _ request: PluginLiveSubscriptionRequest,
+        sessions: any PluginLiveSessionSource
+    ) async throws -> any PluginLiveSubscription {
+        guard let follower = try await sessions.openSession() as? KafkaPluginDriver else {
+            throw KafkaError.notConnected
+        }
+        let cluster = follower.liveCluster
+        var cursors: [KafkaFollowCursor] = []
+        for topicName in request.channels {
+            let metadata = try await cluster.metadata(topics: [topicName], refresh: true)
+            let topic = try metadata.requireTopic(named: topicName)
+            let partitions = topic.partitions.map(\.index).sorted()
+            let ends = try await KafkaOffsetsRequest.listOffsets(
+                topic: topicName,
+                partitions: partitions,
+                timestamp: KafkaOffsetsRequest.latestTimestamp,
+                cluster: cluster
+            )
+            cursors += KafkaFollowCursor.cursors(topic: topicName, ends: ends)
+        }
+        return KafkaLiveSubscription(cluster: cluster, cursors: cursors)
+    }
+}
+
+struct KafkaFollowCursor: Sendable, Equatable {
+    let topic: String
+    let partition: Int32
+    let nextOffset: Int64
+
+    static func cursors(topic: String, ends: [Int32: Int64]) -> [KafkaFollowCursor] {
+        ends.keys.sorted().compactMap { partition in
+            ends[partition].map { KafkaFollowCursor(topic: topic, partition: partition, nextOffset: $0) }
+        }
+    }
+
+    static func message(from record: KafkaRecord, topic: String) -> PluginLiveMessage {
+        PluginLiveMessage(
+            timestamp: Date(timeIntervalSince1970: TimeInterval(record.timestamp) / 1_000),
+            channel: topic,
+            payload: text(record.value) ?? "",
+            key: text(record.key),
+            partition: Int(record.partition),
+            offset: record.offset
+        )
+    }
+
+    // A payload that is not UTF-8 reads as hex, the way the topic grid shows it.
+    static func text(_ data: Data?) -> String? {
+        guard let data else { return nil }
+        return String(data: data, encoding: .utf8) ?? data.map { String(format: "%02X", $0) }.joined()
+    }
+}
+
+final class KafkaLiveSubscription: PluginLiveSubscription, @unchecked Sendable {
+    private static let recordsPerFetch = 500
+    private static let idleWait: Duration = .milliseconds(500)
+    private static let logger = Logger(subsystem: "com.TablePro.KafkaDriver", category: "LiveMessages")
+
+    let messages: AsyncThrowingStream<PluginLiveMessage, Error>
+    private let cluster: KafkaCluster
+    private let reader: Task<Void, Never>
+
+    init(cluster: KafkaCluster, cursors: [KafkaFollowCursor]) {
+        self.cluster = cluster
+        let (stream, continuation) = AsyncThrowingStream<PluginLiveMessage, Error>.makeStream()
+        messages = stream
+        reader = Task {
+            do {
+                // Each Fetch waits up to 500 ms on the broker when the partition has nothing new, so no loop spins.
+                try await withThrowingTaskGroup(of: Void.self) { group in
+                    for cursor in cursors {
+                        group.addTask {
+                            var offset = cursor.nextOffset
+                            while !Task.isCancelled {
+                                let started = ContinuousClock.now
+                                let page = try await KafkaFetchRequest.fetch(
+                                    topic: cursor.topic,
+                                    partition: cursor.partition,
+                                    startOffset: offset,
+                                    maximumRecords: Self.recordsPerFetch,
+                                    cluster: cluster
+                                )
+                                for record in page.records {
+                                    continuation.yield(KafkaFollowCursor.message(from: record, topic: cursor.topic))
+                                }
+                                if let last = page.records.last {
+                                    offset = last.offset + 1
+                                } else if ContinuousClock.now - started < Self.idleWait / 2 {
+                                    // A trailing transaction marker answers at once and reads as no record,
+                                    // where an idle partition makes the broker wait out the long poll.
+                                    try await Task.sleep(for: Self.idleWait)
+                                }
+                            }
+                        }
+                    }
+                    // The first partition that fails ends the follow, and leaving the group cancels the others.
+                    while try await group.next() != nil {}
+                }
+                continuation.finish()
+            } catch {
+                guard !Task.isCancelled else {
+                    continuation.finish()
+                    return
+                }
+                Self.logger.info("Topic follow ended: \(String(describing: type(of: error)), privacy: .public) \(error.localizedDescription, privacy: .private)")
+                continuation.finish(throwing: error)
+            }
+        }
+    }
+
+    func publish(_ payload: String, to channel: String) async throws {
+        let metadata = try await cluster.metadata(topics: [channel])
+        let topic = try metadata.requireTopic(named: channel)
+        guard let partition = topic.partitions.map(\.index).min() else { throw KafkaError.unknownTopic(channel) }
+        _ = try await KafkaProduceRequest.produce(
+            topic: channel,
+            partition: partition,
+            key: nil,
+            value: Data(payload.utf8),
+            headers: [],
+            timestamp: Int64(Date().timeIntervalSince1970 * 1_000),
+            cluster: cluster
+        )
+    }
+
+    func close() async {
+        reader.cancel()
+        _ = await reader.value
+    }
+}
diff --git a/apple/Plugins/KafkaDriverPlugin/KafkaPluginDriver.swift b/apple/Plugins/KafkaDriverPlugin/KafkaPluginDriver.swift
index 85462fc501..d532868e48 100644
--- a/apple/Plugins/KafkaDriverPlugin/KafkaPluginDriver.swift
+++ b/apple/Plugins/KafkaDriverPlugin/KafkaPluginDriver.swift
@@ -13,6 +13,8 @@ final class KafkaPluginDriver: PluginDatabaseDriver, @unchecked Sendable {
     private let state = KafkaDriverState()
     private let anchors = KafkaAnchorStore()
 
+    var liveCluster: KafkaCluster { cluster }
+
     var capabilities: PluginCapabilities { [] }
 
     /// The config is read here and not retained. It carries the password, and nothing after
diff --git a/apple/Plugins/PostgreSQLDriverPlugin/LibPQDriverCore.swift b/apple/Plugins/PostgreSQLDriverPlugin/LibPQDriverCore.swift
index 6355c2c140..6f02976256 100644
--- a/apple/Plugins/PostgreSQLDriverPlugin/LibPQDriverCore.swift
+++ b/apple/Plugins/PostgreSQLDriverPlugin/LibPQDriverCore.swift
@@ -304,6 +304,11 @@ final class LibPQDriverCore: @unchecked Sendable {
         libpqConnection?.observeServerNotices(observer)
     }
 
+    func nextNotifications(waitMicroseconds: Int64) async throws -> [LibPQNotification] {
+        guard let connection = libpqConnection else { throw LibPQPluginError.notConnected }
+        return try await connection.nextNotifications(waitMicroseconds: waitMicroseconds)
+    }
+
     func cancelQueryKeepingReply() {
         libpqConnection?.requestServerCancelKeepingReply()
     }
diff --git a/apple/Plugins/PostgreSQLDriverPlugin/LibPQPluginConnection.swift b/apple/Plugins/PostgreSQLDriverPlugin/LibPQPluginConnection.swift
index abc295d78d..0c2d48583d 100644
--- a/apple/Plugins/PostgreSQLDriverPlugin/LibPQPluginConnection.swift
+++ b/apple/Plugins/PostgreSQLDriverPlugin/LibPQPluginConnection.swift
@@ -628,6 +628,39 @@ final class LibPQPluginConnection: @unchecked Sendable {
         typeNames.merge(names)
     }
 
+    // MARK: - Notifications
+
+    // Waits at most `waitMicroseconds` on the queue, so a NOTIFY sent on this session gets its turn between waits.
+    func nextNotifications(waitMicroseconds: Int64) async throws -> [LibPQNotification] {
+        try await pluginDispatchAsync(on: queue) { [self] in
+            guard !isShuttingDown, let conn = connectionHandle else { throw LibPQPluginError.notConnected }
+            var received = Self.drainNotifications(conn)
+            guard received.isEmpty else { return received }
+            let socket = PQsocket(conn)
+            guard socket >= 0 else { throw connectionError(from: conn) }
+            let deadline = pg_usec_time_t(Int64(PQgetCurrentTimeUSec()) + waitMicroseconds)
+            guard PQsocketPoll(socket, 1, 0, deadline) >= 0 else { throw connectionError(from: conn) }
+            guard PQconsumeInput(conn) != 0, PQstatus(conn) == CONNECTION_OK else {
+                stateLock.withLock { _hasLostConnection = true }
+                throw connectionError(from: conn)
+            }
+            received = Self.drainNotifications(conn)
+            return received
+        }
+    }
+
+    private static func drainNotifications(_ conn: OpaquePointer) -> [LibPQNotification] {
+        var received: [LibPQNotification] = []
+        while let notify = PQnotifies(conn) {
+            received.append(LibPQNotification(
+                channel: String(cString: notify.pointee.relname),
+                payload: notify.pointee.extra.map { String(cString: $0) } ?? ""
+            ))
+            PQfreemem(notify)
+        }
+        return received
+    }
+
     // MARK: - Query Cancellation
 
     func cancelCurrentQuery() {
diff --git a/apple/Plugins/PostgreSQLDriverPlugin/PGlitePluginDriver.swift b/apple/Plugins/PostgreSQLDriverPlugin/PGlitePluginDriver.swift
index 9c29dd68d6..39684d302e 100644
--- a/apple/Plugins/PostgreSQLDriverPlugin/PGlitePluginDriver.swift
+++ b/apple/Plugins/PostgreSQLDriverPlugin/PGlitePluginDriver.swift
@@ -28,6 +28,8 @@ final class PGlitePluginDriver: PostgreSQLPluginDriver {
 
     // pldebugger is a server C extension and needs a second session; PGlite has neither.
     override var canDebugRoutines: Bool { false }
+    // A second session would open a different, empty database, so a NOTIFY from the person's session never arrives.
+    override var canFollowNotifications: Bool { false }
 
     override func connect() async throws {
         do {
diff --git a/apple/Plugins/PostgreSQLDriverPlugin/PostgreSQLPluginDriver+LiveMessages.swift b/apple/Plugins/PostgreSQLDriverPlugin/PostgreSQLPluginDriver+LiveMessages.swift
new file mode 100644
index 0000000000..36d7f6ed9b
--- /dev/null
+++ b/apple/Plugins/PostgreSQLDriverPlugin/PostgreSQLPluginDriver+LiveMessages.swift
@@ -0,0 +1,96 @@
+//
+//  PostgreSQLPluginDriver+LiveMessages.swift
+//  PostgreSQLDriverPlugin
+//
+
+import Foundation
+import os
+import TableProLogRedaction
+import TableProPluginKit
+
+struct LibPQNotification: Sendable, Equatable {
+    let channel: String
+    let payload: String
+}
+
+// Redshift and CockroachDB are siblings of this class and never conform; PGlite turns it off through
+// `canFollowNotifications`.
+extension PostgreSQLPluginDriver: PluginLiveMessaging {
+    var liveMessageKind: PluginLiveMessageKind? {
+        canFollowNotifications ? .notifications : nil
+    }
+
+    func subscribe(
+        _ request: PluginLiveSubscriptionRequest,
+        sessions: any PluginLiveSessionSource
+    ) async throws -> any PluginLiveSubscription {
+        guard canFollowNotifications,
+              let listener = try await sessions.openSession() as? PostgreSQLPluginDriver
+        else { throw LibPQPluginError.notConnected }
+        listener.core.interruptBlockedCallOnDisconnect()
+        for statement in PostgreSQLListenStatements.listen(request.channels) {
+            _ = try await listener.execute(query: statement)
+        }
+        return PostgreSQLLiveSubscription(listener: listener)
+    }
+}
+
+enum PostgreSQLListenStatements {
+    static let notifyStatement = "SELECT pg_catalog.pg_notify($1, $2)"
+
+    // LISTEN takes an identifier, never a parameter, so each channel is quoted as one.
+    static func listen(_ channels: [String]) -> [String] {
+        channels.map { "LISTEN \(PostgreSQLObjectQueries.quoteIdentifier($0))" }
+    }
+}
+
+final class PostgreSQLLiveSubscription: PluginLiveSubscription, @unchecked Sendable {
+    // Short, so a NOTIFY sent on this session runs between two waits instead of behind a long one.
+    private static let waitMicroseconds: Int64 = 250_000
+    private static let logger = Logger(subsystem: "com.TablePro.PostgreSQLDriver", category: "LiveMessages")
+
+    let messages: AsyncThrowingStream<PluginLiveMessage, Error>
+    private let listener: PostgreSQLPluginDriver
+    private let reader: Task<Void, Never>
+
+    init(listener: PostgreSQLPluginDriver) {
+        self.listener = listener
+        let (stream, continuation) = AsyncThrowingStream<PluginLiveMessage, Error>.makeStream()
+        messages = stream
+        reader = Task {
+            do {
+                while !Task.isCancelled {
+                    let batch = try await listener.core.nextNotifications(waitMicroseconds: Self.waitMicroseconds)
+                    let arrived = Date()
+                    for notification in batch {
+                        continuation.yield(PluginLiveMessage(
+                            timestamp: arrived,
+                            channel: notification.channel,
+                            payload: notification.payload
+                        ))
+                    }
+                }
+                continuation.finish()
+            } catch {
+                guard !Task.isCancelled else {
+                    continuation.finish()
+                    return
+                }
+                Self.logger.info("Notification read ended: \(LogRedaction.publicDescription(of: error), privacy: .public) \(error.localizedDescription, privacy: .private)")
+                continuation.finish(throwing: error)
+            }
+        }
+    }
+
+    func publish(_ payload: String, to channel: String) async throws {
+        _ = try await listener.executeParameterized(
+            query: PostgreSQLListenStatements.notifyStatement,
+            parameters: [.text(channel), .text(payload)]
+        )
+    }
+
+    func close() async {
+        reader.cancel()
+        _ = await reader.value
+    }
+}
diff --git a/apple/Plugins/PostgreSQLDriverPlugin/PostgreSQLPluginDriver.swift b/apple/Plugins/PostgreSQLDriverPlugin/PostgreSQLPluginDriver.swift
index 16732a7857..52e28d5670 100644
--- a/apple/Plugins/PostgreSQLDriverPlugin/PostgreSQLPluginDriver.swift
+++ b/apple/Plugins/PostgreSQLDriverPlugin/PostgreSQLPluginDriver.swift
@@ -56,6 +56,7 @@ class PostgreSQLPluginDriver: LibPQBackedDriver, @unchecked Sendable {
 
     // In the class body so PGlite, which inherits the debugger conformance, can turn it off.
     var canDebugRoutines: Bool { true }
+    var canFollowNotifications: Bool { true }
 
     init(config: DriverConnectionConfig, singleConnectionMode: Bool = false) {
         self.core = LibPQDriverCore(config: config, singleConnectionMode: singleConnectionMode)
diff --git a/apple/Plugins/PostgreSQLDriverTests/PostgreSQLListenStatementsTests.swift b/apple/Plugins/PostgreSQLDriverTests/PostgreSQLListenStatementsTests.swift
new file mode 100644
index 0000000000..f1bd371a1e
--- /dev/null
+++ b/apple/Plugins/PostgreSQLDriverTests/PostgreSQLListenStatementsTests.swift
@@ -0,0 +1,18 @@
+import Foundation
+import TableProPluginKit
+import Testing
+
+struct PostgreSQLListenStatementsTests {
+    @Test("Each channel is LISTENed as a quoted identifier, so case and quotes survive")
+    func listenQuotesIdentifiers() {
+        #expect(PostgreSQLListenStatements.listen(["Orders", "a\"b"]) == [
+            "LISTEN \"Orders\"",
+            "LISTEN \"a\"\"b\"",
+        ])
+    }
+
+    @Test("NOTIFY goes through pg_notify with bound parameters, never a spliced payload")
+    func notifyIsParameterized() {
+        #expect(PostgreSQLListenStatements.notifyStatement == "SELECT pg_catalog.pg_notify($1, $2)")
+    }
+}
diff --git a/apple/Plugins/RedisDriverPlugin/RedisClusterChannel.swift b/apple/Plugins/RedisDriverPlugin/RedisClusterChannel.swift
index d8411d804c..9564c102b2 100644
--- a/apple/Plugins/RedisDriverPlugin/RedisClusterChannel.swift
+++ b/apple/Plugins/RedisDriverPlugin/RedisClusterChannel.swift
@@ -76,6 +76,11 @@ final class RedisClusterChannel: RedisCommandChannel, @unchecked Sendable {
 
     var supportsDatabaseSelection: Bool { servedDatabaseCount > 1 }
 
+    /// Any node: the cluster bus delivers every classic PUBLISH to every node's subscribers.
+    func pubSubConnection() -> RedisPluginConnection? {
+        lock.withLock { connections.values.lazy.compactMap { $0 as? RedisPluginConnection }.first }
+    }
+
     private var servedDatabaseCount: Int {
         lock.lock()
         defer { lock.unlock() }
diff --git a/apple/Plugins/RedisDriverPlugin/RedisCommandChannel.swift b/apple/Plugins/RedisDriverPlugin/RedisCommandChannel.swift
index 240d88868e..db79b92cfa 100644
--- a/apple/Plugins/RedisDriverPlugin/RedisCommandChannel.swift
+++ b/apple/Plugins/RedisDriverPlugin/RedisCommandChannel.swift
@@ -85,9 +85,14 @@ protocol RedisCommandChannel: AnyObject, Sendable {
     /// Sentinel needs this because a demoted primary keeps answering `role:master` and keeps
     /// accepting writes for several seconds after the quorum has moved on.
     func verifyStillPrimary() async throws
+
+    /// The one node connection a dedicated driver subscribes on; nil when the channel has none.
+    func pubSubConnection() -> RedisPluginConnection?
 }
 
 extension RedisCommandChannel {
+    func pubSubConnection() -> RedisPluginConnection? { nil }
+
     var supportsDatabaseSelection: Bool { true }
     var supportsTransactions: Bool { true }
     var partitionsKeyspace: Bool { false }
diff --git a/apple/Plugins/RedisDriverPlugin/RedisPluginConnection.swift b/apple/Plugins/RedisDriverPlugin/RedisPluginConnection.swift
index ff0e1fe1e5..c63ca1f85b 100644
--- a/apple/Plugins/RedisDriverPlugin/RedisPluginConnection.swift
+++ b/apple/Plugins/RedisDriverPlugin/RedisPluginConnection.swift
@@ -412,6 +412,80 @@ final class RedisPluginConnection: RedisCommandChannel, @unchecked Sendable {
 
 extension RedisPluginConnection: RedisClusterNodeConnection {}
 
+// MARK: - Pub/Sub
+
+extension RedisPluginConnection {
+    func pubSubConnection() -> RedisPluginConnection? { self }
+
+    /// Sends without reading: a subscribed connection answers with one confirmation per channel,
+    /// which `nextPushReplies` reads with the messages.
+    func sendSubscription(_ args: [Data]) async throws {
+        try await pluginDispatchAsync(on: queue) { [self] in
+            #if canImport(CRedis)
+            let ctx = try subscribedContext()
+            let argc = Int32(args.count)
+            try RedisCommandArgv.withPointers(args) { argv, argvlen in
+                guard redisAppendCommandArgv(ctx, argc, argv, argvlen) == REDIS_OK else {
+                    throw transportFailure(ctx, delivered: false)
+                }
+            }
+            var done: Int32 = 0
+            while done == 0 {
+                guard redisBufferWrite(ctx, &done) == REDIS_OK else {
+                    throw transportFailure(ctx, delivered: false)
+                }
+            }
+            #else
+            throw RedisPluginError.notConnected
+            #endif
+        }
+    }
+
+    /// Waits at most `waitMilliseconds` for the socket, never in hiredis, whose read timeout would
+    /// leave the context unusable.
+    func nextPushReplies(waitMilliseconds: Int32) async throws -> [RedisReply] {
+        try await pluginDispatchAsync(on: queue) { [self] in
+            #if canImport(CRedis)
+            let ctx = try subscribedContext()
+            var replies = try bufferedReplies(ctx)
+            guard replies.isEmpty else { return replies }
+            var descriptor = pollfd(fd: ctx.pointee.fd, events: Int16(POLLIN), revents: 0)
+            let ready = poll(&descriptor, 1, waitMilliseconds)
+            guard ready >= 0 || errno == EINTR else { throw transportFailure(ctx, delivered: true) }
+            guard ready > 0 else { return [] }
+            guard redisBufferRead(ctx) == REDIS_OK else { throw transportFailure(ctx, delivered: true) }
+            replies = try bufferedReplies(ctx)
+            return replies
+            #else
+            throw RedisPluginError.notConnected
+            #endif
+        }
+    }
+
+    #if canImport(CRedis)
+    private func subscribedContext() throws -> UnsafeMutablePointer<redisContext> {
+        guard !isShuttingDown else { throw RedisPluginError.notConnected }
+        stateLock.lock()
+        defer { stateLock.unlock() }
+        guard let ctx = context else { throw RedisPluginError.notConnected }
+        return ctx
+    }
+
+    private func bufferedReplies(_ ctx: UnsafeMutablePointer<redisContext>) throws -> [RedisReply] {
+        var replies: [RedisReply] = []
+        while true {
+            var rawReply: UnsafeMutableRawPointer?
+            guard redisGetReplyFromReader(ctx, &rawReply) == REDIS_OK else {
+                throw transportFailure(ctx, delivered: true)
+            }
+            guard let reply = rawReply else { return replies }
+            replies.append(parseReply(reply.assumingMemoryBound(to: redisReply.self)))
+            freeReplyObject(reply)
+        }
+    }
+    #endif
+}
+
 // MARK: - Synchronous Helpers (must be called on the serial queue)
 
 #if canImport(CRedis)
diff --git a/apple/Plugins/RedisDriverPlugin/RedisPluginDriver+LiveMessages.swift b/apple/Plugins/RedisDriverPlugin/RedisPluginDriver+LiveMessages.swift
new file mode 100644
index 0000000000..43a610d083
--- /dev/null
+++ b/apple/Plugins/RedisDriverPlugin/RedisPluginDriver+LiveMessages.swift
@@ -0,0 +1,127 @@
+//
+//  RedisPluginDriver+LiveMessages.swift
+//  RedisDriverPlugin
+//
+
+import Foundation
+import os
+import TableProPluginKit
+
+extension RedisPluginDriver: PluginLiveMessaging {
+    var liveMessageKind: PluginLiveMessageKind? { .pubSub }
+
+    func subscribe(
+        _ request: PluginLiveSubscriptionRequest,
+        sessions: any PluginLiveSessionSource
+    ) async throws -> any PluginLiveSubscription {
+        guard let subscriber = try await sessions.openSession() as? RedisPluginDriver,
+              let connection = subscriber.pubSubConnection
+        else { throw RedisPluginError.notConnected }
+        for command in RedisLivePush.subscribeCommands(for: request) {
+            try await connection.sendSubscription(command)
+        }
+        return RedisLiveSubscription(subscriber: connection, sessions: sessions)
+    }
+}
+
+enum RedisLivePush {
+    static func subscribeCommands(for request: PluginLiveSubscriptionRequest) -> [[Data]] {
+        var commands: [[Data]] = []
+        if !request.channels.isEmpty {
+            commands.append([Data("SUBSCRIBE".utf8)] + request.channels.map { Data($0.utf8) })
+        }
+        if !request.patterns.isEmpty {
+            commands.append([Data("PSUBSCRIBE".utf8)] + request.patterns.map { Data($0.utf8) })
+        }
+        return commands
+    }
+
+    // nil for a confirmation or anything else a subscribed connection answers that is not a message.
+    static func message(from reply: RedisReply, receivedAt: Date) throws -> PluginLiveMessage? {
+        if let error = reply.errorMessage {
+            throw RedisPluginError(code: 0, message: error, refusedByServer: true)
+        }
+        guard let items = reply.arrayValue, let kind = items.first?.stringValue?.lowercased() else { return nil }
+        switch (kind, items.count) {
+        case ("message", 3):
+            return PluginLiveMessage(
+                timestamp: receivedAt,
+                channel: items[1].displayText,
+                payload: items[2].displayText
+            )
+        case ("pmessage", 4):
+            return PluginLiveMessage(
+                timestamp: receivedAt,
+                channel: items[2].displayText,
+                pattern: items[1].displayText,
+                payload: items[3].displayText
+            )
+        default:
+            return nil
+        }
+    }
+}
+
+final class RedisLiveSubscription: PluginLiveSubscription, @unchecked Sendable {
+    private static let waitMilliseconds: Int32 = 250
+    private static let logger = Logger(subsystem: "com.TablePro.RedisDriver", category: "LiveMessages")
+
+    let messages: AsyncThrowingStream<PluginLiveMessage, Error>
+    private let reader: Task<Void, Never>
+    private let sessions: any PluginLiveSessionSource
+    private let publisherLock = NSLock()
+    // A subscribed connection refuses PUBLISH, so a test message goes out on a second session, opened on first use.
+    private var publisher: RedisPluginConnection?
+
+    init(subscriber: RedisPluginConnection, sessions: any PluginLiveSessionSource) {
+        self.sessions = sessions
+        let (stream, continuation) = AsyncThrowingStream<PluginLiveMessage, Error>.makeStream()
+        messages = stream
+        reader = Task {
+            do {
+                while !Task.isCancelled {
+                    let replies = try await subscriber.nextPushReplies(waitMilliseconds: Self.waitMilliseconds)
+                    let arrived = Date()
+                    for reply in replies {
+                        if let message = try RedisLivePush.message(from: reply, receivedAt: arrived) {
+                            continuation.yield(message)
+                        }
+                    }
+                }
+                continuation.finish()
+            } catch {
+                guard !Task.isCancelled else {
+                    continuation.finish()
+                    return
+                }
+                Self.logger.info("Subscription read ended: \(String(describing: type(of: error)), privacy: .public) \(error.localizedDescription, privacy: .private)")
+                continuation.finish(throwing: error)
+            }
+        }
+    }
+
+    func publish(_ payload: String, to channel: String) async throws {
+        let connection = try await publishingConnection()
+        let reply = try await connection.executeCommand(
+            [Data("PUBLISH".utf8), Data(channel.utf8), Data(payload.utf8)],
+            scope: .outsideBlock
+        )
+        if let error = reply.errorMessage {
+            throw RedisPluginError(code: 0, message: error, refusedByServer: true)
+        }
+    }
+
+    func close() async {
+        reader.cancel()
+        _ = await reader.value
+    }
+
+    private func publishingConnection() async throws -> RedisPluginConnection {
+        if let publisher = publisherLock.withLock({ publisher }) { return publisher }
+        guard let driver = try await sessions.openSession() as? RedisPluginDriver,
+              let connection = driver.pubSubConnection
+        else { throw RedisPluginError.notConnected }
+        publisherLock.withLock { publisher = connection }
+        return connection
+    }
+}
diff --git a/apple/Plugins/RedisDriverPlugin/RedisPluginDriver.swift b/apple/Plugins/RedisDriverPlugin/RedisPluginDriver.swift
index eb64f5e42a..4aabdce4d6 100644
--- a/apple/Plugins/RedisDriverPlugin/RedisPluginDriver.swift
+++ b/apple/Plugins/RedisDriverPlugin/RedisPluginDriver.swift
@@ -33,6 +33,8 @@ final class RedisPluginDriver: PluginDatabaseDriver, @unchecked Sendable {
     private let config: DriverConnectionConfig
     private var redisConnection: (any RedisCommandChannel)?
 
+    var pubSubConnection: RedisPluginConnection? { redisConnection?.pubSubConnection() }
+
     private static let logger = Logger(subsystem: "com.TablePro.RedisDriver", category: "RedisPluginDriver")
 
     static let maxKeyBrowseScan = 10_000
diff --git a/apple/Plugins/RedisDriverPlugin/RedisSentinelChannel.swift b/apple/Plugins/RedisDriverPlugin/RedisSentinelChannel.swift
index 222e7a196e..a7bea86187 100644
--- a/apple/Plugins/RedisDriverPlugin/RedisSentinelChannel.swift
+++ b/apple/Plugins/RedisDriverPlugin/RedisSentinelChannel.swift
@@ -56,6 +56,8 @@ final class RedisSentinelChannel: RedisCommandChannel, @unchecked Sendable {
 
     var isConnected: Bool { current?.isConnected ?? false }
 
+    func pubSubConnection() -> RedisPluginConnection? { current }
+
     private var current: RedisPluginConnection? {
         lock.lock()
         defer { lock.unlock() }
diff --git a/apple/Plugins/RedisDriverTests/RedisLivePushTests.swift b/apple/Plugins/RedisDriverTests/RedisLivePushTests.swift
new file mode 100644
index 0000000000..9e82b31e5a
--- /dev/null
+++ b/apple/Plugins/RedisDriverTests/RedisLivePushTests.swift
@@ -0,0 +1,45 @@
+import Foundation
+import TableProPluginKit
+import Testing
+
+struct RedisLivePushTests {
+    private static let now = Date(timeIntervalSince1970: 0)
+
+    @Test("Channels and patterns go out as SUBSCRIBE and PSUBSCRIBE")
+    func subscribeCommands() {
+        let commands = RedisLivePush.subscribeCommands(
+            for: PluginLiveSubscriptionRequest(channels: ["a", "b"], patterns: ["news.*"])
+        )
+        let text = commands.map { $0.map { String(decoding: $0, as: UTF8.self) } }
+        #expect(text == [["SUBSCRIBE", "a", "b"], ["PSUBSCRIBE", "news.*"]])
+    }
+
+    @Test("A message and a pattern message become live messages; a confirmation does not")
+    func parsesPushReplies() throws {
+        let message = try RedisLivePush.message(
+            from: .array([.data(Data("message".utf8)), .data(Data("a".utf8)), .data(Data("hi".utf8))]),
+            receivedAt: Self.now
+        )
+        #expect(message == PluginLiveMessage(timestamp: Self.now, channel: "a", payload: "hi"))
+
+        let patterned = try RedisLivePush.message(
+            from: .array([.string("pmessage"), .string("news.*"), .string("news.eu"), .string("x")]),
+            receivedAt: Self.now
+        )
+        #expect(patterned?.channel == "news.eu")
+        #expect(patterned?.pattern == "news.*")
+
+        let confirmation = try RedisLivePush.message(
+            from: .array([.string("subscribe"), .string("a"), .integer(1)]),
+            receivedAt: Self.now
+        )
+        #expect(confirmation == nil)
+    }
+
+    @Test("An error reply ends the read with the server's message")
+    func errorReplyThrows() {
+        #expect(throws: RedisPluginError.self) {
+            _ = try RedisLivePush.message(from: .error("NOPERM no access"), receivedAt: Self.now)
+        }
+    }
+}
diff --git a/apple/Plugins/TableProPluginKit/PluginLiveMessaging.swift b/apple/Plugins/TableProPluginKit/PluginLiveMessaging.swift
new file mode 100644
index 0000000000..137d4c369c
--- /dev/null
+++ b/apple/Plugins/TableProPluginKit/PluginLiveMessaging.swift
@@ -0,0 +1,75 @@
+import Foundation
+
+// Messages that flow past a connection rather than being stored: notifications, pub/sub channels, topic tails.
+public protocol PluginLiveMessaging: AnyObject, Sendable {
+    // Synchronous so a menu can ask without I/O; nil means this driver cannot follow live messages.
+    var liveMessageKind: PluginLiveMessageKind? { get }
+    // Every driver the plugin takes from `sessions` is closed by the host once the subscription ends, so the person's
+    // own session is never subscribed. Throws when a channel or topic cannot be followed.
+    func subscribe(
+        _ request: PluginLiveSubscriptionRequest,
+        sessions: any PluginLiveSessionSource
+    ) async throws -> any PluginLiveSubscription
+}
+
+public protocol PluginLiveSessionSource: Sendable {
+    func openSession() async throws -> any PluginDatabaseDriver
+}
+
+// The host never overlaps two `publish` calls; `close()` may arrive at any time and ends `messages`.
+public protocol PluginLiveSubscription: AnyObject, Sendable {
+    // Finishes after `close()`; throws when the connection drops, and the host then subscribes again.
+    var messages: AsyncThrowingStream<PluginLiveMessage, Error> { get }
+    func publish(_ payload: String, to channel: String) async throws
+    func close() async
+}
+
+public enum PluginLiveMessageKind: String, Sendable {
+    // PostgreSQL LISTEN and NOTIFY.
+    case notifications
+    // Redis SUBSCRIBE, PSUBSCRIBE and PUBLISH.
+    case pubSub
+    // Kafka topics read from their end, without a consumer group.
+    case topics
+}
+
+public struct PluginLiveSubscriptionRequest: Sendable, Equatable {
+    public let channels: [String]
+    // Glob patterns, on a kind that matches channels by pattern.
+    public let patterns: [String]
+
+    public init(channels: [String], patterns: [String] = []) {
+        self.channels = channels
+        self.patterns = patterns
+    }
+}
+
+public struct PluginLiveMessage: Sendable, Equatable {
+    // When the engine stamped it, or when it arrived when the engine stamps nothing.
+    public let timestamp: Date
+    public let channel: String
+    // The pattern that matched, for a message a pattern subscription received.
+    public let pattern: String?
+    public let payload: String
+    public let key: String?
+    public let partition: Int?
+    public let offset: Int64?
+
+    public init(
+        timestamp: Date,
+        channel: String,
+        pattern: String? = nil,
+        payload: String,
+        key: String? = nil,
+        partition: Int? = nil,
+        offset: Int64? = nil
+    ) {
+        self.timestamp = timestamp
+        self.channel = channel
+        self.pattern = pattern
+        self.payload = payload
+        self.key = key
+        self.partition = partition
+        self.offset = offset
+    }
+}
diff --git a/apple/TablePro/Core/Database/DatabaseManager+Sessions.swift b/apple/TablePro/Core/Database/DatabaseManager+Sessions.swift
index e845a66cf1..e04b5fd23d 100644
--- a/apple/TablePro/Core/Database/DatabaseManager+Sessions.swift
+++ b/apple/TablePro/Core/Database/DatabaseManager+Sessions.swift
@@ -751,6 +751,7 @@ extension DatabaseManager {
         // Before the tunnel goes: a debug target whose client vanishes keeps its locks until someone aborts it.
         await RoutineDebugSessions.shared.stopAll(connectionId: sessionId, tabsCloseAfter: true)
         defer { RoutineDebugSessions.shared.reopenAdmission(for: sessionId) }
+        LiveMessageSessions.shared.end(connectionId: sessionId)
 
         tabStatePersister?.persistTabState(for: sessionId)
         if origin == .userRequested {
diff --git a/apple/TablePro/Core/LiveMessages/LiveMessageBuffer.swift b/apple/TablePro/Core/LiveMessages/LiveMessageBuffer.swift
new file mode 100644
index 0000000000..9092b1678a
--- /dev/null
+++ b/apple/TablePro/Core/LiveMessages/LiveMessageBuffer.swift
@@ -0,0 +1,84 @@
+//
+//  LiveMessageBuffer.swift
+//  TablePro
+//
+
+import Foundation
+import TableProPluginKit
+
+internal struct LiveMessageEntry: Identifiable, Equatable, Sendable {
+    enum Kind: Equatable, Sendable {
+        case message(PluginLiveMessage)
+        // The connection dropped: messages sent until it came back were not received.
+        case gap(reason: String)
+    }
+
+    let id: Int
+    let kind: Kind
+
+    var message: PluginLiveMessage? {
+        if case .message(let message) = kind { return message }
+        return nil
+    }
+
+    func matches(_ filter: String) -> Bool {
+        guard !filter.isEmpty else { return true }
+        guard let message else { return true }
+        return message.channel.localizedCaseInsensitiveContains(filter)
+            || message.payload.localizedCaseInsensitiveContains(filter)
+            || (message.key?.localizedCaseInsensitiveContains(filter) ?? false)
+    }
+}
+
+// The newest `capacity` entries. A pause freezes what is shown, never what is kept, so resuming loses nothing up to the cap.
+internal struct LiveMessageBuffer: Sendable {
+    static let defaultCapacity = 10_000
+
+    let capacity: Int
+    private(set) var entries: [LiveMessageEntry] = []
+    private(set) var droppedCount = 0
+    private var nextId = 0
+
+    init(capacity: Int = defaultCapacity) {
+        self.capacity = max(1, capacity)
+    }
+
+    mutating func append(_ messages: [PluginLiveMessage]) {
+        for message in messages {
+            push(.message(message))
+        }
+        trim()
+    }
+
+    mutating func appendGap(reason: String) {
+        push(.gap(reason: reason))
+        trim()
+    }
+
+    mutating func clear() {
+        entries.removeAll(keepingCapacity: true)
+        droppedCount = 0
+    }
+
+    func visible(filter: String) -> [LiveMessageEntry] {
+        Self.filter(entries, by: filter)
+    }
+
+    static func filter(_ entries: [LiveMessageEntry], by filter: String) -> [LiveMessageEntry] {
+        let trimmed = filter.trimmingCharacters(in: .whitespaces)
+        guard !trimmed.isEmpty else { return entries }
+        return entries.filter { $0.matches(trimmed) }
+    }
+
+    private mutating func push(_ kind: LiveMessageEntry.Kind) {
+        entries.append(LiveMessageEntry(id: nextId, kind: kind))
+        nextId += 1
+    }
+
+    private mutating func trim() {
+        let excess = entries.count - capacity
+        guard excess > 0 else { return }
+        entries.removeFirst(excess)
+        droppedCount += excess
+    }
+}
diff --git a/apple/TablePro/Core/LiveMessages/LiveMessageController.swift b/apple/TablePro/Core/LiveMessages/LiveMessageController.swift
new file mode 100644
index 0000000000..ce80621f05
--- /dev/null
+++ b/apple/TablePro/Core/LiveMessages/LiveMessageController.swift
@@ -0,0 +1,215 @@
+//
+//  LiveMessageController.swift
+//  TablePro
+//
+
+import Combine
+import Foundation
+import os
+import TableProPluginKit
+
+// One live tab: a subscription on connections of its own, re-opened after a drop, and the messages it kept.
+@MainActor
+internal final class LiveMessageController: ObservableObject {
+    enum Phase: Equatable {
+        case connecting
+        case live
+        case reconnecting(String)
+        case ended(String?)
+    }
+
+    typealias Subscribe = @Sendable (LiveMessageSessionSource) async throws -> any PluginLiveSubscription
+
+    static let refreshInterval: Duration = .milliseconds(100)
+    static let maximumBackoff: Duration = .seconds(30)
+    private static let logger = Logger(subsystem: "com.TablePro", category: "LiveMessages")
+
+    let kind: PluginLiveMessageKind
+    let request: PluginLiveSubscriptionRequest
+
+    @Published private(set) var phase: Phase = .connecting
+    @Published private(set) var shown: [LiveMessageEntry] = []
+    @Published private(set) var isPaused = false
+    // Kept while paused and not yet shown.
+    @Published private(set) var unseenCount = 0
+    @Published var filter = "" {
+        didSet { refreshShown() }
+    }
+
+    private(set) var buffer: LiveMessageBuffer
+    // What the table showed when it was paused; the buffer keeps growing underneath.
+    private var frozen: [LiveMessageEntry] = []
+    private let subscribe: Subscribe
+    private let makeSource: @MainActor () -> LiveMessageSessionSource
+    private let clock: any Clock<Duration>
+    private var runTask: Task<Void, Never>?
+    private var refreshTask: Task<Void, Never>?
+    private var subscription: (any PluginLiveSubscription)?
+    private var source: LiveMessageSessionSource?
+
+    init(
+        kind: PluginLiveMessageKind,
+        request: PluginLiveSubscriptionRequest,
+        capacity: Int = LiveMessageBuffer.defaultCapacity,
+        clock: any Clock<Duration> = ContinuousClock(),
+        makeSource: @escaping @MainActor () -> LiveMessageSessionSource,
+        subscribe: @escaping Subscribe
+    ) {
+        self.kind = kind
+        self.request = request
+        self.buffer = LiveMessageBuffer(capacity: capacity)
+        self.clock = clock
+        self.makeSource = makeSource
+        self.subscribe = subscribe
+    }
+
+    static func title(for request: PluginLiveSubscriptionRequest) -> String {
+        let names = (request.channels + request.patterns).joined(separator: ", ")
+        return String(format: String(localized: "Live: %@"), names)
+    }
+
+        var hasEnded: Bool {
+        if case .ended = phase { return true }
+        return false
+    }
+
+    var canPublish: Bool { phase == .live }
+
+    // MARK: - Lifecycle
+
+    func start() {
+        guard runTask == nil, !hasEnded else { return }
+        runTask = Task { [weak self] in
+            await self?.run()
+        }
+    }
+
+    // Ends the subscription and closes every connection it opened; the messages stay on screen.
+    func stop() {
+        guard !hasEnded else { return }
+        phase = .ended(nil)
+        runTask?.cancel()
+        runTask = nil
+        refreshTask?.cancel()
+        refreshTask = nil
+        refreshShown()
+        let subscription = subscription
+        let source = source
+        self.subscription = nil
+        self.source = nil
+        Task {
+            await subscription?.close()
+            await source?.closeAll()
+        }
+    }
+
+    private func run() async {
+        var hasSubscribed = false
+        var backoff: Duration = .seconds(1)
+        while !Task.isCancelled {
+            let source = makeSource()
+            self.source = source
+            do {
+                let opened = try await subscribe(source)
+                guard !Task.isCancelled else {
+                    await opened.close()
+                    await source.closeAll()
+                    return
+                }
+                subscription = opened
+                phase = .live
+                hasSubscribed = true
+                backoff = .seconds(1)
+                for try await message in opened.messages {
+                    receive(message)
+                }
+                guard !Task.isCancelled else { return }
+                await dropped(subscription: opened, source: source, reason: String(localized: "The connection closed."))
+            } catch {
+                guard !Task.isCancelled else { return }
+                // A first subscribe that fails is a wrong channel, topic or permission: retrying would only repeat it.
+                guard hasSubscribed else {
+                    await source.closeAll()
+                    self.source = nil
+                    phase = .ended(error.localizedDescription)
+                    return
+                }
+                await dropped(subscription: subscription, source: source, reason: error.localizedDescription)
+            }
+            do {
+                try await clock.sleep(for: backoff)
+            } catch {
+                return
+            }
+            backoff = min(backoff * 2, Self.maximumBackoff)
+        }
+    }
+
+    private func dropped(subscription dropped: (any PluginLiveSubscription)?, source: LiveMessageSessionSource, reason: String) async {
+        Self.logger.info("Live subscription dropped: \(reason, privacy: .private)")
+        if case .reconnecting = phase {} else {
+            buffer.appendGap(reason: reason)
+            scheduleRefresh()
+        }
+        phase = .reconnecting(reason)
+        subscription = nil
+        self.source = nil
+        await dropped?.close()
+        await source.closeAll()
+    }
+
+    // MARK: - Messages
+
+    func receive(_ message: PluginLiveMessage) {
+        buffer.append([message])
+        if isPaused {
+            unseenCount += 1
+        } else {
+            scheduleRefresh()
+        }
+    }
+
+    func pause() {
+        guard !isPaused else { return }
+        isPaused = true
+        refreshTask?.cancel()
+        refreshTask = nil
+        frozen = buffer.entries
+    }
+
+    func resume() {
+        guard isPaused else { return }
+        isPaused = false
+        unseenCount = 0
+        frozen = []
+        refreshShown()
+    }
+
+    func clear() {
+        buffer.clear()
+        frozen = []
+        unseenCount = 0
+        refreshShown()
+    }
+
+    func publish(_ payload: String, to channel: String) async throws {
+        guard let subscription, phase == .live else { throw LiveMessageError.ended }
+        try await subscription.publish(payload, to: channel)
+    }
+
+    // Coalesced, so a burst of messages redraws the table once per interval rather than once per message.
+    private func scheduleRefresh() {
+        guard refreshTask == nil, !isPaused else { return }
+        let clock = clock
+        refreshTask = Task { [weak self] in
+            try? await clock.sleep(for: Self.refreshInterval)
+            guard let self, !Task.isCancelled else { return }
+            self.refreshTask = nil
+            self.refreshShown()
+        }
+    }
+
+    private func refreshShown() {
+        shown = LiveMessageBuffer.filter(isPaused ? frozen : buffer.entries, by: filter)
+    }
+}
diff --git a/apple/TablePro/Core/LiveMessages/LiveMessageExport.swift b/apple/TablePro/Core/LiveMessages/LiveMessageExport.swift
new file mode 100644
index 0000000000..96e5f5bbb0
--- /dev/null
+++ b/apple/TablePro/Core/LiveMessages/LiveMessageExport.swift
@@ -0,0 +1,87 @@
+//
+//  LiveMessageExport.swift
+//  TablePro
+//
+
+import Foundation
+import TableProPluginKit
+
+// The columns a kind shows, and the same columns as text for Copy and Export.
+internal enum LiveMessageExport {
+    enum Column: CaseIterable {
+        case time, channel, pattern, partition, offset, key, payload
+
+        var title: String {
+            switch self {
+            case .time: String(localized: "Time")
+            case .channel: String(localized: "Channel")
+            case .pattern: String(localized: "Pattern")
+            case .partition: String(localized: "Partition")
+            case .offset: String(localized: "Offset")
+            case .key: String(localized: "Key")
+            case .payload: String(localized: "Payload")
+            }
+        }
+    }
+
+    static func columns(for kind: PluginLiveMessageKind) -> [Column] {
+        switch kind {
+        case .pubSub: [.time, .channel, .pattern, .payload]
+        case .topics: [.time, .channel, .partition, .offset, .key, .payload]
+        case .notifications: [.time, .channel, .payload]
+        @unknown default: [.time, .channel, .payload]
+        }
+    }
+
+    static func timeText(_ date: Date) -> String {
+        date.formatted(.iso8601.year().month().day().time(includingFractionalSeconds: true))
+    }
+
+    static func value(of column: Column, in entry: LiveMessageEntry) -> String {
+        guard let message = entry.message else {
+            guard case .gap(let reason) = entry.kind else { return "" }
+            switch column {
+            case .channel: return String(localized: "Connection lost")
+            case .payload: return reason
+            default: return ""
+            }
+        }
+        switch column {
+        case .time: return timeText(message.timestamp)
+        case .channel: return message.channel
+        case .pattern: return message.pattern ?? ""
+        case .partition: return message.partition.map(String.init) ?? ""
+        case .offset: return message.offset.map(String.init) ?? ""
+        case .key: return message.key ?? ""
+        case .payload: return message.payload
+        }
+    }
+
+    // Tab-separated with a header row, the way a grid copies, so it pastes into a spreadsheet.
+    static func tsv(_ entries: [LiveMessageEntry], kind: PluginLiveMessageKind) -> String {
+        let columns = columns(for: kind)
+        let header = columns.map(\.title).joined(separator: "\t")
+        let rows = entries.map { entry in
+            columns.map { column in
+                value(of: column, in: entry)
+                    .replacingOccurrences(of: "\t", with: " ")
+                    .replacingOccurrences(of: "\n", with: " ")
+            }
+            .joined(separator: "\t")
+        }
+        return ([header] + rows).joined(separator: "\n")
+    }
+
+    static func csv(_ entries: [LiveMessageEntry], kind: PluginLiveMessageKind) -> String {
+        let columns = columns(for: kind)
+        let lines = [columns.map { quoted($0.title) }.joined(separator: ",")] + entries.map { entry in
+            columns.map { quoted(value(of: $0, in: entry)) }.joined(separator: ",")
+        }
+        return lines.joined(separator: "\r\n") + "\r\n"
+    }
+
+    private static func quoted(_ field: String) -> String {
+        guard field.contains(where: { $0 == "," || $0 == "\"" || $0.isNewline }) else { return field }
+        return "\"" + field.replacingOccurrences(of: "\"", with: "\"\"") + "\""
+    }
+}
diff --git a/apple/TablePro/Core/LiveMessages/LiveMessageSessionSource.swift b/apple/TablePro/Core/LiveMessages/LiveMessageSessionSource.swift
new file mode 100644
index 0000000000..095c7c72c9
--- /dev/null
+++ b/apple/TablePro/Core/LiveMessages/LiveMessageSessionSource.swift
@@ -0,0 +1,96 @@
+//
+//  LiveMessageSessionSource.swift
+//  TablePro
+//
+
+import Foundation
+import TableProPluginKit
+
+internal struct LiveMessageDriverHandle: Sendable {
+    let driver: any PluginDatabaseDriver
+    let close: @Sendable () -> Void
+}
+
+internal enum LiveMessageError: LocalizedError, Equatable {
+    case ended
+    case needsSeparateConnection
+    case unsupported
+
+    var errorDescription: String? {
+        switch self {
+        case .ended:
+            return String(localized: "The live message view has closed.")
+        case .needsSeparateConnection:
+            return String(localized: "This connection can't open the separate session live messages need.")
+        case .unsupported:
+            return String(localized: "This connection's driver can't follow live messages.")
+        }
+    }
+}
+
+// The connections one subscription attempt opened: never pooled, never health-pinged, no query timeout, all closed together.
+internal actor LiveMessageSessionSource: PluginLiveSessionSource {
+    typealias Opener = @Sendable () async throws -> LiveMessageDriverHandle
+
+    // A subscription waits on messages, not on a statement, so any timeout would end it between two messages.
+    static let queryTimeoutSeconds = 0
+
+    private let opener: Opener
+    private var handles: [LiveMessageDriverHandle] = []
+    private var isClosed = false
+
+    init(opener: @escaping Opener) {
+        self.opener = opener
+    }
+
+    func openSession() async throws -> any PluginDatabaseDriver {
+        guard !isClosed else { throw LiveMessageError.ended }
+        let handle = try await opener()
+        // Closed while the driver was being dialed.
+        guard !isClosed else {
+            handle.close()
+            throw LiveMessageError.ended
+        }
+        handles.append(handle)
+        return handle.driver
+    }
+
+    func closeAll() {
+        isClosed = true
+        let closing = handles
+        handles.removeAll()
+        for handle in closing {
+            handle.close()
+        }
+    }
+}
+
+extension LiveMessageSessionSource {
+    // A second connection to an embedded engine opens a different, empty database, so those engines are refused.
+    @MainActor
+    static func live(scope: DatabaseScope) -> LiveMessageSessionSource {
+        LiveMessageSessionSource {
+            let driver = try await openDedicatedDriver(scope: scope)
+            guard let adapter = driver as? PluginDriverAdapter else {
+                driver.disconnect()
+                throw LiveMessageError.unsupported
+            }
+            return LiveMessageDriverHandle(driver: adapter.schemaPluginDriver) { adapter.disconnect() }
+        }
+    }
+
+    @MainActor
+    private static func openDedicatedDriver(scope: DatabaseScope) async throws -> DatabaseDriver {
+        guard let session = DatabaseManager.shared.session(for: scope.connectionId) else {
+            throw DatabaseError.notConnected
+        }
+        guard session.connection.type.supportsConnectionPooling else {
+            throw LiveMessageError.needsSeparateConnection
+        }
+        return try await MetadataConnectionPool.openServerDriver(
+            for: scope,
+            purpose: .session,
+            queryTimeoutSeconds: queryTimeoutSeconds
+        )
+    }
+}
diff --git a/apple/TablePro/Core/LiveMessages/LiveMessageSessions.swift b/apple/TablePro/Core/LiveMessages/LiveMessageSessions.swift
new file mode 100644
index 0000000000..e24dde794f
--- /dev/null
+++ b/apple/TablePro/Core/LiveMessages/LiveMessageSessions.swift
@@ -0,0 +1,57 @@
+//
+//  LiveMessageSessions.swift
+//  TablePro
+//
+
+import Foundation
+
+// Every live tab's subscription, by the tab that shows it. Closing the tab is what ends it.
+@MainActor
+internal final class LiveMessageSessions {
+    static let shared = LiveMessageSessions()
+
+    private struct Entry {
+        let controller: LiveMessageController
+        let connectionId: UUID
+    }
+
+    private var entries: [UUID: Entry] = [:]
+
+    func register(_ controller: LiveMessageController, tabId: UUID, connectionId: UUID) {
+        entries[tabId] = Entry(controller: controller, connectionId: connectionId)
+    }
+
+    func controller(forTab tabId: UUID) -> LiveMessageController? {
+        entries[tabId]?.controller
+    }
+
+    func end(tabId: UUID) {
+        entries.removeValue(forKey: tabId)?.controller.stop()
+    }
+
+    // Before the connection's transport goes, so no subscription keeps redialing through a closed tunnel.
+    func end(connectionId: UUID) {
+        let tabIds = entries.filter { $0.value.connectionId == connectionId }.map(\.key)
+        end(tabIds: tabIds)
+    }
+
+    func end(tabIds: [UUID]) {
+        for tabId in tabIds {
+            end(tabId: tabId)
+        }
+    }
+
+    // The module went off or the app is quitting. With `closingTabs`, every window drops its live tabs too.
+    func endAll(closingTabs: Bool = false) {
+        if closingTabs {
+            for coordinator in MainContentCoordinator.activeCoordinators.values {
+                coordinator.closeLiveMessageTabs()
+            }
+        }
+        let ending = entries.values.map(\.controller)
+        entries.removeAll()
+        for controller in ending {
+            controller.stop()
+        }
+    }
+}
diff --git a/apple/TablePro/Core/LiveMessages/MainContentCoordinator+LiveMessages.swift b/apple/TablePro/Core/LiveMessages/MainContentCoordinator+LiveMessages.swift
new file mode 100644
index 0000000000..627c150b49
--- /dev/null
+++ b/apple/TablePro/Core/LiveMessages/MainContentCoordinator+LiveMessages.swift
@@ -0,0 +1,62 @@
+//
+//  MainContentCoordinator+LiveMessages.swift
+//  TablePro
+//
+
+import AppKit
+import Foundation
+import SwiftUI
+import TableProPluginKit
+
+extension MainContentCoordinator {
+    // Answers without I/O, so a menu validation can ask. The subscription itself runs on connections of its own.
+    private var liveMessaging: (any PluginDatabaseDriver & PluginLiveMessaging)? {
+        guard AppModuleSettings.shared.isEnabled(.liveMessages),
+              connection.type.supportsConnectionPooling,
+              let adapter = services.databaseManager.driver(for: connectionId) as? PluginDriverAdapter,
+              let messaging = adapter.schemaPluginDriver as? any PluginDatabaseDriver & PluginLiveMessaging,
+              messaging.liveMessageKind != nil
+        else { return nil }
+        return messaging
+    }
+
+    func canOpenLiveMessages() -> Bool {
+        liveMessaging != nil && services.databaseManager.session(for: connectionId)?.isConnected == true
+    }
+
+    func openLiveMessages() {
+        guard let kind = liveMessaging?.liveMessageKind, let window = contentWindow else { return }
+        let suggested = kind == .topics ? tabManager.selectedTab?.tableContext.tableName : nil
+        let setup = RoutineDebugSetup()
+        Task {
+            let request = await RoutineDebugSheet.present(on: window, setup: setup) { finish in
+                LiveMessageStartSheet(kind: kind, suggestedChannel: suggested ?? "", onFinish: finish)
+            }
+            guard let request else { return }
+            openLiveMessageTab(kind: kind, request: request)
+        }
+    }
+
+    // Not a user close: nothing goes to Reopen Closed Tab and the saved tab set is left alone.
+    func closeLiveMessageTabs() {
+        let ids = tabManager.tabs.filter { $0.tabType == .liveMessages }.map(\.id)
+        for id in ids where tabManager.tabs.contains(where: { $0.id == id }) {
+            tabManager.closeTab(id: id)
+        }
+    }
+
+    private func openLiveMessageTab(kind: PluginLiveMessageKind, request: PluginLiveSubscriptionRequest) {
+        // Read again: the module may have gone off, or the connection dropped, while the sheet was up.
+        guard let messaging = liveMessaging, let scope = browseScope else { return }
+        let controller = LiveMessageController(
+            kind: kind,
+            request: request,
+            makeSource: { LiveMessageSessionSource.live(scope: scope) },
+            subscribe: { source in try await messaging.subscribe(request, sessions: source) }
+        )
+        let tabId = UUID()
+        LiveMessageSessions.shared.register(controller, tabId: tabId, connectionId: connectionId)
+        tabManager.addLiveMessagesTab(id: tabId, title: LiveMessageController.title(for: request))
+        controller.start()
+    }
+}
diff --git a/apple/TablePro/Core/MCP/Protocol/Tools/MCPTabSnapshotProvider.swift b/apple/TablePro/Core/MCP/Protocol/Tools/MCPTabSnapshotProvider.swift
index 20886d5950..4415afdd91 100644
--- a/apple/TablePro/Core/MCP/Protocol/Tools/MCPTabSnapshotProvider.swift
+++ b/apple/TablePro/Core/MCP/Protocol/Tools/MCPTabSnapshotProvider.swift
@@ -171,7 +171,7 @@ private extension TabType {
         case .objectSource: "objectSource"
         case .versionHistory: "versionHistory"
         case .tableListing: "tableListing"
-        case .routineDebug: nil
+        case .routineDebug, .liveMessages: nil
         }
     }
 }
diff --git a/apple/TablePro/Core/Menu/DatabaseMenuBuilder.swift b/apple/TablePro/Core/Menu/DatabaseMenuBuilder.swift
index fa759c6bfe..f1e617b865 100644
--- a/apple/TablePro/Core/Menu/DatabaseMenuBuilder.swift
+++ b/apple/TablePro/Core/Menu/DatabaseMenuBuilder.swift
@@ -8,8 +8,11 @@ import AppKit
 // Database-or-Schema titles are a closed set, never interpolated: System Settings binds an App Shortcut to the exact title.
 @MainActor
 enum DatabaseMenuBuilder {
-    static func build(keyboard: KeyboardSettings) -> NSMenuItem {
-        MenuItemFactory.menu(String(localized: "Database"), items: [
+    static let identifier = NSUserInterfaceItemIdentifier("menu.database")
+    static let liveMessagesIdentifier = NSUserInterfaceItemIdentifier("menu.database.liveMessages")
+
+    static func build(keyboard: KeyboardSettings, modules: Set<AppModule> = []) -> NSMenuItem {
+        let item = MenuItemFactory.menu(String(localized: "Database"), items: [
             MenuItemFactory.item(
                 String(localized: "Switch Connection…"),
                 action: #selector(MainSplitViewController.switchConnection(_:)),
@@ -161,6 +164,30 @@ enum DatabaseMenuBuilder {
             MenuItemFactory.separator,
             compareSubmenu()
         ])
+        item.identifier = identifier
+        if let menu = item.submenu {
+            applyModules(modules, to: menu)
+        }
+        return item
+    }
+
+    // Live Messages… sits with the other connection-wide tabs, and exists only while its module is on.
+    static func applyModules(_ modules: Set<AppModule>, to menu: NSMenu) {
+        let existing = menu.items.first { $0.identifier == liveMessagesIdentifier }
+        switch (modules.contains(.liveMessages), existing) {
+        case (true, nil):
+            let anchor = menu.items.firstIndex { $0.action == #selector(MainSplitViewController.showTableHealth(_:)) }
+            let item = MenuItemFactory.item(
+                String(localized: "Live Messages…"),
+                action: #selector(MainSplitViewController.openLiveMessages(_:))
+            )
+            item.identifier = liveMessagesIdentifier
+            menu.insertItem(item, at: anchor.map { $0 + 1 } ?? menu.items.count)
+        case (false, let item?):
+            menu.removeItem(item)
+        default:
+            return
+        }
     }
 
     /// The HIG asks that every toolbar item also be a menu-bar command. These are the Compare &
diff --git a/apple/TablePro/Core/Menu/MainMenuBuilder.swift b/apple/TablePro/Core/Menu/MainMenuBuilder.swift
index 2fbc402a49..20c161903c 100644
--- a/apple/TablePro/Core/Menu/MainMenuBuilder.swift
+++ b/apple/TablePro/Core/Menu/MainMenuBuilder.swift
@@ -32,7 +32,7 @@ enum MainMenuBuilder {
         bar.addItem(FileMenuBuilder.build(keyboard: keyboard))
         bar.addItem(EditMenuBuilder.build(keyboard: keyboard))
         bar.addItem(ViewMenuBuilder.build(keyboard: keyboard))
-        bar.addItem(DatabaseMenuBuilder.build(keyboard: keyboard))
+        bar.addItem(DatabaseMenuBuilder.build(keyboard: keyboard, modules: modules))
         bar.addItem(QueryMenuBuilder.build(keyboard: keyboard))
         if modules.contains(.routineDebugger) {
             bar.addItem(DebugMenuBuilder.build(keyboard: keyboard))
@@ -63,6 +63,9 @@ enum MainMenuBuilder {
         to bar: NSMenu,
         windowsMenu: NSMenu?
     ) {
+        if let database = bar.items.first(where: { $0.identifier == DatabaseMenuBuilder.identifier })?.submenu {
+            DatabaseMenuBuilder.applyModules(modules, to: database)
+        }
         let existing = bar.items.first { $0.identifier == DebugMenuBuilder.identifier }
         switch (modules.contains(.routineDebugger), existing) {
         case (true, nil):
diff --git a/apple/TablePro/Core/Modules/AppModule.swift b/apple/TablePro/Core/Modules/AppModule.swift
index e65d22848f..a4dad1f91d 100644
--- a/apple/TablePro/Core/Modules/AppModule.swift
+++ b/apple/TablePro/Core/Modules/AppModule.swift
@@ -2,22 +2,25 @@ import Foundation
 
 enum AppModule: String, CaseIterable, Sendable {
     case routineDebugger
+    case liveMessages
 
     var title: String {
         switch self {
         case .routineDebugger: String(localized: "Routine Debugger")
+        case .liveMessages: String(localized: "Live Messages")
         }
     }
 
     var summary: String {
         switch self {
         case .routineDebugger: String(localized: "Step through PL/pgSQL and PL/SQL routines with breakpoints.")
+        case .liveMessages: String(localized: "Watch PostgreSQL notifications, Redis channels and Kafka topics as messages arrive.")
         }
     }
 
     var defaultEnabled: Bool {
         switch self {
-        case .routineDebugger: false
+        case .routineDebugger, .liveMessages: false
         }
     }
 
diff --git a/apple/TablePro/Core/Modules/AppModuleActivation.swift b/apple/TablePro/Core/Modules/AppModuleActivation.swift
index 14fc7db9e5..01ab20851d 100644
--- a/apple/TablePro/Core/Modules/AppModuleActivation.swift
+++ b/apple/TablePro/Core/Modules/AppModuleActivation.swift
@@ -31,6 +31,8 @@ enum AppModuleActivation {
         switch module {
         case .routineDebugger:
             Task { await RoutineDebugSessions.shared.stopAll() }
+        case .liveMessages:
+            LiveMessageSessions.shared.endAll(closingTabs: true)
         }
     }
 }
diff --git a/apple/TablePro/Core/Modules/AppModuleTermination.swift b/apple/TablePro/Core/Modules/AppModuleTermination.swift
index 9974d64f5b..3f3bcc8a30 100644
--- a/apple/TablePro/Core/Modules/AppModuleTermination.swift
+++ b/apple/TablePro/Core/Modules/AppModuleTermination.swift
@@ -10,6 +10,7 @@ enum AppModuleTermination {
     // the same one the runs close their connections at.
     static func stopModuleWork() async {
         hasBegun = true
+        LiveMessageSessions.shared.endAll()
         guard RoutineDebugSessions.shared.hasActiveSessions else { return }
         await RoutineDebugSessions.shared.stopAll(timeout: stopLimit)
     }
diff --git a/apple/TablePro/Core/Scripting/ScriptEnumerations.swift b/apple/TablePro/Core/Scripting/ScriptEnumerations.swift
index 70707a1a68..18a54c2178 100644
--- a/apple/TablePro/Core/Scripting/ScriptEnumerations.swift
+++ b/apple/TablePro/Core/Scripting/ScriptEnumerations.swift
@@ -46,7 +46,7 @@ internal enum ScriptEnumerations {
         case .objectSource: fourCharCode("TPk8")
         case .versionHistory: fourCharCode("TPk9")
         case .tableListing: fourCharCode("TPkl")
-        case .routineDebug: nil
+        case .routineDebug, .liveMessages: nil
         }
     }
 
diff --git a/apple/TablePro/Core/Services/Infrastructure/EditorTabDetachPolicy.swift b/apple/TablePro/Core/Services/Infrastructure/EditorTabDetachPolicy.swift
index 850d679e24..d83582b9ad 100644
--- a/apple/TablePro/Core/Services/Infrastructure/EditorTabDetachPolicy.swift
+++ b/apple/TablePro/Core/Services/Infrastructure/EditorTabDetachPolicy.swift
@@ -15,8 +15,9 @@ internal enum EditorTabDetachPolicy {
         isBusy: Bool,
         isConnected: Bool
     ) -> Bool {
-        // A debug run's prompts and its close on a declined start act on the coordinator that opened it.
-        guard tabType != .routineDebug else { return false }
+        // A debug run's prompts and its close on a declined start act on the coordinator that opened it; a live
+        // subscription is registered with that coordinator's tab.
+        guard tabType != .routineDebug, tabType != .liveMessages else { return false }
         // Moving the last tab is Move Connection to New Window.
         guard tabCount > 1 else { return false }
         guard !hasUnsavedWork else { return false }
diff --git a/apple/TablePro/Core/Services/Infrastructure/EditorTabOpener.swift b/apple/TablePro/Core/Services/Infrastructure/EditorTabOpener.swift
index bb74abcead..fc2d84e0af 100644
--- a/apple/TablePro/Core/Services/Infrastructure/EditorTabOpener.swift
+++ b/apple/TablePro/Core/Services/Infrastructure/EditorTabOpener.swift
@@ -104,8 +104,8 @@ internal enum EditorTabOpener {
                 databaseName: payload.databaseName ?? openingDatabaseName,
                 schemaName: payload.schemaName
             )
-        case .routineDebug:
-            // A payload cannot carry a live session; the coordinator opens debug tabs itself.
+        case .routineDebug, .liveMessages:
+            // A payload cannot carry a live session; the coordinator opens these tabs itself.
             return
         }
     }
diff --git a/apple/TablePro/Core/Services/Infrastructure/MainSplitViewController+LiveMessageActions.swift b/apple/TablePro/Core/Services/Infrastructure/MainSplitViewController+LiveMessageActions.swift
new file mode 100644
index 0000000000..06ad4f4c72
--- /dev/null
+++ b/apple/TablePro/Core/Services/Infrastructure/MainSplitViewController+LiveMessageActions.swift
@@ -0,0 +1,18 @@
+import AppKit
+
+extension MainSplitViewController {
+    @objc func openLiveMessages(_ sender: Any?) {
+        guard let coordinator = liveMessagesCoordinator, coordinator.canOpenLiveMessages() else { return }
+        coordinator.openLiveMessages()
+    }
+
+    var canOpenLiveMessages: Bool {
+        liveMessagesCoordinator?.canOpenLiveMessages() == true
+    }
+
+    // Nil while the module is off, so no command reaches the subscription code.
+    private var liveMessagesCoordinator: MainContentCoordinator? {
+        guard AppModuleSettings.shared.isEnabled(.liveMessages) else { return nil }
+        return commandActions?.coordinator
+    }
+}
diff --git a/apple/TablePro/Core/Services/Infrastructure/MainSplitViewController+MenuValidation.swift b/apple/TablePro/Core/Services/Infrastructure/MainSplitViewController+MenuValidation.swift
index 40071b509b..ce7ef8a184 100644
--- a/apple/TablePro/Core/Services/Infrastructure/MainSplitViewController+MenuValidation.swift
+++ b/apple/TablePro/Core/Services/Infrastructure/MainSplitViewController+MenuValidation.swift
@@ -137,6 +137,7 @@ struct MenuValidationContext: Equatable {
     var hasSchemaFilter = false
     // Filled only when a Debug menu item is validated; the menu exists only while the module is on.
     var routineDebug = RoutineDebugMenuState()
+    var canOpenLiveMessages = false
 
     // The validation and the window's handler both read it, so a lit Use Selection for Find never lands on an editor with no selection.
     var usesGridSelectionForFind: Bool {
@@ -296,6 +297,8 @@ extension MainSplitViewController: NSMenuItemValidation {
             return context.isConnected
         case #selector(showTableHealth(_:)):
             return context.isConnected && context.supportsTableHealth
+        case #selector(openLiveMessages(_:)):
+            return context.isConnected && !context.isAgentMode && context.canOpenLiveMessages
 
         case #selector(toggleFilterBar(_:)):
             return context.isConnected && context.canUseTableResultCommands
@@ -729,6 +732,9 @@ extension MainSplitViewController: NSMenuItemValidation {
         if Self.routineDebugSelectors.contains(action) {
             context.routineDebug = routineDebugMenuState
         }
+        if action == #selector(openLiveMessages(_:)) {
+            context.canOpenLiveMessages = canOpenLiveMessages
+        }
         guard Self.agentSessionSelectors.contains(action) else { return context }
         context.agentSessionTarget = agentSessionTarget(for: menuItem)?.status
         return context
diff --git a/apple/TablePro/Core/Services/Infrastructure/MainSplitViewController+RecentTabs.swift b/apple/TablePro/Core/Services/Infrastructure/MainSplitViewController+RecentTabs.swift
index ac82152aa8..e9068e4b76 100644
--- a/apple/TablePro/Core/Services/Infrastructure/MainSplitViewController+RecentTabs.swift
+++ b/apple/TablePro/Core/Services/Infrastructure/MainSplitViewController+RecentTabs.swift
@@ -217,6 +217,8 @@ internal extension MainSplitViewController {
             return "list.bullet.rectangle"
         case .routineDebug:
             return "ladybug"
+        case .liveMessages:
+            return "dot.radiowaves.left.and.right"
         }
     }
 }
diff --git a/apple/TablePro/Core/Services/Infrastructure/MainSplitViewController.swift b/apple/TablePro/Core/Services/Infrastructure/MainSplitViewController.swift
index 46aa54f0fb..8550bdab41 100644
--- a/apple/TablePro/Core/Services/Infrastructure/MainSplitViewController.swift
+++ b/apple/TablePro/Core/Services/Infrastructure/MainSplitViewController.swift
@@ -1386,7 +1386,7 @@ internal final class MainSplitViewController: NSSplitViewController {
         case .usersRoles:
             return UsersRolesLayoutMetrics.tabMinimumWidth
         case .query, .table, .createTable, .erDiagram, .serverDashboard, .insights, .tableHealth, .objectSource,
-             .versionHistory, .tableListing, .routineDebug:
+             .versionHistory, .tableListing, .routineDebug, .liveMessages:
             return defaultDetailMinThickness
         }
     }
diff --git a/apple/TablePro/Core/Services/Infrastructure/Toolbar/ToolbarContextResolver.swift b/apple/TablePro/Core/Services/Infrastructure/Toolbar/ToolbarContextResolver.swift
index 8c30c37b00..500b7b548f 100644
--- a/apple/TablePro/Core/Services/Infrastructure/Toolbar/ToolbarContextResolver.swift
+++ b/apple/TablePro/Core/Services/Infrastructure/Toolbar/ToolbarContextResolver.swift
@@ -70,7 +70,7 @@ internal enum ToolbarContextResolver {
         case .usersRoles:
             String(localized: "Apply Changes")
         case .query, .table, .erDiagram, .serverDashboard, .insights, .tableHealth, .objectSource, .versionHistory,
-             .tableListing, .routineDebug, nil:
+             .tableListing, .routineDebug, .liveMessages, nil:
             String(localized: "Save Changes")
         }
     }
@@ -113,8 +113,8 @@ internal enum ToolbarContextResolver {
         case .serverDashboard, .insights, .tableHealth, .objectSource, .versionHistory, .tableListing:
             /// None of these can stage a change, so the commit control could only ever be dim.
             return [MainWindowToolbar.saveChanges]
-        case .routineDebug:
-            // A debug run has nothing to reload or commit; its own controls take that place.
+        case .routineDebug, .liveMessages:
+            // A debug run or a live subscription has nothing to reload or commit; its own controls take that place.
             return [MainWindowToolbar.refresh, MainWindowToolbar.saveChanges]
         case .query, .table, .usersRoles, .erDiagram:
             return []
diff --git a/apple/TablePro/Core/Services/Infrastructure/WindowTitleResolver.swift b/apple/TablePro/Core/Services/Infrastructure/WindowTitleResolver.swift
index 0b57548852..26df4f0481 100644
--- a/apple/TablePro/Core/Services/Infrastructure/WindowTitleResolver.swift
+++ b/apple/TablePro/Core/Services/Infrastructure/WindowTitleResolver.swift
@@ -184,6 +184,11 @@ enum WindowTitleResolver {
                 return explicitTitle
             }
             return String(localized: "Debug")
+        case .liveMessages:
+            if let explicitTitle, !explicitTitle.isBlank {
+                return explicitTitle
+            }
+            return String(localized: "Live Messages")
         default:
             break
         }
diff --git a/apple/TablePro/Models/Query/QueryTabManager.swift b/apple/TablePro/Models/Query/QueryTabManager.swift
index 27fbb60c23..7f1d233ecf 100644
--- a/apple/TablePro/Models/Query/QueryTabManager.swift
+++ b/apple/TablePro/Models/Query/QueryTabManager.swift
@@ -586,6 +586,16 @@ final class QueryTabManager: ObservableObject {
         String(format: String(localized: "Debug %@"), name)
     }
 
+    @discardableResult
+    func addLiveMessagesTab(id: UUID = UUID(), title: String) -> UUID {
+        var newTab = QueryTab(id: id, title: title, tabType: .liveMessages)
+        newTab.tableContext.isEditable = false
+        newTab.hasUserInteraction = true
+        tabs.append(newTab)
+        selectedTabId = newTab.id
+        return newTab.id
+    }
+
     // Qualified by schema where tables are listed per schema, or two listings in one database would share a label.
     static func tableListingTitle(kindTitle: String, schemaName: String?) -> String {
         guard let schemaName, !schemaName.isEmpty else { return kindTitle }
@@ -807,7 +817,7 @@ extension QueryTabManager {
         case .query, .table, .createTable, .erDiagram, .objectSource, .tableListing:
             return true
         // A paused debug target holds locks, so its tab stays in sight whichever entry the sidebar shows.
-        case .serverDashboard, .usersRoles, .insights, .tableHealth, .versionHistory, .routineDebug:
+        case .serverDashboard, .usersRoles, .insights, .tableHealth, .versionHistory, .routineDebug, .liveMessages:
             return false
         }
     }
diff --git a/apple/TablePro/Models/Query/QueryTabState.swift b/apple/TablePro/Models/Query/QueryTabState.swift
index fa48f0b88c..57639beb68 100644
--- a/apple/TablePro/Models/Query/QueryTabState.swift
+++ b/apple/TablePro/Models/Query/QueryTabState.swift
@@ -22,12 +22,13 @@ enum TabType: Equatable, Codable, Hashable {
     // Read-only grid of one database's or schema's tables, rebuilt from the catalog on each load, never from saved SQL.
     case tableListing
     case routineDebug
+    case liveMessages
 }
 
 extension TabType {
-    // A live debug session cannot be restored, and the debugger is a module, so its tab never reaches disk or an outside client.
+    // A live session cannot be restored, and both tools are modules, so their tabs never reach disk or an outside client.
     var isTransient: Bool {
-        self == .routineDebug
+        self == .routineDebug || self == .liveMessages
     }
 }
 
diff --git a/apple/TablePro/Models/UI/GridSelectionOwner.swift b/apple/TablePro/Models/UI/GridSelectionOwner.swift
index 4364959955..7b1c633c4d 100644
--- a/apple/TablePro/Models/UI/GridSelectionOwner.swift
+++ b/apple/TablePro/Models/UI/GridSelectionOwner.swift
@@ -33,7 +33,7 @@ internal enum GridSelectionOwner: Equatable {
             // Map selects through the same model: a clicked shape selects its row.
             return .dataGrid
         case .createTable, .erDiagram, .serverDashboard, .usersRoles, .insights, .tableHealth, .objectSource,
-             .versionHistory, .tableListing, .routineDebug:
+             .versionHistory, .tableListing, .routineDebug, .liveMessages:
             return .none
         }
     }
diff --git a/apple/TablePro/Models/UI/PendingChangeKind.swift b/apple/TablePro/Models/UI/PendingChangeKind.swift
index 74c21d62be..e0bbe94d95 100644
--- a/apple/TablePro/Models/UI/PendingChangeKind.swift
+++ b/apple/TablePro/Models/UI/PendingChangeKind.swift
@@ -37,7 +37,7 @@ internal enum PendingChangeKind: Equatable, Hashable, Sendable {
         case .usersRoles:
             return hasPrincipalChanges ? .principals : nil
         case .query, .table, .erDiagram, .serverDashboard, .insights, .tableHealth, .objectSource, .versionHistory,
-             .tableListing, .routineDebug:
+             .tableListing, .routineDebug, .liveMessages:
             return contentKind(
                 hasDataChanges: hasDataChanges,
                 hasStructureChanges: hasStructureChanges,
diff --git a/apple/TablePro/Resources/Localizable.xcstrings b/apple/TablePro/Resources/Localizable.xcstrings
index b51e6f29db..8b55f9a598 100644
--- a/apple/TablePro/Resources/Localizable.xcstrings
+++ b/apple/TablePro/Resources/Localizable.xcstrings
@@ -9284,6 +9284,9 @@
           }
         }
       }
+    },
+    "%d messages" : {
+
     },
     "%d more" : {
       "localizations" : {
@@ -9376,6 +9379,9 @@
           }
         }
       }
+    },
+    "%d new" : {
+
     },
     "%d of %d" : {
       "localizations" : {
@@ -33117,6 +33123,9 @@
     },
     "Can't Debug %@" : {
 
+    },
+    "Can't Export Messages" : {
+
     },
     "Can't List Routines" : {
 
@@ -36168,6 +36177,12 @@
           }
         }
       }
+    },
+    "Channel" : {
+
+    },
+    "Channels" : {
+
     },
     "Character Set" : {
       "extractionState" : "manual",
@@ -83884,6 +83899,12 @@
           }
         }
       }
+    },
+    "Follow" : {
+
+    },
+    "Follow Topics" : {
+
     },
     "Follow-up on step %d \"%@\" (%@): %@" : {
       "localizations" : {
@@ -84962,6 +84983,9 @@
           }
         }
       }
+    },
+    "Freeze the list; messages keep arriving underneath" : {
+
     },
     "Freeze Up to This Column" : {
 
@@ -104146,6 +104170,12 @@
           }
         }
       }
+    },
+    "Listen" : {
+
+    },
+    "Listen for Notifications" : {
+
     },
     "Listens on all interfaces (0.0.0.0), reachable from your local network." : {
       "localizations" : {
@@ -104296,6 +104326,18 @@
           }
         }
       }
+    },
+    "Live Messages" : {
+
+    },
+    "Live Messages Ended" : {
+
+    },
+    "Live Messages…" : {
+
+    },
+    "Live: %@" : {
+
     },
     "Load" : {
       "localizations" : {
@@ -132292,6 +132334,9 @@
     },
     "Pattern" : {
 
+    },
+    "Patterns" : {
+
     },
     "Pause" : {
       "extractionState" : "manual",
@@ -132432,6 +132477,9 @@
     "Paw Print" : {
       "extractionState" : "manual",
       "generatesSymbol" : false
+    },
+    "Payload" : {
+
     },
     "Pending approval for" : {
       "localizations" : {
@@ -137603,6 +137651,9 @@
     "Processor" : {
       "extractionState" : "manual",
       "generatesSymbol" : false
+    },
+    "Produce" : {
+
     },
     "PRODUCE needs a topic name." : {
       "extractionState" : "manual",
@@ -143615,6 +143666,9 @@
           }
         }
       }
+    },
+    "Reconnecting…" : {
+
     },
     "Record %@ has no fields list." : {
 
@@ -160983,6 +161037,9 @@
           }
         }
       }
+    },
+    "Send" : {
+
     },
     "Send Message" : {
       "localizations" : {
@@ -161376,6 +161433,18 @@
           }
         }
       }
+    },
+    "Separate names with commas." : {
+
+    },
+    "Separate names with commas. A separate session runs LISTEN and closes with the tab." : {
+
+    },
+    "Separate names with commas. Patterns use glob syntax, such as news.*. A separate connection subscribes and closes with the tab." : {
+
+    },
+    "Separate names with commas. Reading starts at the end of each partition, without a consumer group, so no committed offset moves." : {
+
     },
     "Separator" : {
       "localizations" : {
@@ -165514,6 +165583,9 @@
           }
         }
       }
+    },
+    "Show the messages that arrived while paused" : {
+
     },
     "Show Toolbar" : {
       "localizations" : {
@@ -174933,6 +175005,12 @@
           }
         }
       }
+    },
+    "Subscribe" : {
+
+    },
+    "Subscribe to Channels" : {
+
     },
     "Subtype" : {
 
@@ -184790,6 +184868,9 @@
           }
         }
       }
+    },
+    "The connection closed." : {
+
     },
     "The connection failed while Spanner was committing, so the changes may have been saved. Check the data before running it again." : {
       "extractionState" : "manual",
@@ -187705,6 +187786,9 @@
           }
         }
       }
+    },
+    "The live message view has closed." : {
+
     },
     "The map cannot draw %@." : {
 
@@ -194590,6 +194674,9 @@
           }
         }
       }
+    },
+    "This connection can't open the separate session live messages need." : {
+
     },
     "This connection can't open the separate sessions the debugger needs." : {
 
@@ -195173,6 +195260,9 @@
     },
     "This connection's driver can't debug routines." : {
 
+    },
+    "This connection's driver can't follow live messages." : {
+
     },
     "This connection's driver does not list its tables." : {
       "localizations" : {
@@ -202246,6 +202336,9 @@
           }
         }
       }
+    },
+    "Topics" : {
+
     },
     "Tortoise" : {
       "extractionState" : "manual",
@@ -211944,6 +212037,9 @@
     },
     "Was %@" : {
 
+    },
+    "Watch PostgreSQL notifications, Redis channels and Kafka topics as messages arrive." : {
+
     },
     "Watch shared folders for connection files." : {
       "localizations" : {
diff --git a/apple/TablePro/Views/LiveMessages/LiveMessageStartSheet.swift b/apple/TablePro/Views/LiveMessages/LiveMessageStartSheet.swift
new file mode 100644
index 0000000000..b847b95c9e
--- /dev/null
+++ b/apple/TablePro/Views/LiveMessages/LiveMessageStartSheet.swift
@@ -0,0 +1,120 @@
+//
+//  LiveMessageStartSheet.swift
+//  TablePro
+//
+
+import SwiftUI
+import TableProPluginKit
+
+internal struct LiveMessageStartSheet: View {
+    let kind: PluginLiveMessageKind
+    let onFinish: @MainActor (PluginLiveSubscriptionRequest?) -> Void
+
+    @State private var channels: String
+    @State private var patterns = ""
+
+    init(
+        kind: PluginLiveMessageKind,
+        suggestedChannel: String,
+        onFinish: @escaping @MainActor (PluginLiveSubscriptionRequest?) -> Void
+    ) {
+        self.kind = kind
+        self.onFinish = onFinish
+        _channels = State(initialValue: suggestedChannel)
+    }
+
+    // Commas or new lines separate names; a name itself may hold spaces.
+    nonisolated static func names(in text: String) -> [String] {
+        var seen = Set<String>()
+        return text
+            .split(whereSeparator: { $0 == "," || $0.isNewline })
+            .map { $0.trimmingCharacters(in: .whitespaces) }
+            .filter { !$0.isEmpty && seen.insert($0).inserted }
+    }
+
+    private var request: PluginLiveSubscriptionRequest {
+        PluginLiveSubscriptionRequest(
+            channels: Self.names(in: channels),
+            patterns: kind == .pubSub ? Self.names(in: patterns) : []
+        )
+    }
+
+    private var canStart: Bool {
+        !request.channels.isEmpty || !request.patterns.isEmpty
+    }
+
+    var body: some View {
+        VStack(alignment: .leading, spacing: 16) {
+            Text(LiveMessageKindText.sheetTitle(kind))
+                .font(.headline)
+            Form {
+                TextField(LiveMessageKindText.channelsLabel(kind), text: $channels)
+                    .accessibilityIdentifier("live-messages-channels")
+                if kind == .pubSub {
+                    TextField(String(localized: "Patterns"), text: $patterns)
+                        .accessibilityIdentifier("live-messages-patterns")
+                }
+            }
+            Text(LiveMessageKindText.sheetNote(kind))
+                .font(.callout)
+                .foregroundStyle(.secondary)
+                .fixedSize(horizontal: false, vertical: true)
+            HStack {
+                Spacer()
+                Button(String(localized: "Cancel"), role: .cancel) { onFinish(nil) }
+                    .keyboardShortcut(.cancelAction)
+                Button(LiveMessageKindText.startTitle(kind)) { onFinish(request) }
+                    .keyboardShortcut(.defaultAction)
+                    .disabled(!canStart)
+            }
+        }
+        .padding(20)
+        .frame(width: 440)
+    }
+}
+
+internal enum LiveMessageKindText {
+    static func sheetTitle(_ kind: PluginLiveMessageKind) -> String {
+        switch kind {
+        case .pubSub: String(localized: "Subscribe to Channels")
+        case .topics: String(localized: "Follow Topics")
+        case .notifications: String(localized: "Listen for Notifications")
+        @unknown default: String(localized: "Listen for Notifications")
+        }
+    }
+
+    static func channelsLabel(_ kind: PluginLiveMessageKind) -> String {
+        kind == .topics ? String(localized: "Topics") : String(localized: "Channels")
+    }
+
+    static func startTitle(_ kind: PluginLiveMessageKind) -> String {
+        switch kind {
+        case .pubSub: String(localized: "Subscribe")
+        case .topics: String(localized: "Follow")
+        case .notifications: String(localized: "Listen")
+        @unknown default: String(localized: "Listen")
+        }
+    }
+
+    static func sheetNote(_ kind: PluginLiveMessageKind) -> String {
+        switch kind {
+        case .pubSub:
+            String(localized: "Separate names with commas. Patterns use glob syntax, such as news.*. A separate connection subscribes and closes with the tab.")
+        case .topics:
+            String(localized: "Separate names with commas. Reading starts at the end of each partition, without a consumer group, so no committed offset moves.")
+        case .notifications:
+            String(localized: "Separate names with commas. A separate session runs LISTEN and closes with the tab.")
+        @unknown default:
+            String(localized: "Separate names with commas.")
+        }
+    }
+
+    static func publishTitle(_ kind: PluginLiveMessageKind) -> String {
+        switch kind {
+        case .pubSub: "PUBLISH"
+        case .topics: String(localized: "Produce")
+        case .notifications: "NOTIFY"
+        @unknown default: String(localized: "Send")
+        }
+    }
+}
diff --git a/apple/TablePro/Views/LiveMessages/LiveMessageTabView.swift b/apple/TablePro/Views/LiveMessages/LiveMessageTabView.swift
new file mode 100644
index 0000000000..4c33e9c04c
--- /dev/null
+++ b/apple/TablePro/Views/LiveMessages/LiveMessageTabView.swift
@@ -0,0 +1,265 @@
+//
+//  LiveMessageTabView.swift
+//  TablePro
+//
+
+import AppKit
+import SwiftUI
+import TableProPluginKit
+import UniformTypeIdentifiers
+
+// A live tab exists only with a live controller, so a tab that has none says so rather than drawing an empty table.
+internal struct LiveMessageTabContent: View {
+    let tabId: UUID
+
+    var body: some View {
+        if let controller = LiveMessageSessions.shared.controller(forTab: tabId) {
+            LiveMessageTabView(controller: controller)
+        } else {
+            UnavailableStateView(String(localized: "Live Messages Ended"), systemImage: "dot.radiowaves.left.and.right")
+                .frame(maxWidth: .infinity, maxHeight: .infinity)
+        }
+    }
+}
+
+internal struct LiveMessageTabView: View {
+    @ObservedObject var controller: LiveMessageController
+    @State private var selection = Set<LiveMessageEntry.ID>()
+    @State private var publishChannel = ""
+    @State private var publishPayload = ""
+    @State private var isPublishing = false
+    @State private var publishError: String?
+
+    // Newest first, so an arriving message is in sight without scrolling.
+    private var rows: [LiveMessageEntry] {
+        controller.shown.reversed()
+    }
+
+    var body: some View {
+        VStack(spacing: 0) {
+            controlBar
+                .paneBarBackground()
+            Divider()
+            table
+                .frame(maxWidth: .infinity, maxHeight: .infinity)
+            Divider()
+            publishBar
+                .paneBarBackground()
+        }
+        .onAppear {
+            if publishChannel.isEmpty {
+                publishChannel = controller.request.channels.first ?? ""
+            }
+        }
+    }
+
+    // MARK: - Control bar
+
+    private var controlBar: some View {
+        HStack(spacing: 10) {
+            status
+            Spacer(minLength: 8)
+            if controller.isPaused, controller.unseenCount > 0 {
+                Text(String(format: String(localized: "%d new"), controller.unseenCount))
+                    .font(.callout)
+                    .foregroundStyle(.secondary)
+                    .monospacedDigit()
+            }
+            Text(String(format: String(localized: "%d messages"), controller.shown.count))
+                .font(.callout)
+                .foregroundStyle(.secondary)
+                .monospacedDigit()
+            NativeSearchField(
+                text: $controller.filter,
+                placeholder: String(localized: "Filter"),
+                controlSize: .small,
+                accessibilityIdentifier: "live-messages-filter"
+            )
+            .frame(width: 180)
+            Button {
+                controller.isPaused ? controller.resume() : controller.pause()
+            } label: {
+                Label(
+                    controller.isPaused ? String(localized: "Resume") : String(localized: "Pause"),
+                    systemImage: controller.isPaused ? "play.fill" : "pause.fill"
+                )
+            }
+            .help(controller.isPaused
+                ? String(localized: "Show the messages that arrived while paused")
+                : String(localized: "Freeze the list; messages keep arriving underneath"))
+            Button(String(localized: "Clear")) {
+                selection.removeAll()
+                controller.clear()
+            }
+            Button(String(localized: "Copy")) { copy() }
+                .disabled(controller.shown.isEmpty)
+            Button(String(localized: "Export…")) { export() }
+                .disabled(controller.shown.isEmpty)
+        }
+        .controlSize(.small)
+        .padding(.horizontal, 12)
+        .padding(.vertical, 6)
+    }
+
+    @ViewBuilder
+    private var status: some View {
+        switch controller.phase {
+        case .connecting:
+            Label(String(localized: "Connecting…"), systemImage: "circle.dotted")
+                .foregroundStyle(.secondary)
+        case .live:
+            Label(LiveMessageController.title(for: controller.request), systemImage: "dot.radiowaves.left.and.right")
+                .lineLimit(1)
+                .truncationMode(.tail)
+        case .reconnecting(let reason):
+            Label(String(localized: "Reconnecting…"), systemImage: "arrow.triangle.2.circlepath")
+                .foregroundStyle(.secondary)
+                .help(reason)
+        case .ended(let error):
+            if let error {
+                Label(error, systemImage: "exclamationmark.triangle")
+                    .foregroundStyle(.secondary)
+                    .lineLimit(2)
+                    .help(error)
+            } else {
+                Label(String(localized: "Stopped"), systemImage: "stop.circle")
+                    .foregroundStyle(.secondary)
+            }
+        }
+    }
+
+    // MARK: - Table
+
+    @ViewBuilder
+    private var table: some View {
+        switch controller.kind {
+        case .pubSub:
+            Table(rows, selection: $selection) {
+                timeColumn
+                textColumn(.channel)
+                textColumn(.pattern)
+                textColumn(.payload)
+            }
+            .onCopyCommand { copyProviders() }
+        case .topics:
+            Table(rows, selection: $selection) {
+                timeColumn
+                textColumn(.channel)
+                textColumn(.partition)
+                textColumn(.offset)
+                textColumn(.key)
+                textColumn(.payload)
+            }
+            .onCopyCommand { copyProviders() }
+        case .notifications:
+            channelTable
+        @unknown default:
+            channelTable
+        }
+    }
+
+    private var channelTable: some View {
+        Table(rows, selection: $selection) {
+            timeColumn
+            textColumn(.channel)
+            textColumn(.payload)
+        }
+        .onCopyCommand { copyProviders() }
+    }
+
+    private var timeColumn: TableColumn<LiveMessageEntry, Never, Text, Text> {
+        TableColumn(LiveMessageExport.Column.time.title) { entry in
+            Text(entry.message.map { $0.timestamp.formatted(.dateTime.hour().minute().second().secondFraction(.fractional(3))) } ?? "")
+                .monospacedDigit()
+        }
+    }
+
+    private func textColumn(_ column: LiveMessageExport.Column) -> TableColumn<LiveMessageEntry, Never, Text, Text> {
+        TableColumn(column.title) { entry in
+            let text = Text(LiveMessageExport.value(of: column, in: entry))
+            if entry.message == nil {
+                return text.foregroundColor(.secondary).italic()
+            }
+            return text
+        }
+    }
+
+    // MARK: - Publish
+
+    private var publishBar: some View {
+        HStack(spacing: 8) {
+            Text(LiveMessageKindText.publishTitle(controller.kind))
+                .font(.callout.weight(.medium))
+                .foregroundStyle(.secondary)
+            TextField(LiveMessageKindText.channelsLabel(controller.kind), text: $publishChannel)
+                .frame(width: 160)
+                .accessibilityIdentifier("live-messages-publish-channel")
+            TextField(String(localized: "Payload"), text: $publishPayload)
+                .onSubmit { send() }
+                .accessibilityIdentifier("live-messages-publish-payload")
+            Button(String(localized: "Send")) { send() }
+                .disabled(!controller.canPublish || isPublishing
+                    || publishChannel.trimmingCharacters(in: .whitespaces).isEmpty)
+            if let publishError {
+                Image(systemName: "exclamationmark.triangle")
+                    .foregroundStyle(.secondary)
+                    .help(publishError)
+            }
+        }
+        .controlSize(.small)
+        .padding(.horizontal, 12)
+        .padding(.vertical, 6)
+    }
+
+    private func send() {
+        let channel = publishChannel.trimmingCharacters(in: .whitespaces)
+        guard controller.canPublish, !isPublishing, !channel.isEmpty else { return }
+        let payload = publishPayload
+        isPublishing = true
+        publishError = nil
+        Task {
+            do {
+                try await controller.publish(payload, to: channel)
+                publishPayload = ""
+            } catch {
+                publishError = error.localizedDescription
+            }
+            isPublishing = false
+        }
+    }
+
+    // MARK: - Copy and export
+
+    private var selectedOrAll: [LiveMessageEntry] {
+        let chosen = controller.shown.filter { selection.contains($0.id) }
+        return chosen.isEmpty ? controller.shown : chosen
+    }
+
+    private func copy() {
+        ClipboardService.shared.writeText(LiveMessageExport.tsv(selectedOrAll, kind: controller.kind))
+    }
+
+    private func copyProviders() -> [NSItemProvider] {
+        [NSItemProvider(object: LiveMessageExport.tsv(selectedOrAll, kind: controller.kind) as NSString)]
+    }
+
+    private func export() {
+        let text = LiveMessageExport.csv(controller.shown, kind: controller.kind)
+        let panel = NSSavePanel()
+        panel.allowedContentTypes = [.commaSeparatedText]
+        panel.nameFieldStringValue = "live-messages.csv"
+        guard let window = NSApp.keyWindow else { return }
+        panel.beginSheetModal(for: window) { response in
+            guard response == .OK, let url = panel.url else { return }
+            do {
+                try Data(text.utf8).write(to: url, options: .atomic)
+            } catch {
+                AlertHelper.showErrorSheet(
+                    title: String(localized: "Can't Export Messages"),
+                    message: error.localizedDescription,
+                    window: window
+                )
+            }
+        }
+    }
+}
diff --git a/apple/TablePro/Views/Main/Child/MainEditorContentView.swift b/apple/TablePro/Views/Main/Child/MainEditorContentView.swift
index 458ab4de9a..a3de00ddeb 100644
--- a/apple/TablePro/Views/Main/Child/MainEditorContentView.swift
+++ b/apple/TablePro/Views/Main/Child/MainEditorContentView.swift
@@ -217,6 +217,9 @@ struct MainEditorContentView: View {
         case .routineDebug:
             RoutineDebugTabContent(tabId: tab.id, databaseType: connection.type)
                 .id(tab.id)
+        case .liveMessages:
+            LiveMessageTabContent(tabId: tab.id)
+                .id(tab.id)
         }
     }
 
diff --git a/apple/TablePro/Views/Main/Extensions/MainContentCoordinator+Refresh.swift b/apple/TablePro/Views/Main/Extensions/MainContentCoordinator+Refresh.swift
index e95d3fb3c2..b08726e990 100644
--- a/apple/TablePro/Views/Main/Extensions/MainContentCoordinator+Refresh.swift
+++ b/apple/TablePro/Views/Main/Extensions/MainContentCoordinator+Refresh.swift
@@ -55,7 +55,7 @@ extension MainContentCoordinator {
         case .usersRoles:
             usersRolesActions?.refresh()
             return
-        case .query, .table, .createTable, .routineDebug:
+        case .query, .table, .createTable, .routineDebug, .liveMessages:
             break
         }
         if tab.display.resultsViewMode == .structure {
diff --git a/apple/TablePro/Views/Main/Extensions/MainContentCoordinator+TabClosing.swift b/apple/TablePro/Views/Main/Extensions/MainContentCoordinator+TabClosing.swift
index b2bb6f89cc..3a7f159b30 100644
--- a/apple/TablePro/Views/Main/Extensions/MainContentCoordinator+TabClosing.swift
+++ b/apple/TablePro/Views/Main/Extensions/MainContentCoordinator+TabClosing.swift
@@ -15,6 +15,7 @@ extension MainContentCoordinator {
             saveLastFilters(of: tab)
             if tab.tabType.isTransient {
                 RoutineDebugSessions.shared.endSession(forTab: id, window: contentWindow)
+                LiveMessageSessions.shared.end(tabId: id)
             } else {
                 RecentlyClosedTabStore.shared.push(tab: tab, connection: connection)
             }
diff --git a/apple/TablePro/Views/Main/MainContentCoordinator.swift b/apple/TablePro/Views/Main/MainContentCoordinator.swift
index 5c95425d6b..c2356729e2 100644
--- a/apple/TablePro/Views/Main/MainContentCoordinator.swift
+++ b/apple/TablePro/Views/Main/MainContentCoordinator.swift
@@ -879,6 +879,7 @@ final class MainContentCoordinator: ObservableObject {
         let debugTabIds = tabManager.tabs.filter { $0.tabType.isTransient }.map(\.id)
         if !debugTabIds.isEmpty {
             RoutineDebugSessions.shared.endSessions(forTabs: debugTabIds, window: contentWindow)
+            LiveMessageSessions.shared.end(tabIds: debugTabIds)
         }
         tabManager.tabs.removeAll()
         tabManager.selectedTabId = nil
diff --git a/apple/TableProTests/Core/LiveMessages/LiveMessageBufferTests.swift b/apple/TableProTests/Core/LiveMessages/LiveMessageBufferTests.swift
new file mode 100644
index 0000000000..c86488a1da
--- /dev/null
+++ b/apple/TableProTests/Core/LiveMessages/LiveMessageBufferTests.swift
@@ -0,0 +1,58 @@
+import Foundation
+@testable import TablePro
+import TableProPluginKit
+import Testing
+
+struct LiveMessageBufferTests {
+    private static func message(_ payload: String, channel: String = "orders") -> PluginLiveMessage {
+        PluginLiveMessage(timestamp: Date(timeIntervalSince1970: 0), channel: channel, payload: payload)
+    }
+
+    @Test("Keeps only the newest messages up to the cap and counts the ones it dropped")
+    func keepsNewestUpToCap() {
+        var buffer = LiveMessageBuffer(capacity: 3)
+        buffer.append((1...5).map { Self.message("m\($0)") })
+
+        #expect(buffer.entries.compactMap { $0.message?.payload } == ["m3", "m4", "m5"])
+        #expect(buffer.droppedCount == 2)
+    }
+
+    @Test("Entry ids keep growing after a trim and a clear, so a table never reuses one")
+    func idsNeverRepeat() {
+        var buffer = LiveMessageBuffer(capacity: 2)
+        buffer.append([Self.message("a"), Self.message("b"), Self.message("c")])
+        let before = buffer.entries.map(\.id)
+        buffer.clear()
+        buffer.append([Self.message("d")])
+
+        #expect(buffer.entries.isEmpty == false)
+        #expect(!before.contains(buffer.entries[0].id))
+    }
+
+    @Test("A gap marker is kept in order and always passes the filter")
+    func gapKeptAndUnfiltered() {
+        var buffer = LiveMessageBuffer()
+        buffer.append([Self.message("before")])
+        buffer.appendGap(reason: "server closed the connection")
+        buffer.append([Self.message("after", channel: "audit")])
+
+        let visible = buffer.visible(filter: "audit")
+        #expect(visible.count == 2)
+        #expect(visible.first?.message == nil)
+        #expect(visible.last?.message?.payload == "after")
+    }
+
+    @Test("The filter matches channel, payload and key, ignoring case")
+    func filterMatchesFields() {
+        var buffer = LiveMessageBuffer()
+        buffer.append([
+            Self.message("hello", channel: "orders"),
+            PluginLiveMessage(timestamp: Date(), channel: "t", payload: "x", key: "Customer-7"),
+            Self.message("other", channel: "misc"),
+        ])
+
+        #expect(buffer.visible(filter: "ORDERS").count == 1)
+        #expect(buffer.visible(filter: "customer").count == 1)
+        #expect(buffer.visible(filter: "  ").count == 3)
+    }
+}
diff --git a/apple/TableProTests/Core/LiveMessages/LiveMessageControllerTests.swift b/apple/TableProTests/Core/LiveMessages/LiveMessageControllerTests.swift
new file mode 100644
index 0000000000..5d26de889a
--- /dev/null
+++ b/apple/TableProTests/Core/LiveMessages/LiveMessageControllerTests.swift
@@ -0,0 +1,174 @@
+import Foundation
+@testable import TablePro
+import TableProPluginKit
+import TableProTestSupport
+import Testing
+
+private final class FakeSubscription: PluginLiveSubscription, @unchecked Sendable {
+    let messages: AsyncThrowingStream<PluginLiveMessage, Error>
+    let continuation: AsyncThrowingStream<PluginLiveMessage, Error>.Continuation
+    private let lock = NSLock()
+    private var _published: [String] = []
+    private var _isClosed = false
+
+    init() {
+        (messages, continuation) = AsyncThrowingStream<PluginLiveMessage, Error>.makeStream()
+    }
+
+    var published: [String] { lock.withLock { _published } }
+    var isClosed: Bool { lock.withLock { _isClosed } }
+
+    func publish(_ payload: String, to channel: String) async throws {
+        lock.withLock { _published.append("\(channel):\(payload)") }
+    }
+
+    func close() async {
+        lock.withLock { _isClosed = true }
+        continuation.finish()
+    }
+}
+
+private struct Dropped: Error, LocalizedError {
+    var errorDescription: String? { "server closed the connection" }
+}
+
+@MainActor
+private final class Harness {
+    var subscriptions: [FakeSubscription] = []
+    var failFirst = false
+    var opened = 0
+    var closedSources = 0
+    let clock = ManualClock()
+    private(set) var controller: LiveMessageController!
+
+    init(capacity: Int = LiveMessageBuffer.defaultCapacity) {
+        controller = LiveMessageController(
+            kind: .notifications,
+            request: PluginLiveSubscriptionRequest(channels: ["orders"]),
+            capacity: capacity,
+            clock: clock,
+            makeSource: { LiveMessageSessionSource { throw LiveMessageError.unsupported } },
+            subscribe: { [weak self] _ in try await self!.next() }
+        )
+    }
+
+    func next() async throws -> any PluginLiveSubscription {
+        opened += 1
+        if failFirst { throw Dropped() }
+        let subscription = FakeSubscription()
+        subscriptions.append(subscription)
+        return subscription
+    }
+
+    static func message(_ payload: String) -> PluginLiveMessage {
+        PluginLiveMessage(timestamp: Date(), channel: "orders", payload: payload)
+    }
+}
+
+@MainActor
+struct LiveMessageControllerTests {
+    @Test("Messages that arrive while paused are kept and shown on resume")
+    func pauseLosesNothing() async throws {
+        let harness = Harness()
+        let controller = harness.controller!
+        controller.start()
+        try await waitUntil { controller.phase == .live }
+
+        controller.pause()
+        harness.subscriptions[0].continuation.yield(Harness.message("one"))
+        harness.subscriptions[0].continuation.yield(Harness.message("two"))
+        try await waitUntil { controller.unseenCount == 2 }
+        #expect(controller.shown.isEmpty)
+
+        controller.resume()
+        #expect(controller.shown.compactMap { $0.message?.payload } == ["one", "two"])
+        #expect(controller.unseenCount == 0)
+        controller.stop()
+    }
+
+    @Test("A dropped connection leaves one gap row and subscribes again after the backoff")
+    func dropReconnects() async throws {
+        let harness = Harness()
+        let controller = harness.controller!
+        controller.start()
+        try await waitUntil { controller.phase == .live }
+
+        harness.subscriptions[0].continuation.finish(throwing: Dropped())
+        try await waitUntil { controller.phase == .reconnecting("server closed the connection") }
+        #expect(harness.subscriptions[0].isClosed)
+        #expect(controller.buffer.entries.count == 1)
+        #expect(controller.buffer.entries.first?.message == nil)
+
+        // The table refresh and the backoff both sleep on the clock.
+        try await waitUntil { harness.clock.pendingSleeperCount == 2 }
+        harness.clock.advance(by: .seconds(1))
+        try await waitUntil { harness.opened == 2 && controller.phase == .live }
+        controller.stop()
+    }
+
+    @Test("A first subscribe that fails ends the tab with the error instead of retrying")
+    func firstFailureEnds() async throws {
+        let harness = Harness()
+        harness.failFirst = true
+        let controller = harness.controller!
+        controller.start()
+
+        try await waitUntil { controller.hasEnded }
+        #expect(controller.phase == .ended("server closed the connection"))
+        #expect(harness.opened == 1)
+    }
+
+    @Test("Stopping closes the subscription and refuses a publish")
+    func stopCloses() async throws {
+        let harness = Harness()
+        let controller = harness.controller!
+        controller.start()
+        try await waitUntil { controller.phase == .live }
+
+        try await controller.publish("hi", to: "orders")
+        #expect(harness.subscriptions[0].published == ["orders:hi"])
+
+        controller.stop()
+        try await waitUntil { harness.subscriptions[0].isClosed }
+        await #expect(throws: LiveMessageError.ended) {
+            try await controller.publish("late", to: "orders")
+        }
+    }
+
+    @Test("Closing a live tab's tab id stops its controller")
+    func sessionsEndStopsController() async throws {
+        let harness = Harness()
+        let controller = harness.controller!
+        let sessions = LiveMessageSessions()
+        let tabId = UUID()
+        sessions.register(controller, tabId: tabId, connectionId: UUID())
+        controller.start()
+        try await waitUntil { controller.phase == .live }
+
+        sessions.end(tabId: tabId)
+        #expect(controller.hasEnded)
+        #expect(sessions.controller(forTab: tabId) == nil)
+        try await waitUntil { harness.subscriptions[0].isClosed }
+    }
+}
+
+@MainActor
+struct LiveMessageSessionsTests {
+    @Test("Disconnecting a connection stops only that connection's live tabs")
+    func endByConnection() async throws {
+        let first = Harness()
+        let second = Harness()
+        let sessions = LiveMessageSessions()
+        let connection = UUID()
+        sessions.register(first.controller, tabId: UUID(), connectionId: connection)
+        sessions.register(second.controller, tabId: UUID(), connectionId: UUID())
+        first.controller.start()
+        second.controller.start()
+        try await waitUntil { first.controller.phase == .live && second.controller.phase == .live }
+
+        sessions.end(connectionId: connection)
+        #expect(first.controller.hasEnded)
+        #expect(!second.controller.hasEnded)
+        second.controller.stop()
+    }
+}
diff --git a/apple/TableProTests/Core/LiveMessages/LiveMessageExportTests.swift b/apple/TableProTests/Core/LiveMessages/LiveMessageExportTests.swift
new file mode 100644
index 0000000000..24c3e60418
--- /dev/null
+++ b/apple/TableProTests/Core/LiveMessages/LiveMessageExportTests.swift
@@ -0,0 +1,41 @@
+import Foundation
+@testable import TablePro
+import TableProPluginKit
+import Testing
+
+struct LiveMessageExportTests {
+    private static let kafka = LiveMessageEntry(id: 0, kind: .message(PluginLiveMessage(
+        timestamp: Date(timeIntervalSince1970: 0),
+        channel: "orders",
+        payload: "a,\"b\"",
+        key: "k1",
+        partition: 2,
+        offset: 41
+    )))
+
+    @Test("Each kind shows its own columns")
+    func columnsPerKind() {
+        #expect(LiveMessageExport.columns(for: .notifications) == [.time, .channel, .payload])
+        #expect(LiveMessageExport.columns(for: .pubSub).contains(.pattern))
+        #expect(LiveMessageExport.columns(for: .topics) == [.time, .channel, .partition, .offset, .key, .payload])
+    }
+
+    @Test("CSV quotes a field holding a comma or a quote")
+    func csvQuotes() {
+        let csv = LiveMessageExport.csv([Self.kafka], kind: .topics)
+        let row = csv.components(separatedBy: "\r\n")[1]
+        #expect(row.hasSuffix(",orders,2,41,k1,\"a,\"\"b\"\"\""))
+    }
+
+    @Test("A gap row copies as Connection lost with its reason")
+    func gapCopies() {
+        let gap = LiveMessageEntry(id: 1, kind: .gap(reason: "timeout"))
+        let tsv = LiveMessageExport.tsv([gap], kind: .notifications)
+        #expect(tsv.components(separatedBy: "\n")[1] == "\t\(String(localized: "Connection lost"))\ttimeout")
+    }
+
+    @Test("Names split on commas and new lines, trimmed and without repeats")
+    func startSheetNames() {
+        #expect(LiveMessageStartSheet.names(in: " orders, audit log\norders,, ") == ["orders", "audit log"])
+    }
+}
diff --git a/apple/TableProTests/Core/LiveMessages/LiveMessagesModuleResidueTests.swift b/apple/TableProTests/Core/LiveMessages/LiveMessagesModuleResidueTests.swift
new file mode 100644
index 0000000000..68cd0d6a71
--- /dev/null
+++ b/apple/TableProTests/Core/LiveMessages/LiveMessagesModuleResidueTests.swift
@@ -0,0 +1,40 @@
+import AppKit
+import Foundation
+@testable import TablePro
+import Testing
+
+// An off module leaves nothing behind (apple/AGENTS.md, Scope): no Live Messages… command anywhere in the menu bar.
+@MainActor
+struct LiveMessagesModuleResidueTests {
+    private func liveItems(in menu: NSMenu) -> [NSMenuItem] {
+        menu.items.flatMap { item -> [NSMenuItem] in
+            let own = item.action == #selector(MainSplitViewController.openLiveMessages(_:))
+                || item.identifier == DatabaseMenuBuilder.liveMessagesIdentifier ? [item] : []
+            return own + (item.submenu.map(liveItems(in:)) ?? [])
+        }
+    }
+
+    @Test("Off: the menu bar has no Live Messages… command")
+    func offLeavesNoMenuItem() {
+        let bar = MainMenuBuilder.build(keyboard: KeyboardSettings(), modules: []).bar
+        #expect(liveItems(in: bar).isEmpty)
+    }
+
+    @Test("On: Live Messages… sits in the Database menu, and turning the module off removes it")
+    func togglesWithModule() throws {
+        let built = MainMenuBuilder.build(keyboard: KeyboardSettings(), modules: [.liveMessages])
+        let database = try #require(built.bar.items.first { $0.identifier == DatabaseMenuBuilder.identifier }?.submenu)
+        #expect(liveItems(in: database).count == 1)
+
+        MainMenuBuilder.applyModules([], keyboard: KeyboardSettings(), to: built.bar, windowsMenu: built.windowsMenu)
+        #expect(liveItems(in: built.bar).isEmpty)
+
+        MainMenuBuilder.applyModules([.liveMessages], keyboard: KeyboardSettings(), to: built.bar, windowsMenu: built.windowsMenu)
+        #expect(liveItems(in: built.bar).count == 1)
+    }
+
+    @Test("The module is off by default")
+    func offByDefault() {
+        #expect(!AppModule.liveMessages.defaultEnabled)
+    }
+}
diff --git a/apple/TableProTests/Core/Services/EditorTabConnectionMovePolicyTests.swift b/apple/TableProTests/Core/Services/EditorTabConnectionMovePolicyTests.swift
index 401701638c..5346e27113 100644
--- a/apple/TableProTests/Core/Services/EditorTabConnectionMovePolicyTests.swift
+++ b/apple/TableProTests/Core/Services/EditorTabConnectionMovePolicyTests.swift
@@ -17,7 +17,7 @@ struct EditorTabConnectionMovePolicyTests {
     @Test("Every other tab type names an object of its own connection and stays")
     func nonQueryTabsStay() {
         for tabType in [TabType.table, .createTable, .erDiagram, .serverDashboard, .usersRoles, .insights,
-                        .tableHealth, .objectSource, .versionHistory, .tableListing, .routineDebug] {
+                        .tableHealth, .objectSource, .versionHistory, .tableListing, .routineDebug, .liveMessages] {
             #expect(!EditorTabConnectionMovePolicy.canMove(tabType: tabType, isBusy: false, hasPendingGridEdits: false))
         }
     }
diff --git a/apple/changelog.d/3981.added.md b/apple/changelog.d/3981.added.md
new file mode 100644
index 0000000000..fa0761e735
--- /dev/null
+++ b/apple/changelog.d/3981.added.md
@@ -0,0 +1 @@
+Live Messages module: watch PostgreSQL notifications, Redis channels and Kafka topics as messages arrive.