diff --git a/CHANGELOG.md b/CHANGELOG.md
index 38611973..6981c247 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,9 @@
 
 ## 0.13.2 - Unreleased
 
+### Read Commands
+- fix: coalesce GUID-linked Apple URL previews even when the originating text row omits the URL, preserving one logical message across history, search, stats, unread counts, and at-least-once live watch with a no-skip physical `cursor` (#181, thanks @omarshahine).
+
 ## 0.13.1 - 2026-07-17
 
 ### Highlights
diff --git a/README.md b/README.md
index e884ad40..4c72b74e 100644
--- a/README.md
+++ b/README.md
@@ -233,8 +233,18 @@ Before handing the file to Messages, `imsg` stages it under
 ## Watch Behavior
 
 `imsg watch` starts at the newest message by default and streams messages
-written after it starts. Use `--since-rowid <id>` to resume from a stored
-cursor.
+written after it starts. Use `--since-rowid <cursor>` to resume from a stored
+`cursor` field. The cursor normally matches `id`, but advances past a folded
+URL-preview row only after applicable interleaved rows have been emitted, so
+resuming does not skip part of the stream. Until that frontier is complete, an
+interleaved event can repeat the prior cursor and may be replayed after a
+reconnect; deduplicate at-least-once delivery by stable `chat_id` plus `id` or
+`guid`.
+
+New live text rows are held for up to two seconds so a delayed, GUID-linked
+Apple URL preview can be folded into the same logical event. Backlog and resume
+reads are not delayed; when both physical rows already exist, they are
+coalesced immediately.
 
 The watcher listens for filesystem events on `chat.db`, `chat.db-wal`,
 `chat.db-shm`, and the containing Messages directory, then backs that up with
diff --git a/Sources/IMsgCore/Message+URLPreview.swift b/Sources/IMsgCore/Message+URLPreview.swift
index b18fc4d9..a22a7357 100644
--- a/Sources/IMsgCore/Message+URLPreview.swift
+++ b/Sources/IMsgCore/Message+URLPreview.swift
@@ -1,6 +1,10 @@
 import Foundation
 
 extension Message {
+  var physicalCompletionRowID: Int64 {
+    max(rowID, urlPreview?.rowID ?? rowID)
+  }
+
   public struct URLPreviewMetadata: Sendable, Equatable {
     public let rowID: Int64
     public let guid: String
@@ -15,12 +19,14 @@ extension Message {
     }
   }
 
-  public func withURLPreview(_ preview: URLPreviewMetadata) -> Message {
+  public func withURLPreview(_ preview: URLPreviewMetadata, text replacementText: String? = nil)
+    -> Message
+  {
     Message(
       rowID: rowID,
       chatID: chatID,
       sender: sender,
-      text: text,
+      text: replacementText ?? text,
       date: date,
       isFromMe: isFromMe,
       service: service,
@@ -37,6 +43,42 @@ extension Message {
       ),
       balloonBundleID: balloonBundleID,
       urlPreview: preview,
+      cursorRowID: cursorRowID,
+      reaction: ReactionMetadata(
+        isReaction: isReaction,
+        reactionType: reactionType,
+        isReactionAdd: isReactionAdd,
+        reactedToGUID: reactedToGUID
+      ),
+      poll: poll,
+      isRead: isRead,
+      dateRead: dateRead
+    )
+  }
+
+  func withCursorRowID(_ cursorRowID: Int64) -> Message {
+    Message(
+      rowID: rowID,
+      chatID: chatID,
+      sender: sender,
+      text: text,
+      date: date,
+      isFromMe: isFromMe,
+      service: service,
+      handleID: handleID,
+      attachmentsCount: attachmentsCount,
+      guid: guid,
+      routing: RoutingMetadata(
+        replyToGUID: replyToGUID,
+        threadOriginatorGUID: threadOriginatorGUID,
+        threadOriginatorPart: threadOriginatorPart,
+        destinationCallerID: destinationCallerID,
+        replyToText: replyToText,
+        replyToSender: replyToSender
+      ),
+      balloonBundleID: balloonBundleID,
+      urlPreview: urlPreview,
+      cursorRowID: cursorRowID,
       reaction: ReactionMetadata(
         isReaction: isReaction,
         reactionType: reactionType,
diff --git a/Sources/IMsgCore/MessageStatsQueries.swift b/Sources/IMsgCore/MessageStatsQueries.swift
index 7d6d95cf..e2a7795d 100644
--- a/Sources/IMsgCore/MessageStatsQueries.swift
+++ b/Sources/IMsgCore/MessageStatsQueries.swift
@@ -110,6 +110,18 @@ struct StatsPreviewQuery {
       ? "previous.balloon_bundle_id" : "NULL"
     let previewBody = schema.hasAttributedBody ? "preview.attributedBody" : "NULL"
     let previousBody = schema.hasAttributedBody ? "previous.attributedBody" : "NULL"
+    let previewGUID = schema.hasGUIDColumn ? "preview.guid" : "NULL"
+    let previousGUID = schema.hasGUIDColumn ? "previous.guid" : "NULL"
+    let previewReplyToGUID = schema.hasReplyToGUIDColumn ? "preview.reply_to_guid" : "NULL"
+    let previousReplyToGUID = schema.hasReplyToGUIDColumn ? "previous.reply_to_guid" : "NULL"
+    let previewAssociatedGUID =
+      schema.hasReactionColumns ? "preview.associated_message_guid" : "NULL"
+    let previousAssociatedGUID =
+      schema.hasReactionColumns ? "previous.associated_message_guid" : "NULL"
+    let previewAssociatedType =
+      schema.hasReactionColumns ? "preview.associated_message_type" : "NULL"
+    let previousAssociatedType =
+      schema.hasReactionColumns ? "previous.associated_message_type" : "NULL"
     self.sql = """
       WITH base AS (
         SELECT cmj.chat_id, m.ROWID AS message_id
@@ -124,6 +136,10 @@ struct StatsPreviewQuery {
       )
       SELECT s.chat_id AS chat_id,
              preview.ROWID AS preview_rowid,
+             IFNULL(\(previewGUID), '') AS preview_guid,
+             IFNULL(\(previewReplyToGUID), '') AS preview_reply_to_guid,
+             IFNULL(\(previewAssociatedGUID), '') AS preview_associated_guid,
+             \(previewAssociatedType) AS preview_associated_type,
              preview.handle_id AS preview_handle_id,
              IFNULL(preview_handle.id, '') AS preview_sender,
              IFNULL(preview.text, '') AS preview_text,
@@ -134,6 +150,10 @@ struct StatsPreviewQuery {
              IFNULL(\(previewDestination), '') AS preview_destination_caller_id,
              IFNULL(\(previewBalloon), '') AS preview_balloon_bundle_id,
              previous.ROWID AS previous_rowid,
+             IFNULL(\(previousGUID), '') AS previous_guid,
+             IFNULL(\(previousReplyToGUID), '') AS previous_reply_to_guid,
+             IFNULL(\(previousAssociatedGUID), '') AS previous_associated_guid,
+             \(previousAssociatedType) AS previous_associated_type,
              previous.handle_id AS previous_handle_id,
              IFNULL(previous_handle.id, '') AS previous_sender,
              IFNULL(previous.text, '') AS previous_text,
diff --git a/Sources/IMsgCore/MessageStore+Chats.swift b/Sources/IMsgCore/MessageStore+Chats.swift
index d80a0c69..0f2a49c0 100644
--- a/Sources/IMsgCore/MessageStore+Chats.swift
+++ b/Sources/IMsgCore/MessageStore+Chats.swift
@@ -263,6 +263,9 @@ extension MessageStore {
           handleID: decoded.handleID,
           attachmentsCount: decoded.attachments,
           guid: decoded.guid,
+          routing: Message.RoutingMetadata(
+            replyToGUID: routedReplyToGUID(decoded)
+          ),
           balloonBundleID: decoded.balloonBundleID.nilIfEmpty,
           reaction: Message.ReactionMetadata(
             isReaction: reaction.isReaction,
diff --git a/Sources/IMsgCore/MessageStore+MessageConstruction.swift b/Sources/IMsgCore/MessageStore+MessageConstruction.swift
index 78121641..89b1362f 100644
--- a/Sources/IMsgCore/MessageStore+MessageConstruction.swift
+++ b/Sources/IMsgCore/MessageStore+MessageConstruction.swift
@@ -113,4 +113,84 @@ extension MessageStore {
     }
     return nil
   }
+
+  func linkedURLPreviewLookahead(
+    afterRowID: Int64,
+    candidates: [Message],
+    db: Connection
+  ) throws -> [Message] {
+    guard schema.hasBalloonBundleIDColumn else { return [] }
+
+    let textCandidates = candidates.filter { !isURLPreviewBalloon($0) }
+    guard !textCandidates.isEmpty else { return [] }
+
+    let candidateChatIDsByRowID = Dictionary(grouping: textCandidates, by: \.rowID)
+      .mapValues { Set($0.map(\.chatID)) }
+    let selection = MessageRowSelection(store: self, includeChatID: true)
+    var previews: [Message] = []
+    var seenPreviewChatIDsByRowID: [Int64: Set<Int64>] = [:]
+    var parentCache: ReplyParentCache = [:]
+    var pollOptionCache = PollOptionTextCache()
+    let maximumDateWindowsPerQuery = 200
+
+    for startIndex in stride(
+      from: 0,
+      to: textCandidates.count,
+      by: maximumDateWindowsPerQuery
+    ) {
+      let endIndex = min(startIndex + maximumDateWindowsPerQuery, textCandidates.count)
+      let chunk = textCandidates[startIndex..<endIndex]
+      let datePredicate = Array(
+        repeating: "(m.date >= ? AND m.date <= ?)",
+        count: chunk.count
+      ).joined(separator: " OR ")
+      let sql = """
+        SELECT \(selection.selectList)
+        FROM message m
+        LEFT JOIN chat_message_join cmj ON m.ROWID = cmj.message_id
+        LEFT JOIN handle h ON m.handle_id = h.ROWID
+        WHERE m.ROWID > ?
+          AND m.balloon_bundle_id = ?
+          AND (\(datePredicate))
+        ORDER BY m.ROWID ASC
+        """
+      var bindings: [Binding?] = [
+        afterRowID,
+        MessageStore.urlPreviewBalloonBundleID,
+      ]
+      for candidate in chunk {
+        bindings.append(MessageStore.appleEpoch(candidate.date))
+        bindings.append(
+          MessageStore.appleEpoch(
+            candidate.date.addingTimeInterval(MessageStore.urlPreviewCoalescingWindow)
+          ))
+      }
+
+      let rows = try db.prepareRowIterator(sql, bindings: bindings)
+      while let row = try rows.failableNext() {
+        let decoded = try decodeMessageRow(
+          row,
+          columns: selection.columns,
+          fallbackChatID: nil
+        )
+        let preview = try message(
+          from: decoded,
+          db,
+          parentCache: &parentCache,
+          pollOptionCache: &pollOptionCache
+        )
+        guard
+          let preceding = try precedingTextMessageForURLPreview(preview, db: db),
+          candidateChatIDsByRowID[preceding.rowID]?.contains(preview.chatID) == true
+        else {
+          continue
+        }
+        guard
+          seenPreviewChatIDsByRowID[decoded.rowID, default: []].insert(preview.chatID).inserted
+        else { continue }
+        previews.append(preview)
+      }
+    }
+    return previews
+  }
 }
diff --git a/Sources/IMsgCore/MessageStore+MessageRows.swift b/Sources/IMsgCore/MessageStore+MessageRows.swift
index ea7de9e3..859899d6 100644
--- a/Sources/IMsgCore/MessageStore+MessageRows.swift
+++ b/Sources/IMsgCore/MessageStore+MessageRows.swift
@@ -99,7 +99,7 @@ struct MessageRowSelection {
     let columns = MessageRowColumns.message(chatID: includeChatID ? "chat_id" : nil)
     let schema = store.schema
     let bodyColumn = schema.hasAttributedBody ? "m.attributedBody" : "NULL"
-    let guidColumn = schema.hasReactionColumns ? "m.guid" : "NULL"
+    let guidColumn = schema.hasGUIDColumn ? "m.guid" : "NULL"
     let associatedGuidColumn = schema.hasReactionColumns ? "m.associated_message_guid" : "NULL"
     let associatedTypeColumn = schema.hasReactionColumns ? "m.associated_message_type" : "NULL"
     let destinationCallerColumn =
diff --git a/Sources/IMsgCore/MessageStore+Messages.swift b/Sources/IMsgCore/MessageStore+Messages.swift
index 7bcc17ef..143cee69 100644
--- a/Sources/IMsgCore/MessageStore+Messages.swift
+++ b/Sources/IMsgCore/MessageStore+Messages.swift
@@ -9,6 +9,32 @@ extension MessageStore {
     }
   }
 
+  func maxRowID(chatID: Int64?, includeReactions: Bool) throws -> Int64 {
+    return try withConnection { db in
+      let reactionFilter =
+        includeReactions || !schema.hasReactionColumns
+        ? ""
+        : " AND (m.associated_message_type IS NULL OR m.associated_message_type < 2000 OR m.associated_message_type > 3006)"
+      let value: Binding?
+      if let chatID {
+        value = try db.scalar(
+          """
+          SELECT MAX(m.ROWID)
+          FROM message m
+          JOIN chat_message_join cmj ON m.ROWID = cmj.message_id
+          WHERE cmj.chat_id = ?\(reactionFilter)
+          """,
+          chatID
+        )
+      } else {
+        value = try db.scalar(
+          "SELECT MAX(m.ROWID) FROM message m WHERE 1 = 1\(reactionFilter)"
+        )
+      }
+      return int64Value(value) ?? 0
+    }
+  }
+
   public func messages(chatID: Int64, limit: Int) throws -> [Message] {
     return try messages(chatID: chatID, limit: limit, filter: nil)
   }
@@ -112,7 +138,7 @@ extension MessageStore {
         includeReactions: includeReactions
       )
       if !batch.messages.isEmpty {
-        return batch.messages
+        return Array(batch.messages.prefix(limit))
       }
       guard batch.maxScannedRowID > cursor else {
         return []
@@ -125,7 +151,9 @@ extension MessageStore {
     afterRowID: Int64,
     chatID: Int64?,
     limit: Int,
-    includeReactions: Bool
+    includeReactions: Bool,
+    suppressLateURLPreviews: Bool = true,
+    deduplicateURLBalloons: Bool = true
   ) throws -> MessagesAfterBatch {
     let query = MessagesAfterQuery(
       store: self,
@@ -157,6 +185,37 @@ extension MessageStore {
             pollOptionCache: &pollOptionCache
           ))
       }
+      let previewLookahead = try linkedURLPreviewLookahead(
+        afterRowID: maxScannedRowID,
+        candidates: messages,
+        db: db
+      )
+      if let lookaheadMaxRowID = previewLookahead.map(\.rowID).max() {
+        let gapQuery = MessagesAfterQuery(
+          store: self,
+          afterRowID: MessageID(rawValue: maxScannedRowID),
+          chatID: chatID.map { ChatID(rawValue: $0) },
+          limit: Int.max,
+          includeReactions: includeReactions,
+          throughRowID: MessageID(rawValue: lookaheadMaxRowID)
+        )
+        let gapRows = try db.prepareRowIterator(gapQuery.sql, bindings: gapQuery.bindings)
+        while let row = try gapRows.failableNext() {
+          let decoded = try decodeMessageRow(
+            row,
+            columns: gapQuery.selection.columns,
+            fallbackChatID: gapQuery.fallbackChatID
+          )
+          maxScannedRowID = max(maxScannedRowID, decoded.rowID)
+          messages.append(
+            try message(
+              from: decoded,
+              db,
+              parentCache: &parentCache,
+              pollOptionCache: &pollOptionCache
+            ))
+        }
+      }
       let coalesced = try coalesceURLPreviewMessages(
         messages,
         validateExistingCoalescence: { text, preview in
@@ -166,10 +225,11 @@ extension MessageStore {
           guard try self.precedingTextMessageForURLPreview(preview, db: db) != nil else {
             return nil
           }
-          return .suppress
+          return suppressLateURLPreviews ? .suppress : nil
         }
       )
       let visibleMessages = coalesced.filter { message in
+        guard deduplicateURLBalloons else { return true }
         guard isURLPreviewBalloon(message) else { return true }
         return !shouldSkipURLBalloonDuplicate(
           chatID: message.chatID,
@@ -396,16 +456,28 @@ extension MessageStore {
   }
 
   func routedReplyToGUID(_ row: DecodedMessageRow) -> String? {
-    if let associatedType = row.associatedType, ReactionType.isReaction(associatedType) {
+    routedReplyToGUID(
+      databaseReplyToGUID: row.databaseReplyToGUID,
+      associatedGUID: row.associatedGUID,
+      associatedType: row.associatedType
+    )
+  }
+
+  func routedReplyToGUID(
+    databaseReplyToGUID: String,
+    associatedGUID: String,
+    associatedType: Int?
+  ) -> String? {
+    if let associatedType, ReactionType.isReaction(associatedType) {
       return nil
     }
-    let databaseReplyToGUID = normalizeAssociatedGUID(row.databaseReplyToGUID)
-    if !databaseReplyToGUID.isEmpty {
-      return databaseReplyToGUID
+    let normalizedDatabaseGUID = normalizeAssociatedGUID(databaseReplyToGUID)
+    if !normalizedDatabaseGUID.isEmpty {
+      return normalizedDatabaseGUID
     }
     return replyToGUID(
-      associatedGuid: row.associatedGUID,
-      associatedType: row.associatedType
+      associatedGuid: associatedGUID,
+      associatedType: associatedType
     )
   }
 }
diff --git a/Sources/IMsgCore/MessageStore+Queries.swift b/Sources/IMsgCore/MessageStore+Queries.swift
index a244d685..34036804 100644
--- a/Sources/IMsgCore/MessageStore+Queries.swift
+++ b/Sources/IMsgCore/MessageStore+Queries.swift
@@ -64,7 +64,8 @@ struct MessagesAfterQuery {
     afterRowID: MessageID,
     chatID: ChatID?,
     limit: Int,
-    includeReactions: Bool
+    includeReactions: Bool,
+    throughRowID: MessageID? = nil
   ) {
     self.selection = MessageRowSelection(store: store, includeChatID: true)
     let reactionFilter: String
@@ -82,6 +83,10 @@ struct MessagesAfterQuery {
       WHERE m.ROWID > ?\(reactionFilter)
       """
     var bindings: [Binding?] = [afterRowID.rawValue]
+    if let throughRowID {
+      sql += " AND m.ROWID <= ?"
+      bindings.append(throughRowID.rawValue)
+    }
     if let chatID {
       sql += " AND cmj.chat_id = ?"
       bindings.append(chatID.rawValue)
diff --git a/Sources/IMsgCore/MessageStore+Search.swift b/Sources/IMsgCore/MessageStore+Search.swift
index ac4c5b34..84f45d31 100644
--- a/Sources/IMsgCore/MessageStore+Search.swift
+++ b/Sources/IMsgCore/MessageStore+Search.swift
@@ -76,6 +76,20 @@ extension MessageStore {
               pollOptionCache: &pollOptionCache
             ))
         }
+        let queriedMessageCount = messages.count
+        if let firstTextRowID = messages.filter({ !isURLPreviewBalloon($0) }).map(\.rowID).min() {
+          let existingChatIDsByRowID = Dictionary(grouping: messages, by: \.rowID)
+            .mapValues { Set($0.map(\.chatID)) }
+          let linkedPreviews = try linkedURLPreviewLookahead(
+            afterRowID: firstTextRowID,
+            candidates: messages,
+            db: db
+          )
+          messages.append(
+            contentsOf: linkedPreviews.filter {
+              existingChatIDsByRowID[$0.rowID]?.contains($0.chatID) != true
+            })
+        }
         var usedFallbackReplacement = false
         let coalesced = try coalesceURLPreviewMessages(
           messages,
@@ -86,7 +100,15 @@ extension MessageStore {
             guard let previous = try self.precedingTextMessageForURLPreview(preview, db: db) else {
               return nil
             }
-            guard self.searchMessage(previous, matches: trimmed, exact: exact) else {
+            guard
+              self.searchMessage(previous, matches: trimmed, exact: exact)
+                || (self.searchMessage(preview, matches: trimmed, exact: exact)
+                  && (!exact
+                    || self.previewMessageTargetsTextMessage(
+                      preview,
+                      textMessage: previous
+                    )))
+            else {
               return nil
             }
             return .replace(previous)
@@ -96,7 +118,8 @@ extension MessageStore {
           }
         ).sorted(by: searchMessagesNewestFirst)
 
-        if messages.count < physicalLimit || (coalesced.count >= limit && !usedFallbackReplacement)
+        if queriedMessageCount < physicalLimit
+          || (coalesced.count >= limit && !usedFallbackReplacement)
         {
           return Array(coalesced.prefix(limit))
         }
diff --git a/Sources/IMsgCore/MessageStore+Stats.swift b/Sources/IMsgCore/MessageStore+Stats.swift
index 39920f44..a06b7c9e 100644
--- a/Sources/IMsgCore/MessageStore+Stats.swift
+++ b/Sources/IMsgCore/MessageStore+Stats.swift
@@ -330,6 +330,11 @@ extension MessageStore {
       ? TypedStreamParser.parseAttributedBody(try dataValue(row, "\(prefix)_body")) : rawText
     let rawSender = try stringValue(row, "\(prefix)_sender")
     let destination = try stringValue(row, "\(prefix)_destination_caller_id")
+    let replyToGUID = routedReplyToGUID(
+      databaseReplyToGUID: try stringValue(row, "\(prefix)_reply_to_guid"),
+      associatedGUID: try stringValue(row, "\(prefix)_associated_guid"),
+      associatedType: try intValue(row, "\(prefix)_associated_type")
+    )
     return Message(
       rowID: rowID,
       chatID: chatID,
@@ -340,6 +345,8 @@ extension MessageStore {
       service: try stringValue(row, "\(prefix)_service").nilIfEmpty ?? "unknown",
       handleID: try int64Value(row, "\(prefix)_handle_id"),
       attachmentsCount: 0,
+      guid: try stringValue(row, "\(prefix)_guid"),
+      routing: Message.RoutingMetadata(replyToGUID: replyToGUID),
       balloonBundleID: try stringValue(row, "\(prefix)_balloon_bundle_id").nilIfEmpty
     )
   }
diff --git a/Sources/IMsgCore/MessageStore+URLPreviews.swift b/Sources/IMsgCore/MessageStore+URLPreviews.swift
index c51df019..282ec50c 100644
--- a/Sources/IMsgCore/MessageStore+URLPreviews.swift
+++ b/Sources/IMsgCore/MessageStore+URLPreviews.swift
@@ -7,7 +7,7 @@ enum URLPreviewCoalescingFallback {
 
 extension MessageStore {
   static let urlPreviewBalloonBundleID = "com.apple.messages.URLBalloonProvider"
-  private static let urlPreviewCoalescingWindow: TimeInterval = 5
+  static let urlPreviewCoalescingWindow: TimeInterval = 5
 
   func coalesceURLPreviewMessages(
     _ messages: [Message],
@@ -41,7 +41,11 @@ extension MessageStore {
           && canCoalesceURLPreview(textMessage: textMessage, previewMessage: preview.element)
         {
           replacements[candidate.offset] = textMessage.withURLPreview(
-            urlPreviewMetadata(from: preview.element)
+            urlPreviewMetadata(from: preview.element),
+            text: coalescedURLPreviewText(
+              textMessage: textMessage,
+              previewMessage: preview.element
+            )
           )
           suppressed.insert(preview.offset)
           continue
@@ -57,7 +61,11 @@ extension MessageStore {
       case .replace(let textMessage):
         fallbackReplacementUsed?()
         replacements[preview.offset] = textMessage.withURLPreview(
-          urlPreviewMetadata(from: preview.element)
+          urlPreviewMetadata(from: preview.element),
+          text: coalescedURLPreviewText(
+            textMessage: textMessage,
+            previewMessage: preview.element
+          )
         )
       }
     }
@@ -86,6 +94,9 @@ extension MessageStore {
     guard delta >= 0 && delta <= MessageStore.urlPreviewCoalescingWindow else {
       return false
     }
+    if previewMessageTargetsTextMessage(previewMessage, textMessage: textMessage) {
+      return true
+    }
     return textMessageContainsPreviewURL(
       textMessage.text,
       previewText: previewMessage.text
@@ -126,6 +137,29 @@ extension MessageStore {
     }
   }
 
+  private func coalescedURLPreviewText(
+    textMessage: Message,
+    previewMessage: Message
+  ) -> String {
+    let previewText = previewMessage.text.trimmingCharacters(in: .whitespacesAndNewlines)
+    guard isLikelyURLPreviewText(previewText) else { return textMessage.text }
+    guard !textMessageContainsPreviewURL(textMessage.text, previewText: previewText) else {
+      return textMessage.text
+    }
+    guard !textMessage.text.isEmpty else { return previewText }
+    return "\(textMessage.text)\n\(previewText)"
+  }
+
+  func previewMessageTargetsTextMessage(
+    _ previewMessage: Message,
+    textMessage: Message
+  ) -> Bool {
+    guard let replyToGUID = previewMessage.replyToGUID else { return false }
+    let previewTarget = normalizeAssociatedGUID(replyToGUID)
+    let textGUID = normalizeAssociatedGUID(textMessage.guid)
+    return !previewTarget.isEmpty && !textGUID.isEmpty && previewTarget == textGUID
+  }
+
   private func isLikelyURLPreviewText(_ text: String) -> Bool {
     let lowercased = text.lowercased()
     return lowercased.hasPrefix("http://")
diff --git a/Sources/IMsgCore/MessageStoreSchema.swift b/Sources/IMsgCore/MessageStoreSchema.swift
index 22853769..27b88670 100644
--- a/Sources/IMsgCore/MessageStoreSchema.swift
+++ b/Sources/IMsgCore/MessageStoreSchema.swift
@@ -6,6 +6,7 @@ struct MessageStoreSchema: Sendable {
   let hasThreadOriginatorGUIDColumn: Bool
   let hasThreadOriginatorPartColumn: Bool
   let hasDestinationCallerID: Bool
+  let hasGUIDColumn: Bool
   let hasAudioMessageColumn: Bool
   let hasAttachmentUserInfo: Bool
   let hasBalloonBundleIDColumn: Bool
@@ -35,6 +36,7 @@ struct MessageStoreSchema: Sendable {
     self.hasThreadOriginatorGUIDColumn = messageColumns.contains("thread_originator_guid")
     self.hasThreadOriginatorPartColumn = messageColumns.contains("thread_originator_part")
     self.hasDestinationCallerID = messageColumns.contains("destination_caller_id")
+    self.hasGUIDColumn = messageColumns.contains("guid")
     self.hasAudioMessageColumn = messageColumns.contains("is_audio_message")
     self.hasAttachmentUserInfo = attachmentColumns.contains("user_info")
     self.hasBalloonBundleIDColumn = messageColumns.contains("balloon_bundle_id")
@@ -78,6 +80,7 @@ struct MessageStoreSchema: Sendable {
     self.hasThreadOriginatorPartColumn =
       hasThreadOriginatorPartColumn ?? base.hasThreadOriginatorPartColumn
     self.hasDestinationCallerID = hasDestinationCallerID ?? base.hasDestinationCallerID
+    self.hasGUIDColumn = base.hasGUIDColumn
     self.hasAudioMessageColumn = hasAudioMessageColumn ?? base.hasAudioMessageColumn
     self.hasAttachmentUserInfo = hasAttachmentUserInfo ?? base.hasAttachmentUserInfo
     self.hasBalloonBundleIDColumn = hasBalloonBundleIDColumn ?? base.hasBalloonBundleIDColumn
diff --git a/Sources/IMsgCore/MessageWatcher.swift b/Sources/IMsgCore/MessageWatcher.swift
index 79b509dc..a68dc374 100644
--- a/Sources/IMsgCore/MessageWatcher.swift
+++ b/Sources/IMsgCore/MessageWatcher.swift
@@ -7,6 +7,9 @@ import Foundation
 public struct MessageWatcherConfiguration: Sendable, Equatable {
   public var debounceInterval: TimeInterval
   public var fallbackPollInterval: TimeInterval?
+  /// Holds newly observed rows briefly because Messages can commit a linked
+  /// URL-preview row after its text row in a separate filesystem event.
+  public var urlPreviewSettleInterval: TimeInterval
   public var batchLimit: Int
   /// When true, reaction events (tapback add/remove) are included in the stream
   public var includeReactions: Bool
@@ -14,11 +17,13 @@ public struct MessageWatcherConfiguration: Sendable, Equatable {
   public init(
     debounceInterval: TimeInterval = 0.25,
     fallbackPollInterval: TimeInterval? = 5,
+    urlPreviewSettleInterval: TimeInterval = 2,
     batchLimit: Int = 100,
     includeReactions: Bool = false
   ) {
     self.debounceInterval = debounceInterval
     self.fallbackPollInterval = fallbackPollInterval
+    self.urlPreviewSettleInterval = urlPreviewSettleInterval
     self.batchLimit = batchLimit
     self.includeReactions = includeReactions
   }
@@ -55,12 +60,22 @@ public final class MessageWatcher: @unchecked Sendable {
 private final class WatchState: @unchecked Sendable {
   private static let unresolvedChatRetryLimit = 20
 
-  private enum MessageYieldDecision {
+  fileprivate enum MessageYieldDecision {
     case yield
     case retry
     case skip
   }
 
+  private struct URLPreviewSettleCohort {
+    let throughRowID: Int64
+    let deadline: Date
+  }
+
+  fileprivate struct URLPreviewDeliveryKey: Hashable {
+    let rowID: Int64
+    let chatID: Int64
+  }
+
   private let store: MessageStore
   private let chatID: Int64?
   private let configuration: MessageWatcherConfiguration
@@ -68,6 +83,7 @@ private final class WatchState: @unchecked Sendable {
   private let queue = DispatchQueue(label: "imsg.watch", qos: .userInitiated)
 
   private var cursor: Int64
+  private var startupTailRowID: Int64 = 0
   #if os(macOS)
     private struct FileWatchIdentity: Equatable {
       let device: UInt64
@@ -85,6 +101,9 @@ private final class WatchState: @unchecked Sendable {
   private var pending = false
   private var stopped = false
   private var unresolvedChatAttempts: [Int64: Int] = [:]
+  private var terminallySkippedRowIDs = Set<Int64>()
+  private var urlPreviewSettleCohorts: [URLPreviewSettleCohort] = []
+  private var deliveredURLPreviews = Set<URLPreviewDeliveryKey>()
 
   init(
     store: MessageStore,
@@ -101,20 +120,25 @@ private final class WatchState: @unchecked Sendable {
   }
 
   func start() {
-    queue.async {
-      do {
+    do {
+      // Capture the tail before returning the stream. Otherwise a message can
+      // arrive between subscription and asynchronous startup, be mistaken for
+      // pre-existing history, and bypass the URL-preview settle window.
+      let tailRowID = try store.maxRowID()
+      queue.async {
         if self.cursor == 0 {
-          self.cursor = try self.store.maxRowID()
+          self.cursor = tailRowID
         }
+        self.startupTailRowID = tailRowID
         #if os(macOS)
           self.refreshFileSources()
           self.installDirectorySource()
         #endif
         self.poll()
         self.scheduleFallbackPoll()
-      } catch {
-        self.continuation.finish(throwing: error)
       }
+    } catch {
+      continuation.finish(throwing: error)
     }
   }
 
@@ -236,35 +260,181 @@ private final class WatchState: @unchecked Sendable {
   private func poll() {
     if stopped { return }
     do {
+      let queryLimit =
+        configuration.batchLimit == Int.max
+        ? Int.max : configuration.batchLimit + 1
       let batch = try store.messagesAfterBatch(
         afterRowID: cursor,
         chatID: chatID,
-        limit: configuration.batchLimit,
+        limit: queryLimit,
+        includeReactions: configuration.includeReactions,
+        suppressLateURLPreviews: false,
+        deduplicateURLBalloons: false
+      )
+      let observedTailRowID = try store.maxRowID(
+        chatID: chatID,
         includeReactions: configuration.includeReactions
       )
-      for message in batch.messages {
+      registerURLPreviewSettleCohort(
+        throughRowID: max(observedTailRowID, batch.maxScannedRowID),
+        observedAt: Date()
+      )
+      let settlingTextRowID = urlPreviewSettlingBoundary(messages: batch.messages)
+      let firstHeldRowID = settlingTextRowID.flatMap { boundary in
+        batch.messages
+          .filter { $0.physicalCompletionRowID >= boundary }
+          .map(\.rowID)
+          .min()
+      }
+      let messagesToDeliver = batch.messages
+        .filter { message in
+          guard let settlingTextRowID else { return true }
+          guard message.physicalCompletionRowID < settlingTextRowID else { return false }
+          guard let firstHeldRowID else { return false }
+          return message.rowID < firstHeldRowID
+        }
+        .sorted { lhs, rhs in
+          if lhs.physicalCompletionRowID == rhs.physicalCompletionRowID {
+            return lhs.rowID < rhs.rowID
+          }
+          return lhs.physicalCompletionRowID < rhs.physicalCompletionRowID
+        }
+      var deliverableMessages: [Message] = []
+      var pendingURLPreviews = deliveredURLPreviews
+      for message in messagesToDeliver {
         switch yieldDecision(for: message) {
         case .yield:
           break
         case .retry:
+          // Abort before any yield or cursor update so this unresolved row
+          // remains inside both the internal and published frontiers.
           return
         case .skip:
           continue
         }
-        continuation.yield(message)
-        if message.rowID > cursor {
-          cursor = message.rowID
+        let urlPreviewKey = urlPreviewDeliveryKey(for: message)
+        if let urlPreviewKey, deliveredURLPreviews.contains(urlPreviewKey) {
+          continue
+        }
+        if store.isURLPreviewBalloon(message),
+          store.shouldSkipURLBalloonDuplicate(
+            chatID: message.chatID,
+            sender: message.sender,
+            text: message.text,
+            isFromMe: message.isFromMe,
+            date: message.date,
+            rowID: message.rowID
+          )
+        {
+          continue
         }
+        if let urlPreviewKey, !pendingURLPreviews.insert(urlPreviewKey).inserted {
+          continue
+        }
+        deliverableMessages.append(message)
       }
-      if batch.maxScannedRowID > cursor {
+
+      var eventCursors = Array(repeating: cursor, count: deliverableMessages.count)
+      var nextUndeliveredRowID = firstHeldRowID
+      for index in deliverableMessages.indices.reversed() {
+        if let nextUndeliveredRowID {
+          eventCursors[index] = max(cursor, nextUndeliveredRowID - 1)
+        } else {
+          eventCursors[index] = max(cursor, batch.maxScannedRowID)
+        }
+        let rowID = deliverableMessages[index].rowID
+        nextUndeliveredRowID = min(nextUndeliveredRowID ?? rowID, rowID)
+      }
+
+      for (index, message) in deliverableMessages.enumerated() {
+        continuation.yield(message.withCursorRowID(eventCursors[index]))
+        if let urlPreviewKey = urlPreviewDeliveryKey(for: message) {
+          deliveredURLPreviews.insert(urlPreviewKey)
+        }
+      }
+      if let firstHeldRowID {
+        cursor = max(cursor, firstHeldRowID - 1)
+      } else if let settlingTextRowID {
+        cursor = max(cursor, settlingTextRowID - 1)
+      } else if batch.maxScannedRowID > cursor {
         cursor = batch.maxScannedRowID
       }
+      pruneURLPreviewSettleState()
     } catch {
       continuation.finish(throwing: error)
     }
   }
+}
 
-  private func yieldDecision(for message: Message) -> MessageYieldDecision {
+extension WatchState {
+  fileprivate func registerURLPreviewSettleCohort(throughRowID: Int64, observedAt: Date) {
+    let interval = configuration.urlPreviewSettleInterval
+    guard interval > 0 else { return }
+    let coveredThroughRowID = urlPreviewSettleCohorts.last?.throughRowID ?? startupTailRowID
+    guard throughRowID > max(startupTailRowID, coveredThroughRowID) else { return }
+    urlPreviewSettleCohorts.append(
+      URLPreviewSettleCohort(
+        throughRowID: throughRowID,
+        deadline: observedAt.addingTimeInterval(interval)
+      ))
+  }
+
+  fileprivate func urlPreviewSettlingBoundary(messages: [Message]) -> Int64? {
+    guard configuration.urlPreviewSettleInterval > 0 else {
+      return nil
+    }
+
+    // Reactions do not create their own settle gap. The poll's firstHeldRowID
+    // frontier still keeps every later event behind an earlier pending text.
+    let uncoalescedLiveTextRows = messages.filter {
+      $0.rowID > startupTailRowID && !$0.isReaction && !store.isURLPreviewBalloon($0)
+        && $0.urlPreview == nil
+    }
+
+    let now = Date()
+    let waitingRows = uncoalescedLiveTextRows.compactMap { message -> (Int64, Date)? in
+      guard
+        let deadline = urlPreviewSettleCohorts.first(where: {
+          message.rowID <= $0.throughRowID
+        })?.deadline,
+        now < deadline
+      else {
+        return nil
+      }
+      return (message.rowID, deadline)
+    }
+    guard let boundary = waitingRows.min(by: { $0.0 < $1.0 }) else { return nil }
+    if let nextDeadline = waitingRows.map(\.1).min() {
+      scheduleURLPreviewSettlePoll(at: nextDeadline)
+    }
+    return boundary.0
+  }
+
+  fileprivate func scheduleURLPreviewSettlePoll(at date: Date) {
+    let delay = max(0, date.timeIntervalSinceNow)
+    queue.asyncAfter(deadline: .now() + delay) { [weak self] in
+      guard let self, !self.stopped else { return }
+      self.poll()
+    }
+  }
+
+  fileprivate func pruneURLPreviewSettleState() {
+    urlPreviewSettleCohorts.removeAll { $0.throughRowID <= cursor }
+    deliveredURLPreviews = deliveredURLPreviews.filter { $0.rowID > cursor }
+    terminallySkippedRowIDs = terminallySkippedRowIDs.filter { $0 > cursor }
+  }
+
+  fileprivate func urlPreviewDeliveryKey(for message: Message) -> URLPreviewDeliveryKey? {
+    let rowID =
+      message.urlPreview?.rowID
+      ?? (store.isURLPreviewBalloon(message) ? message.rowID : nil)
+    return rowID.map { URLPreviewDeliveryKey(rowID: $0, chatID: message.chatID) }
+  }
+
+  fileprivate func yieldDecision(for message: Message) -> MessageYieldDecision {
+    if terminallySkippedRowIDs.contains(message.rowID) {
+      return .skip
+    }
     guard message.chatID <= 0 else {
       unresolvedChatAttempts.removeValue(forKey: message.rowID)
       return .yield
@@ -278,9 +448,7 @@ private final class WatchState: @unchecked Sendable {
     }
 
     unresolvedChatAttempts.removeValue(forKey: message.rowID)
-    if message.rowID > cursor {
-      cursor = message.rowID
-    }
+    terminallySkippedRowIDs.insert(message.rowID)
     return .skip
   }
 }
diff --git a/Sources/IMsgCore/Models.swift b/Sources/IMsgCore/Models.swift
index 5ed13636..831248ce 100644
--- a/Sources/IMsgCore/Models.swift
+++ b/Sources/IMsgCore/Models.swift
@@ -327,9 +327,12 @@ public struct Message: Sendable, Equatable {
   /// or a Polls vote update.
   public let poll: MessagePollEvent?
   /// Metadata for an Apple URL preview balloon row that was folded into this
-  /// message. The message itself still uses the originating text row's id,
-  /// guid, text, and timestamp.
+  /// message. The message keeps the originating row's id, guid, and timestamp;
+  /// its text also carries the preview URL when Apple omitted it from that row.
   public let urlPreview: URLPreviewMetadata?
+  /// Exclusive resume cursor for a live watch event. Batch read results leave
+  /// this nil because cursor safety depends on stream emission order.
+  public let cursorRowID: Int64?
 
   // Reaction metadata (populated when message is a reaction event)
   /// Whether this message is a reaction event (tapback add/remove)
@@ -359,6 +362,7 @@ public struct Message: Sendable, Equatable {
     routing: RoutingMetadata = RoutingMetadata(),
     balloonBundleID: String? = nil,
     urlPreview: URLPreviewMetadata? = nil,
+    cursorRowID: Int64? = nil,
     reaction: ReactionMetadata = ReactionMetadata(),
     poll: MessagePollEvent? = nil,
     isRead: Bool? = nil,
@@ -383,6 +387,7 @@ public struct Message: Sendable, Equatable {
     self.balloonBundleID = balloonBundleID
     self.poll = poll
     self.urlPreview = urlPreview
+    self.cursorRowID = cursorRowID
     self.isReaction = reaction.isReaction
     self.reactionType = reaction.reactionType
     self.isReactionAdd = reaction.isReactionAdd
@@ -408,6 +413,7 @@ public struct Message: Sendable, Equatable {
     destinationCallerID: String? = nil,
     balloonBundleID: String? = nil,
     urlPreview: URLPreviewMetadata? = nil,
+    cursorRowID: Int64? = nil,
     replyToText: String? = nil,
     replyToSender: String? = nil,
     isReaction: Bool = false,
@@ -439,6 +445,7 @@ public struct Message: Sendable, Equatable {
       ),
       balloonBundleID: balloonBundleID,
       urlPreview: urlPreview,
+      cursorRowID: cursorRowID,
       reaction: ReactionMetadata(
         isReaction: isReaction,
         reactionType: reactionType,
diff --git a/Sources/imsg/Commands/WatchCommand.swift b/Sources/imsg/Commands/WatchCommand.swift
index 0246bfe7..c48a4ea0 100644
--- a/Sources/imsg/Commands/WatchCommand.swift
+++ b/Sources/imsg/Commands/WatchCommand.swift
@@ -90,12 +90,15 @@ enum WatchCommand {
     let store = try storeFactory(dbPath)
     let watcher = MessageWatcher(store: store)
     let cache = ChatCache(store: store)
-    let contacts = await contactResolverFactory()
     let config = MessageWatcherConfiguration(
       debounceInterval: debounceInterval,
       batchLimit: 100,
       includeReactions: includeReactions
     )
+    // Subscribe before contact discovery so messages arriving during that
+    // asynchronous startup work are buffered instead of becoming history.
+    let stream = streamProvider(watcher, chatID, sinceRowID, config)
+    let contacts = await contactResolverFactory()
 
     let bbEvents = values.flag("bbEvents")
     if bbEvents {
@@ -119,7 +122,6 @@ enum WatchCommand {
       }
     }
 
-    let stream = streamProvider(watcher, chatID, sinceRowID, config)
     for try await message in stream {
       if !filter.allows(message) {
         continue
diff --git a/Sources/imsg/OutputModels.swift b/Sources/imsg/OutputModels.swift
index 3b45add7..f5f0649e 100644
--- a/Sources/imsg/OutputModels.swift
+++ b/Sources/imsg/OutputModels.swift
@@ -79,6 +79,7 @@ struct MessagePayload: Codable {
   }
 
   let id: Int64
+  let cursor: Int64?
   let chatID: Int64
   let guid: String
   let replyToGUID: String?
@@ -122,6 +123,7 @@ struct MessagePayload: Codable {
     reactionSenderNames: [Int64: String] = [:]
   ) {
     self.id = message.rowID
+    self.cursor = message.cursorRowID
     self.chatID = message.chatID
     self.guid = message.guid
     self.replyToGUID = message.replyToGUID
@@ -171,6 +173,7 @@ struct MessagePayload: Codable {
 
   enum CodingKeys: String, CodingKey {
     case id
+    case cursor
     case chatID = "chat_id"
     case guid
     case replyToGUID = "reply_to_guid"
diff --git a/Tests/IMsgCoreTests/MessageDatabaseFixture.swift b/Tests/IMsgCoreTests/MessageDatabaseFixture.swift
index 53bbba3f..031be7cd 100644
--- a/Tests/IMsgCoreTests/MessageDatabaseFixture.swift
+++ b/Tests/IMsgCoreTests/MessageDatabaseFixture.swift
@@ -9,6 +9,7 @@ enum MessageDatabaseFixture {
     var includeThreadOriginatorGUID = false
     var includeThreadOriginatorPart = false
     var includeDestinationCallerID = false
+    var includeGUID = false
     var includeAudioMessage = false
     var includeBalloonBundleID = false
     var includePayloadData = false
@@ -34,6 +35,8 @@ enum MessageDatabaseFixture {
       options.includeThreadOriginatorPart ? "thread_originator_part TEXT," : ""
     let destinationCallerColumn =
       options.includeDestinationCallerID ? "destination_caller_id TEXT," : ""
+    let guidColumn =
+      options.includeGUID && !options.includeReactionColumns ? "guid TEXT," : ""
     let audioMessageColumn = options.includeAudioMessage ? "is_audio_message INTEGER," : ""
     let balloonColumn = options.includeBalloonBundleID ? "balloon_bundle_id TEXT," : ""
     let payloadDataColumn = options.includePayloadData ? "payload_data BLOB," : ""
@@ -57,6 +60,7 @@ enum MessageDatabaseFixture {
         \(threadOriginatorColumn)
         \(threadOriginatorPartColumn)
         \(destinationCallerColumn)
+        \(guidColumn)
         \(audioMessageColumn)
         \(balloonColumn)
         \(payloadDataColumn)
diff --git a/Tests/IMsgCoreTests/MessageStoreSchemaDetectionTests.swift b/Tests/IMsgCoreTests/MessageStoreSchemaDetectionTests.swift
index 3b99a23b..d2a6e449 100644
--- a/Tests/IMsgCoreTests/MessageStoreSchemaDetectionTests.swift
+++ b/Tests/IMsgCoreTests/MessageStoreSchemaDetectionTests.swift
@@ -49,6 +49,7 @@ func schemaDetectsOptionalMessageColumns() throws {
   let store = try MessageStore(connection: db, path: ":memory:")
   #expect(store.schema.hasAttributedBody)
   #expect(store.schema.hasReactionColumns)
+  #expect(store.schema.hasGUIDColumn)
   #expect(store.schema.hasThreadOriginatorGUIDColumn)
   #expect(store.schema.hasThreadOriginatorPartColumn)
   #expect(store.schema.hasDestinationCallerID)
diff --git a/Tests/IMsgCoreTests/MessageStoreStatsTests.swift b/Tests/IMsgCoreTests/MessageStoreStatsTests.swift
index 42cac461..70424d13 100644
--- a/Tests/IMsgCoreTests/MessageStoreStatsTests.swift
+++ b/Tests/IMsgCoreTests/MessageStoreStatsTests.swift
@@ -224,6 +224,77 @@ func messageStatsCoalescesConsecutivePreviewsLikeHistory() throws {
   #expect(try store.messageStats(timeZoneIdentifier: "UTC").totalMessages == 1)
 }
 
+@Test
+func messageStatsCoalescesGUIDLinkedPreviewWhenTextOmitsURL() throws {
+  let db = try Connection(.inMemory)
+  try MessageDatabaseFixture.createSchema(
+    db,
+    options: MessageDatabaseFixture.SchemaOptions(
+      includeGUID: true,
+      includeBalloonBundleID: true,
+      includeReplyToGUID: true
+    )
+  )
+  let date = TestDatabase.appleEpoch(Date(timeIntervalSince1970: 1_735_691_400))
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+111')")
+  try db.run(
+    "INSERT INTO chat(ROWID, chat_identifier, guid, display_name, service_name) VALUES (1, '+111', 'iMessage;-;+111', 'Alpha', 'iMessage')"
+  )
+  try db.run(
+    """
+    INSERT INTO message(
+      ROWID, handle_id, text, guid, reply_to_guid, balloon_bundle_id,
+      date, is_from_me, service
+    )
+    VALUES
+      (1, 1, 'Check this out', 'text-guid', NULL, NULL, ?, 0, 'iMessage'),
+      (2, 1, 'https://example.com', 'preview-guid', 'p:0/text-guid', ?, ?, 0, 'iMessage')
+    """,
+    date,
+    MessageStore.urlPreviewBalloonBundleID,
+    date + 399_000_000
+  )
+  try db.run("INSERT INTO chat_message_join(chat_id, message_id) VALUES (1, 1), (1, 2)")
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  #expect(try store.messageStats(timeZoneIdentifier: "UTC").totalMessages == 1)
+}
+
+@Test
+func messageStatsCoalescesAssociatedGUIDLinkedPreviewWhenTextOmitsURL() throws {
+  let db = try Connection(.inMemory)
+  try MessageDatabaseFixture.createSchema(
+    db,
+    options: MessageDatabaseFixture.SchemaOptions(
+      includeReactionColumns: true,
+      includeBalloonBundleID: true
+    )
+  )
+  let date = TestDatabase.appleEpoch(Date(timeIntervalSince1970: 1_735_691_400))
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+111')")
+  try db.run(
+    "INSERT INTO chat(ROWID, chat_identifier, guid, display_name, service_name) VALUES (1, '+111', 'iMessage;-;+111', 'Alpha', 'iMessage')"
+  )
+  try db.run(
+    """
+    INSERT INTO message(
+      ROWID, handle_id, text, guid, associated_message_guid, associated_message_type,
+      balloon_bundle_id, date, is_from_me, service
+    )
+    VALUES
+      (1, 1, 'Check this out', 'text-guid', NULL, NULL, NULL, ?, 0, 'iMessage'),
+      (2, 1, 'https://example.com', 'preview-guid', 'p:0/text-guid', 0, ?, ?, 0, 'iMessage')
+    """,
+    date,
+    MessageStore.urlPreviewBalloonBundleID,
+    date + 399_000_000
+  )
+  try db.run("INSERT INTO chat_message_join(chat_id, message_id) VALUES (1, 1), (1, 2)")
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  #expect(try store.messageStats(timeZoneIdentifier: "UTC").totalMessages == 1)
+}
+
 @Test
 func messageStatsOnlyRequiresMediaTablesWhenRequested() throws {
   let db = try Connection(.inMemory)
diff --git a/Tests/IMsgCoreTests/MessageStoreURLPreviewRoutingTests.swift b/Tests/IMsgCoreTests/MessageStoreURLPreviewRoutingTests.swift
new file mode 100644
index 00000000..2aac28ef
--- /dev/null
+++ b/Tests/IMsgCoreTests/MessageStoreURLPreviewRoutingTests.swift
@@ -0,0 +1,362 @@
+import Foundation
+import SQLite
+import Testing
+
+@testable import IMsgCore
+
+@Test
+func messagesAfterCoalescesGUIDLinkedURLPreviewWhenTextOmitsURL() throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 1,
+    text: "Check this out",
+    guid: "text-guid",
+    date: now
+  )
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 2,
+    text: "https://example.com",
+    guid: "preview-guid",
+    replyToGUID: "p:0/text-guid",
+    balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+    date: now.addingTimeInterval(0.399)
+  )
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let messages = try store.messagesAfter(afterRowID: 0, chatID: 1, limit: 10)
+
+  #expect(messages.map(\.rowID) == [1])
+  #expect(messages.first?.text == "Check this out\nhttps://example.com")
+  #expect(messages.first?.urlPreview?.rowID == 2)
+}
+
+@Test
+func messagesAfterKeepsUnlinkedURLPreviewWhenTextOmitsURL() throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 1,
+    text: "Check this out",
+    guid: "text-guid",
+    date: now
+  )
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 2,
+    text: "https://example.com",
+    guid: "preview-guid",
+    balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+    date: now.addingTimeInterval(0.206)
+  )
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let messages = try store.messagesAfter(afterRowID: 0, chatID: 1, limit: 10)
+
+  #expect(messages.map(\.rowID) == [1, 2])
+  #expect(messages.allSatisfy { $0.urlPreview == nil })
+}
+
+@Test
+func messagesAfterCoalescesGUIDLinkedPreviewWithoutReactionColumns() throws {
+  let db = try Connection(.inMemory)
+  try MessageDatabaseFixture.createSchema(
+    db,
+    options: MessageDatabaseFixture.SchemaOptions(
+      includeGUID: true,
+      includeBalloonBundleID: true,
+      includeReplyToGUID: true
+    )
+  )
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+  try db.run(
+    """
+    INSERT INTO message(
+      ROWID, handle_id, text, guid, reply_to_guid, balloon_bundle_id,
+      date, is_from_me, service
+    )
+    VALUES
+      (1, 1, 'Check this out', 'text-guid', NULL, NULL, ?, 0, 'iMessage'),
+      (2, 1, 'https://example.com', 'preview-guid', 'text-guid', ?, ?, 0, 'iMessage')
+    """,
+    TestDatabase.appleEpoch(now),
+    MessageStore.urlPreviewBalloonBundleID,
+    TestDatabase.appleEpoch(now.addingTimeInterval(0.399))
+  )
+  try db.run("INSERT INTO chat_message_join(chat_id, message_id) VALUES (1, 1), (1, 2)")
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let messages = try store.messagesAfter(afterRowID: 0, chatID: 1, limit: 10)
+
+  #expect(messages.map(\.rowID) == [1])
+  #expect(messages.first?.urlPreview?.rowID == 2)
+}
+
+@Test
+func messagesAfterCoalescesAssociatedGUIDLinkedURLPreviewWhenTextOmitsURL() throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 1,
+    text: "Check this out",
+    guid: "text-guid",
+    date: now
+  )
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 2,
+    text: "https://example.com",
+    guid: "preview-guid",
+    associatedMessageGUID: "p:0/text-guid",
+    associatedMessageType: 0,
+    balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+    date: now.addingTimeInterval(0.399)
+  )
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let messages = try store.messagesAfter(afterRowID: 0, chatID: 1, limit: 10)
+
+  #expect(messages.map(\.rowID) == [1])
+  #expect(messages.first?.text == "Check this out\nhttps://example.com")
+  #expect(messages.first?.urlPreview?.rowID == 2)
+}
+
+@Test
+func searchMessagesByURLCoalescesGUIDLinkedPreviewWhenTextOmitsURL() throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 1,
+    text: "Check this out",
+    guid: "text-guid",
+    date: now
+  )
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 2,
+    text: "https://example.com",
+    guid: "preview-guid",
+    replyToGUID: "text-guid",
+    balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+    date: now.addingTimeInterval(1)
+  )
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let messages = try store.searchMessages(query: "example.com", match: "contains", limit: 10)
+
+  #expect(messages.map(\.rowID) == [1])
+  #expect(messages.first?.text == "Check this out\nhttps://example.com")
+  #expect(messages.first?.urlPreview?.rowID == 2)
+}
+
+@Test
+func searchMessagesByExactURLCoalescesGUIDLinkedPreviewWhenTextOmitsURL() throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 1,
+    text: "Check this out",
+    guid: "text-guid",
+    date: now
+  )
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 2,
+    text: "https://example.com",
+    guid: "preview-guid",
+    replyToGUID: "text-guid",
+    balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+    date: now.addingTimeInterval(1)
+  )
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let messages = try store.searchMessages(
+    query: "https://example.com",
+    match: "exact",
+    limit: 10
+  )
+
+  #expect(messages.map(\.rowID) == [1])
+  #expect(messages.first?.text == "Check this out\nhttps://example.com")
+  #expect(messages.first?.urlPreview?.rowID == 2)
+}
+
+@Test
+func searchMessagesByCaptionIncludesGUIDLinkedPreviewWhenTextOmitsURL() throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 1,
+    text: "Check this out",
+    guid: "text-guid",
+    date: now
+  )
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 2,
+    text: "https://example.com",
+    guid: "preview-guid",
+    replyToGUID: "text-guid",
+    balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+    date: now.addingTimeInterval(1)
+  )
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let messages = try store.searchMessages(query: "Check this", match: "contains", limit: 10)
+
+  #expect(messages.map(\.rowID) == [1])
+  #expect(messages.first?.text == "Check this out\nhttps://example.com")
+  #expect(messages.first?.urlPreview?.rowID == 2)
+}
+
+@Test
+func messagesAfterKeepsLimitWhenPreviewLookaheadCrossesInterleavedRows() throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 1,
+    text: "Check this out",
+    guid: "text-guid",
+    date: now
+  )
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 2,
+    chatID: 2,
+    text: "interleaved",
+    guid: "interleaved-guid",
+    date: now.addingTimeInterval(0.5)
+  )
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 3,
+    text: "https://example.com",
+    guid: "preview-guid",
+    replyToGUID: "text-guid",
+    balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+    date: now.addingTimeInterval(1)
+  )
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let messages = try store.messagesAfter(afterRowID: 0, chatID: 1, limit: 1)
+
+  #expect(messages.map(\.rowID) == [1])
+  #expect(messages.first?.urlPreview?.rowID == 3)
+}
+
+@Test
+func messagesAfterFindsEligibleAssociationForMultiChatPreview() throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("PRAGMA automatic_index = OFF")
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 1,
+    chatID: 1,
+    text: "Check this out",
+    guid: "text-guid",
+    date: now
+  )
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 2,
+    chatID: 2,
+    text: "https://example.com",
+    guid: "preview-guid",
+    replyToGUID: "text-guid",
+    balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+    date: now.addingTimeInterval(1)
+  )
+  try db.run("INSERT INTO chat_message_join(chat_id, message_id) VALUES (1, 2)")
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let messages = try store.messagesAfter(afterRowID: 0, chatID: 1, limit: 1)
+
+  #expect(messages.map(\.rowID) == [1])
+  #expect(messages.first?.urlPreview?.rowID == 2)
+}
+
+@Test
+func searchLookaheadKeepsCandidateChatAssociation() throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("PRAGMA automatic_index = OFF")
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 1,
+    chatID: 1,
+    text: "Check this out",
+    guid: "text-guid",
+    date: now
+  )
+  try db.run("INSERT INTO chat_message_join(chat_id, message_id) VALUES (2, 1)")
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 2,
+    chatID: 2,
+    text: "https://example.com",
+    guid: "preview-guid",
+    replyToGUID: "text-guid",
+    balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+    date: now.addingTimeInterval(1)
+  )
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let messages = try store.searchMessages(query: "Check this", match: "contains", limit: 1)
+
+  #expect(messages.map(\.rowID) == [1])
+  #expect(messages.first?.chatID == 1)
+  #expect(messages.first?.urlPreview == nil)
+}
+
+@Test
+func searchLookaheadCoalescesEveryMultiChatAssociation() throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 1,
+    chatID: 1,
+    text: "Check this out",
+    guid: "text-guid",
+    date: now
+  )
+  try db.run("INSERT INTO chat_message_join(chat_id, message_id) VALUES (2, 1)")
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 2,
+    chatID: 1,
+    text: "https://example.com",
+    guid: "preview-guid",
+    replyToGUID: "text-guid",
+    balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+    date: now.addingTimeInterval(1)
+  )
+  try db.run("INSERT INTO chat_message_join(chat_id, message_id) VALUES (2, 2)")
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let messages = try store.searchMessages(query: "Check this", match: "contains", limit: 10)
+
+  #expect(messages.count == 2)
+  #expect(Set(messages.map(\.chatID)) == Set([1, 2]))
+  #expect(messages.allSatisfy { $0.urlPreview?.rowID == 2 })
+}
diff --git a/Tests/IMsgCoreTests/MessageStoreURLPreviewTestSupport.swift b/Tests/IMsgCoreTests/MessageStoreURLPreviewTestSupport.swift
new file mode 100644
index 00000000..bb80eab8
--- /dev/null
+++ b/Tests/IMsgCoreTests/MessageStoreURLPreviewTestSupport.swift
@@ -0,0 +1,75 @@
+import Foundation
+import SQLite
+
+@testable import IMsgCore
+
+func makeURLPreviewTestDB() throws -> Connection {
+  let db = try Connection(.inMemory)
+  try db.execute(
+    """
+    CREATE TABLE message (
+      ROWID INTEGER PRIMARY KEY,
+      handle_id INTEGER,
+      text TEXT,
+      guid TEXT,
+      associated_message_guid TEXT,
+      associated_message_type INTEGER,
+      reply_to_guid TEXT,
+      balloon_bundle_id TEXT,
+      is_read INTEGER, date_read INTEGER,
+      date INTEGER,
+      is_from_me INTEGER,
+      service TEXT
+    );
+    """
+  )
+  try db.execute("CREATE TABLE handle (ROWID INTEGER PRIMARY KEY, id TEXT);")
+  try db.execute("CREATE TABLE chat_message_join (chat_id INTEGER, message_id INTEGER);")
+  try db.execute(
+    "CREATE TABLE message_attachment_join (message_id INTEGER, attachment_id INTEGER);")
+  return db
+}
+
+func insertURLPreviewTestMessage(
+  _ db: Connection,
+  rowID: Int64,
+  chatID: Int64 = 1,
+  handleID: Int64 = 1,
+  text: String,
+  guid: String,
+  associatedMessageGUID: String? = nil,
+  associatedMessageType: Int? = nil,
+  replyToGUID: String? = nil,
+  balloonBundleID: String? = nil,
+  date: Date,
+  isFromMe: Bool = false,
+  isRead: Bool = false,
+  dateRead: Date? = nil
+) throws {
+  try db.run(
+    """
+    INSERT INTO message(
+      ROWID, handle_id, text, guid, associated_message_guid, associated_message_type,
+      reply_to_guid, balloon_bundle_id, is_read, date_read, date, is_from_me, service
+    )
+    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'iMessage')
+    """,
+    rowID,
+    handleID,
+    text,
+    guid,
+    associatedMessageGUID,
+    associatedMessageType,
+    replyToGUID,
+    balloonBundleID,
+    isRead ? 1 : 0,
+    dateRead.map(TestDatabase.appleEpoch) ?? 0,
+    TestDatabase.appleEpoch(date),
+    isFromMe ? 1 : 0
+  )
+  try db.run(
+    "INSERT INTO chat_message_join(chat_id, message_id) VALUES (?, ?)",
+    chatID,
+    rowID
+  )
+}
diff --git a/Tests/IMsgCoreTests/MessageStoreURLPreviewTests.swift b/Tests/IMsgCoreTests/MessageStoreURLPreviewTests.swift
index 2d770dbd..fa4835f5 100644
--- a/Tests/IMsgCoreTests/MessageStoreURLPreviewTests.swift
+++ b/Tests/IMsgCoreTests/MessageStoreURLPreviewTests.swift
@@ -4,74 +4,6 @@ import Testing
 
 @testable import IMsgCore
 
-func makeURLPreviewTestDB() throws -> Connection {
-  let db = try Connection(.inMemory)
-  try db.execute(
-    """
-    CREATE TABLE message (
-      ROWID INTEGER PRIMARY KEY,
-      handle_id INTEGER,
-      text TEXT,
-      guid TEXT,
-      associated_message_guid TEXT,
-      associated_message_type INTEGER,
-      balloon_bundle_id TEXT,
-      is_read INTEGER, date_read INTEGER,
-      date INTEGER,
-      is_from_me INTEGER,
-      service TEXT
-    );
-    """
-  )
-  try db.execute("CREATE TABLE handle (ROWID INTEGER PRIMARY KEY, id TEXT);")
-  try db.execute("CREATE TABLE chat_message_join (chat_id INTEGER, message_id INTEGER);")
-  try db.execute(
-    "CREATE TABLE message_attachment_join (message_id INTEGER, attachment_id INTEGER);")
-  return db
-}
-
-func insertURLPreviewTestMessage(
-  _ db: Connection,
-  rowID: Int64,
-  chatID: Int64 = 1,
-  handleID: Int64 = 1,
-  text: String,
-  guid: String,
-  associatedMessageGUID: String? = nil,
-  associatedMessageType: Int? = nil,
-  balloonBundleID: String? = nil,
-  date: Date,
-  isFromMe: Bool = false,
-  isRead: Bool = false,
-  dateRead: Date? = nil
-) throws {
-  try db.run(
-    """
-    INSERT INTO message(
-      ROWID, handle_id, text, guid, associated_message_guid, associated_message_type,
-      balloon_bundle_id, is_read, date_read, date, is_from_me, service
-    )
-    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'iMessage')
-    """,
-    rowID,
-    handleID,
-    text,
-    guid,
-    associatedMessageGUID,
-    associatedMessageType,
-    balloonBundleID,
-    isRead ? 1 : 0,
-    dateRead.map(TestDatabase.appleEpoch) ?? 0,
-    TestDatabase.appleEpoch(date),
-    isFromMe ? 1 : 0
-  )
-  try db.run(
-    "INSERT INTO chat_message_join(chat_id, message_id) VALUES (?, ?)",
-    chatID,
-    rowID
-  )
-}
-
 @Test
 func messagesAfterSkipsPreviewOnlyBatchForPublicCursorlessAPI() throws {
   let db = try makeURLPreviewTestDB()
diff --git a/Tests/IMsgCoreTests/MessageStoreUnreadStateTests.swift b/Tests/IMsgCoreTests/MessageStoreUnreadStateTests.swift
index f13be68f..60eda96c 100644
--- a/Tests/IMsgCoreTests/MessageStoreUnreadStateTests.swift
+++ b/Tests/IMsgCoreTests/MessageStoreUnreadStateTests.swift
@@ -148,6 +148,87 @@ func listChatsUsesTextRowReadStateForSplitURLPreview() throws {
   #expect(try store.listChats(limit: 1, unreadOnly: true).isEmpty)
 }
 
+@Test
+func listChatsCoalescesGUIDLinkedPreviewWhenTextOmitsURL() throws {
+  let db = try Connection(.inMemory)
+  try MessageDatabaseFixture.createSchema(
+    db,
+    options: MessageDatabaseFixture.SchemaOptions(
+      includeGUID: true,
+      includeBalloonBundleID: true,
+      includeReplyToGUID: true,
+      includeReadState: true
+    )
+  )
+  let now = Date()
+  try db.run(
+    """
+    INSERT INTO chat(ROWID, chat_identifier, guid, display_name, service_name)
+    VALUES (1, '+111', 'iMessage;-;+111', 'Links', 'iMessage')
+    """
+  )
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+111')")
+  try db.run(
+    """
+    INSERT INTO message(
+      ROWID, handle_id, text, guid, reply_to_guid, balloon_bundle_id,
+      date, is_from_me, service, is_read, date_read
+    )
+    VALUES
+      (1, 1, 'Check this out', 'text-guid', NULL, NULL, ?, 0, 'iMessage', 0, 0),
+      (2, 1, 'https://example.com', 'preview-guid', 'p:0/text-guid', ?, ?, 0, 'iMessage', 0, 0)
+    """,
+    TestDatabase.appleEpoch(now),
+    MessageStore.urlPreviewBalloonBundleID,
+    TestDatabase.appleEpoch(now.addingTimeInterval(0.399))
+  )
+  try db.run("INSERT INTO chat_message_join(chat_id, message_id) VALUES (1, 1), (1, 2)")
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  #expect(try store.listChats(limit: 1).first?.unreadCount == 1)
+  #expect(try store.listChats(limit: 1, unreadOnly: true).first?.unreadCount == 1)
+}
+
+@Test
+func listChatsCoalescesAssociatedGUIDLinkedPreviewWhenTextOmitsURL() throws {
+  let db = try Connection(.inMemory)
+  try MessageDatabaseFixture.createSchema(
+    db,
+    options: MessageDatabaseFixture.SchemaOptions(
+      includeReactionColumns: true,
+      includeBalloonBundleID: true,
+      includeReadState: true
+    )
+  )
+  let now = Date()
+  try db.run(
+    """
+    INSERT INTO chat(ROWID, chat_identifier, guid, display_name, service_name)
+    VALUES (1, '+111', 'iMessage;-;+111', 'Links', 'iMessage')
+    """
+  )
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+111')")
+  try db.run(
+    """
+    INSERT INTO message(
+      ROWID, handle_id, text, guid, associated_message_guid, associated_message_type,
+      balloon_bundle_id, date, is_from_me, service, is_read, date_read
+    )
+    VALUES
+      (1, 1, 'Check this out', 'text-guid', NULL, NULL, NULL, ?, 0, 'iMessage', 0, 0),
+      (2, 1, 'https://example.com', 'preview-guid', 'p:0/text-guid', 0, ?, ?, 0, 'iMessage', 0, 0)
+    """,
+    TestDatabase.appleEpoch(now),
+    MessageStore.urlPreviewBalloonBundleID,
+    TestDatabase.appleEpoch(now.addingTimeInterval(0.399))
+  )
+  try db.run("INSERT INTO chat_message_join(chat_id, message_id) VALUES (1, 1), (1, 2)")
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  #expect(try store.listChats(limit: 1).first?.unreadCount == 1)
+  #expect(try store.listChats(limit: 1, unreadOnly: true).first?.unreadCount == 1)
+}
+
 @Test
 func listChatsDoesNotCoalesceURLPreviewAcrossInterveningMessage() throws {
   let db = try Connection(.inMemory)
diff --git a/Tests/IMsgCoreTests/MessageWatcherTests.swift b/Tests/IMsgCoreTests/MessageWatcherTests.swift
index f35367cc..698d9f14 100644
--- a/Tests/IMsgCoreTests/MessageWatcherTests.swift
+++ b/Tests/IMsgCoreTests/MessageWatcherTests.swift
@@ -113,7 +113,7 @@ private enum WatcherTestDatabase {
   }
 }
 
-private func nextMessage(
+func nextMessage(
   from stream: AsyncThrowingStream<Message, Error>,
   timeoutNanoseconds: UInt64 = 2_000_000_000
 ) async throws -> Message? {
@@ -152,6 +152,93 @@ func messageWatcherYieldsExistingMessages() async throws {
   #expect(message?.text == "hello")
 }
 
+@Test
+func messageWatcherYieldsOneLogicalEventForGUIDLinkedURLPreview() async throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 1,
+    text: "Check this out",
+    guid: "text-guid",
+    date: now
+  )
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 2,
+    text: "https://example.com",
+    guid: "preview-guid",
+    replyToGUID: "text-guid",
+    balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+    date: now.addingTimeInterval(0.399)
+  )
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let watcher = MessageWatcher(store: store)
+  let stream = watcher.stream(
+    chatID: 1,
+    sinceRowID: -1,
+    configuration: MessageWatcherConfiguration(
+      debounceInterval: 0.01,
+      fallbackPollInterval: nil,
+      batchLimit: 10
+    )
+  )
+
+  let first = try await nextMessage(from: stream)
+  let second = try await nextMessage(from: stream, timeoutNanoseconds: 100_000_000)
+
+  #expect(first?.rowID == 1)
+  #expect(first?.urlPreview?.rowID == 2)
+  #expect(first?.text == "Check this out\nhttps://example.com")
+  #expect(second == nil)
+}
+
+@Test
+func messageWatcherYieldsOneLogicalEventForAssociatedGUIDLinkedURLPreview() async throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 1,
+    text: "Check this out",
+    guid: "text-guid",
+    date: now
+  )
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 2,
+    text: "https://example.com",
+    guid: "preview-guid",
+    associatedMessageGUID: "p:0/text-guid",
+    associatedMessageType: 0,
+    balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+    date: now.addingTimeInterval(0.399)
+  )
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let watcher = MessageWatcher(store: store)
+  let stream = watcher.stream(
+    chatID: 1,
+    sinceRowID: -1,
+    configuration: MessageWatcherConfiguration(
+      debounceInterval: 0.01,
+      fallbackPollInterval: nil,
+      batchLimit: 10
+    )
+  )
+
+  let first = try await nextMessage(from: stream)
+  let second = try await nextMessage(from: stream, timeoutNanoseconds: 100_000_000)
+
+  #expect(first?.rowID == 1)
+  #expect(first?.urlPreview?.rowID == 2)
+  #expect(first?.text == "Check this out\nhttps://example.com")
+  #expect(second == nil)
+}
+
 @Test
 func messageWatcherFallbackPollYieldsMessagesWithoutFileEvents() async throws {
   let fixture = try WatcherTestDatabase.makeMutableStore()
@@ -162,6 +249,7 @@ func messageWatcherFallbackPollYieldsMessagesWithoutFileEvents() async throws {
     configuration: MessageWatcherConfiguration(
       debounceInterval: 60,
       fallbackPollInterval: 0.01,
+      urlPreviewSettleInterval: 0,
       batchLimit: 10
     )
   )
@@ -179,6 +267,31 @@ func messageWatcherFallbackPollYieldsMessagesWithoutFileEvents() async throws {
   #expect(message?.text == "fallback")
 }
 
+@Test
+func messageWatcherDoesNotSettleEachBacklogPage() async throws {
+  let fixture = try WatcherTestDatabase.makeMutableStore()
+  try fixture.insertMessage(2, "first")
+  try fixture.insertMessage(3, "second")
+  try fixture.insertMessage(4, "third")
+
+  let watcher = MessageWatcher(store: fixture.store)
+  let stream = watcher.stream(
+    chatID: nil,
+    sinceRowID: 1,
+    configuration: MessageWatcherConfiguration(
+      debounceInterval: 0.005,
+      fallbackPollInterval: 0.005,
+      urlPreviewSettleInterval: 1,
+      batchLimit: 1
+    )
+  )
+
+  let first = try await nextMessage(from: stream, timeoutNanoseconds: 200_000_000)
+
+  #expect(first?.rowID == 2)
+  #expect(first?.text == "first")
+}
+
 @Test
 func messageWatcherRetriesUnresolvedChatMetadata() async throws {
   let fixture = try WatcherTestDatabase.makeMutableStore()
@@ -189,6 +302,7 @@ func messageWatcherRetriesUnresolvedChatMetadata() async throws {
     configuration: MessageWatcherConfiguration(
       debounceInterval: 0.01,
       fallbackPollInterval: 0.01,
+      urlPreviewSettleInterval: 0,
       batchLimit: 10
     )
   )
@@ -216,6 +330,7 @@ func messageWatcherSkipsPersistentlyUnresolvedChatMetadata() async throws {
     configuration: MessageWatcherConfiguration(
       debounceInterval: 0.001,
       fallbackPollInterval: 0.01,
+      urlPreviewSettleInterval: 0,
       batchLimit: 10
     )
   )
@@ -232,6 +347,32 @@ func messageWatcherSkipsPersistentlyUnresolvedChatMetadata() async throws {
   #expect(message?.text == "after orphan")
 }
 
+@Test
+func messageWatcherMakesProgressPastMultipleUnresolvedRows() async throws {
+  let fixture = try WatcherTestDatabase.makeMutableStore()
+  let watcher = MessageWatcher(store: fixture.store)
+  let stream = watcher.stream(
+    chatID: nil,
+    sinceRowID: 0,
+    configuration: MessageWatcherConfiguration(
+      debounceInterval: 0.001,
+      fallbackPollInterval: 0.01,
+      urlPreviewSettleInterval: 0,
+      batchLimit: 10
+    )
+  )
+
+  try await Task.sleep(nanoseconds: 20_000_000)
+  try fixture.insertUnjoinedMessage(2, "first orphan")
+  try fixture.insertUnjoinedMessage(3, "second orphan")
+  try fixture.insertMessage(4, "after orphans")
+
+  let message = try await nextMessage(from: stream, timeoutNanoseconds: 1_000_000_000)
+  #expect(message?.rowID == 4)
+  #expect(message?.chatID == 1)
+  #expect(message?.text == "after orphans")
+}
+
 #if os(macOS)
   @Test
   func messageWatcherRearmsSidecarAfterRotation() async throws {
@@ -262,6 +403,7 @@ func messageWatcherSkipsPersistentlyUnresolvedChatMetadata() async throws {
       configuration: MessageWatcherConfiguration(
         debounceInterval: 0.01,
         fallbackPollInterval: nil,
+        urlPreviewSettleInterval: 0,
         batchLimit: 10
       )
     )
diff --git a/Tests/IMsgCoreTests/MessageWatcherURLPreviewSettleTests.swift b/Tests/IMsgCoreTests/MessageWatcherURLPreviewSettleTests.swift
new file mode 100644
index 00000000..0d3ca1f8
--- /dev/null
+++ b/Tests/IMsgCoreTests/MessageWatcherURLPreviewSettleTests.swift
@@ -0,0 +1,273 @@
+import Foundation
+import SQLite
+import Testing
+
+@testable import IMsgCore
+
+@Test
+func messageWatcherKeepsUnlinkedLiveURLPreviewSeparateWhileSettling() async throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let watcher = MessageWatcher(store: store)
+  let stream = watcher.stream(
+    chatID: 1,
+    sinceRowID: 0,
+    configuration: MessageWatcherConfiguration(
+      debounceInterval: 0.005,
+      fallbackPollInterval: 0.005,
+      urlPreviewSettleInterval: 0.05,
+      batchLimit: 10
+    )
+  )
+
+  try await Task.sleep(nanoseconds: 10_000_000)
+  _ = try store.withConnection { connection in
+    try insertURLPreviewTestMessage(
+      connection,
+      rowID: 1,
+      text: "first message",
+      guid: "text-guid",
+      date: now
+    )
+    try insertURLPreviewTestMessage(
+      connection,
+      rowID: 2,
+      text: "https://example.com",
+      guid: "unlinked-preview-guid",
+      balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+      date: now.addingTimeInterval(1)
+    )
+  }
+
+  let first = try await nextMessage(from: stream)
+  let second = try await nextMessage(from: stream)
+  let third = try await nextMessage(from: stream, timeoutNanoseconds: 100_000_000)
+
+  #expect(first?.rowID == 1)
+  #expect(first?.urlPreview == nil)
+  #expect(second?.rowID == 2)
+  #expect(second?.balloonBundleID == MessageStore.urlPreviewBalloonBundleID)
+  #expect(third == nil)
+}
+
+@Test
+func messageWatcherDoesNotExtendOlderSettleDeadlineForNewerLiveText() async throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let watcher = MessageWatcher(store: store)
+  let stream = watcher.stream(
+    chatID: 1,
+    sinceRowID: 0,
+    configuration: MessageWatcherConfiguration(
+      debounceInterval: 0.005,
+      fallbackPollInterval: 0.005,
+      urlPreviewSettleInterval: 0.1,
+      batchLimit: 10
+    )
+  )
+
+  try await Task.sleep(nanoseconds: 10_000_000)
+  _ = try store.withConnection { connection in
+    try insertURLPreviewTestMessage(
+      connection,
+      rowID: 1,
+      text: "plain first",
+      guid: "first-guid",
+      date: now
+    )
+  }
+  try await Task.sleep(nanoseconds: 70_000_000)
+  _ = try store.withConnection { connection in
+    try insertURLPreviewTestMessage(
+      connection,
+      rowID: 2,
+      text: "Check this out",
+      guid: "second-guid",
+      date: now.addingTimeInterval(0.5)
+    )
+  }
+
+  let first = try await nextMessage(from: stream, timeoutNanoseconds: 70_000_000)
+
+  _ = try store.withConnection { connection in
+    try insertURLPreviewTestMessage(
+      connection,
+      rowID: 3,
+      text: "https://example.com",
+      guid: "preview-guid",
+      replyToGUID: "second-guid",
+      balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+      date: now.addingTimeInterval(1)
+    )
+  }
+
+  let second = try await nextMessage(from: stream)
+  let third = try await nextMessage(from: stream, timeoutNanoseconds: 100_000_000)
+
+  #expect(first?.rowID == 1)
+  #expect(second?.rowID == 2)
+  #expect(second?.urlPreview?.rowID == 3)
+  #expect(third == nil)
+}
+
+@Test
+func messageWatcherDoesNotRestartSettleDeadlineAcrossLimitedPages() async throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let watcher = MessageWatcher(store: store)
+  let stream = watcher.stream(
+    chatID: 1,
+    sinceRowID: 0,
+    configuration: MessageWatcherConfiguration(
+      debounceInterval: 0.005,
+      fallbackPollInterval: 0.005,
+      urlPreviewSettleInterval: 0.1,
+      batchLimit: 1
+    )
+  )
+
+  try await Task.sleep(nanoseconds: 10_000_000)
+  _ = try store.withConnection { connection in
+    for rowID in Int64(1)...4 {
+      try insertURLPreviewTestMessage(
+        connection,
+        rowID: rowID,
+        text: "message \(rowID)",
+        guid: "guid-\(rowID)",
+        date: now.addingTimeInterval(Double(rowID))
+      )
+    }
+  }
+
+  let first = try await nextMessage(from: stream)
+  let second = try await nextMessage(from: stream)
+  let third = try await nextMessage(from: stream, timeoutNanoseconds: 70_000_000)
+  let fourth = try await nextMessage(from: stream)
+
+  #expect([first?.rowID, second?.rowID, third?.rowID, fourth?.rowID] == [1, 2, 3, 4])
+}
+
+@Test
+func watcherTailQueryExcludesReactionsWhenConfigured() throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 1,
+    text: "message",
+    guid: "message-guid",
+    date: now
+  )
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 2,
+    text: "Loved message",
+    guid: "reaction-guid",
+    associatedMessageGUID: "message-guid",
+    associatedMessageType: 2001,
+    date: now.addingTimeInterval(1)
+  )
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+
+  #expect(try store.maxRowID(chatID: 1, includeReactions: false) == 1)
+  #expect(try store.maxRowID(chatID: 1, includeReactions: true) == 2)
+  #expect(try store.maxRowID(chatID: nil, includeReactions: false) == 1)
+  #expect(try store.maxRowID(chatID: nil, includeReactions: true) == 2)
+}
+
+@Test
+func messageWatcherDoesNotSettleReactionEvents() async throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let stream = MessageWatcher(store: store).stream(
+    chatID: 1,
+    sinceRowID: 0,
+    configuration: MessageWatcherConfiguration(
+      debounceInterval: 0.005,
+      fallbackPollInterval: 0.005,
+      urlPreviewSettleInterval: 0.2,
+      batchLimit: 10,
+      includeReactions: true
+    )
+  )
+
+  try await Task.sleep(nanoseconds: 10_000_000)
+  _ = try store.withConnection { connection in
+    try insertURLPreviewTestMessage(
+      connection,
+      rowID: 1,
+      text: "Loved message",
+      guid: "reaction-guid",
+      associatedMessageGUID: "target-guid",
+      associatedMessageType: 2001,
+      date: now
+    )
+  }
+
+  let reaction = try await nextMessage(from: stream, timeoutNanoseconds: 100_000_000)
+  #expect(reaction?.rowID == 1)
+  #expect(reaction?.isReaction == true)
+}
+
+@Test
+func messageWatcherKeepsReactionBehindEarlierSettleGap() async throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let stream = MessageWatcher(store: store).stream(
+    chatID: 1,
+    sinceRowID: 0,
+    configuration: MessageWatcherConfiguration(
+      debounceInterval: 0.005,
+      fallbackPollInterval: 0.005,
+      urlPreviewSettleInterval: 0.05,
+      batchLimit: 10,
+      includeReactions: true
+    )
+  )
+
+  try await Task.sleep(nanoseconds: 10_000_000)
+  _ = try store.withConnection { connection in
+    try insertURLPreviewTestMessage(
+      connection,
+      rowID: 1,
+      text: "plain text",
+      guid: "text-guid",
+      date: now
+    )
+    try insertURLPreviewTestMessage(
+      connection,
+      rowID: 2,
+      text: "Loved message",
+      guid: "reaction-guid",
+      associatedMessageGUID: "text-guid",
+      associatedMessageType: 2001,
+      date: now.addingTimeInterval(0.01)
+    )
+  }
+
+  let first = try await nextMessage(from: stream)
+  let second = try await nextMessage(from: stream)
+  let third = try await nextMessage(from: stream, timeoutNanoseconds: 100_000_000)
+
+  #expect(first?.rowID == 1)
+  #expect(second?.rowID == 2)
+  #expect(second?.isReaction == true)
+  #expect(third == nil)
+}
diff --git a/Tests/IMsgCoreTests/MessageWatcherURLPreviewTests.swift b/Tests/IMsgCoreTests/MessageWatcherURLPreviewTests.swift
new file mode 100644
index 00000000..cfa9b81f
--- /dev/null
+++ b/Tests/IMsgCoreTests/MessageWatcherURLPreviewTests.swift
@@ -0,0 +1,431 @@
+import Foundation
+import SQLite
+import Testing
+
+@testable import IMsgCore
+
+@Test
+func messageWatcherCoalescesGUIDLinkedURLPreviewAcrossBatchBoundary() async throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 1,
+    text: "Check this out",
+    guid: "text-guid",
+    date: now
+  )
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 2,
+    text: "https://example.com",
+    guid: "preview-guid",
+    replyToGUID: "text-guid",
+    balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+    date: now.addingTimeInterval(1)
+  )
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let watcher = MessageWatcher(store: store)
+  let stream = watcher.stream(
+    chatID: 1,
+    sinceRowID: -1,
+    configuration: MessageWatcherConfiguration(
+      debounceInterval: 0.01,
+      fallbackPollInterval: nil,
+      batchLimit: 1
+    )
+  )
+
+  let first = try await nextMessage(from: stream)
+  let second = try await nextMessage(from: stream, timeoutNanoseconds: 100_000_000)
+
+  #expect(first?.rowID == 1)
+  #expect(first?.cursorRowID == 2)
+  #expect(first?.urlPreview?.rowID == 2)
+  #expect(first?.text == "Check this out\nhttps://example.com")
+  #expect(second == nil)
+
+  let resumedStream = MessageWatcher(store: store).stream(
+    chatID: 1,
+    sinceRowID: first?.cursorRowID,
+    configuration: MessageWatcherConfiguration(
+      debounceInterval: 0.01,
+      fallbackPollInterval: nil,
+      batchLimit: 1
+    )
+  )
+  let replay = try await nextMessage(from: resumedStream, timeoutNanoseconds: 100_000_000)
+  #expect(replay == nil)
+}
+
+@Test
+func messageWatcherKeepsMultiChatPreviewAssociationsDistinct() async throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 1,
+    chatID: 1,
+    text: "https://example.com",
+    guid: "preview-guid",
+    balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+    date: now
+  )
+  try db.run("INSERT INTO chat_message_join(chat_id, message_id) VALUES (2, 1)")
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let stream = MessageWatcher(store: store).stream(
+    chatID: nil,
+    sinceRowID: -1,
+    configuration: MessageWatcherConfiguration(
+      debounceInterval: 0.01,
+      fallbackPollInterval: nil,
+      batchLimit: 10
+    )
+  )
+
+  let first = try await nextMessage(from: stream)
+  let second = try await nextMessage(from: stream)
+  let third = try await nextMessage(from: stream, timeoutNanoseconds: 100_000_000)
+
+  #expect(Set([first?.chatID, second?.chatID].compactMap { $0 }) == Set([1, 2]))
+  #expect(first?.rowID == 1)
+  #expect(second?.rowID == 1)
+  #expect(third == nil)
+}
+
+@Test
+func messageWatcherCoalescesLivePreviewPastInterleavedChatRow() async throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let watcher = MessageWatcher(store: store)
+  let stream = watcher.stream(
+    chatID: nil,
+    sinceRowID: 0,
+    configuration: MessageWatcherConfiguration(
+      debounceInterval: 0.005,
+      fallbackPollInterval: 0.005,
+      urlPreviewSettleInterval: 0.05,
+      batchLimit: 1
+    )
+  )
+
+  try await Task.sleep(nanoseconds: 10_000_000)
+  _ = try store.withConnection { connection in
+    try insertURLPreviewTestMessage(
+      connection,
+      rowID: 1,
+      chatID: 1,
+      text: "Check this out",
+      guid: "text-guid",
+      date: now
+    )
+    try insertURLPreviewTestMessage(
+      connection,
+      rowID: 2,
+      chatID: 2,
+      text: "other chat",
+      guid: "other-guid",
+      date: now.addingTimeInterval(0.5)
+    )
+    try insertURLPreviewTestMessage(
+      connection,
+      rowID: 3,
+      chatID: 1,
+      text: "https://example.com",
+      guid: "preview-guid",
+      replyToGUID: "text-guid",
+      balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+      date: now.addingTimeInterval(1)
+    )
+  }
+
+  let first = try await nextMessage(from: stream)
+  let second = try await nextMessage(from: stream)
+  let third = try await nextMessage(from: stream, timeoutNanoseconds: 100_000_000)
+
+  #expect(first?.rowID == 2)
+  #expect(first?.cursorRowID == 0)
+  #expect(second?.rowID == 1)
+  #expect(second?.cursorRowID == 3)
+  #expect(second?.urlPreview?.rowID == 3)
+  #expect(second?.text == "Check this out\nhttps://example.com")
+  #expect(third == nil)
+}
+
+@Test
+func messageWatcherDoesNotSkipRowsBeforeLookedAheadPreview() async throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  _ = try store.withConnection { connection in
+    try insertURLPreviewTestMessage(
+      connection,
+      rowID: 1,
+      chatID: 1,
+      text: "Check this out",
+      guid: "text-guid",
+      date: now
+    )
+    try insertURLPreviewTestMessage(
+      connection,
+      rowID: 2,
+      chatID: 2,
+      text: "first interleaved",
+      guid: "first-interleaved-guid",
+      date: now.addingTimeInterval(0.25)
+    )
+    try insertURLPreviewTestMessage(
+      connection,
+      rowID: 3,
+      chatID: 2,
+      text: "second interleaved",
+      guid: "second-interleaved-guid",
+      date: now.addingTimeInterval(0.5)
+    )
+    try insertURLPreviewTestMessage(
+      connection,
+      rowID: 4,
+      chatID: 1,
+      text: "https://example.com",
+      guid: "preview-guid",
+      replyToGUID: "text-guid",
+      balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+      date: now.addingTimeInterval(1)
+    )
+  }
+
+  let watcher = MessageWatcher(store: store)
+  let stream = watcher.stream(
+    chatID: nil,
+    sinceRowID: -1,
+    configuration: MessageWatcherConfiguration(
+      debounceInterval: 0.005,
+      fallbackPollInterval: nil,
+      urlPreviewSettleInterval: 1,
+      batchLimit: 1
+    )
+  )
+
+  let first = try await nextMessage(from: stream)
+  let second = try await nextMessage(from: stream)
+  let third = try await nextMessage(from: stream)
+  let fourth = try await nextMessage(from: stream, timeoutNanoseconds: 100_000_000)
+
+  #expect(first?.rowID == 2)
+  #expect(first?.cursorRowID == 0)
+  #expect(second?.rowID == 3)
+  #expect(second?.cursorRowID == 0)
+  #expect(third?.rowID == 1)
+  #expect(third?.cursorRowID == 4)
+  #expect(third?.urlPreview?.rowID == 4)
+  #expect(fourth == nil)
+}
+
+@Test
+func messageWatcherCoalescesURLContainingTextAcrossBatchBoundary() async throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 1,
+    text: "Dump https://example.com",
+    guid: "text-guid",
+    date: now
+  )
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 2,
+    text: "https://example.com",
+    guid: "preview-guid",
+    balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+    date: now.addingTimeInterval(1)
+  )
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let watcher = MessageWatcher(store: store)
+  let stream = watcher.stream(
+    chatID: 1,
+    sinceRowID: -1,
+    configuration: MessageWatcherConfiguration(
+      debounceInterval: 0.01,
+      fallbackPollInterval: nil,
+      batchLimit: 1
+    )
+  )
+
+  let first = try await nextMessage(from: stream)
+  let second = try await nextMessage(from: stream, timeoutNanoseconds: 100_000_000)
+
+  #expect(first?.rowID == 1)
+  #expect(first?.urlPreview?.rowID == 2)
+  #expect(second == nil)
+}
+
+@Test
+func messageWatcherCoalescesBacklogPreviewPastInterleavedChatRow() async throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 2,
+    chatID: 1,
+    text: "Check this out",
+    guid: "text-guid",
+    date: now
+  )
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 3,
+    chatID: 2,
+    text: "other chat",
+    guid: "other-guid",
+    date: now.addingTimeInterval(0.5)
+  )
+  try insertURLPreviewTestMessage(
+    db,
+    rowID: 4,
+    chatID: 1,
+    text: "https://example.com",
+    guid: "preview-guid",
+    replyToGUID: "text-guid",
+    balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+    date: now.addingTimeInterval(1)
+  )
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let watcher = MessageWatcher(store: store)
+  let stream = watcher.stream(
+    chatID: nil,
+    sinceRowID: 1,
+    configuration: MessageWatcherConfiguration(
+      debounceInterval: 0.005,
+      fallbackPollInterval: nil,
+      urlPreviewSettleInterval: 1,
+      batchLimit: 1
+    )
+  )
+
+  let first = try await nextMessage(from: stream, timeoutNanoseconds: 200_000_000)
+  let second = try await nextMessage(from: stream, timeoutNanoseconds: 200_000_000)
+  let third = try await nextMessage(from: stream, timeoutNanoseconds: 100_000_000)
+
+  #expect(first?.rowID == 3)
+  #expect(first?.cursorRowID == 1)
+  #expect(second?.rowID == 2)
+  #expect(second?.cursorRowID == 4)
+  #expect(second?.urlPreview?.rowID == 4)
+  #expect(second?.text == "Check this out\nhttps://example.com")
+  #expect(third == nil)
+}
+
+@Test
+func messageWatcherCoalescesGUIDLinkedURLPreviewInsertedAfterTextRow() async throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let watcher = MessageWatcher(store: store)
+  let stream = watcher.stream(
+    chatID: 1,
+    sinceRowID: 0,
+    configuration: MessageWatcherConfiguration(
+      debounceInterval: 0.01,
+      fallbackPollInterval: 0.01,
+      urlPreviewSettleInterval: 2,
+      batchLimit: 10
+    )
+  )
+
+  _ = try store.withConnection { connection in
+    try insertURLPreviewTestMessage(
+      connection,
+      rowID: 1,
+      text: "Check this out",
+      guid: "text-guid",
+      date: now
+    )
+  }
+  try await Task.sleep(nanoseconds: 30_000_000)
+  _ = try store.withConnection { connection in
+    try insertURLPreviewTestMessage(
+      connection,
+      rowID: 2,
+      text: "https://example.com",
+      guid: "preview-guid",
+      replyToGUID: "text-guid",
+      balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+      date: now.addingTimeInterval(1)
+    )
+  }
+
+  let first = try await nextMessage(from: stream)
+  let second = try await nextMessage(from: stream, timeoutNanoseconds: 100_000_000)
+
+  #expect(first?.rowID == 1)
+  #expect(first?.urlPreview?.rowID == 2)
+  #expect(first?.text == "Check this out\nhttps://example.com")
+  #expect(second == nil)
+}
+
+@Test
+func messageWatcherEmitsLateURLPreviewInsteadOfDroppingItsURL() async throws {
+  let db = try makeURLPreviewTestDB()
+  let now = Date()
+  try db.run("INSERT INTO handle(ROWID, id) VALUES (1, '+123')")
+
+  let store = try MessageStore(connection: db, path: ":memory:")
+  let watcher = MessageWatcher(store: store)
+  let stream = watcher.stream(
+    chatID: 1,
+    sinceRowID: 0,
+    configuration: MessageWatcherConfiguration(
+      debounceInterval: 0.005,
+      fallbackPollInterval: 0.005,
+      urlPreviewSettleInterval: 0.02,
+      batchLimit: 10
+    )
+  )
+
+  try await Task.sleep(nanoseconds: 10_000_000)
+  _ = try store.withConnection { connection in
+    try insertURLPreviewTestMessage(
+      connection,
+      rowID: 1,
+      text: "Check this out",
+      guid: "text-guid",
+      date: now
+    )
+  }
+  let first = try await nextMessage(from: stream)
+
+  _ = try store.withConnection { connection in
+    try insertURLPreviewTestMessage(
+      connection,
+      rowID: 2,
+      text: "https://example.com",
+      guid: "preview-guid",
+      replyToGUID: "text-guid",
+      balloonBundleID: MessageStore.urlPreviewBalloonBundleID,
+      date: now.addingTimeInterval(3)
+    )
+  }
+  let second = try await nextMessage(from: stream)
+
+  #expect(first?.rowID == 1)
+  #expect(first?.text == "Check this out")
+  #expect(second?.rowID == 2)
+  #expect(second?.text == "https://example.com")
+  #expect(second?.balloonBundleID == MessageStore.urlPreviewBalloonBundleID)
+}
diff --git a/Tests/imsgTests/WatchCommandTests.swift b/Tests/imsgTests/WatchCommandTests.swift
index 3c1a1144..a00d18d3 100644
--- a/Tests/imsgTests/WatchCommandTests.swift
+++ b/Tests/imsgTests/WatchCommandTests.swift
@@ -6,6 +6,23 @@ import Testing
 @testable import IMsgCore
 @testable import imsg
 
+private final class WatchStartupRecorder: @unchecked Sendable {
+  private let lock = NSLock()
+  private var events: [String] = []
+
+  func append(_ event: String) {
+    lock.lock()
+    events.append(event)
+    lock.unlock()
+  }
+
+  func snapshot() -> [String] {
+    lock.lock()
+    defer { lock.unlock() }
+    return events
+  }
+}
+
 private func singleMessageStreamProvider(
   _ message: Message
 ) -> (
@@ -22,6 +39,38 @@ private func singleMessageStreamProvider(
   }
 }
 
+@Test
+func watchCommandSubscribesBeforeResolvingContacts() async throws {
+  let values = ParsedValues(
+    positional: [],
+    options: ["db": ["/tmp/unused"], "debounce": ["1ms"]],
+    flags: []
+  )
+  let runtime = RuntimeOptions(parsedValues: values)
+  let store = try CommandTestDatabase.makeStoreForRPC()
+  let recorder = WatchStartupRecorder()
+
+  _ = try await StdoutCapture.capture {
+    try await WatchCommand.run(
+      values: values,
+      runtime: runtime,
+      storeFactory: { _ in store },
+      contactResolverFactory: {
+        recorder.append("contacts")
+        return NoOpContactResolver()
+      },
+      streamProvider: { _, _, _, _ in
+        recorder.append("stream")
+        return AsyncThrowingStream { continuation in
+          continuation.finish()
+        }
+      }
+    )
+  }
+
+  #expect(recorder.snapshot() == ["stream", "contacts"])
+}
+
 @Test
 func watchCommandRejectsInvalidDebounce() async {
   let values = ParsedValues(
@@ -118,6 +167,33 @@ func watchCommandRunsWithJsonOutput() async throws {
   #expect(payload["balloon_bundle_id"] as? String == "com.apple.messages.URLBalloonProvider")
 }
 
+@Test
+func messagePayloadUsesFoldedPreviewRowAsCursor() throws {
+  let message = Message(
+    rowID: 5,
+    chatID: 1,
+    sender: "+123",
+    text: "hello\nhttps://example.com",
+    date: Date(),
+    isFromMe: false,
+    service: "iMessage",
+    handleID: nil,
+    attachmentsCount: 0,
+    urlPreview: Message.URLPreviewMetadata(
+      rowID: 6,
+      guid: "preview-guid",
+      balloonBundleID: "com.apple.messages.URLBalloonProvider",
+      date: Date()
+    ),
+    cursorRowID: 6
+  )
+
+  let payload = try MessagePayload(message: message, attachments: []).asDictionary()
+
+  #expect(payload["id"] as? Int == 5)
+  #expect(payload["cursor"] as? Int == 6)
+}
+
 @Test
 func watchCommandJsonReportsDirectChatMetadata() async throws {
   let values = ParsedValues(
diff --git a/docs/json.md b/docs/json.md
index ca4bb72f..eabb2678 100644
--- a/docs/json.md
+++ b/docs/json.md
@@ -40,7 +40,8 @@ Returned by `imsg history`, `imsg search`, `imsg watch`, and the JSON-RPC `messa
 
 | Field | Type | Notes |
 |-------|------|-------|
-| `id` | int | rowid. Use as the `--since-rowid` cursor in watch. |
+| `id` | int | Logical message rowid. A coalesced URL preview keeps the originating text rowid. |
+| `cursor` | int | Live watch only. Monotonic exclusive resume cursor covering every applicable physical row emitted so far. Persist this value for `--since-rowid` or `watch.subscribe` resumes. |
 | `chat_id` | int | Always present. Preferred routing handle. |
 | `chat_identifier` | string | Portable handle. |
 | `chat_guid` | string | Portable GUID. |
@@ -51,7 +52,7 @@ Returned by `imsg history`, `imsg search`, `imsg watch`, and the JSON-RPC `messa
 | `reply_to_guid` | string | When set, this message is an inline reply to that GUID. |
 | `destination_caller_id` | string | Outgoing only — which of your numbers Messages routed through. |
 | `balloon_bundle_id` | string | Raw Messages `message.balloon_bundle_id`, when present. URL preview rows use `com.apple.messages.URLBalloonProvider`, which lets consumers recognize link-preview payload rows without inferring from message text. |
-| `url_preview` | object | Present when imsg folds an Apple URL-preview balloon row into its originating text row. The outer message keeps the text row's `id`, `guid`, `text`, and `created_at`. |
+| `url_preview` | object | Present when imsg folds an Apple URL-preview balloon row into its originating text row. The outer message keeps the text row's `id`, `guid`, and `created_at`; its `text` also includes the preview URL when Apple omitted it from that row. |
 | `sender` | string | Raw handle. Empty for some self-sent messages. |
 | `sender_name` | string | Resolved Contacts name when permission granted. |
 | `is_from_me` | bool | True for outbound. |
@@ -108,7 +109,7 @@ Returned by `imsg chat-background status --chat-id <id> --json`. This surface is
 
 ### URL preview coalescing
 
-Messages may store a link send as two rows: the user's text row and a later `com.apple.messages.URLBalloonProvider` preview row. `history`, `search`, `watch`, `messages.history`, and `watch.subscribe` coalesce those rows into one logical message when the preview immediately follows a same-chat/same-sender text row containing the preview URL. In batch reads the coalesced message includes:
+Messages may store a link send as two rows: the user's text row and a later `com.apple.messages.URLBalloonProvider` preview row. `history`, `search`, `watch`, `messages.history`, and `watch.subscribe` coalesce those rows into one logical message when the preview follows a compatible same-chat/same-sender text row and either contains a URL already present in that text or structurally targets the text row's GUID. The sender, direction, handle, row order, and five-second time-window safeguards still apply. When Apple omitted the URL from the text row, the coalesced message appends it to `text`. In batch reads the coalesced message includes:
 
 | Field | Type | Notes |
 |-------|------|-------|
@@ -117,7 +118,14 @@ Messages may store a link send as two rows: the user's text row and a later `com
 | `balloon_bundle_id` | string | `com.apple.messages.URLBalloonProvider`. |
 | `created_at` | ISO8601 | Preview row timestamp. |
 
-Live watch calls do not delay the text message waiting for a preview. If the preview row arrives in a later poll after the text row was already emitted, imsg suppresses the preview row so consumers still receive one notification.
+Live watch holds newly observed text rows for up to two seconds so a delayed linked preview can arrive before emission. Backlog rows are not held. If a linked preview arrives after the settling window, it is emitted separately rather than dropping the URL.
+For coalesced events, the outer message keeps the text row's `id`. Live watch
+orders events by physical completion and emits a monotonic `cursor` only after
+every applicable row through that point has been delivered, so resuming cannot
+skip an interleaved message. An early interleaved event can repeat the prior
+cursor until the frontier is complete and can therefore replay after a
+reconnect. Watch delivery is at-least-once; deduplicate by stable `chat_id` plus
+`id` or `guid`. Batch reads omit `cursor`.
 
 ### Reaction extensions
 
diff --git a/docs/quickstart.md b/docs/quickstart.md
index 3599c984..27ca6fe5 100644
--- a/docs/quickstart.md
+++ b/docs/quickstart.md
@@ -66,7 +66,7 @@ Leave it running. Send yourself a message from another device — you'll see the
 imsg watch --chat-id 42 --reactions --json
 ```
 
-To resume from a saved cursor (useful for agents that store the last seen `id`):
+To resume from a saved cursor (useful for agents that store the last seen `cursor` field):
 
 ```bash
 imsg watch --chat-id 42 --since-rowid 9000 --json
diff --git a/docs/rpc.md b/docs/rpc.md
index 54b5e68b..41f1be96 100644
--- a/docs/rpc.md
+++ b/docs/rpc.md
@@ -101,7 +101,7 @@ Older Messages database schemas without scheduling columns return an invalid-par
 Params:
 
 - `chat_id` (int, optional) — omit for all-chat stream.
-- `since_rowid` (int, optional) — exclusive cursor.
+- `since_rowid` (int, optional) — exclusive cursor; resume from the last message's `cursor` field, not its `id`. Repeated cursors can replay an event, so deduplicate notifications by stable message identity.
 - `participants` (array, optional)
 - `start` / `end` (ISO 8601, optional)
 - `attachments` (bool, default `false`)
diff --git a/docs/watch.md b/docs/watch.md
index 02db5b88..68aea6d8 100644
--- a/docs/watch.md
+++ b/docs/watch.md
@@ -3,7 +3,7 @@ title: Watch
 description: "Stream new iMessage and SMS rows live, with filesystem-event triggers and a poll-based fallback."
 ---
 
-`imsg watch` follows `chat.db` and emits each new message as soon as Messages writes it. It's the right primitive for agents, dashboards, notifiers, and anything that wants near-real-time inbound.
+`imsg watch` follows `chat.db` and emits new logical messages in near real time. Fresh text rows are briefly settled so a delayed Apple URL-preview row can be folded into the same event. It's the right primitive for agents, dashboards, notifiers, and anything that wants near-real-time inbound.
 
 ## Stream all chats
 
@@ -23,13 +23,21 @@ imsg watch --chat-id 42 --json
 
 ## Resuming from a cursor
 
-For long-lived consumers — agents, sync jobs — store the last `id` (rowid) you successfully processed and resume:
+For long-lived consumers — agents, sync jobs — store the last `cursor` you successfully processed and resume:
 
 ```bash
 imsg watch --chat-id 42 --since-rowid 9000 --json
 ```
 
 `--since-rowid` is exclusive: `9000` means "everything strictly after rowid 9000."
+The `cursor` normally equals `id`. For a logical message with a folded URL
+preview, watch first emits any applicable interleaved rows, then advances past
+both physical rows so reconnecting cannot skip another message. An interleaved
+event's cursor can therefore remain at the prior value until that frontier is
+complete. A reconnect from a repeated cursor can replay an already processed
+event, so treat watch as at-least-once and deduplicate by stable `chat_id` plus
+`id` or `guid`. Keep `id` as the message identity; use only `cursor` for
+resuming.
 
 If you don't pass `--since-rowid`, watch starts at the newest message at the moment of launch. Messages written before then are not replayed; use [`history`](history.md) for that.
 
@@ -89,6 +97,10 @@ could look like a direct message.
 
 Lower the debounce if you need lower latency and can tolerate occasional duplicate emissions during database churn. Raise it if downstream consumers can't keep up.
 
+The debounce is separate from URL-preview settling. A newly observed text row
+may wait up to two seconds for a structurally linked preview companion; backlog
+rows are not held.
+
 `--debounce` accepts Go-style durations: `100ms`, `1s`, `2s500ms`.
 
 ## How it knows when to read
@@ -114,9 +126,19 @@ Each fallback poll also refreshes the file watches, so a rotated `chat.db-wal` o
 
 This is the fix for the long-standing "watch goes silent after a while" class of bug. See `CHANGELOG.md` 0.6.0 entry.
 
-## URL preview deduplication
+## URL preview coalescing
+
+Messages may store one composition as a text row followed by a
+`com.apple.messages.URLBalloonProvider` row. On newer macOS versions the text
+row can omit the URL while the preview row targets it by GUID. `imsg watch`
+holds a new text row for up to two seconds and coalesces a matching preview when
+the chat, sender, direction, handle, row order, time window, and GUID or URL
+relationship all agree. If the text omitted the URL, the logical message text
+includes it.
 
-When you send a link, Messages writes a "balloon" placeholder row first, then later replaces it once the preview metadata is fetched. Without dedup, watch would emit both. `imsg watch` deduplicates these without dropping unrelated messages from other chats — the dedup is keyed precisely on the balloon update path, not on text similarity.
+Unlinked preview rows remain separate. A structurally linked preview that
+arrives after the settling window is also emitted separately rather than
+silently losing its URL.
 
 ## Output schema