feat: add cross-server direct messaging via Velocity

Allow /tell and /reply to reach players on other Paper servers behind a
Velocity proxy using the "<player>@<server>" target syntax.

Engine (protocol bumped 1.0.0 -> 1.0.1, optional sub-channels):
- Add DirectMessageRelay, DirectMessageError, PresenceSnapshot/PresenceEntry
  and PresenceRequest messages plus codec branches.

Velocity:
- CrossServerDirectMessageRelay routes a DM to the target server (or returns
  a delivery error to the source).
- PresenceTracker broadcasts proxy-wide presence snapshots on join/quit/switch
  and on request.

Paper:
- RemotePlayerRegistry caches proxy presence for completion and remote target
  resolution.
- CrossServerDirectMessageManager handles send/receive/error and dedup.
- DirectMessageHandler reply state generalized to ReplyTarget (Local/Remote)
  so /reply works across servers.
- TellCommand parses "name@server", completes local names and remote
  name@server targets, and uses exact local name matching.
- New crossServerDirectMessage config flag and i18n keys (en/ja).

Co-Authored-By: Claude <noreply@anthropic.com>

27 files changed, 1493 insertions, 75 deletions

diff --git a/engine/src/main/kotlin/dev/m1sk9/lunaticChat/engine/protocol/PluginMessage.kt b/engine/src/main/kotlin/dev/m1sk9/lunaticChat/engine/protocol/PluginMessage.kt
index f7090da..1ba5c25 100644
--- a/engine/src/main/kotlin/dev/m1sk9/lunaticChat/engine/protocol/PluginMessage.kt
+++ b/engine/src/main/kotlin/dev/m1sk9/lunaticChat/engine/protocol/PluginMessage.kt
@@ -78,4 +78,89 @@ sealed interface PluginMessage {
         val message: String,
         val timestamp: Long = System.currentTimeMillis(),
     ) : PluginMessage
+
+    /**
+     * Cross-server direct message relay (Paper → Velocity → target Paper)
+     *
+     * Routed to a single target server instead of broadcast. Romaji conversion is
+     * applied on the sending side, so [message] carries the already-converted text.
+     *
+     * @property messageId UUID for deduplication
+     * @property sourceServerName Sender's server identifier (used for reply routing)
+     * @property senderId Sender UUID as string
+     * @property senderName Sender name
+     * @property targetServerName Target server identifier (routing key)
+     * @property targetName Target player name
+     * @property message Message content (romaji-converted if applicable)
+     * @property timestamp Message timestamp (milliseconds since epoch)
+     */
+    @Serializable
+    data class DirectMessageRelay(
+        val messageId: String,
+        val sourceServerName: String,
+        val senderId: String,
+        val senderName: String,
+        val targetServerName: String,
+        val targetName: String,
+        val message: String,
+        val timestamp: Long = System.currentTimeMillis(),
+    ) : PluginMessage
+
+    /**
+     * Cross-server direct message delivery failure (Velocity → source Paper)
+     *
+     * @property messageId UUID of the failed relay
+     * @property senderId Sender UUID as string (used to locate the sender to notify)
+     * @property targetName Target player name that was requested
+     * @property targetServerName Target server name that was requested
+     * @property reason Failure reason: [Reason.TARGET_OFFLINE] or [Reason.SERVER_NOT_FOUND]
+     */
+    @Serializable
+    data class DirectMessageError(
+        val messageId: String,
+        val senderId: String,
+        val targetName: String,
+        val targetServerName: String,
+        val reason: String,
+    ) : PluginMessage {
+        object Reason {
+            const val TARGET_OFFLINE = "TARGET_OFFLINE"
+            const val SERVER_NOT_FOUND = "SERVER_NOT_FOUND"
+        }
+    }
+
+    /**
+     * Proxy-wide player presence snapshot (Velocity → Paper)
+     *
+     * Sent on join/quit/server-switch and in response to [PresenceRequest].
+     * Replaces the receiver's cached roster entirely.
+     *
+     * @property players All players currently connected to the proxy
+     * @property timestamp Snapshot timestamp (milliseconds since epoch)
+     */
+    @Serializable
+    data class PresenceSnapshot(
+        val players: List<PresenceEntry>,
+        val timestamp: Long = System.currentTimeMillis(),
+    ) : PluginMessage
+
+    /**
+     * Presence snapshot request (Paper → Velocity)
+     *
+     * Sent after a successful handshake to obtain the initial roster.
+     */
+    @Serializable
+    data object PresenceRequest : PluginMessage
 }
+
+/**
+ * A single player presence entry used in [PluginMessage.PresenceSnapshot].
+ *
+ * @property playerName Player name
+ * @property serverName Name of the server the player is currently connected to
+ */
+@Serializable
+data class PresenceEntry(
+    val playerName: String,
+    val serverName: String,
+)
diff --git a/engine/src/main/kotlin/dev/m1sk9/lunaticChat/engine/protocol/PluginMessageCodec.kt b/engine/src/main/kotlin/dev/m1sk9/lunaticChat/engine/protocol/PluginMessageCodec.kt
index 85b51f0..14d1f82 100644
--- a/engine/src/main/kotlin/dev/m1sk9/lunaticChat/engine/protocol/PluginMessageCodec.kt
+++ b/engine/src/main/kotlin/dev/m1sk9/lunaticChat/engine/protocol/PluginMessageCodec.kt
@@ -24,6 +24,10 @@ object PluginMessageCodec {
         const val STATUS_REQUEST = "status_request"
         const val STATUS_RESPONSE = "status_response"
         const val GLOBAL_CHAT = "global_chat"
+        const val DIRECT_MESSAGE = "direct_message"
+        const val DIRECT_MESSAGE_ERROR = "direct_message_error"
+        const val PRESENCE_SNAPSHOT = "presence_snapshot"
+        const val PRESENCE_REQUEST = "presence_request"
     }
 
     /**
@@ -53,6 +57,18 @@ object PluginMessageCodec {
                 is PluginMessage.GlobalChatMessage -> {
                     SubChannel.GLOBAL_CHAT to json.encodeToString(message)
                 }
+                is PluginMessage.DirectMessageRelay -> {
+                    SubChannel.DIRECT_MESSAGE to json.encodeToString(message)
+                }
+                is PluginMessage.DirectMessageError -> {
+                    SubChannel.DIRECT_MESSAGE_ERROR to json.encodeToString(message)
+                }
+                is PluginMessage.PresenceSnapshot -> {
+                    SubChannel.PRESENCE_SNAPSHOT to json.encodeToString(message)
+                }
+                is PluginMessage.PresenceRequest -> {
+                    SubChannel.PRESENCE_REQUEST to "{}"
+                }
             }
 
         dataOut.writeUTF(subChannel)
@@ -91,6 +107,18 @@ object PluginMessageCodec {
             SubChannel.GLOBAL_CHAT -> {
                 json.decodeFromString<PluginMessage.GlobalChatMessage>(messageJson)
             }
+            SubChannel.DIRECT_MESSAGE -> {
+                json.decodeFromString<PluginMessage.DirectMessageRelay>(messageJson)
+            }
+            SubChannel.DIRECT_MESSAGE_ERROR -> {
+                json.decodeFromString<PluginMessage.DirectMessageError>(messageJson)
+            }
+            SubChannel.PRESENCE_SNAPSHOT -> {
+                json.decodeFromString<PluginMessage.PresenceSnapshot>(messageJson)
+            }
+            SubChannel.PRESENCE_REQUEST -> {
+                PluginMessage.PresenceRequest
+            }
             else -> throw IllegalArgumentException("Unknown sub-channel: $subChannel")
         }
     }
diff --git a/engine/src/main/kotlin/dev/m1sk9/lunaticChat/engine/protocol/ProtocolVersion.kt b/engine/src/main/kotlin/dev/m1sk9/lunaticChat/engine/protocol/ProtocolVersion.kt
index 3cbb2cd..6585c86 100644
--- a/engine/src/main/kotlin/dev/m1sk9/lunaticChat/engine/protocol/ProtocolVersion.kt
+++ b/engine/src/main/kotlin/dev/m1sk9/lunaticChat/engine/protocol/ProtocolVersion.kt
@@ -33,7 +33,7 @@ package dev.m1sk9.lunaticChat.engine.protocol
 object ProtocolVersion {
     const val MAJOR = 1
     const val MINOR = 0
-    const val PATCH = 0
+    const val PATCH = 1
 
     /**
      * Minimum MINOR version this build can interoperate with (same MAJOR).
diff --git a/engine/src/test/kotlin/dev/m1sk9/lunaticChat/engine/protocol/PluginMessageCodecTest.kt b/engine/src/test/kotlin/dev/m1sk9/lunaticChat/engine/protocol/PluginMessageCodecTest.kt
index e848780..0eb56c1 100644
--- a/engine/src/test/kotlin/dev/m1sk9/lunaticChat/engine/protocol/PluginMessageCodecTest.kt
+++ b/engine/src/test/kotlin/dev/m1sk9/lunaticChat/engine/protocol/PluginMessageCodecTest.kt
@@ -136,6 +136,101 @@ class PluginMessageCodecTest {
     }
 
     @Test
+    fun `encode and decode DirectMessageRelay round-trip`() {
+        val original =
+            PluginMessage.DirectMessageRelay(
+                messageId = "dm-id-123",
+                sourceServerName = "survival",
+                senderId = "00000004-0000-0000-0000-000000000000",
+                senderName = "Sender",
+                targetServerName = "lobby",
+                targetName = "Recipient",
+                message = "Hello across servers!",
+                timestamp = 4000L,
+            )
+
+        val encoded = PluginMessageCodec.encode(original)
+        val decoded = PluginMessageCodec.decode(encoded)
+
+        assertIs<PluginMessage.DirectMessageRelay>(decoded)
+        assertEquals(original.messageId, decoded.messageId)
+        assertEquals(original.sourceServerName, decoded.sourceServerName)
+        assertEquals(original.senderId, decoded.senderId)
+        assertEquals(original.senderName, decoded.senderName)
+        assertEquals(original.targetServerName, decoded.targetServerName)
+        assertEquals(original.targetName, decoded.targetName)
+        assertEquals(original.message, decoded.message)
+        assertEquals(original.timestamp, decoded.timestamp)
+    }
+
+    @Test
+    fun `encode and decode DirectMessageError round-trip`() {
+        val original =
+            PluginMessage.DirectMessageError(
+                messageId = "dm-id-456",
+                senderId = "00000005-0000-0000-0000-000000000000",
+                targetName = "Ghost",
+                targetServerName = "lobby",
+                reason = PluginMessage.DirectMessageError.Reason.TARGET_OFFLINE,
+            )
+
+        val encoded = PluginMessageCodec.encode(original)
+        val decoded = PluginMessageCodec.decode(encoded)
+
+        assertIs<PluginMessage.DirectMessageError>(decoded)
+        assertEquals(original.messageId, decoded.messageId)
+        assertEquals(original.senderId, decoded.senderId)
+        assertEquals(original.targetName, decoded.targetName)
+        assertEquals(original.targetServerName, decoded.targetServerName)
+        assertEquals(PluginMessage.DirectMessageError.Reason.TARGET_OFFLINE, decoded.reason)
+    }
+
+    @Test
+    fun `encode and decode PresenceSnapshot round-trip`() {
+        val original =
+            PluginMessage.PresenceSnapshot(
+                players =
+                    listOf(
+                        PresenceEntry("Alice", "lobby"),
+                        PresenceEntry("Bob", "survival"),
+                    ),
+                timestamp = 5000L,
+            )
+
+        val encoded = PluginMessageCodec.encode(original)
+        val decoded = PluginMessageCodec.decode(encoded)
+
+        assertIs<PluginMessage.PresenceSnapshot>(decoded)
+        assertEquals(2, decoded.players.size)
+        assertEquals("Alice", decoded.players[0].playerName)
+        assertEquals("lobby", decoded.players[0].serverName)
+        assertEquals("Bob", decoded.players[1].playerName)
+        assertEquals("survival", decoded.players[1].serverName)
+        assertEquals(original.timestamp, decoded.timestamp)
+    }
+
+    @Test
+    fun `encode and decode PresenceSnapshot with empty list round-trip`() {
+        val original = PluginMessage.PresenceSnapshot(players = emptyList(), timestamp = 6000L)
+
+        val encoded = PluginMessageCodec.encode(original)
+        val decoded = PluginMessageCodec.decode(encoded)
+
+        assertIs<PluginMessage.PresenceSnapshot>(decoded)
+        assertEquals(0, decoded.players.size)
+    }
+
+    @Test
+    fun `encode and decode PresenceRequest round-trip`() {
+        val original = PluginMessage.PresenceRequest
+
+        val encoded = PluginMessageCodec.encode(original)
+        val decoded = PluginMessageCodec.decode(encoded)
+
+        assertIs<PluginMessage.PresenceRequest>(decoded)
+    }
+
+    @Test
     fun `decode should throw on unknown sub-channel`() {
         val out = java.io.ByteArrayOutputStream()
         val dataOut = java.io.DataOutputStream(out)
@@ -180,6 +275,10 @@ class PluginMessageCodecTest {
         assertEquals("status_request", PluginMessageCodec.SubChannel.STATUS_REQUEST)
         assertEquals("status_response", PluginMessageCodec.SubChannel.STATUS_RESPONSE)
         assertEquals("global_chat", PluginMessageCodec.SubChannel.GLOBAL_CHAT)
+        assertEquals("direct_message", PluginMessageCodec.SubChannel.DIRECT_MESSAGE)
+        assertEquals("direct_message_error", PluginMessageCodec.SubChannel.DIRECT_MESSAGE_ERROR)
+        assertEquals("presence_snapshot", PluginMessageCodec.SubChannel.PRESENCE_SNAPSHOT)
+        assertEquals("presence_request", PluginMessageCodec.SubChannel.PRESENCE_REQUEST)
     }
 
     @Test
@@ -191,6 +290,10 @@ class PluginMessageCodecTest {
                 PluginMessage.StatusRequest,
                 PluginMessage.StatusResponse("1.0.0", "1.0.0", true),
                 PluginMessage.GlobalChatMessage("id", "srv", "pid", "name", "msg", 0L),
+                PluginMessage.DirectMessageRelay("id", "src", "sid", "sname", "tsrv", "tname", "msg", 0L),
+                PluginMessage.DirectMessageError("id", "sid", "tname", "tsrv", "TARGET_OFFLINE"),
+                PluginMessage.PresenceSnapshot(listOf(PresenceEntry("p", "s")), 0L),
+                PluginMessage.PresenceRequest,
             )
 
         messages.forEach { message ->
diff --git a/engine/src/test/kotlin/dev/m1sk9/lunaticChat/engine/protocol/ProtocolBackwardCompatibilityTest.kt b/engine/src/test/kotlin/dev/m1sk9/lunaticChat/engine/protocol/ProtocolBackwardCompatibilityTest.kt
index 217baef..c9bc6c0 100644
--- a/engine/src/test/kotlin/dev/m1sk9/lunaticChat/engine/protocol/ProtocolBackwardCompatibilityTest.kt
+++ b/engine/src/test/kotlin/dev/m1sk9/lunaticChat/engine/protocol/ProtocolBackwardCompatibilityTest.kt
@@ -32,6 +32,16 @@ class ProtocolBackwardCompatibilityTest {
 
         const val GLOBAL_CHAT_V1_0_0 =
             """{"messageId":"abc-123","serverName":"lobby","playerId":"00000001-0000-0000-0000-000000000000","playerName":"TestPlayer","message":"Hello, world!","timestamp":1000}"""
+
+        // Protocol 1.0.1 snapshots — NEVER MODIFY after commit
+        const val DIRECT_MESSAGE_V1_0_1 =
+            """{"messageId":"dm-123","sourceServerName":"survival","senderId":"00000004-0000-0000-0000-000000000000","senderName":"Sender","targetServerName":"lobby","targetName":"Recipient","message":"Hello!","timestamp":4000}"""
+
+        const val DIRECT_MESSAGE_ERROR_V1_0_1 =
+            """{"messageId":"dm-456","senderId":"00000005-0000-0000-0000-000000000000","targetName":"Ghost","targetServerName":"lobby","reason":"TARGET_OFFLINE"}"""
+
+        const val PRESENCE_SNAPSHOT_V1_0_1 =
+            """{"players":[{"playerName":"Alice","serverName":"lobby"},{"playerName":"Bob","serverName":"survival"}],"timestamp":5000}"""
     }
 
     private fun buildRawMessage(
@@ -137,4 +147,53 @@ class ProtocolBackwardCompatibilityTest {
 
         assertIs<PluginMessage.StatusRequest>(decoded)
     }
+
+    @Test
+    fun `current codec can decode protocol 1_0_1 DirectMessageRelay`() {
+        val data = buildRawMessage("direct_message", DIRECT_MESSAGE_V1_0_1)
+        val decoded = PluginMessageCodec.decode(data)
+
+        assertIs<PluginMessage.DirectMessageRelay>(decoded)
+        assertEquals("dm-123", decoded.messageId)
+        assertEquals("survival", decoded.sourceServerName)
+        assertEquals("00000004-0000-0000-0000-000000000000", decoded.senderId)
+        assertEquals("Sender", decoded.senderName)
+        assertEquals("lobby", decoded.targetServerName)
+        assertEquals("Recipient", decoded.targetName)
+        assertEquals("Hello!", decoded.message)
+        assertEquals(4000L, decoded.timestamp)
+    }
+
+    @Test
+    fun `current codec can decode protocol 1_0_1 DirectMessageError`() {
+        val data = buildRawMessage("direct_message_error", DIRECT_MESSAGE_ERROR_V1_0_1)
+        val decoded = PluginMessageCodec.decode(data)
+
+        assertIs<PluginMessage.DirectMessageError>(decoded)
+        assertEquals("dm-456", decoded.messageId)
+        assertEquals("00000005-0000-0000-0000-000000000000", decoded.senderId)
+        assertEquals("Ghost", decoded.targetName)
+        assertEquals("lobby", decoded.targetServerName)
+        assertEquals("TARGET_OFFLINE", decoded.reason)
+    }
+
+    @Test
+    fun `current codec can decode protocol 1_0_1 PresenceSnapshot`() {
+        val data = buildRawMessage("presence_snapshot", PRESENCE_SNAPSHOT_V1_0_1)
+        val decoded = PluginMessageCodec.decode(data)
+
+        assertIs<PluginMessage.PresenceSnapshot>(decoded)
+        assertEquals(2, decoded.players.size)
+        assertEquals("Alice", decoded.players[0].playerName)
+        assertEquals("lobby", decoded.players[0].serverName)
+        assertEquals(5000L, decoded.timestamp)
+    }
+
+    @Test
+    fun `current codec can decode protocol 1_0_1 PresenceRequest with empty JSON`() {
+        val data = buildRawMessage("presence_request", "{}")
+        val decoded = PluginMessageCodec.decode(data)
+
+        assertIs<PluginMessage.PresenceRequest>(decoded)
+    }
 }
diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/LunaticChat.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/LunaticChat.kt
index addc3a3..035c578 100644
--- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/LunaticChat.kt
+++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/LunaticChat.kt
@@ -139,14 +139,26 @@ class LunaticChat :
 
         // Register core commands
         commandRegistry.registerAll(
-            TellCommand(this, services.directMessageHandler, services.languageManager),
+            TellCommand(
+                this,
+                services.directMessageHandler,
+                services.languageManager,
+                services.crossServerDirectMessageManager,
+                services.remotePlayerRegistry,
+                configuration.features.velocityIntegration.serverName,
+            ),
             LunaticChatCommand(this, settingHandlerRegistry, services.languageManager, configuration),
         )
 
         // Conditionally register /reply command if quick replies are enabled
         if (configuration.features.quickReplies.enabled) {
             commandRegistry.registerAll(
-                ReplyCommand(this, services.directMessageHandler, services.languageManager),
+                ReplyCommand(
+                    this,
+                    services.directMessageHandler,
+                    services.languageManager,
+                    services.crossServerDirectMessageManager,
+                ),
             )
         }
 
diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/ServiceContainer.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/ServiceContainer.kt
index 0607d82..4d668f2 100644
--- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/ServiceContainer.kt
+++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/ServiceContainer.kt
@@ -9,6 +9,8 @@ import dev.m1sk9.lunaticChat.paper.converter.RomanjiConverter
 import dev.m1sk9.lunaticChat.paper.i18n.LanguageManager
 import dev.m1sk9.lunaticChat.paper.settings.PlayerSettingsManager
 import dev.m1sk9.lunaticChat.paper.velocity.CrossServerChatManager
+import dev.m1sk9.lunaticChat.paper.velocity.CrossServerDirectMessageManager
+import dev.m1sk9.lunaticChat.paper.velocity.RemotePlayerRegistry
 import dev.m1sk9.lunaticChat.paper.velocity.VelocityConnectionManager
 
 /**
@@ -27,6 +29,8 @@ import dev.m1sk9.lunaticChat.paper.velocity.VelocityConnectionManager
  * @property channelNotificationHandler Optional (only when channel chat feature is enabled)
  * @property velocityConnectionManager Optional (only when Velocity integration is enabled)
  * @property crossServerChatManager Optional (only when Velocity integration and cross-server chat are enabled)
+ * @property crossServerDirectMessageManager Optional (only when Velocity integration and cross-server DM are enabled)
+ * @property remotePlayerRegistry Optional (only when Velocity integration and cross-server DM are enabled)
  */
 data class ServiceContainer(
     val languageManager: LanguageManager,
@@ -39,4 +43,6 @@ data class ServiceContainer(
     val channelNotificationHandler: ChannelNotificationHandler? = null,
     val velocityConnectionManager: VelocityConnectionManager? = null,
     val crossServerChatManager: CrossServerChatManager? = null,
+    val crossServerDirectMessageManager: CrossServerDirectMessageManager? = null,
+    val remotePlayerRegistry: RemotePlayerRegistry? = null,
 )
diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/ServiceInitializer.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/ServiceInitializer.kt
index 606273b..5ed8ed6 100644
--- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/ServiceInitializer.kt
+++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/ServiceInitializer.kt
@@ -15,6 +15,8 @@ import dev.m1sk9.lunaticChat.paper.i18n.LanguageManager
 import dev.m1sk9.lunaticChat.paper.settings.PlayerSettingsManager
 import dev.m1sk9.lunaticChat.paper.settings.YamlPlayerSettingsStorage
 import dev.m1sk9.lunaticChat.paper.velocity.CrossServerChatManager
+import dev.m1sk9.lunaticChat.paper.velocity.CrossServerDirectMessageManager
+import dev.m1sk9.lunaticChat.paper.velocity.RemotePlayerRegistry
 import dev.m1sk9.lunaticChat.paper.velocity.VelocityConnectionManager
 import io.ktor.client.HttpClient
 import org.bukkit.event.EventHandler
@@ -56,6 +58,8 @@ class ServiceInitializer(
     private var channelMessageLogger: ChannelMessageLogger? = null
     private var velocityConnectionManager: VelocityConnectionManager? = null
     private var crossServerChatManager: CrossServerChatManager? = null
+    private var crossServerDirectMessageManager: CrossServerDirectMessageManager? = null
+    private var remotePlayerRegistry: RemotePlayerRegistry? = null
     private val handshakeCompleted = AtomicBoolean(false)
 
     /**
@@ -132,6 +136,14 @@ class ServiceInitializer(
                 null
             }
 
+        // 8. Initialize cross-server direct message manager and presence registry (optional)
+        if (configuration.features.velocityIntegration.enabled &&
+            configuration.features.velocityIntegration.crossServerDirectMessage &&
+            velocityManager != null
+        ) {
+            initializeCrossServerDirectMessage(velocityManager, directMessageHandler, languageManager)
+        }
+
         return ServiceContainer(
             languageManager = languageManager,
             playerSettingsManager = playerSettingsManager,
@@ -143,6 +155,8 @@ class ServiceInitializer(
             channelNotificationHandler = channelNotificationHandler,
             velocityConnectionManager = velocityManager,
             crossServerChatManager = crossServerManager,
+            crossServerDirectMessageManager = crossServerDirectMessageManager,
+            remotePlayerRegistry = remotePlayerRegistry,
         )
     }
 
@@ -358,6 +372,36 @@ class ServiceInitializer(
     }
 
     /**
+     * Initializes cross-server direct messaging: the presence registry and the
+     * direct message manager, wiring them into the Velocity connection manager
+     * and the direct message handler.
+     */
+    private fun initializeCrossServerDirectMessage(
+        velocityManager: VelocityConnectionManager,
+        directMessageHandler: DirectMessageHandler,
+        languageManager: LanguageManager,
+    ) {
+        val registry = RemotePlayerRegistry(configuration.features.velocityIntegration.serverName)
+        remotePlayerRegistry = registry
+        directMessageHandler.remotePlayerRegistry = registry
+
+        val manager =
+            CrossServerDirectMessageManager(
+                plugin = plugin,
+                logger = logger,
+                configuration = configuration,
+                directMessageHandler = directMessageHandler,
+                languageManager = languageManager,
+                cacheSize = configuration.features.velocityIntegration.messageDeduplicationCacheSize,
+            )
+        crossServerDirectMessageManager = manager
+
+        velocityManager.setCrossServerDirectMessageManager(manager, registry)
+
+        logger.info("Cross-server direct messages initialized")
+    }
+
+    /**
      * Performs handshake with Velocity proxy.
      */
     private fun performVelocityHandshake(
diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/chat/handler/DirectMessageHandler.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/chat/handler/DirectMessageHandler.kt
index 1963528..f92f01d 100644
--- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/chat/handler/DirectMessageHandler.kt
+++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/chat/handler/DirectMessageHandler.kt
@@ -8,6 +8,7 @@ import dev.m1sk9.lunaticChat.paper.converter.RomanjiConverter
 import dev.m1sk9.lunaticChat.paper.converter.convertWithRomaji
 import dev.m1sk9.lunaticChat.paper.i18n.LanguageManager
 import dev.m1sk9.lunaticChat.paper.settings.PlayerSettingsManager
+import dev.m1sk9.lunaticChat.paper.velocity.RemotePlayerRegistry
 import net.kyori.adventure.text.Component
 import net.kyori.adventure.text.event.ClickEvent
 import net.kyori.adventure.text.event.HoverEvent
@@ -17,6 +18,23 @@ import java.util.UUID
 import java.util.concurrent.ConcurrentHashMap
 
 /**
+ * A reply destination for the /reply command.
+ *
+ * Generalized to support both same-server players ([Local]) and players on
+ * another server behind the Velocity proxy ([Remote]).
+ */
+sealed interface ReplyTarget {
+    data class Local(
+        val uuid: UUID,
+    ) : ReplyTarget
+
+    data class Remote(
+        val playerName: String,
+        val serverName: String,
+    ) : ReplyTarget
+}
+
+/**
  * Manages direct message state including reply targets.
  * Tracks the last player who messaged each player for /reply functionality.
  */
@@ -26,43 +44,49 @@ class DirectMessageHandler(
     private val romanjiConverter: RomanjiConverter?,
     private val languageManager: LanguageManager,
 ) {
-    private val lastMessager: ConcurrentHashMap<UUID, UUID> = ConcurrentHashMap()
-    private val lastRecipient: ConcurrentHashMap<UUID, UUID> = ConcurrentHashMap()
+    private val lastMessager: ConcurrentHashMap<UUID, ReplyTarget> = ConcurrentHashMap()
+    private val lastRecipient: ConcurrentHashMap<UUID, ReplyTarget> = ConcurrentHashMap()
 
     /**
-     * Records a direct message between two players.
+     * Registry of proxy-wide player presence. Set when cross-server direct
+     * messages are enabled; used to validate [ReplyTarget.Remote] reply targets.
+     */
+    var remotePlayerRegistry: RemotePlayerRegistry? = null
+
+    /**
+     * Records a local direct message between two players.
      * Updates both the sender's last recipient and the receiver's last messager.
      */
     fun recordMessage(
         sender: Player,
         recipient: Player,
     ) {
-        lastRecipient[sender.uniqueId] = recipient.uniqueId
-        lastMessager[recipient.uniqueId] = sender.uniqueId
+        lastRecipient[sender.uniqueId] = ReplyTarget.Local(recipient.uniqueId)
+        lastMessager[recipient.uniqueId] = ReplyTarget.Local(sender.uniqueId)
     }
 
     /**
-     * Gets the player to reply to.
+     * Gets the target to reply to.
      * First checks if someone has messaged this player, otherwise falls back
-     * to the last person they messaged.
+     * to the last person they messaged. Targets that are no longer reachable
+     * (offline locally, or absent from the proxy roster) are skipped.
      */
-    fun getReplyTarget(player: Player): Player? {
-        val messager = lastMessager[player.uniqueId]?.let { Bukkit.getPlayer(it) }
-        if (messager != null && messager.isOnline) {
-            return messager
-        }
-
-        val recipient = lastRecipient[player.uniqueId]?.let { Bukkit.getPlayer(it) }
-        if (recipient != null && recipient.isOnline) {
-            return recipient
-        }
-
+    fun getReplyTarget(player: Player): ReplyTarget? {
+        resolveValidTarget(lastMessager[player.uniqueId])?.let { return it }
+        resolveValidTarget(lastRecipient[player.uniqueId])?.let { return it }
         return null
     }
 
+    private fun resolveValidTarget(target: ReplyTarget?): ReplyTarget? =
+        when (target) {
+            is ReplyTarget.Local -> target.takeIf { Bukkit.getPlayer(it.uuid)?.isOnline == true }
+            is ReplyTarget.Remote -> target.takeIf { remotePlayerRegistry?.serverOf(it.playerName) == it.serverName }
+            null -> null
+        }
+
     /**
      * Clears message history for a player (called on disconnect).
-     * Removes entries where this player is either the sender or recipient.
+     * Removes entries where this player is either the sender or a local target.
      */
     fun clearPlayer(player: Player) {
         val playerId = player.uniqueId
@@ -71,13 +95,13 @@ class DirectMessageHandler(
         lastMessager.remove(playerId)
         lastRecipient.remove(playerId)
 
-        // Remove entries where this player is the recipient
-        lastMessager.entries.removeIf { it.value == playerId }
-        lastRecipient.entries.removeIf { it.value == playerId }
+        // Remove entries where this player is the local target
+        lastMessager.entries.removeIf { (it.value as? ReplyTarget.Local)?.uuid == playerId }
+        lastRecipient.entries.removeIf { (it.value as? ReplyTarget.Local)?.uuid == playerId }
     }
 
     /**
-     * Sends a direct message from one player to another.
+     * Sends a direct message from one player to another on the same server.
      * Handles formatting and recording the conversation.
      * Applies romaji-to-Japanese conversion if sender has it enabled.
      *
@@ -93,31 +117,13 @@ class DirectMessageHandler(
         val senderSettings = settingsManager?.getSettings(sender.uniqueId)
         val recipientSettings = settingsManager?.getSettings(recipient.uniqueId)
 
-        val displayMessage =
-            if (senderSettings?.japaneseConversionEnabled == true && romanjiConverter != null) {
-                convertWithRomaji(message, romanjiConverter)
-            } else {
-                message
-            }
+        val displayMessage = convertIfEnabled(message, senderSettings?.japaneseConversionEnabled == true)
 
         val format = configuration.messageFormat.directMessageFormat
 
-        val spyMessage = formatMessage(format, sender.name, recipient.name, message)
-        SpyPermissionManager
-            .getDirectMessageSpyPlayers()
-            .values
-            .filter { it.isOnline && it.uniqueId !in setOf(sender.uniqueId, recipient.uniqueId) }
-            .forEach {
-                it.sendMessage(
-                    spyMessage.hoverEvent(
-                        HoverEvent.showText(
-                            Component.text(languageManager.getMessage("general.spyMessage")),
-                        ),
-                    ),
-                )
-            }
+        notifySpies(format, sender.name, recipient.name, message)
 
-        val userMessage = formatMessage(format, sender.name, recipient.name, displayMessage)
+        val userMessage = formatMessage(format, sender.name, recipient.name, displayMessage, replyTo = sender.name)
         sender.apply {
             sendMessage(userMessage)
             takeIf { senderSettings?.directMessageNotificationEnabled == true }
@@ -131,11 +137,103 @@ class DirectMessageHandler(
         return true
     }
 
+    /**
+     * Handles the sender-side display of an outgoing cross-server direct message.
+     * Applies romaji conversion, shows the message to the sender, notifies spies,
+     * and records the remote reply target.
+     *
+     * @return the message body to relay (romaji-converted if applicable), since the
+     *   receiving server has no access to the sender's settings.
+     */
+    fun handleOutgoingCrossServerMessage(
+        sender: Player,
+        targetName: String,
+        targetServerName: String,
+        message: String,
+    ): String {
+        val senderSettings = settingsManager?.getSettings(sender.uniqueId)
+        val displayMessage = convertIfEnabled(message, senderSettings?.japaneseConversionEnabled == true)
+
+        val format = configuration.messageFormat.directMessageFormat
+        val recipientDisplay = "$targetName@$targetServerName"
+
+        notifySpies(format, sender.name, recipientDisplay, message)
+
+        val userMessage =
+            formatMessage(format, sender.name, recipientDisplay, displayMessage, replyTo = recipientDisplay)
+        sender.apply {
+            sendMessage(userMessage)
+            takeIf { senderSettings?.directMessageNotificationEnabled == true }
+                ?.playMessageSendNotification()
+        }
+
+        lastRecipient[sender.uniqueId] = ReplyTarget.Remote(targetName, targetServerName)
+        return displayMessage
+    }
+
+    /**
+     * Handles the receiver-side display of an incoming cross-server direct message.
+     * The message body is already romaji-converted by the sending server.
+     */
+    fun handleIncomingCrossServerMessage(
+        recipient: Player,
+        senderName: String,
+        sourceServerName: String,
+        message: String,
+    ) {
+        val recipientSettings = settingsManager?.getSettings(recipient.uniqueId)
+        val format = configuration.messageFormat.directMessageFormat
+        val senderDisplay = "$senderName@$sourceServerName"
+
+        val userMessage =
+            formatMessage(format, senderDisplay, recipient.name, message, replyTo = senderDisplay)
+        recipient.apply {
+            sendMessage(userMessage)
+            takeIf { recipientSettings?.directMessageNotificationEnabled == true }
+                ?.playDirectMessageNotification()
+        }
+
+        lastMessager[recipient.uniqueId] = ReplyTarget.Remote(senderName, sourceServerName)
+    }
+
+    private fun convertIfEnabled(
+        message: String,
+        enabled: Boolean,
+    ): String =
+        if (enabled && romanjiConverter != null) {
+            convertWithRomaji(message, romanjiConverter)
+        } else {
+            message
+        }
+
+    private fun notifySpies(
+        format: String,
+        senderName: String,
+        recipientName: String,
+        rawMessage: String,
+    ) {
+        val spyMessage = formatMessage(format, senderName, recipientName, rawMessage, replyTo = senderName)
+        SpyPermissionManager
+            .getDirectMessageSpyPlayers()
+            .values
+            .filter { it.isOnline && it.name !in setOf(senderName, recipientName) }
+            .forEach {
+                it.sendMessage(
+                    spyMessage.hoverEvent(
+                        HoverEvent.showText(
+                            Component.text(languageManager.getMessage("general.spyMessage")),
+                        ),
+                    ),
+                )
+            }
+    }
+
     private fun formatMessage(
         format: String,
         senderName: String,
         recipientName: String,
         message: String,
+        replyTo: String,
     ): Component {
         val text =
             format
@@ -145,6 +243,6 @@ class DirectMessageHandler(
 
         return Component
             .text(text)
-            .clickEvent(ClickEvent.suggestCommand("/tell $senderName "))
+            .clickEvent(ClickEvent.suggestCommand("/tell $replyTo "))
     }
 }
diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/ReplyCommand.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/ReplyCommand.kt
index e1a997e..1e24430 100644
--- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/ReplyCommand.kt
+++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/ReplyCommand.kt
@@ -6,6 +6,7 @@ import dev.m1sk9.lunaticChat.engine.command.CommandResult
 import dev.m1sk9.lunaticChat.engine.permission.LunaticChatPermissionNode
 import dev.m1sk9.lunaticChat.paper.LunaticChat
 import dev.m1sk9.lunaticChat.paper.chat.handler.DirectMessageHandler
+import dev.m1sk9.lunaticChat.paper.chat.handler.ReplyTarget
 import dev.m1sk9.lunaticChat.paper.command.annotation.Command
 import dev.m1sk9.lunaticChat.paper.command.annotation.Permission
 import dev.m1sk9.lunaticChat.paper.command.annotation.PlayerOnly
@@ -13,8 +14,10 @@ import dev.m1sk9.lunaticChat.paper.command.core.CommandContext
 import dev.m1sk9.lunaticChat.paper.command.core.LunaticCommand
 import dev.m1sk9.lunaticChat.paper.i18n.LanguageManager
 import dev.m1sk9.lunaticChat.paper.i18n.MessageFormatter
+import dev.m1sk9.lunaticChat.paper.velocity.CrossServerDirectMessageManager
 import io.papermc.paper.command.brigadier.CommandSourceStack
 import io.papermc.paper.command.brigadier.Commands
+import org.bukkit.Bukkit
 
 @Command(
     name = "reply",
@@ -27,6 +30,7 @@ class ReplyCommand(
     plugin: LunaticChat,
     private val dmHandler: DirectMessageHandler,
     private val languageManager: LanguageManager,
+    private val crossServerDirectMessageManager: CrossServerDirectMessageManager? = null,
 ) : LunaticCommand(plugin) {
     override val description: String
         get() = languageManager.getMessage("commandDescription.reply")
@@ -61,8 +65,29 @@ class ReplyCommand(
                     ),
                 )
 
-        dmHandler.sendDirectMessage(sender, target, message)
-
-        return CommandResult.Success
+        return when (target) {
+            is ReplyTarget.Local -> {
+                val recipient =
+                    Bukkit.getPlayer(target.uuid)
+                        ?: return CommandResult.Failure(
+                            MessageFormatter.formatError(
+                                languageManager.getMessage("directMessage.replyTargetNotFound"),
+                            ),
+                        )
+                dmHandler.sendDirectMessage(sender, recipient, message)
+                CommandResult.Success
+            }
+            is ReplyTarget.Remote -> {
+                val manager =
+                    crossServerDirectMessageManager
+                        ?: return CommandResult.Failure(
+                            MessageFormatter.formatError(
+                                languageManager.getMessage("directMessage.replyTargetNotFound"),
+                            ),
+                        )
+                manager.sendCrossServerMessage(sender, target.playerName, target.serverName, message)
+                CommandResult.Success
+            }
+        }
     }
 }
diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommand.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommand.kt
index 5375b08..ed2312d 100644
--- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommand.kt
+++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommand.kt
@@ -15,10 +15,14 @@ import dev.m1sk9.lunaticChat.paper.command.core.CommandContext
 import dev.m1sk9.lunaticChat.paper.command.core.LunaticCommand
 import dev.m1sk9.lunaticChat.paper.i18n.LanguageManager
 import dev.m1sk9.lunaticChat.paper.i18n.MessageFormatter
+import dev.m1sk9.lunaticChat.paper.velocity.CrossServerDirectMessageManager
+import dev.m1sk9.lunaticChat.paper.velocity.RemotePlayerRegistry
 import io.papermc.paper.command.brigadier.CommandSourceStack
 import io.papermc.paper.command.brigadier.Commands
 import org.bukkit.Bukkit
+import org.bukkit.entity.Player
 import java.util.concurrent.CompletableFuture
+import com.mojang.brigadier.context.CommandContext as BrigadierCommandContext
 
 @Command(
     name = "tell",
@@ -31,41 +35,93 @@ class TellCommand(
     plugin: LunaticChat,
     private val directMessageHandler: DirectMessageHandler,
     private val languageManager: LanguageManager,
+    private val crossServerDirectMessageManager: CrossServerDirectMessageManager? = null,
+    private val remotePlayerRegistry: RemotePlayerRegistry? = null,
+    private val localServerName: String = "",
 ) : LunaticCommand(plugin) {
     override val description: String
         get() = languageManager.getMessage("commandDescription.tell")
 
+    private companion object {
+        val WHITESPACE = Regex("\\s+")
+    }
+
     override fun buildCommand(): LiteralArgumentBuilder<CommandSourceStack> =
         Commands
             .literal(name)
             .then(
+                // A single greedy argument is used so the target token may contain '@'
+                // (e.g. "<player>@<server>"), which Brigadier's word()/string() reject.
                 Commands
-                    .argument("player", StringArgumentType.word())
-                    .suggests { _, builder -> suggestOnlinePlayers(builder) }
-                    .then(
-                        Commands
-                            .argument("message", StringArgumentType.greedyString())
-                            .executes { ctx ->
-                                val context = wrapContext(ctx)
-                                checkPlayerOnly(context)?.let { return@executes handleResult(context, it) }
-
-                                val targetName = StringArgumentType.getString(ctx, "player")
-                                val message = StringArgumentType.getString(ctx, "message")
-
-                                val result = execute(context, targetName, message)
-                                handleResult(context, result)
-                            },
-                    ),
+                    .argument("input", StringArgumentType.greedyString())
+                    .suggests { ctx, builder -> suggestTargets(ctx, builder) }
+                    .executes { ctx ->
+                        val context = wrapContext(ctx)
+                        checkPlayerOnly(context)?.let { return@executes handleResult(context, it) }
+
+                        val input = StringArgumentType.getString(ctx, "input")
+                        val result = parseAndExecute(context, input)
+                        handleResult(context, result)
+                    },
+            )
+
+    internal fun parseAndExecute(
+        ctx: CommandContext,
+        input: String,
+    ): CommandResult {
+        val parts = input.trim().split(WHITESPACE, limit = 2)
+        val targetName = parts[0]
+        val message = parts.getOrNull(1)
+        if (targetName.isEmpty() || message.isNullOrBlank()) {
+            return CommandResult.Failure(
+                MessageFormatter.formatError(
+                    languageManager.getMessage("directMessage.usage"),
+                ),
             )
+        }
+        return execute(ctx, targetName, message)
+    }
 
-    private fun execute(
+    internal fun execute(
         ctx: CommandContext,
         targetName: String,
         message: String,
     ): CommandResult {
         val sender = ctx.requirePlayer()
+
+        // Cross-server target: "<playerName>@<serverName>"
+        if (targetName.contains('@')) {
+            val manager =
+                crossServerDirectMessageManager
+                    ?: return CommandResult.Failure(
+                        MessageFormatter.formatError(
+                            languageManager.getMessage("directMessage.crossServerDisabled"),
+                        ),
+                    )
+            val name = targetName.substringBefore('@')
+            val server = targetName.substringAfter('@')
+            if (name.isEmpty() || server.isEmpty()) {
+                return CommandResult.Failure(
+                    MessageFormatter.formatError(
+                        languageManager.getMessage("directMessage.targetOffline", mapOf("target" to targetName)),
+                    ),
+                )
+            }
+            if (name.equals(sender.name, ignoreCase = true) &&
+                server.equals(localServerName, ignoreCase = true)
+            ) {
+                return CommandResult.Failure(
+                    MessageFormatter.formatError(
+                        languageManager.getMessage("directMessage.yourself"),
+                    ),
+                )
+            }
+            manager.sendCrossServerMessage(sender, name, server, message)
+            return CommandResult.Success
+        }
+
         val recipient =
-            Bukkit.getPlayer(targetName)
+            Bukkit.getPlayerExact(targetName)
                 ?: return CommandResult.Failure(
                     MessageFormatter.formatError(
                         languageManager.getMessage("directMessage.targetOffline", mapOf("target" to targetName)),
@@ -85,12 +141,34 @@ class TellCommand(
         return CommandResult.Success
     }
 
-    private fun suggestOnlinePlayers(builder: SuggestionsBuilder): CompletableFuture<Suggestions> {
-        val input = builder.remaining.lowercase()
+    private fun suggestTargets(
+        ctx: BrigadierCommandContext<CommandSourceStack>,
+        builder: SuggestionsBuilder,
+    ): CompletableFuture<Suggestions> {
+        // Only complete the target token; once the message part begins, stop suggesting.
+        val remaining = builder.remaining
+        if (remaining.contains(' ')) {
+            return builder.buildFuture()
+        }
+        val input = remaining.lowercase()
+        val senderName = (ctx.source.sender as? Player)?.name
+
+        // Local online players (same-server targets), excluding the sender themselves
         Bukkit
             .getOnlinePlayers()
-            .filter { it.name.lowercase().startsWith(input) }
+            .filter { it.name != senderName && it.name.lowercase().startsWith(input) }
             .forEach { builder.suggest(it.name) }
+
+        // Remote players (cross-server targets) as "<name>@<server>"
+        val registry = remotePlayerRegistry
+        if (crossServerDirectMessageManager != null && registry != null) {
+            registry
+                .remotePlayers()
+                .map { "${it.playerName}@${it.serverName}" }
+                .filter { it.lowercase().startsWith(input) }
+                .forEach { builder.suggest(it) }
+        }
+
         return builder.buildFuture()
     }
 }
diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/config/ConfigManager.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/config/ConfigManager.kt
index 33952be..37764bf 100644
--- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/config/ConfigManager.kt
+++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/config/ConfigManager.kt
@@ -60,6 +60,11 @@ class ConfigManager {
                                         "features.velocityIntegration.crossServerGlobalChat",
                                         false,
                                     ),
+                                crossServerDirectMessage =
+                                    configFile.getBoolean(
+                                        "features.velocityIntegration.crossServerDirectMessage",
+                                        false,
+                                    ),
                                 serverName =
                                     configFile.getString(
                                         "features.velocityIntegration.serverName",
diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/config/key/VelocityIntegrationConfig.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/config/key/VelocityIntegrationConfig.kt
index 51a67b6..2b2babb 100644
--- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/config/key/VelocityIntegrationConfig.kt
+++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/config/key/VelocityIntegrationConfig.kt
@@ -3,6 +3,7 @@ package dev.m1sk9.lunaticChat.paper.config.key
 data class VelocityIntegrationConfig(
     val enabled: Boolean = false,
     val crossServerGlobalChat: Boolean = false,
+    val crossServerDirectMessage: Boolean = false,
     val serverName: String = "Unknown",
     val messageDeduplicationCacheSize: Int = 100,
 )
diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/CrossServerDirectMessageManager.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/CrossServerDirectMessageManager.kt
new file mode 100644
index 0000000..b8390a7
--- /dev/null
+++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/CrossServerDirectMessageManager.kt
@@ -0,0 +1,185 @@
+package dev.m1sk9.lunaticChat.paper.velocity
+
+import dev.m1sk9.lunaticChat.engine.protocol.PluginMessage
+import dev.m1sk9.lunaticChat.engine.protocol.PluginMessageCodec
+import dev.m1sk9.lunaticChat.paper.chat.handler.DirectMessageHandler
+import dev.m1sk9.lunaticChat.paper.config.LunaticChatConfiguration
+import dev.m1sk9.lunaticChat.paper.i18n.LanguageManager
+import dev.m1sk9.lunaticChat.paper.i18n.MessageFormatter
+import org.bukkit.entity.Player
+import org.bukkit.plugin.Plugin
+import java.util.UUID
+import java.util.concurrent.ConcurrentHashMap
+import java.util.logging.Level
+import java.util.logging.Logger
+
+/**
+ * Manages cross-server direct messages.
+ *
+ * Handles:
+ * - Sending direct messages to a player on another server via Velocity
+ * - Processing incoming relayed direct messages from Velocity
+ * - Surfacing delivery errors returned by Velocity
+ * - Message deduplication using an LRU-style cache
+ */
+class CrossServerDirectMessageManager(
+    private val plugin: Plugin,
+    private val logger: Logger,
+    private val configuration: LunaticChatConfiguration,
+    private val directMessageHandler: DirectMessageHandler,
+    private val languageManager: LanguageManager,
+    private val cacheSize: Int = 100,
+) {
+    companion object {
+        private const val CHANNEL = "lunaticchat:main"
+        private const val CLEANUP_THRESHOLD_MILLIS = 60_000L
+    }
+
+    private val processedMessages = ConcurrentHashMap<String, Long>()
+
+    /**
+     * Sends a direct message to a player on another server through Velocity.
+     *
+     * Must be called on the main server thread. The sender-side display, spy
+     * notification and reply recording are handled by [DirectMessageHandler];
+     * the (possibly romaji-converted) body is what gets relayed.
+     */
+    fun sendCrossServerMessage(
+        sender: Player,
+        targetName: String,
+        targetServerName: String,
+        message: String,
+    ) {
+        try {
+            val messageId = UUID.randomUUID().toString()
+            processedMessages[messageId] = System.currentTimeMillis()
+
+            val relayedMessage =
+                directMessageHandler.handleOutgoingCrossServerMessage(
+                    sender = sender,
+                    targetName = targetName,
+                    targetServerName = targetServerName,
+                    message = message,
+                )
+
+            val relay =
+                PluginMessage.DirectMessageRelay(
+                    messageId = messageId,
+                    sourceServerName = configuration.features.velocityIntegration.serverName,
+                    senderId = sender.uniqueId.toString(),
+                    senderName = sender.name,
+                    targetServerName = targetServerName,
+                    targetName = targetName,
+                    message = relayedMessage,
+                )
+
+            sender.sendPluginMessage(plugin, CHANNEL, PluginMessageCodec.encode(relay))
+            logger.info(
+                "Sent direct message to Velocity: messageId=$messageId, " +
+                    "target=$targetName@$targetServerName",
+            )
+
+            if (processedMessages.size > cacheSize) {
+                cleanupOldMessages()
+            }
+        } catch (e: Exception) {
+            logger.log(Level.SEVERE, "Failed to send cross-server direct message", e)
+        }
+    }
+
+    /**
+     * Handles a relayed direct message arriving from Velocity for a local recipient.
+     */
+    fun handleIncomingMessage(message: PluginMessage.DirectMessageRelay) {
+        try {
+            if (!shouldProcessMessage(message.messageId)) {
+                logger.fine("Ignoring duplicate direct message: messageId=${message.messageId}")
+                return
+            }
+            processedMessages[message.messageId] = System.currentTimeMillis()
+
+            plugin.server.scheduler.runTask(
+                plugin,
+                Runnable {
+                    val recipient = plugin.server.getPlayer(message.targetName)
+                    if (recipient == null) {
+                        logger.warning(
+                            "Received direct message for offline player: ${message.targetName} " +
+                                "(messageId=${message.messageId})",
+                        )
+                        return@Runnable
+                    }
+                    directMessageHandler.handleIncomingCrossServerMessage(
+                        recipient = recipient,
+                        senderName = message.senderName,
+                        sourceServerName = message.sourceServerName,
+                        message = message.message,
+                    )
+                },
+            )
+
+            if (processedMessages.size > cacheSize) {
+                cleanupOldMessages()
+            }
+        } catch (e: Exception) {
+            logger.log(Level.SEVERE, "Failed to handle incoming direct message", e)
+        }
+    }
+
+    /**
+     * Handles a delivery error returned by Velocity and notifies the sender.
+     */
+    fun handleError(error: PluginMessage.DirectMessageError) {
+        try {
+            val senderId = runCatching { UUID.fromString(error.senderId) }.getOrNull() ?: return
+            plugin.server.scheduler.runTask(
+                plugin,
+                Runnable {
+                    val sender = plugin.server.getPlayer(senderId) ?: return@Runnable
+                    val messageKey =
+                        when (error.reason) {
+                            PluginMessage.DirectMessageError.Reason.SERVER_NOT_FOUND -> "directMessage.remoteServerNotFound"
+                            else -> "directMessage.remoteTargetOffline"
+                        }
+                    val text =
+                        languageManager.getMessage(
+                            messageKey,
+                            mapOf("target" to error.targetName, "server" to error.targetServerName),
+                        )
+                    sender.sendMessage(MessageFormatter.formatError(text))
+                },
+            )
+        } catch (e: Exception) {
+            logger.log(Level.SEVERE, "Failed to handle direct message error", e)
+        }
+    }
+
+    private fun shouldProcessMessage(messageId: String): Boolean = !processedMessages.containsKey(messageId)
+
+    private fun cleanupOldMessages() {
+        try {
+            val cutoffTime = System.currentTimeMillis() - CLEANUP_THRESHOLD_MILLIS
+
+            val keysToRemove = processedMessages.entries.filter { it.value < cutoffTime }.map { it.key }
+            keysToRemove.forEach { processedMessages.remove(it) }
+            var removedCount = keysToRemove.size
+
+            if (processedMessages.size > cacheSize) {
+                val toRemove = processedMessages.size - cacheSize
+                processedMessages.entries
+                    .sortedBy { it.value }
+                    .take(toRemove)
+                    .forEach {
+                        processedMessages.remove(it.key)
+                        removedCount++
+                    }
+            }
+
+            if (removedCount > 0) {
+                logger.fine("Cleaned up $removedCount old messages from direct message dedup cache")
+            }
+        } catch (e: Exception) {
+            logger.log(Level.WARNING, "Failed to cleanup old direct messages", e)
+        }
+    }
+}
diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/RemotePlayerRegistry.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/RemotePlayerRegistry.kt
new file mode 100644
index 0000000..cf57dba
--- /dev/null
+++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/RemotePlayerRegistry.kt
@@ -0,0 +1,49 @@
+package dev.m1sk9.lunaticChat.paper.velocity
+
+import dev.m1sk9.lunaticChat.engine.protocol.PresenceEntry
+import java.util.concurrent.ConcurrentHashMap
+
+/**
+ * Caches proxy-wide player presence (player name -> server name) on this Paper server.
+ *
+ * Velocity is the source of truth; this registry is replaced wholesale on each
+ * [dev.m1sk9.lunaticChat.engine.protocol.PluginMessage.PresenceSnapshot]. It backs
+ * cross-server direct message tab-completion and remote target resolution.
+ *
+ * @param localServerName This server's name (as configured in velocityIntegration.serverName),
+ *   used to exclude local players from remote lookups.
+ */
+class RemotePlayerRegistry(
+    private val localServerName: String,
+) {
+    // lowercase player name -> presence entry (preserves original-cased name)
+    private val players = ConcurrentHashMap<String, PresenceEntry>()
+
+    /**
+     * Replaces the entire roster with the given snapshot entries.
+     */
+    fun replaceAll(entries: List<PresenceEntry>) {
+        players.clear()
+        entries.forEach { players[it.playerName.lowercase()] = it }
+    }
+
+    /**
+     * Clears the roster (e.g. when the Velocity connection is lost).
+     */
+    fun clear() {
+        players.clear()
+    }
+
+    /**
+     * Returns the server a player is currently on, or null if unknown.
+     */
+    fun serverOf(playerName: String): String? = players[playerName.lowercase()]?.serverName
+
+    /**
+     * Returns players on servers other than this one, with original-cased names.
+     */
+    fun remotePlayers(): List<PresenceEntry> =
+        players.values
+            .filter { !it.serverName.equals(localServerName, ignoreCase = true) }
+            .toList()
+}
diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/VelocityConnectionManager.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/VelocityConnectionManager.kt
index a93cb9f..1f4ccb4 100644
--- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/VelocityConnectionManager.kt
+++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/VelocityConnectionManager.kt
@@ -17,6 +17,8 @@ class VelocityConnectionManager(
     private val pluginVersion: String,
     private val logger: Logger,
     private var crossServerChatManager: CrossServerChatManager? = null,
+    private var crossServerDirectMessageManager: CrossServerDirectMessageManager? = null,
+    private var remotePlayerRegistry: RemotePlayerRegistry? = null,
 ) : PluginMessageListener {
     companion object {
         private const val CHANNEL = "lunaticchat:main"
@@ -51,6 +53,7 @@ class VelocityConnectionManager(
     private var statusFuture: CompletableFuture<PluginMessage.StatusResponse>? = null
     private var velocityVersion: String? = null
     private var lastError: String? = null
+    private var handshakePlayer: Player? = null
 
     /**
      * Initialize
@@ -72,6 +75,18 @@ class VelocityConnectionManager(
     }
 
     /**
+     * Sets the cross-server direct message manager and remote player registry.
+     * Called after initialization to avoid circular dependency.
+     */
+    fun setCrossServerDirectMessageManager(
+        manager: CrossServerDirectMessageManager,
+        registry: RemotePlayerRegistry,
+    ) {
+        this.crossServerDirectMessageManager = manager
+        this.remotePlayerRegistry = registry
+    }
+
+    /**
      * Performs handshake
      *
      * @param player Player to use for sending messages
@@ -86,6 +101,7 @@ class VelocityConnectionManager(
 
         state = ConnectionState.HANDSHAKING
         lastError = null
+        handshakePlayer = player
 
         val future = CompletableFuture<HandshakeResult>()
         handshakeFuture = future
@@ -178,6 +194,9 @@ class VelocityConnectionManager(
                 is PluginMessage.HandshakeResponse -> handleHandshakeResponse(pluginMessage)
                 is PluginMessage.StatusResponse -> handleStatusResponse(pluginMessage)
                 is PluginMessage.GlobalChatMessage -> handleGlobalChatMessage(pluginMessage)
+                is PluginMessage.DirectMessageRelay -> handleDirectMessageRelay(pluginMessage)
+                is PluginMessage.DirectMessageError -> handleDirectMessageError(pluginMessage)
+                is PluginMessage.PresenceSnapshot -> handlePresenceSnapshot(pluginMessage)
                 else -> logger.warning("Unexpected message type: ${pluginMessage::class.simpleName}")
             }
         } catch (e: Exception) {
@@ -201,6 +220,7 @@ class VelocityConnectionManager(
                 "Successfully connected to Velocity (version: ${response.velocityVersion}, " +
                     "protocol: ${response.protocolMajor}.${response.protocolMinor}.${response.protocolPatch})",
             )
+            requestInitialPresence()
             future.complete(HandshakeResult.Success(response.velocityVersion))
         } else {
             state = ConnectionState.FAILED
@@ -246,6 +266,43 @@ class VelocityConnectionManager(
     }
 
     /**
+     * Handles a relayed direct message from Velocity
+     */
+    private fun handleDirectMessageRelay(message: PluginMessage.DirectMessageRelay) {
+        val manager = crossServerDirectMessageManager
+        if (manager != null) {
+            manager.handleIncomingMessage(message)
+        } else {
+            logger.warning("Received direct message but CrossServerDirectMessageManager is not initialized")
+        }
+    }
+
+    /**
+     * Handles a direct message delivery error from Velocity
+     */
+    private fun handleDirectMessageError(message: PluginMessage.DirectMessageError) {
+        crossServerDirectMessageManager?.handleError(message)
+    }
+
+    /**
+     * Handles a proxy-wide presence snapshot from Velocity
+     */
+    private fun handlePresenceSnapshot(message: PluginMessage.PresenceSnapshot) {
+        remotePlayerRegistry?.replaceAll(message.players)
+    }
+
+    /**
+     * Requests the initial presence snapshot after a successful handshake.
+     */
+    private fun requestInitialPresence() {
+        if (remotePlayerRegistry == null) return
+        val player = handshakePlayer?.takeIf { it.isOnline } ?: return
+        val data = PluginMessageCodec.encode(PluginMessage.PresenceRequest)
+        player.sendPluginMessage(plugin, CHANNEL, data)
+        logger.info("Requested initial presence snapshot from Velocity")
+    }
+
+    /**
      * Shutdown
      */
     fun shutdown() {
diff --git a/platform-paper/src/main/resources/config.yml b/platform-paper/src/main/resources/config.yml
index 1952c7f..111d90b 100644
--- a/platform-paper/src/main/resources/config.yml
+++ b/platform-paper/src/main/resources/config.yml
@@ -72,6 +72,10 @@ features:
     # If enabled, global chat messages will be shared across all Paper servers connected to the Velocity proxy.
     # Players on different servers can communicate through the GLOBAL chat mode.
     crossServerGlobalChat: false
+    # If enabled, direct messages (/tell, /msg) can be sent to players on other Paper servers
+    # by specifying "<playerName>@<serverName>" as the target. The server name must match the
+    # serverName configured on the destination Paper server.
+    crossServerDirectMessage: false
     # Server name to display in cross-server chat (e.g., "survival", "creative", "lobby").
     # This should match the server name defined in your Velocity configuration.
     serverName: "Unknown"
diff --git a/platform-paper/src/main/resources/languages/en.yml b/platform-paper/src/main/resources/languages/en.yml
index 0e781b9..5a76910 100644
--- a/platform-paper/src/main/resources/languages/en.yml
+++ b/platform-paper/src/main/resources/languages/en.yml
@@ -26,6 +26,10 @@ directMessage:
   targetOffline: "Player '{target}' is currently offline."
   yourself: "You cannot send a message to yourself."
   replyTargetNotFound: "Reply target player not found."
+  crossServerDisabled: "Cross-server direct messages are disabled on this server."
+  remoteServerNotFound: "Server '{server}' was not found on the proxy."
+  remoteTargetOffline: "Player '{target}' is not online on server '{server}'."
+  usage: "Usage: /tell <player[@server]> <message>"
 
 romajiConversion:
   toggle: "Romaji-to-kana conversion has been set to {toggle}"
diff --git a/platform-paper/src/main/resources/languages/ja.yml b/platform-paper/src/main/resources/languages/ja.yml
index f118f82..b2582c4 100644
--- a/platform-paper/src/main/resources/languages/ja.yml
+++ b/platform-paper/src/main/resources/languages/ja.yml
@@ -26,6 +26,10 @@ directMessage:
   targetOffline: "プレイヤー '{target}' は現在オフラインです"
   yourself: "自分自身にメッセージを送信することはできません"
   replyTargetNotFound: "返信対象のプレイヤーが見つかりません"
+  crossServerDisabled: "このサーバーではクロスサーバーダイレクトメッセージが無効です"
+  remoteServerNotFound: "サーバー '{server}' はプロキシ内に見つかりませんでした"
+  remoteTargetOffline: "プレイヤー '{target}' はサーバー '{server}' にオンラインではありません"
+  usage: "使い方: /tell <プレイヤー[@サーバー]> <メッセージ>"
 
 romajiConversion:
   toggle: "かな・ローマ字変換機能を{toggle}にしました"
diff --git a/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommandTest.kt b/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommandTest.kt
new file mode 100644
index 0000000..208d4bf
--- /dev/null
+++ b/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommandTest.kt
@@ -0,0 +1,107 @@
+package dev.m1sk9.lunaticChat.paper.command.impl
+
+import dev.m1sk9.lunaticChat.engine.command.CommandResult
+import dev.m1sk9.lunaticChat.paper.LunaticChat
+import dev.m1sk9.lunaticChat.paper.TestUtils
+import dev.m1sk9.lunaticChat.paper.chat.handler.DirectMessageHandler
+import dev.m1sk9.lunaticChat.paper.command.core.CommandContext
+import dev.m1sk9.lunaticChat.paper.i18n.LanguageManager
+import dev.m1sk9.lunaticChat.paper.velocity.CrossServerDirectMessageManager
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.verify
+import kotlin.test.Test
+import kotlin.test.assertIs
+
+class TellCommandTest {
+    private fun createCommand(
+        crossServerManager: CrossServerDirectMessageManager? = null,
+        localServerName: String = "lobby",
+    ): TellDeps {
+        val plugin = mockk<LunaticChat>(relaxed = true)
+        val dmHandler = mockk<DirectMessageHandler>(relaxed = true)
+        val languageManager = mockk<LanguageManager>(relaxed = true)
+        every { languageManager.getMessage(any()) } returns "msg"
+        every { languageManager.getMessage(any(), any()) } returns "msg"
+
+        val sender = TestUtils.createMockPlayer(name = "Alice")
+        val ctx = mockk<CommandContext>(relaxed = true)
+        every { ctx.requirePlayer() } returns sender
+
+        val command = TellCommand(plugin, dmHandler, languageManager, crossServerManager, null, localServerName)
+        return TellDeps(command, ctx, dmHandler, crossServerManager, sender)
+    }
+
+    private data class TellDeps(
+        val command: TellCommand,
+        val ctx: CommandContext,
+        val dmHandler: DirectMessageHandler,
+        val crossServerManager: CrossServerDirectMessageManager?,
+        val sender: org.bukkit.entity.Player,
+    )
+
+    @Test
+    fun `cross-server target fails when cross-server DM is disabled`() {
+        val deps = createCommand(crossServerManager = null)
+
+        val result = deps.command.execute(deps.ctx, "Bob@survival", "hello")
+
+        assertIs<CommandResult.Failure>(result)
+    }
+
+    @Test
+    fun `cross-server target delegates to the manager when enabled`() {
+        val manager = mockk<CrossServerDirectMessageManager>(relaxed = true)
+        val deps = createCommand(crossServerManager = manager)
+
+        val result = deps.command.execute(deps.ctx, "Bob@survival", "hello")
+
+        assertIs<CommandResult.Success>(result)
+        verify { manager.sendCrossServerMessage(deps.sender, "Bob", "survival", "hello") }
+    }
+
+    @Test
+    fun `cross-server target to self on local server is rejected`() {
+        val manager = mockk<CrossServerDirectMessageManager>(relaxed = true)
+        val deps = createCommand(crossServerManager = manager, localServerName = "lobby")
+
+        // Sender is "Alice", local server is "lobby"
+        val result = deps.command.execute(deps.ctx, "Alice@lobby", "hello")
+
+        assertIs<CommandResult.Failure>(result)
+        verify(exactly = 0) { manager.sendCrossServerMessage(any(), any(), any(), any()) }
+    }
+
+    @Test
+    fun `cross-server target with empty server part fails`() {
+        val manager = mockk<CrossServerDirectMessageManager>(relaxed = true)
+        val deps = createCommand(crossServerManager = manager)
+
+        val result = deps.command.execute(deps.ctx, "Bob@", "hello")
+
+        assertIs<CommandResult.Failure>(result)
+        verify(exactly = 0) { manager.sendCrossServerMessage(any(), any(), any(), any()) }
+    }
+
+    @Test
+    fun `parseAndExecute splits target and message on first whitespace`() {
+        val manager = mockk<CrossServerDirectMessageManager>(relaxed = true)
+        val deps = createCommand(crossServerManager = manager)
+
+        val result = deps.command.parseAndExecute(deps.ctx, "Bob@survival hello there")
+
+        assertIs<CommandResult.Success>(result)
+        verify { manager.sendCrossServerMessage(deps.sender, "Bob", "survival", "hello there") }
+    }
+
+    @Test
+    fun `parseAndExecute without a message returns usage failure`() {
+        val manager = mockk<CrossServerDirectMessageManager>(relaxed = true)
+        val deps = createCommand(crossServerManager = manager)
+
+        val result = deps.command.parseAndExecute(deps.ctx, "Bob@survival")
+
+        assertIs<CommandResult.Failure>(result)
+        verify(exactly = 0) { manager.sendCrossServerMessage(any(), any(), any(), any()) }
+    }
+}
diff --git a/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/velocity/RemotePlayerRegistryTest.kt b/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/velocity/RemotePlayerRegistryTest.kt
new file mode 100644
index 0000000..e4737d6
--- /dev/null
+++ b/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/velocity/RemotePlayerRegistryTest.kt
@@ -0,0 +1,82 @@
+package dev.m1sk9.lunaticChat.paper.velocity
+
+import dev.m1sk9.lunaticChat.engine.protocol.PresenceEntry
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+
+class RemotePlayerRegistryTest {
+    @Test
+    fun `serverOf returns the server for a known player case-insensitively`() {
+        val registry = RemotePlayerRegistry(localServerName = "lobby")
+        registry.replaceAll(listOf(PresenceEntry("Alice", "survival")))
+
+        assertEquals("survival", registry.serverOf("Alice"))
+        assertEquals("survival", registry.serverOf("alice"))
+    }
+
+    @Test
+    fun `serverOf returns null for an unknown player`() {
+        val registry = RemotePlayerRegistry(localServerName = "lobby")
+        registry.replaceAll(listOf(PresenceEntry("Alice", "survival")))
+
+        assertNull(registry.serverOf("Bob"))
+    }
+
+    @Test
+    fun `remotePlayers excludes players on the local server`() {
+        val registry = RemotePlayerRegistry(localServerName = "lobby")
+        registry.replaceAll(
+            listOf(
+                PresenceEntry("Alice", "survival"),
+                PresenceEntry("Bob", "lobby"),
+                PresenceEntry("Carol", "creative"),
+            ),
+        )
+
+        val remoteNames = registry.remotePlayers().map { it.playerName }.toSet()
+        assertEquals(setOf("Alice", "Carol"), remoteNames)
+    }
+
+    @Test
+    fun `remotePlayers preserves original-cased names`() {
+        val registry = RemotePlayerRegistry(localServerName = "lobby")
+        registry.replaceAll(listOf(PresenceEntry("AliceCased", "survival")))
+
+        assertEquals("AliceCased", registry.remotePlayers().single().playerName)
+    }
+
+    @Test
+    fun `replaceAll replaces the entire roster`() {
+        val registry = RemotePlayerRegistry(localServerName = "lobby")
+        registry.replaceAll(listOf(PresenceEntry("Alice", "survival")))
+        registry.replaceAll(listOf(PresenceEntry("Bob", "creative")))
+
+        assertNull(registry.serverOf("Alice"))
+        assertEquals("creative", registry.serverOf("Bob"))
+    }
+
+    @Test
+    fun `clear empties the roster`() {
+        val registry = RemotePlayerRegistry(localServerName = "lobby")
+        registry.replaceAll(listOf(PresenceEntry("Alice", "survival")))
+        registry.clear()
+
+        assertNull(registry.serverOf("Alice"))
+        assertTrue(registry.remotePlayers().isEmpty())
+    }
+
+    @Test
+    fun `local server name match is case-insensitive`() {
+        val registry = RemotePlayerRegistry(localServerName = "Lobby")
+        registry.replaceAll(
+            listOf(
+                PresenceEntry("Alice", "lobby"),
+                PresenceEntry("Bob", "survival"),
+            ),
+        )
+
+        assertEquals(listOf("Bob"), registry.remotePlayers().map { it.playerName })
+    }
+}
diff --git a/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/LunaticChat.kt b/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/LunaticChat.kt
index be5a4cd..df62b2f 100644
--- a/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/LunaticChat.kt
+++ b/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/LunaticChat.kt
@@ -8,7 +8,9 @@ import com.velocitypowered.api.plugin.Plugin
 import com.velocitypowered.api.plugin.PluginContainer
 import com.velocitypowered.api.proxy.ProxyServer
 import dev.m1sk9.lunaticChat.velocity.messaging.CrossServerChatRelay
+import dev.m1sk9.lunaticChat.velocity.messaging.CrossServerDirectMessageRelay
 import dev.m1sk9.lunaticChat.velocity.messaging.PluginMessageHandler
+import dev.m1sk9.lunaticChat.velocity.presence.PresenceTracker
 import org.slf4j.Logger
 
 /**
@@ -36,6 +38,8 @@ class LunaticChat
     ) {
         private var messageHandler: PluginMessageHandler? = null
         private var crossServerChatRelay: CrossServerChatRelay? = null
+        private var crossServerDirectMessageRelay: CrossServerDirectMessageRelay? = null
+        private var presenceTracker: PresenceTracker? = null
 
         @Subscribe
         fun onProxyInitialization(event: ProxyInitializeEvent) {
@@ -54,6 +58,22 @@ class LunaticChat
                     logger = logger,
                 )
 
+            // Initialize cross-server direct message relay
+            crossServerDirectMessageRelay =
+                CrossServerDirectMessageRelay(
+                    server = server,
+                    logger = logger,
+                )
+
+            // Initialize presence tracker
+            presenceTracker =
+                PresenceTracker(
+                    plugin = this@LunaticChat,
+                    server = server,
+                    logger = logger,
+                )
+            presenceTracker?.initialize()
+
             // Initialize plugin message handler
             messageHandler =
                 PluginMessageHandler(
@@ -62,6 +82,8 @@ class LunaticChat
                     logger = logger,
                     pluginVersion = pluginVersion,
                     crossServerChatRelay = crossServerChatRelay!!,
+                    crossServerDirectMessageRelay = crossServerDirectMessageRelay!!,
+                    presenceTracker = presenceTracker!!,
                 )
             messageHandler?.initialize()
 
@@ -72,5 +94,6 @@ class LunaticChat
         fun onProxyShutdown(event: ProxyShutdownEvent) {
             logger.info("Shutting down LunaticChat Velocity plugin")
             messageHandler?.shutdown()
+            presenceTracker?.shutdown()
         }
     }
diff --git a/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/CrossServerDirectMessageRelay.kt b/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/CrossServerDirectMessageRelay.kt
new file mode 100644
index 0000000..89cc15f
--- /dev/null
+++ b/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/CrossServerDirectMessageRelay.kt
@@ -0,0 +1,88 @@
+package dev.m1sk9.lunaticChat.velocity.messaging
+
+import com.velocitypowered.api.proxy.ProxyServer
+import com.velocitypowered.api.proxy.messages.MinecraftChannelIdentifier
+import com.velocitypowered.api.proxy.server.RegisteredServer
+import dev.m1sk9.lunaticChat.engine.protocol.PluginMessage
+import dev.m1sk9.lunaticChat.engine.protocol.PluginMessageCodec
+import org.slf4j.Logger
+
+/**
+ * Routes cross-server direct messages to a single target server.
+ *
+ * Unlike [CrossServerChatRelay] which broadcasts, this resolves the requested
+ * target server and player, then forwards the message only to that server.
+ * On failure it sends a [PluginMessage.DirectMessageError] back to the source.
+ */
+class CrossServerDirectMessageRelay(
+    private val server: ProxyServer,
+    private val logger: Logger,
+) {
+    companion object {
+        private val CHANNEL = MinecraftChannelIdentifier.create("lunaticchat", "main")
+    }
+
+    /**
+     * Relays a direct message to the target server, or returns an error to the source.
+     *
+     * @param message Direct message relay to route
+     * @param sourceServer The server that sent the message
+     */
+    fun relay(
+        message: PluginMessage.DirectMessageRelay,
+        sourceServer: RegisteredServer,
+    ) {
+        try {
+            val targetServer =
+                server.allServers.firstOrNull { it.serverInfo.name == message.targetServerName }
+            if (targetServer == null) {
+                logger.info(
+                    "Direct message target server not found: ${message.targetServerName} " +
+                        "(messageId=${message.messageId})",
+                )
+                sendError(sourceServer, message, PluginMessage.DirectMessageError.Reason.SERVER_NOT_FOUND)
+                return
+            }
+
+            val targetPlayer = server.getPlayer(message.targetName).orElse(null)
+            val onTargetServer =
+                targetPlayer
+                    ?.currentServer
+                    ?.orElse(null)
+                    ?.serverInfo
+                    ?.name == message.targetServerName
+            if (targetPlayer == null || !onTargetServer) {
+                logger.info(
+                    "Direct message target offline or on different server: ${message.targetName}@${message.targetServerName} " +
+                        "(messageId=${message.messageId})",
+                )
+                sendError(sourceServer, message, PluginMessage.DirectMessageError.Reason.TARGET_OFFLINE)
+                return
+            }
+
+            targetServer.sendPluginMessage(CHANNEL, PluginMessageCodec.encode(message))
+            logger.info(
+                "Relayed direct message from ${message.senderName}@${message.sourceServerName} " +
+                    "to ${message.targetName}@${message.targetServerName} (messageId=${message.messageId})",
+            )
+        } catch (e: Exception) {
+            logger.error("Failed to relay direct message: ${e.message}", e)
+        }
+    }
+
+    private fun sendError(
+        sourceServer: RegisteredServer,
+        message: PluginMessage.DirectMessageRelay,
+        reason: String,
+    ) {
+        val error =
+            PluginMessage.DirectMessageError(
+                messageId = message.messageId,
+                senderId = message.senderId,
+                targetName = message.targetName,
+                targetServerName = message.targetServerName,
+                reason = reason,
+            )
+        sourceServer.sendPluginMessage(CHANNEL, PluginMessageCodec.encode(error))
+    }
+}
diff --git a/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/PluginMessageHandler.kt b/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/PluginMessageHandler.kt
index 382eba4..4472d1c 100644
--- a/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/PluginMessageHandler.kt
+++ b/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/PluginMessageHandler.kt
@@ -8,6 +8,7 @@ import com.velocitypowered.api.proxy.messages.MinecraftChannelIdentifier
 import dev.m1sk9.lunaticChat.engine.protocol.PluginMessage
 import dev.m1sk9.lunaticChat.engine.protocol.PluginMessageCodec
 import dev.m1sk9.lunaticChat.engine.protocol.ProtocolVersion
+import dev.m1sk9.lunaticChat.velocity.presence.PresenceTracker
 import org.slf4j.Logger
 
 /**
@@ -25,6 +26,8 @@ class PluginMessageHandler(
     private val logger: Logger,
     private val pluginVersion: String,
     private val crossServerChatRelay: CrossServerChatRelay,
+    private val crossServerDirectMessageRelay: CrossServerDirectMessageRelay,
+    private val presenceTracker: PresenceTracker,
 ) {
     companion object {
         private val CHANNEL = MinecraftChannelIdentifier.create("lunaticchat", "main")
@@ -63,6 +66,12 @@ class PluginMessageHandler(
                 is PluginMessage.GlobalChatMessage -> {
                     handleGlobalChatMessage(source, message)
                 }
+                is PluginMessage.DirectMessageRelay -> {
+                    crossServerDirectMessageRelay.relay(message, source.server)
+                }
+                is PluginMessage.PresenceRequest -> {
+                    presenceTracker.sendSnapshotTo(source.server)
+                }
                 else -> {
                     logger.warn("Unexpected message type from Paper: ${message::class.simpleName}")
                 }
diff --git a/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/presence/PresenceTracker.kt b/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/presence/PresenceTracker.kt
new file mode 100644
index 0000000..f9442ee
--- /dev/null
+++ b/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/presence/PresenceTracker.kt
@@ -0,0 +1,98 @@
+package dev.m1sk9.lunaticChat.velocity.presence
+
+import com.velocitypowered.api.event.Subscribe
+import com.velocitypowered.api.event.connection.DisconnectEvent
+import com.velocitypowered.api.event.player.ServerPostConnectEvent
+import com.velocitypowered.api.proxy.Player
+import com.velocitypowered.api.proxy.ProxyServer
+import com.velocitypowered.api.proxy.messages.MinecraftChannelIdentifier
+import com.velocitypowered.api.proxy.server.RegisteredServer
+import dev.m1sk9.lunaticChat.engine.protocol.PluginMessage
+import dev.m1sk9.lunaticChat.engine.protocol.PluginMessageCodec
+import dev.m1sk9.lunaticChat.engine.protocol.PresenceEntry
+import org.slf4j.Logger
+
+/**
+ * Tracks proxy-wide player presence and pushes snapshots to Paper servers.
+ *
+ * Velocity is the source of truth for presence. Paper servers keep a cache that
+ * is replaced wholesale on each [PluginMessage.PresenceSnapshot].
+ */
+class PresenceTracker(
+    /**
+     * Plugin instance for event registration. Type is [Any] because Velocity's
+     * EventManager.register() accepts Object (mirrors [PluginMessageHandler]).
+     */
+    private val plugin: Any,
+    private val server: ProxyServer,
+    private val logger: Logger,
+) {
+    companion object {
+        private val CHANNEL = MinecraftChannelIdentifier.create("lunaticchat", "main")
+    }
+
+    /**
+     * Registers presence event listeners.
+     */
+    fun initialize() {
+        server.eventManager.register(plugin, this)
+        logger.info("Presence tracker registered")
+    }
+
+    /**
+     * Broadcasts the current snapshot when a player joins or switches servers.
+     */
+    @Subscribe
+    fun onServerPostConnect(event: ServerPostConnectEvent) {
+        broadcastSnapshot()
+    }
+
+    /**
+     * Broadcasts the current snapshot when a player disconnects, excluding them.
+     */
+    @Subscribe
+    fun onDisconnect(event: DisconnectEvent) {
+        broadcastSnapshot(exclude = event.player)
+    }
+
+    /**
+     * Unregisters presence event listeners.
+     */
+    fun shutdown() {
+        server.eventManager.unregisterListener(plugin, this)
+        logger.info("Presence tracker unregistered")
+    }
+
+    /**
+     * Sends the current snapshot to a single server (initial sync on request).
+     */
+    fun sendSnapshotTo(target: RegisteredServer) {
+        val data = PluginMessageCodec.encode(buildSnapshot())
+        target.sendPluginMessage(CHANNEL, data)
+    }
+
+    private fun broadcastSnapshot(exclude: Player? = null) {
+        try {
+            val data = PluginMessageCodec.encode(buildSnapshot(exclude))
+            server.allServers.forEach { it.sendPluginMessage(CHANNEL, data) }
+        } catch (e: Exception) {
+            logger.error("Failed to broadcast presence snapshot: ${e.message}", e)
+        }
+    }
+
+    private fun buildSnapshot(exclude: Player? = null): PluginMessage.PresenceSnapshot {
+        val entries =
+            server.allPlayers
+                .asSequence()
+                .filter { exclude == null || it.uniqueId != exclude.uniqueId }
+                .mapNotNull { player ->
+                    val serverName =
+                        player.currentServer
+                            .orElse(null)
+                            ?.serverInfo
+                            ?.name ?: return@mapNotNull null
+                    PresenceEntry(playerName = player.username, serverName = serverName)
+                }.toList()
+        return PluginMessage.PresenceSnapshot(players = entries)
+    }
+}
diff --git a/platform-velocity/src/test/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/CrossServerDirectMessageRelayTest.kt b/platform-velocity/src/test/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/CrossServerDirectMessageRelayTest.kt
new file mode 100644
index 0000000..afbf6ca
--- /dev/null
+++ b/platform-velocity/src/test/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/CrossServerDirectMessageRelayTest.kt
@@ -0,0 +1,116 @@
+package dev.m1sk9.lunaticChat.velocity.messaging
+
+import com.velocitypowered.api.proxy.Player
+import com.velocitypowered.api.proxy.ProxyServer
+import com.velocitypowered.api.proxy.ServerConnection
+import com.velocitypowered.api.proxy.messages.ChannelIdentifier
+import com.velocitypowered.api.proxy.server.RegisteredServer
+import com.velocitypowered.api.proxy.server.ServerInfo
+import dev.m1sk9.lunaticChat.engine.protocol.PluginMessage
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.verify
+import org.slf4j.Logger
+import java.util.Optional
+import kotlin.test.Test
+
+class CrossServerDirectMessageRelayTest {
+    private fun createRelay(): Pair<CrossServerDirectMessageRelay, ProxyServer> {
+        val server = mockk<ProxyServer>(relaxed = true)
+        val logger = mockk<Logger>(relaxed = true)
+        val relay = CrossServerDirectMessageRelay(server, logger)
+        return relay to server
+    }
+
+    private fun createRegisteredServer(name: String): RegisteredServer {
+        val server = mockk<RegisteredServer>(relaxed = true)
+        val serverInfo = mockk<ServerInfo>(relaxed = true)
+        every { serverInfo.name } returns name
+        every { server.serverInfo } returns serverInfo
+        return server
+    }
+
+    private fun createPlayer(currentServerName: String?): Player {
+        val player = mockk<Player>(relaxed = true)
+        if (currentServerName == null) {
+            every { player.currentServer } returns Optional.empty()
+        } else {
+            val connection = mockk<ServerConnection>(relaxed = true)
+            val info = mockk<ServerInfo>(relaxed = true)
+            every { info.name } returns currentServerName
+            every { connection.serverInfo } returns info
+            every { player.currentServer } returns Optional.of(connection)
+        }
+        return player
+    }
+
+    private fun createMessage(
+        targetName: String = "Recipient",
+        targetServerName: String = "survival",
+    ): PluginMessage.DirectMessageRelay =
+        PluginMessage.DirectMessageRelay(
+            messageId = "dm-1",
+            sourceServerName = "lobby",
+            senderId = "00000001-0000-0000-0000-000000000000",
+            senderName = "Sender",
+            targetServerName = targetServerName,
+            targetName = targetName,
+            message = "Hello!",
+            timestamp = 1000L,
+        )
+
+    @Test
+    fun `relay should forward to target server only`() {
+        val (relay, proxyServer) = createRelay()
+        val sourceServer = createRegisteredServer("lobby")
+        val targetServer = createRegisteredServer("survival")
+        every { proxyServer.allServers } returns listOf(sourceServer, targetServer)
+        every { proxyServer.getPlayer("Recipient") } returns Optional.of(createPlayer("survival"))
+
+        relay.relay(createMessage(), sourceServer)
+
+        verify(exactly = 1) { targetServer.sendPluginMessage(any<ChannelIdentifier>(), any<ByteArray>()) }
+        verify(exactly = 0) { sourceServer.sendPluginMessage(any<ChannelIdentifier>(), any<ByteArray>()) }
+    }
+
+    @Test
+    fun `relay should return SERVER_NOT_FOUND error when target server missing`() {
+        val (relay, proxyServer) = createRelay()
+        val sourceServer = createRegisteredServer("lobby")
+        every { proxyServer.allServers } returns listOf(sourceServer)
+
+        relay.relay(createMessage(targetServerName = "ghost"), sourceServer)
+
+        // Error returned to source; nothing relayed to a target
+        verify(exactly = 1) { sourceServer.sendPluginMessage(any<ChannelIdentifier>(), any<ByteArray>()) }
+    }
+
+    @Test
+    fun `relay should return TARGET_OFFLINE error when player not online`() {
+        val (relay, proxyServer) = createRelay()
+        val sourceServer = createRegisteredServer("lobby")
+        val targetServer = createRegisteredServer("survival")
+        every { proxyServer.allServers } returns listOf(sourceServer, targetServer)
+        every { proxyServer.getPlayer("Recipient") } returns Optional.empty()
+
+        relay.relay(createMessage(), sourceServer)
+
+        verify(exactly = 1) { sourceServer.sendPluginMessage(any<ChannelIdentifier>(), any<ByteArray>()) }
+        verify(exactly = 0) { targetServer.sendPluginMessage(any<ChannelIdentifier>(), any<ByteArray>()) }
+    }
+
+    @Test
+    fun `relay should return TARGET_OFFLINE error when player on a different server`() {
+        val (relay, proxyServer) = createRelay()
+        val sourceServer = createRegisteredServer("lobby")
+        val targetServer = createRegisteredServer("survival")
+        every { proxyServer.allServers } returns listOf(sourceServer, targetServer)
+        // Player exists but is connected to "creative", not the requested "survival"
+        every { proxyServer.getPlayer("Recipient") } returns Optional.of(createPlayer("creative"))
+
+        relay.relay(createMessage(), sourceServer)
+
+        verify(exactly = 1) { sourceServer.sendPluginMessage(any<ChannelIdentifier>(), any<ByteArray>()) }
+        verify(exactly = 0) { targetServer.sendPluginMessage(any<ChannelIdentifier>(), any<ByteArray>()) }
+    }
+}
diff --git a/platform-velocity/src/test/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/PluginMessageHandlerTest.kt b/platform-velocity/src/test/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/PluginMessageHandlerTest.kt
index 3342331..ed564eb 100644
--- a/platform-velocity/src/test/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/PluginMessageHandlerTest.kt
+++ b/platform-velocity/src/test/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/PluginMessageHandlerTest.kt
@@ -12,6 +12,7 @@ import com.velocitypowered.api.proxy.server.ServerInfo
 import dev.m1sk9.lunaticChat.engine.protocol.PluginMessage
 import dev.m1sk9.lunaticChat.engine.protocol.PluginMessageCodec
 import dev.m1sk9.lunaticChat.engine.protocol.ProtocolVersion
+import dev.m1sk9.lunaticChat.velocity.presence.PresenceTracker
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
@@ -21,14 +22,24 @@ import kotlin.test.Test
 class PluginMessageHandlerTest {
     private val channel = MinecraftChannelIdentifier.create("lunaticchat", "main")
 
-    private fun createHandler(pluginVersion: String = "0.10.0"): Triple<PluginMessageHandler, ProxyServer, CrossServerChatRelay> {
+    private data class Handlers(
+        val handler: PluginMessageHandler,
+        val server: ProxyServer,
+        val relay: CrossServerChatRelay,
+        val dmRelay: CrossServerDirectMessageRelay,
+        val presenceTracker: PresenceTracker,
+    )
+
+    private fun createHandler(pluginVersion: String = "0.10.0"): Handlers {
         val plugin = Any()
         val server = mockk<ProxyServer>(relaxed = true)
         val logger = mockk<Logger>(relaxed = true)
         val relay = mockk<CrossServerChatRelay>(relaxed = true)
+        val dmRelay = mockk<CrossServerDirectMessageRelay>(relaxed = true)
+        val presenceTracker = mockk<PresenceTracker>(relaxed = true)
 
-        val handler = PluginMessageHandler(plugin, server, logger, pluginVersion, relay)
-        return Triple(handler, server, relay)
+        val handler = PluginMessageHandler(plugin, server, logger, pluginVersion, relay, dmRelay, presenceTracker)
+        return Handlers(handler, server, relay, dmRelay, presenceTracker)
     }
 
     private fun createServerConnection(serverName: String = "lobby"): ServerConnection {
@@ -145,6 +156,43 @@ class PluginMessageHandlerTest {
     }
 
     @Test
+    fun `onPluginMessage should relay direct message`() {
+        val handlers = createHandler()
+        val connection = createServerConnection()
+
+        val dm =
+            PluginMessage.DirectMessageRelay(
+                messageId = "dm-1",
+                sourceServerName = "lobby",
+                senderId = "00000001-0000-0000-0000-000000000000",
+                senderName = "Sender",
+                targetServerName = "survival",
+                targetName = "Recipient",
+                message = "Hi!",
+                timestamp = 1000L,
+            )
+        val data = PluginMessageCodec.encode(dm)
+        val event = createPluginMessageEvent(connection, mockk(relaxed = true), channel, data)
+
+        handlers.handler.onPluginMessage(event)
+
+        verify { handlers.dmRelay.relay(any<PluginMessage.DirectMessageRelay>(), any<RegisteredServer>()) }
+    }
+
+    @Test
+    fun `onPluginMessage should send presence snapshot on request`() {
+        val handlers = createHandler()
+        val connection = createServerConnection()
+
+        val data = PluginMessageCodec.encode(PluginMessage.PresenceRequest)
+        val event = createPluginMessageEvent(connection, mockk(relaxed = true), channel, data)
+
+        handlers.handler.onPluginMessage(event)
+
+        verify { handlers.presenceTracker.sendSnapshotTo(any<RegisteredServer>()) }
+    }
+
+    @Test
     fun `onPluginMessage should ignore messages from wrong channel`() {
         val (handler, _, relay) = createHandler()
         val connection = createServerConnection()