Add plugin player commands and harden proxy ingress

2026-10-25 18:46:18 +00:00 · 2026-09-25 22:18:48 +00:00
1 parent f052712679
commit 72086ccb5d
@@ -6,7 +6,7 @@ StrataProxy 是面向 Minecraft 1.7.10 Forge 的玩家会话代理。项目正

 - **玩家会话**：代理负责入服、连接后端、转发、切换后端和断线清理。玩家身份包含 UUID 与连接代次，避免旧连接的异步结果作用到新连接。在线玩家视图由 Session 持有，监听器只保留按 UUID 查找会话的索引。
 - **服务器目录**：静态配置与插件调用同一个注册接口。`ServerView` 只包含名称、地址、标签和插件元数据。目录不采集或暴露后端负载、连接占用和容量；玩法插件自行获取选服所需的数据。同名重注册生成新句柄，旧异步结果不能修改新条目。一次连接使用选定时的地址快照，后续注册更新只影响新选择。
- **插件 API**：插件可查询 `PlayerView`、`ServerView`，动态注册后端，并以一个异步回调决定初始落点。空岛分配、实例唤醒和玩法数据由插件自行实现。单次选服失败或超时只结束当前请求，不停用插件；玩家断线会取消未决请求并移除尚未执行的回调，已启动的插件异步任务只做尽力取消。宿主关闭时会结束未决选服请求并撤销插件注册。插件返回的拒绝原因会作为登录断开消息发给玩家。
+- **插件 API**：插件可查询 `PlayerView`、`ServerView`，动态注册后端、注册玩家命令，并以一个异步回调决定初始落点。空岛分配、实例唤醒和玩法数据由插件自行实现。单次选服失败或超时只结束当前请求，不停用插件；玩家断线会取消未决请求并移除尚未执行的回调，已启动的插件异步任务只做尽力取消。宿主关闭时会结束未决选服请求并撤销插件注册。插件返回的拒绝原因会作为登录断开消息发给玩家。
 - **发现方式**：DNS 是独立插件，核心不认识 DNS 或 Agent。配置只启用列出的插件；插件关闭时其注册会清理。普通 DNS 主机名通过独立解析器查询 A 与 AAAA 地址，不使用 JVM 的进程级地址缓存；`localhost` 按系统回环地址解析。最小默认选路按注册顺序选取后端，因此 DNS 插件先注册 IPv4 地址。一次查询失败会保留已有注册，连续三次失败会移除旧地址，恢复解析后重新注册。
 - **协议热路径**：同一玩家的前端和后端通道绑定同一个 Netty 事件循环，会话控制状态在该循环上串行更新。会话始终保留协议 5 的帧边界，普通数据包不重新编码并尽量复用 Netty `ByteBuf`；Keep Alive ID 在会话内转换，可同时跟踪多个未回复请求，转服后旧后端的迟到回复会被丢弃；转服后只改写少数与玩家实体 ID 相关的包，并按目标通道可写状态控制读取。后端主机名由 Netty 异步解析，避免把 DNS 等待放在玩家 I/O 线程上。
 - **日志**：SLF4J API 与 Logback 运行时。
@@ -23,6 +23,8 @@ StrataProxy 是面向 Minecraft 1.7.10 Forge 的玩家会话代理。项目正

 默认配置：`proxy-core/src/main/resources/config/strataproxy.yml`。安装包包含静态后端示例和 `strataproxy-dns.example.yml`；后者通过 `plugins.enabled` 启用 DNS 插件，`backends: []` 表示不使用静态后端。相对插件目录从配置文件所在目录解析；安装包示例中的 `../plugins` 指向同一安装包的插件目录。`ONLINE_BUNGEE` 只能连接已启用旧版 Bungee 身份转发、且限制直连的可信后端。

+安装包示例使用 `OFFLINE`，因此只监听 `127.0.0.1`，适合本机试运行。离线模式信任客户端提交的玩家名；未开启公开监听选项时必须使用字面量回环 IP，避免主机名解析变化后绑定到公网地址。对外服务应选择 `ONLINE_BUNGEE` 并禁止玩家直连后端。确实需要在受控网络中公开离线监听时，必须显式配置 `allowOfflinePublicAccess: true`。`maxConnections` 默认 4096，限制包括尚未登录的所有客户端连接；达到上限的新连接会直接关闭。部署时按内存、文件描述符与预期在线人数调整。
+
 `--validate-config` 检查核心配置和静态后端注册；启用插件的自定义设置由插件在启动时检查。

 `plugins.initialPlacementTimeoutSeconds` 控制初始选服回调的等待时间，默认 15 秒，可配置为 1–120 秒。握手与身份验证、后端连接与登录各有独立的 15 秒期限；选服阶段不会消耗这两个阶段的时间。后端 Login Success 后，客户端必须在 2 分钟内完成初次 PLAY/Forge 握手，否则代理断开会话。插件可以异步查询或准备后端；超过时限的当前玩家登录会失败，后续玩家仍可调用该插件。客户端和后端自身也可能提前断开等待中的连接。
@@ -31,6 +33,20 @@ StrataProxy 是面向 Minecraft 1.7.10 Forge 的玩家会话代理。项目正

 在线模式最多同时保留 256 个 Mojang 会话校验请求（包括等待响应和解析的请求）；超过上限的新登录会立即失败，避免请求与解析任务无限积压。此限制只约束代理的认证资源，与后端负载无关。

+### 插件玩家命令
+
+插件在 `onLoad` 或 `onEnable` 中调用 `context.commands().register("where", handler)` 注册根命令。名称不区分大小写，并在所有插件之间唯一；需要别名时分别注册。处理器收到 `CommandInvocation`，可读取 `player()`、`name()`、未拆词的 `arguments()`，并通过 `reply(String)` 给该玩家发送纯文本聊天消息。处理器在有界插件工作线程上执行，可以自行查询数据库；需要稍后回复时可保留本次 invocation。插件停用后注册自动撤销，返回的回复也会被忽略。
+
+```java
+context.commands().register("where", command -> {
+    command.reply("当前服务器：" + command.player().currentServer().orElse("未连接"));
+});
+```
+
+代理只消费已注册的斜杠命令；普通聊天和未知命令按原字节转发给后端。每个玩家的代理命令最多突发 10 次，随后每 200 毫秒恢复一次额度；超额命令被消费，限速提示最多每两秒发送一次。命令不提供内置权限判断，插件必须在处理器中根据自己的权限来源检查玩家身份。回调异常会记录日志并向玩家发送通用失败消息；工作队列已满时会立即告知玩家稍后重试。单条回复最多 1024 个 Java 字符；玩家已断线、前端不可写或单会话待发送回复超过 64 条时会丢弃回复，避免无限排队。
+
+代理关闭时所有插件的 `onDisable` 共享 10 秒期限；超时后会中断钩子并继续清理。如果插件忽略中断，它的代码可能在资源清理后仍短暂运行，插件应让关闭钩子可及时退出。
+
 候选后端就绪后的切换阶段另有 15 秒期限。如果旧链路的写入持续未完成，代理会以失败结果结束转服并断开该会话，避免玩家与插件调用无限等待。Forge 切换包写出后，目标握手和 Join Game 最多再等待 30 秒；超时会结束玩家会话并返回失败。

 ```powershell
@@ -115,6 +131,8 @@ PR 的 `.gitea/workflows/verify.yml` 独立运行 `gradlew check`，不使用 Ma

 当前会话路径的 [JFR 采样记录](benchmarks/results/2026-09-25-session-jfr-screening.md) 只提供了分配线索，代理 I/O 线程的 CPU 执行样本不足以定位热点；因此没有据此改动普通转发路径。

+[玩家命令拦截初筛](benchmarks/results/2026-09-26-command-interceptor.md) 比较同一合成 PLAY 回声流量下启用与关闭命令拦截器的结果；三轮吞吐和延迟范围重叠。该实验没有发送命令，也不能替代目标整合包的容量验收。
+
 将 `-Window` 改为 `16` 可测每连接最多 16 个在途往返的情形。

 原 stop-and-wait 小样本见 `benchmarks/results/proxy-session-benchmark-smoke-2026-09-25.md`；Window 参数的小样本和 Window=1/16 重复测量见 `benchmarks/results/2026-09-25-proxy-session-window.md`。这些是环回网络上的合成帧对照，不能代表 Forge 整合包、真实后端或跨主机部署的性能。
@@ -46,11 +46,13 @@ public final class ProxySessionBenchmark {
        int payload = intArg(args, "payload", 1024);
        int repeats = intArg(args, "repeats", 2);
        int window = intArg(args, "window", 1);
+        int commandInterceptor = intArg(args, "command-interceptor", 1);
        String mode = modeArg(args);
        boolean afterTransfer = mode.equals("post-transfer");
        if (connections < 1 || connections > 32 || messages < 1 || messages > 100_000
                || warmup < 0 || warmup > 10_000 || payload < 5 || payload > 1_048_576
-                || repeats < 1 || repeats > 10 || window < 1 || window > 1024) {
+                || repeats < 1 || repeats > 10 || window < 1 || window > 1024
+                || (commandInterceptor != 0 && commandInterceptor != 1)) {
            throw new IllegalArgumentException("bounds: connections 1..32, messages 1..100000, warmup 0..10000, "
                    + "payload 5..1048576, repeats 1..10, window 1..1024");
        }
@@ -75,13 +77,15 @@ public final class ProxySessionBenchmark {
                    new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), catalog);
            proxy.setPlacement(player -> java.util.concurrent.CompletableFuture.completedFuture(
                    Optional.of(PlacementDecision.select("bench"))));
+            if (commandInterceptor == 1) proxy.setCommandDispatcher((player, message, reply, admission) -> false);
            try {
                int proxyPort = ((InetSocketAddress) proxy.start().toCompletableFuture()
                        .get(SETUP_TIMEOUT_SECONDS, TimeUnit.SECONDS).localAddress()).getPort();
                System.out.printf("Offline synthetic Minecraft 1.7.10 PLAY echo; mode=%s payload=%d bytes frame=%d bytes "
                                + "connections=%d measured_roundtrips_per_connection=%d warmup_per_connection=%d "
-                                + "window_per_connection=%d repeats=%d%n",
-                        mode, playPayload.length, playFrame.length, connections, messages, warmup, window, repeats);
+                                + "window_per_connection=%d repeats=%d command_interceptor=%d%n",
+                        mode, playPayload.length, playFrame.length, connections, messages, warmup, window, repeats,
+                        commandInterceptor);
                System.out.println("Direct and proxy phases use the same JVM, fake backend, client code, framed payload, "
                        + "connection concurrency, and per-client in-flight window. Login/setup and warmup are excluded from timing.");
                if (afterTransfer) System.out.println("Proxy clients switch from bench to replacement before warmup; "
@@ -462,7 +466,8 @@ public final class ProxySessionBenchmark {
    private static int intArg(String[] args, String key, int fallback) {
        int found = fallback;
        boolean seen = false;
-        List<String> known = List.of("connections", "messages", "warmup", "payload", "repeats", "window", "mode");
+        List<String> known = List.of("connections", "messages", "warmup", "payload", "repeats", "window",
+                "mode", "command-interceptor");
        for (int i = 0; i < args.length; i++) {
            if (args[i].startsWith("--") && !known.contains(args[i].substring(2)))
                throw new IllegalArgumentException("unknown option " + args[i]);
@@ -0,0 +1,32 @@
+Reusing configuration cache.
+> Task :proxy-plugin-api:processResources NO-SOURCE
+> Task :proxy-core:processResources UP-TO-DATE
+> Task :dns-discovery:processResources UP-TO-DATE
+> Task :agent-discovery:processResources UP-TO-DATE
+> Task :proxy-plugin-api:compileJava UP-TO-DATE
+> Task :proxy-plugin-api:classes UP-TO-DATE
+> Task :proxy-plugin-api:jar UP-TO-DATE
+> Task :dns-discovery:compileJava UP-TO-DATE
+> Task :proxy-core:compileJava UP-TO-DATE
+> Task :dns-discovery:classes UP-TO-DATE
+> Task :agent-discovery:compileJava UP-TO-DATE
+> Task :proxy-core:classes UP-TO-DATE
+> Task :agent-discovery:classes UP-TO-DATE
+> Task :dns-discovery:jar UP-TO-DATE
+> Task :proxy-core:jar UP-TO-DATE
+> Task :agent-discovery:jar UP-TO-DATE
+> Task :proxy-core:startScripts UP-TO-DATE
+> Task :proxy-core:installDist UP-TO-DATE
+
+BUILD SUCCESSFUL in 15s
+13 actionable tasks: 13 up-to-date
+Configuration cache entry reused.
+Offline synthetic Minecraft 1.7.10 PLAY echo; mode=initial payload=4096 bytes frame=4098 bytes connections=4 measured_roundtrips_per_connection=5000 warmup_per_connection=500 window_per_connection=16 repeats=3 command_interceptor=0
+Direct and proxy phases use the same JVM, fake backend, client code, framed payload, connection concurrency, and per-client in-flight window. Login/setup and warmup are excluded from timing.
+No Forge, real pack, or real client is involved.
+direct-1: roundtrips=20000 echoes=20000 correct=20000 errors=0 roundtrips/s=110546.8 warmup_roundtrips=2000 window=16 one_way_MiB/s=431.82 latency_ms_p50=0.511 p95=0.729 p99=0.824 elapsed_ms=180.919
+proxy-1: roundtrips=20000 echoes=20000 correct=20000 errors=0 roundtrips/s=73819.8 warmup_roundtrips=2000 window=16 one_way_MiB/s=288.36 latency_ms_p50=0.744 p95=1.450 p99=2.075 elapsed_ms=270.930
+proxy-2: roundtrips=20000 echoes=20000 correct=20000 errors=0 roundtrips/s=84262.3 warmup_roundtrips=2000 window=16 one_way_MiB/s=329.15 latency_ms_p50=0.645 p95=1.241 p99=2.418 elapsed_ms=237.354
+direct-2: roundtrips=20000 echoes=20000 correct=20000 errors=0 roundtrips/s=120449.1 warmup_roundtrips=2000 window=16 one_way_MiB/s=470.50 latency_ms_p50=0.489 p95=0.700 p99=0.753 elapsed_ms=166.045
+direct-3: roundtrips=20000 echoes=20000 correct=20000 errors=0 roundtrips/s=111637.3 warmup_roundtrips=2000 window=16 one_way_MiB/s=436.08 latency_ms_p50=0.514 p95=0.711 p99=0.801 elapsed_ms=179.152
+proxy-3: roundtrips=20000 echoes=20000 correct=20000 errors=0 roundtrips/s=96942.4 warmup_roundtrips=2000 window=16 one_way_MiB/s=378.68 latency_ms_p50=0.627 p95=0.768 p99=0.926 elapsed_ms=206.308
@@ -0,0 +1,32 @@
+Reusing configuration cache.
+> Task :proxy-plugin-api:processResources NO-SOURCE
+> Task :dns-discovery:processResources UP-TO-DATE
+> Task :agent-discovery:processResources UP-TO-DATE
+> Task :proxy-core:processResources UP-TO-DATE
+> Task :proxy-plugin-api:compileJava UP-TO-DATE
+> Task :proxy-plugin-api:classes UP-TO-DATE
+> Task :agent-discovery:compileJava UP-TO-DATE
+> Task :agent-discovery:classes UP-TO-DATE
+> Task :proxy-plugin-api:jar UP-TO-DATE
+> Task :dns-discovery:compileJava UP-TO-DATE
+> Task :proxy-core:compileJava UP-TO-DATE
+> Task :proxy-core:classes UP-TO-DATE
+> Task :dns-discovery:classes UP-TO-DATE
+> Task :agent-discovery:jar UP-TO-DATE
+> Task :proxy-core:jar UP-TO-DATE
+> Task :dns-discovery:jar UP-TO-DATE
+> Task :proxy-core:startScripts UP-TO-DATE
+> Task :proxy-core:installDist UP-TO-DATE
+
+BUILD SUCCESSFUL in 15s
+13 actionable tasks: 13 up-to-date
+Configuration cache entry reused.
+Offline synthetic Minecraft 1.7.10 PLAY echo; mode=initial payload=4096 bytes frame=4098 bytes connections=4 measured_roundtrips_per_connection=5000 warmup_per_connection=500 window_per_connection=16 repeats=3 command_interceptor=1
+Direct and proxy phases use the same JVM, fake backend, client code, framed payload, connection concurrency, and per-client in-flight window. Login/setup and warmup are excluded from timing.
+No Forge, real pack, or real client is involved.
+direct-1: roundtrips=20000 echoes=20000 correct=20000 errors=0 roundtrips/s=104076.5 warmup_roundtrips=2000 window=16 one_way_MiB/s=406.55 latency_ms_p50=0.582 p95=0.793 p99=1.092 elapsed_ms=192.166
+proxy-1: roundtrips=20000 echoes=20000 correct=20000 errors=0 roundtrips/s=69443.8 warmup_roundtrips=2000 window=16 one_way_MiB/s=271.27 latency_ms_p50=0.742 p95=1.773 p99=2.812 elapsed_ms=288.003
+proxy-2: roundtrips=20000 echoes=20000 correct=20000 errors=0 roundtrips/s=65450.0 warmup_roundtrips=2000 window=16 one_way_MiB/s=255.66 latency_ms_p50=0.688 p95=1.951 p99=4.450 elapsed_ms=305.577
+direct-2: roundtrips=20000 echoes=20000 correct=20000 errors=0 roundtrips/s=106529.6 warmup_roundtrips=2000 window=16 one_way_MiB/s=416.13 latency_ms_p50=0.597 p95=0.723 p99=0.815 elapsed_ms=187.741
+direct-3: roundtrips=20000 echoes=20000 correct=20000 errors=0 roundtrips/s=102517.6 warmup_roundtrips=2000 window=16 one_way_MiB/s=400.46 latency_ms_p50=0.612 p95=0.732 p99=0.828 elapsed_ms=195.089
+proxy-3: roundtrips=20000 echoes=20000 correct=20000 errors=0 roundtrips/s=92875.1 warmup_roundtrips=2000 window=16 one_way_MiB/s=362.79 latency_ms_p50=0.655 p95=0.853 p99=1.100 elapsed_ms=215.343
@@ -0,0 +1,15 @@
+# Player command interception screening
+
+Run date: 2026-09-26 (Asia/Shanghai). Local Windows loopback, synthetic Minecraft 1.7.10 PLAY echo, four connections, 4,096-byte bodies, 500 warmup and 5,000 measured exchanges per connection, three repetitions, in-flight window 16. Each command started a fresh benchmark JVM. The direct and proxy phases in each command shared the JVM and fake backend. The only proxy configuration difference was whether the frontend command interceptor was installed; no command packets were sent.
+
+```powershell
+.\benchmarks\run-proxy-session.ps1 -Connections 4 -Messages 5000 -Warmup 500 -Payload 4096 -Repeats 3 -Window 16 -CommandInterceptor 1
+.\benchmarks\run-proxy-session.ps1 -Connections 4 -Messages 5000 -Warmup 500 -Payload 4096 -Repeats 3 -Window 16 -CommandInterceptor 0
+```
+
+| Interceptor | Proxy roundtrips/s, rounds 1–3 | Proxy p95 latency, ms | Correct echoes |
+| --- | --- | --- | --- |
+| Enabled | 69,444; 65,450; 92,875 | 1.773; 1.951; 0.853 | 60,000 / 60,000 |
+| Disabled | 73,820; 84,262; 96,942 | 1.450; 1.241; 0.768 | 60,000 / 60,000 |
+
+The ranges overlap, though all three enabled rounds were slower in this run. Separate JVM runs and visible throughput variation prevent attributing that difference to the interceptor. This short screening run does not resolve a small overhead or establish target-pack capacity. It verifies ordinary framed traffic remained correct with the interceptor installed under these conditions. Raw output: [enabled](2026-09-26-command-interceptor-on.txt), [disabled](2026-09-26-command-interceptor-off.txt).
@@ -5,6 +5,7 @@ param(
    [ValidateRange(5, 1048576)][int]$Payload = 1024,
    [ValidateRange(1, 10)][int]$Repeats = 2,
    [ValidateRange(1, 1024)][int]$Window = 1,
+    [ValidateSet(0, 1)][int]$CommandInterceptor = 1,
    [ValidateSet('initial', 'post-transfer')][string]$Mode = 'initial'
 )

@@ -24,7 +25,7 @@ try {

    & java -cp "$classes;$installLib\*" dev.strataproxy.core.session.ProxySessionBenchmark `
        --connections $Connections --messages $Messages --warmup $Warmup --payload $Payload --repeats $Repeats `
-        --window $Window --mode $Mode
+        --window $Window --mode $Mode --command-interceptor $CommandInterceptor
    if ($LASTEXITCODE -ne 0) { throw "benchmark failed with exit code $LASTEXITCODE" }
 } finally {
    Pop-Location
@@ -208,6 +208,9 @@ class AgentDiscoveryPluginTest {
                };
            }
            @Override public Servers servers() { return servers; }
+            @Override public dev.strataproxy.api.Commands commands() {
+                return (name, handler) -> { throw new UnsupportedOperationException(); };
+            }
            @Override public org.slf4j.Logger logger() { return org.slf4j.LoggerFactory.getLogger("agent-test"); }
            @Override public Map<String, String> settings() {
                return Map.of("secret", new String(SECRET, StandardCharsets.UTF_8), "port", Integer.toString(port),
@@ -305,6 +305,10 @@ class DnsDiscoveryPluginTest {
                return servers;
            }

+            @Override public dev.strataproxy.api.Commands commands() {
+                return (name, handler) -> { throw new UnsupportedOperationException(); };
+            }
+
            @Override
            public org.slf4j.Logger logger() {
                return LoggerFactory.getLogger("dns-discovery-test");
@@ -1,11 +1,14 @@
 package dev.strataproxy.app;

+import io.netty.util.NetUtil;
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.util.List;
 import java.util.Map;

 /** Configuration understood by the new proxy runtime. */
-public record ProxyConfiguration(String listen, Authentication authentication, List<Backend> backends, Plugins plugins) {
+public record ProxyConfiguration(String listen, Authentication authentication, List<Backend> backends,
+                                 Plugins plugins, Integer maxConnections, boolean allowOfflinePublicAccess) {
    public enum Authentication { OFFLINE, ONLINE_BUNGEE }

    public ProxyConfiguration {
@@ -16,8 +19,16 @@ public record ProxyConfiguration(String listen, Authentication authentication, L
        if (authentication == null) {
            throw new IllegalArgumentException("authentication is required: OFFLINE or ONLINE_BUNGEE");
        }
+        if (authentication == Authentication.OFFLINE && !allowOfflinePublicAccess
+                && !isLiteralLoopback(listen)) {
+            throw new IllegalArgumentException("OFFLINE requires a literal loopback listen IP or allowOfflinePublicAccess: true");
+        }
        backends = backends == null ? List.of() : List.copyOf(backends);
        plugins = plugins == null ? new Plugins("plugins", Map.of(), null) : plugins;
+        maxConnections = maxConnections == null ? 4096 : maxConnections;
+        if (maxConnections < 1 || maxConnections > 1_000_000) {
+            throw new IllegalArgumentException("maxConnections must be between 1 and 1000000");
+        }
        var names = new java.util.HashSet<String>();
        for (var backend : backends) {
            if (!names.add(backend.name())) {
@@ -27,13 +38,30 @@ public record ProxyConfiguration(String listen, Authentication authentication, L
    }

    public ProxyConfiguration(String listen, Authentication authentication, List<Backend> backends) {
-        this(listen, authentication, backends, null);
+        this(listen, authentication, backends, null, null, false);
+    }
+
+    public ProxyConfiguration(String listen, Authentication authentication, List<Backend> backends, Plugins plugins) {
+        this(listen, authentication, backends, plugins, null, false);
    }

    public InetSocketAddress listenAddress() {
        return parseAddress(listen);
    }

+    private static boolean isLiteralLoopback(String listen) {
+        String value = listen.trim();
+        String host = value.substring(0, value.lastIndexOf(':'));
+        if (host.startsWith("[") && host.endsWith("]")) host = host.substring(1, host.length() - 1);
+        byte[] bytes = NetUtil.createByteArrayFromIpAddressString(host);
+        if (bytes == null) return false;
+        try {
+            return InetAddress.getByAddress(bytes).isLoopbackAddress();
+        } catch (java.net.UnknownHostException impossible) {
+            throw new AssertionError(impossible);
+        }
+    }
+
    public static InetSocketAddress parseAddress(String text) {
        if (text == null || text.isBlank()) {
            throw new IllegalArgumentException("address is required");
@@ -75,10 +75,12 @@ public final class ProxyMain {
                : new ProxySessionListener(configuration.listenAddress(), catalog,
                        new MojangSessionVerifier(Duration.ofSeconds(5), verifierWorkers), placementTimeout);
        try (var plugins = new PluginHost(catalog, listener, placementTimeout)) {
+            listener.setMaxConnections(configuration.maxConnections());
            plugins.loadPlugins(pluginDirectory(configPath, configuration.plugins().directory()),
                    configuration.plugins().enabled());
            plugins.enable();
            listener.setPlacement(plugins::placeInitial);
+            listener.setCommandDispatcher(plugins::dispatchCommand);
            var serverChannel = listener.start().toCompletableFuture().join();
            LOGGER.info("Listening on {}", serverChannel.localAddress());
            var shutdown = new Thread(() -> {
@@ -1,6 +1,10 @@
 package dev.strataproxy.core.plugin;

 import dev.strataproxy.api.InitialPlacementHandler;
+import dev.strataproxy.api.CommandHandler;
+import dev.strataproxy.api.CommandInvocation;
+import dev.strataproxy.api.CommandRegistration;
+import dev.strataproxy.api.Commands;
 import dev.strataproxy.api.PlacementDecision;
 import dev.strataproxy.api.PlayerIdentity;
 import dev.strataproxy.api.PlayerView;
@@ -32,6 +36,7 @@ import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
@@ -43,15 +48,20 @@ import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.FutureTask;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+import java.util.function.BooleanSupplier;
 import java.util.jar.JarFile;

 /** Owns plugin lifecycles and exposes only the proxy services in the plugin API. */
@@ -65,12 +75,15 @@ public final class PluginHost implements AutoCloseable {
    private final BackendCatalog catalog;
    private final Players players;
    private final Duration placementTimeout;
+    private final Duration shutdownTimeout;
    private final ThreadPoolExecutor callbacks;
+    private final ThreadPoolExecutor commandWorkers;
+    private final ConcurrentHashMap<String, RegisteredCommand> commands = new ConcurrentHashMap<>();
    private final ScheduledExecutorService timer;
    private final List<URLClassLoader> classLoaders = new ArrayList<>();
    private final List<LoadedPlugin> plugins = new ArrayList<>();
    private final Set<PendingPlacement> pendingPlacements = ConcurrentHashMap.newKeySet();
-    private State state = State.LOADING;
+    private volatile State state = State.LOADING;
    private InitialPlacementHandler placementHandler;

    public PluginHost(BackendCatalog catalog, Players players, Duration placementTimeout) {
@@ -79,18 +92,30 @@ public final class PluginHost implements AutoCloseable {

    PluginHost(BackendCatalog catalog, Players players, Duration placementTimeout,
               int callbackThreads, int callbackQueueCapacity) {
+        this(catalog, players, placementTimeout, callbackThreads, callbackQueueCapacity, Duration.ofSeconds(10));
+    }
+
+    PluginHost(BackendCatalog catalog, Players players, Duration placementTimeout,
+               int callbackThreads, int callbackQueueCapacity, Duration shutdownTimeout) {
        this.catalog = Objects.requireNonNull(catalog, "catalog");
        this.players = Objects.requireNonNull(players, "players");
        this.placementTimeout = Objects.requireNonNull(placementTimeout, "placementTimeout");
+        this.shutdownTimeout = Objects.requireNonNull(shutdownTimeout, "shutdownTimeout");
        if (placementTimeout.isZero() || placementTimeout.isNegative()) {
            throw new IllegalArgumentException("placementTimeout must be positive");
        }
+        if (shutdownTimeout.isZero() || shutdownTimeout.isNegative()) {
+            throw new IllegalArgumentException("shutdownTimeout must be positive");
+        }
        if (callbackThreads < 1 || callbackQueueCapacity < 1) {
            throw new IllegalArgumentException("callback executor sizes must be positive");
        }
        this.callbacks = new ThreadPoolExecutor(callbackThreads, callbackThreads, 0, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(callbackQueueCapacity), namedThreads("strataproxy-plugin-callback"),
                new ThreadPoolExecutor.AbortPolicy());
+        this.commandWorkers = new ThreadPoolExecutor(callbackThreads, callbackThreads, 0, TimeUnit.MILLISECONDS,
+                new ArrayBlockingQueue<>(callbackQueueCapacity), namedThreads("strataproxy-plugin-command"),
+                new ThreadPoolExecutor.AbortPolicy());
        this.timer = Executors.newSingleThreadScheduledExecutor(namedThreads("strataproxy-plugin-timeout"));
    }

@@ -296,6 +321,44 @@ public final class PluginHost implements AutoCloseable {
        return result;
    }

+    /** Claims only known root commands; caller retains and forwards every other chat frame. */
+    public boolean dispatchCommand(PlayerView player, String message, Consumer<String> reply) {
+        return dispatchCommand(player, message, reply, () -> true);
+    }
+
+    /** Admission is checked only after the command name is known, preserving unknown-command passthrough. */
+    public boolean dispatchCommand(PlayerView player, String message, Consumer<String> reply,
+                                   BooleanSupplier admission) {
+        Objects.requireNonNull(player, "player");
+        Objects.requireNonNull(message, "message");
+        Objects.requireNonNull(reply, "reply");
+        Objects.requireNonNull(admission, "admission");
+        if (state != State.ENABLED || !message.startsWith("/") || message.length() < 2) return false;
+        int end = 1;
+        while (end < message.length() && !Character.isWhitespace(message.charAt(end))) end++;
+        String name = message.substring(1, end).toLowerCase(Locale.ROOT);
+        RegisteredCommand registered = commands.get(name);
+        if (registered == null || !registered.context.active) return false;
+        if (!admission.getAsBoolean()) return true;
+        String arguments = end == message.length() ? "" : message.substring(end).stripLeading();
+        CommandInvocation invocation = new CommandInvocation(player, name, arguments,
+                text -> { if (registered.context.active) reply.accept(text); });
+        try {
+            commandWorkers.execute(() -> {
+                if (state != State.ENABLED || !registered.context.active || commands.get(name) != registered) return;
+                try {
+                    registered.handler.execute(invocation);
+                } catch (Throwable failure) {
+                    LOGGER.warn("Plugin {} command /{} failed", registered.context.owner.id(), name, failure);
+                    invocation.reply("Proxy command failed.");
+                }
+            });
+        } catch (RejectedExecutionException overloaded) {
+            reply.accept("Proxy command service is busy. Please try again.");
+        }
+        return true;
+    }
+
    @Override
    public synchronized void close() {
        if (state == State.CLOSED) {
@@ -307,15 +370,13 @@ public final class PluginHost implements AutoCloseable {
            request.result().completeExceptionally(new IllegalStateException("Plugin host closed"));
        }
        callbacks.shutdownNow();
+        commandWorkers.shutdownNow();
        timer.shutdownNow();
+        long deadline = System.nanoTime() + shutdownTimeout.toNanos();
        for (int index = plugins.size() - 1; index >= 0; index--) {
            LoadedPlugin loaded = plugins.get(index);
            if (loaded.enabled) {
-                try {
-                    loaded.plugin.onDisable();
-                } catch (Throwable failure) {
-                    LOGGER.warn("Plugin {} failed during onDisable", loaded.owner.id(), failure);
-                }
+                disableWithinDeadline(loaded, deadline);
            }
            loaded.context.deactivateAndRemove();
        }
@@ -330,6 +391,30 @@ public final class PluginHost implements AutoCloseable {
        placementHandler = null;
    }

+    private void disableWithinDeadline(LoadedPlugin loaded, long deadline) {
+        long remaining = deadline - System.nanoTime();
+        if (remaining <= 0) {
+            LOGGER.warn("Skipping onDisable for plugin {} after shutdown deadline", loaded.owner.id());
+            return;
+        }
+        FutureTask<Void> stop = new FutureTask<>(() -> {
+            loaded.plugin.onDisable();
+            return null;
+        });
+        Thread.ofVirtual().name("strataproxy-plugin-disable-" + loaded.owner.id()).start(stop);
+        try {
+            stop.get(remaining, TimeUnit.NANOSECONDS);
+        } catch (TimeoutException timeout) {
+            stop.cancel(true);
+            LOGGER.warn("Plugin {} onDisable exceeded shutdown deadline", loaded.owner.id());
+        } catch (ExecutionException failure) {
+            LOGGER.warn("Plugin {} failed during onDisable", loaded.owner.id(), failure.getCause());
+        } catch (InterruptedException interrupted) {
+            stop.cancel(true);
+            Thread.currentThread().interrupt();
+        }
+    }
+
    private void loadOne(Plugin plugin, String id, Map<String, String> settings) {
        long generation = NEXT_HOST_GENERATION.updateAndGet(value -> {
            if (value == Long.MAX_VALUE) {
@@ -378,6 +463,8 @@ public final class PluginHost implements AutoCloseable {
    private record PendingPlacement(CompletableFuture<Optional<PlacementDecision>> result,
                                    AtomicBoolean decided) { }

+    private record RegisteredCommand(String name, PluginContextImpl context, CommandHandler handler) { }
+
    private static final class LoadedPlugin {
        private final Plugin plugin;
        private final BackendOwner owner;
@@ -395,6 +482,7 @@ public final class PluginHost implements AutoCloseable {
        private final BackendOwner owner;
        private final Players pluginPlayers;
        private final PluginServers servers;
+        private final Commands pluginCommands;
        private final Logger logger;
        private final Map<String, String> settings;
        private volatile boolean active = true;
@@ -403,17 +491,20 @@ public final class PluginHost implements AutoCloseable {
            this.owner = owner;
            this.pluginPlayers = new PluginPlayers(this);
            this.servers = new PluginServers(this);
+            this.pluginCommands = new PluginCommands(this);
            this.logger = LoggerFactory.getLogger("plugin." + owner.id());
            this.settings = Map.copyOf(settings);
        }

        @Override public Players players() { return pluginPlayers; }
        @Override public Servers servers() { return servers; }
+        @Override public Commands commands() { return pluginCommands; }
        @Override public Logger logger() { return logger; }
        @Override public Map<String, String> settings() { return settings; }

        private synchronized void deactivateAndRemove() {
            active = false;
+            commands.entrySet().removeIf(entry -> entry.getValue().context == this);
            catalog.removeOwner(owner);
        }

@@ -422,6 +513,29 @@ public final class PluginHost implements AutoCloseable {
        }
    }

+    private final class PluginCommands implements Commands {
+        private final PluginContextImpl context;
+
+        private PluginCommands(PluginContextImpl context) { this.context = context; }
+
+        @Override public CommandRegistration register(String name, CommandHandler handler) {
+            Objects.requireNonNull(name, "name");
+            Objects.requireNonNull(handler, "handler");
+            String normalized = name.toLowerCase(Locale.ROOT);
+            if (!normalized.matches("[a-z0-9_.:-]{1,32}")) {
+                throw new IllegalArgumentException("invalid command name: " + name);
+            }
+            synchronized (context) {
+                context.requireActive();
+                RegisteredCommand registration = new RegisteredCommand(normalized, context, handler);
+                if (commands.putIfAbsent(normalized, registration) != null) {
+                    throw new IllegalArgumentException("Command already registered: /" + normalized);
+                }
+                return () -> commands.remove(normalized, registration);
+            }
+        }
+    }
+
    private final class PluginPlayers implements Players {
        private final PluginContextImpl context;

@@ -1,5 +1,6 @@
 package dev.strataproxy.core.protocol;

+import com.fasterxml.jackson.databind.ObjectMapper;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;

@@ -8,6 +9,9 @@ import java.util.OptionalInt;

 /** The few protocol 5 PLAY packets a backend switch must understand or synthesize. */
 public final class Minecraft1710PlayPackets {
+    private static final ObjectMapper JSON = new ObjectMapper();
+    public static final int CLIENT_CHAT = 0x01;
+    public static final int SERVER_CHAT = 0x02;
    public static final int JOIN_GAME = 0x01;
    public static final int RESPAWN = 0x07;
    public static final int CLIENT_CUSTOM_PAYLOAD = 0x17;
@@ -16,6 +20,43 @@ public final class Minecraft1710PlayPackets {

    private Minecraft1710PlayPackets() { }

+    /** Reads a complete client PLAY frame. Non-chat frames are left untouched. */
+    public static Optional<String> playerChat(ByteBuf frame) {
+        // The frame decoder already verified the prefix. Peek at the one-byte chat ID
+        // before creating a duplicate so ordinary gameplay packets stay allocation-free.
+        int offset = frame.readerIndex();
+        for (int index = 0; index < 3; index++) {
+            int current = frame.getUnsignedByte(offset++);
+            if ((current & 0x80) == 0) {
+                if (frame.getUnsignedByte(offset) != CLIENT_CHAT) return Optional.empty();
+                break;
+            }
+        }
+        ByteBuf input = frame.duplicate();
+        int length = ProtocolVarInt.read(input);
+        if (length < 1 || length != input.readableBytes()) throw new ProtocolException("invalid PLAY frame length");
+        if (ProtocolVarInt.read(input) != CLIENT_CHAT) return Optional.empty();
+        String message = ProtocolStrings.read(input, 100);
+        if (input.isReadable()) throw new ProtocolException("trailing client chat bytes");
+        return Optional.of(message);
+    }
+
+    /** Encodes a plain-text proxy reply as a protocol 5 clientbound PLAY chat frame. */
+    public static ByteBuf chatReply(ByteBufAllocator allocator, String message) {
+        if (message == null || message.length() > 1024) {
+            throw new IllegalArgumentException("chat reply must contain at most 1024 characters");
+        }
+        String component = JSON.createObjectNode().put("text", message).toString();
+        ByteBuf packet = allocator.buffer();
+        try {
+            ProtocolVarInt.write(packet, SERVER_CHAT);
+            ProtocolStrings.write(packet, component, 32767);
+            return frame(allocator, packet);
+        } finally {
+            packet.release();
+        }
+    }
+
    public record JoinGame(int entityId, int gameMode, int dimension, int difficulty, String levelType) { }

    public record ForgeServerHello(int protocolVersion, int dimensionOverride) { }
@@ -0,0 +1,45 @@
+package dev.strataproxy.core.session;
+
+import dev.strataproxy.core.protocol.Minecraft1710PlayPackets;
+import dev.strataproxy.core.relay.RawRelay;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.util.ReferenceCountUtil;
+
+import java.util.function.Predicate;
+
+/** Consumes only registered proxy commands; all other PLAY frames keep their wire bytes. */
+final class PlayerCommandInterceptor extends ChannelInboundHandlerAdapter {
+    private final Predicate<String> dispatch;
+    private final Runnable closeSession;
+
+    PlayerCommandInterceptor(Predicate<String> dispatch, Runnable closeSession) {
+        this.dispatch = dispatch;
+        this.closeSession = closeSession;
+    }
+
+    @Override public void channelRead(ChannelHandlerContext ctx, Object message) {
+        if (!(message instanceof ByteBuf frame)) {
+            ReferenceCountUtil.release(message);
+            closeSession.run();
+            return;
+        }
+        ByteBuf outgoing = frame;
+        try {
+            var chat = Minecraft1710PlayPackets.playerChat(frame);
+            if (chat.isPresent() && chat.get().startsWith("/") && dispatch.test(chat.get())) {
+                outgoing = null;
+                frame.release();
+                RawRelay.continueAfterDrop(ctx.channel());
+            } else {
+                outgoing = null;
+                ctx.fireChannelRead(frame);
+            }
+        } catch (RuntimeException malformed) {
+            closeSession.run();
+        } finally {
+            if (outgoing != null) outgoing.release();
+        }
+    }
+}
@@ -40,10 +40,16 @@ import java.util.UUID;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
+import java.util.function.Consumer;
+import java.util.function.BooleanSupplier;
 import java.util.concurrent.ThreadFactory;

 /** Owns protocol 5 client sessions from login through backend disconnect. */
 public final class ProxySessionListener implements Players {
+    @FunctionalInterface public interface CommandDispatcher {
+        /** Returns true only if this proxy owns the command. */
+        boolean dispatch(PlayerView player, String message, Consumer<String> reply, BooleanSupplier admission);
+    }
    private final SocketAddress bindAddress;
    private final BackendCatalog catalog;
    private final SessionVerifier verifier;
@@ -60,7 +66,10 @@ public final class ProxySessionListener implements Players {
    private final Set<Session> allSessions = ConcurrentHashMap.newKeySet();
    private final AtomicLong nextConnectionId = new AtomicLong();
    private final AtomicInteger onlineCount = new AtomicInteger();
+    private final AtomicInteger connectionCount = new AtomicInteger();
    private volatile Function<PlayerView, CompletionStage<Optional<PlacementDecision>>> placement;
+    private volatile CommandDispatcher commandDispatcher;
+    private int maxConnections = 4096;
    private volatile Channel listener;
    private volatile boolean closed;
    private boolean started;
@@ -149,6 +158,21 @@ public final class ProxySessionListener implements Players {
        this.placement = Objects.requireNonNull(placement, "placement");
    }

+    /** Configures optional plugin command dispatch before the listener starts. */
+    public synchronized void setCommandDispatcher(CommandDispatcher dispatcher) {
+        if (started || closed || commandDispatcher != null) {
+            throw new IllegalStateException("Command dispatcher must be configured once before listener start");
+        }
+        commandDispatcher = Objects.requireNonNull(dispatcher, "dispatcher");
+    }
+
+    /** Bounds accepted sessions, including clients that have not completed login. */
+    public synchronized void setMaxConnections(int limit) {
+        if (started || closed) throw new IllegalStateException("Connection limit must be set before listener start");
+        if (limit < 1 || limit > 1_000_000) throw new IllegalArgumentException("invalid connection limit");
+        maxConnections = limit;
+    }
+
    public synchronized CompletionStage<Channel> start() {
        if (closed) return CompletableFuture.failedFuture(new IllegalStateException("listener is closed"));
        if (started) return CompletableFuture.failedFuture(new IllegalStateException("listener already started"));
@@ -190,12 +214,13 @@ public final class ProxySessionListener implements Players {

    /** Serializes admission with close() so every accepted session is either closed or rejected. */
    private synchronized Session registerSession(SocketChannel channel) {
-        if (closed) {
+        if (closed || connectionCount.get() >= maxConnections) {
            channel.close();
            return null;
        }
        Session session = new Session(this, channel);
        allSessions.add(session);
+        connectionCount.incrementAndGet();
        return session;
    }

@@ -266,6 +291,7 @@ public final class ProxySessionListener implements Players {
        return MinecraftEncryptionRequest.create("", encryptionKeys.getPublic(), random);
    }
    Function<PlayerView, CompletionStage<Optional<PlacementDecision>>> placement() { return placement; }
+    CommandDispatcher commandDispatcher() { return commandDispatcher; }
    boolean claimIdentity(UUID uuid, Session session) { return sessionsByPlayerId.putIfAbsent(uuid, session) == null; }
    void releaseIdentity(UUID uuid, Session session) { sessionsByPlayerId.remove(uuid, session); }
    Set<Session> allSessions() { return allSessions; }
@@ -273,4 +299,5 @@ public final class ProxySessionListener implements Players {
    int onlineCount() { return onlineCount.get(); }
    void sessionPublished() { onlineCount.incrementAndGet(); }
    void sessionUnpublished() { onlineCount.decrementAndGet(); }
+    void sessionClosed() { connectionCount.decrementAndGet(); }
 }
@@ -55,6 +55,7 @@ import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;

 /** Mutable session control state is confined to the frontend event loop. */
@io.netty.channel.ChannelHandler.Sharable
@@ -62,6 +63,10 @@ final class Session extends ChannelInboundHandlerAdapter {
    private static final Logger LOGGER = LoggerFactory.getLogger(Session.class);
    private static final int MAX_TRANSITION_BUFFER_BYTES = ProtocolProfile.minecraft1710().maxFrameBytes();
    private static final int MAX_TRANSITION_BUFFER_FRAMES = 1024;
+    private static final int MAX_PENDING_COMMAND_REPLIES = 64;
+    private static final int COMMAND_BURST = 10;
+    private static final long COMMAND_TOKEN_NANOS = TimeUnit.MILLISECONDS.toNanos(200);
+    private static final long COMMAND_NOTICE_NANOS = TimeUnit.SECONDS.toNanos(2);
    private static final Duration FORGE_TRANSFER_HANDSHAKE_TIMEOUT = Duration.ofSeconds(30);
    private static final Duration LOGIN_DISCONNECT_DRAIN_TIMEOUT = Duration.ofSeconds(5);
    private final ProxySessionListener owner;
@@ -84,6 +89,11 @@ final class Session extends ChannelInboundHandlerAdapter {
    private int transitionBufferBytes;
    private final ArrayDeque<PendingFrame> transitionBuffer = new ArrayDeque<>();
    private final AtomicBoolean closed = new AtomicBoolean();
+    private final AtomicInteger pendingCommandReplies = new AtomicInteger();
+    private int commandTokens = COMMAND_BURST;
+    private long commandTokenRefillNanos = System.nanoTime();
+    private long lastCommandNoticeNanos;
+    private boolean commandNoticeSent;
    private ScheduledFuture<?> initialLoginDeadline;
    private ScheduledFuture<?> initialPlayDeadline;
    private PlayObservation playObservation;
@@ -500,6 +510,10 @@ final class Session extends ChannelInboundHandlerAdapter {
        try {
            var observation = playObservation;
            frontend.pipeline().addLast("keep-alive-bridge", new KeepAliveBridge(keepAlives, true, this::closePair));
+            if (owner.commandDispatcher() != null) {
+                frontend.pipeline().addLast("player-commands", new PlayerCommandInterceptor(
+                        this::dispatchPlayerCommand, this::closePair));
+            }
            target.pipeline().addLast("keep-alive-bridge", new KeepAliveBridge(keepAlives, false, this::closePair));
            var link = RawRelay.attach(frontend, target,
                    bytes -> observation.observeFrame(false, bytes),
@@ -523,6 +537,59 @@ final class Session extends ChannelInboundHandlerAdapter {
        }
    }

+    private boolean dispatchPlayerCommand(String message) {
+        if (closed.get() || !published) return false;
+        return owner.commandDispatcher().dispatch(view, message, this::sendCommandReply, this::admitPlayerCommand);
+    }
+
+    private boolean admitPlayerCommand() {
+        long now = System.nanoTime();
+        long elapsed = now - commandTokenRefillNanos;
+        if (elapsed >= COMMAND_TOKEN_NANOS) {
+            long replenished = Math.min(COMMAND_BURST, elapsed / COMMAND_TOKEN_NANOS);
+            commandTokens = (int) Math.min(COMMAND_BURST, commandTokens + replenished);
+            commandTokenRefillNanos = now;
+        }
+        if (commandTokens > 0) {
+            commandTokens--;
+            return true;
+        }
+        if (!commandNoticeSent || now - lastCommandNoticeNanos >= COMMAND_NOTICE_NANOS) {
+            commandNoticeSent = true;
+            lastCommandNoticeNanos = now;
+            sendCommandReply("Too many proxy commands. Please slow down.");
+        }
+        return false;
+    }
+
+    private void sendCommandReply(String message) {
+        if (message == null || message.length() > 1024) {
+            throw new IllegalArgumentException("chat reply must contain at most 1024 characters");
+        }
+        if (pendingCommandReplies.incrementAndGet() > MAX_PENDING_COMMAND_REPLIES) {
+            pendingCommandReplies.decrementAndGet();
+            return;
+        }
+        Runnable send = () -> {
+            try {
+                if (closed.get() || !frontend.isActive() || !frontend.isWritable()) return;
+                ByteBuf frame = Minecraft1710PlayPackets.chatReply(frontend.alloc(), message);
+                frontend.writeAndFlush(frame).addListener(write -> {
+                    if (!write.isSuccess()) closePair();
+                });
+            } finally {
+                pendingCommandReplies.decrementAndGet();
+            }
+        };
+        if (frontend.eventLoop().inEventLoop()) send.run();
+        else {
+            try { frontend.eventLoop().execute(send); }
+            catch (RejectedExecutionException ignored) {
+                pendingCommandReplies.decrementAndGet();
+            }
+        }
+    }
+
    private void bufferTransitionFrame(boolean fromFrontend, ByteBuf packet) {
        int bytes = packet.readableBytes();
        if (transitionBuffer.size() >= MAX_TRANSITION_BUFFER_FRAMES
@@ -1075,6 +1142,7 @@ final class Session extends ChannelInboundHandlerAdapter {
            if (identityClaimed) owner.releaseIdentity(identity.playerId(), this);
            if (published) owner.sessionUnpublished();
            owner.allSessions().remove(this);
+            owner.sessionClosed();
            PendingFrame pending;
            while ((pending = transitionBuffer.pollFirst()) != null) pending.payload.release();
            transitionBufferBytes = 0;
@@ -1,7 +1,8 @@
 # Example: enable the DNS plugin instead of configured static backends.
 # Replace the host with a DNS name that resolves to your backend nodes.
-listen: "0.0.0.0:25577"
+listen: "127.0.0.1:25577"
 authentication: OFFLINE
+maxConnections: 4096
 backends: []
 plugins:
  directory: "../plugins"
@@ -1,5 +1,6 @@
-listen: "0.0.0.0:25577"
+listen: "127.0.0.1:25577"
 authentication: OFFLINE
+maxConnections: 4096
 plugins:
  directory: "../plugins"
  initialPlacementTimeoutSeconds: 15
@@ -27,6 +27,43 @@ final class ProxyConfigurationTest {
        try {
            Files.writeString(config, "listen: \"127.0.0.1:25577\"\nauthentication: OFFLINE\nbackends: []\n");
            assertEquals(0, new ProxyConfigurationLoader().load(config).backends().size());
+            assertEquals(4096, new ProxyConfigurationLoader().load(config).maxConnections());
+        } finally {
+            Files.deleteIfExists(config);
+        }
+    }
+
+    @Test
+    void rejectsInvalidConnectionLimit() throws Exception {
+        var config = Files.createTempFile("strataproxy-cap", ".yml");
+        try {
+            for (int invalid : new int[]{0, -1, 1_000_001}) {
+                Files.writeString(config, "listen: \"127.0.0.1:25577\"\nauthentication: OFFLINE\n"
+                        + "maxConnections: " + invalid + "\n");
+                assertThrows(com.fasterxml.jackson.databind.JsonMappingException.class,
+                        () -> new ProxyConfigurationLoader().load(config));
+            }
+        } finally {
+            Files.deleteIfExists(config);
+        }
+    }
+
+    @Test
+    void offlinePublicBindRequiresExplicitOptIn() throws Exception {
+        var config = Files.createTempFile("strataproxy-public-offline", ".yml");
+        try {
+            String publicOffline = "listen: \"0.0.0.0:25577\"\nauthentication: OFFLINE\n";
+            Files.writeString(config, publicOffline);
+            assertThrows(com.fasterxml.jackson.databind.JsonMappingException.class,
+                    () -> new ProxyConfigurationLoader().load(config));
+            Files.writeString(config, "listen: \"localhost:25577\"\nauthentication: OFFLINE\n");
+            assertThrows(com.fasterxml.jackson.databind.JsonMappingException.class,
+                    () -> new ProxyConfigurationLoader().load(config));
+            Files.writeString(config, publicOffline + "allowOfflinePublicAccess: true\n");
+            assertEquals(true, new ProxyConfigurationLoader().load(config).allowOfflinePublicAccess());
+            Files.writeString(config, "listen: \"0.0.0.0:25577\"\nauthentication: ONLINE_BUNGEE\n");
+            assertEquals(ProxyConfiguration.Authentication.ONLINE_BUNGEE,
+                    new ProxyConfigurationLoader().load(config).authentication());
        } finally {
            Files.deleteIfExists(config);
        }
@@ -1,6 +1,7 @@
 package dev.strataproxy.core.plugin;

 import dev.strataproxy.api.InitialPlacementHandler;
+import dev.strataproxy.api.CommandInvocation;
 import dev.strataproxy.api.PlacementDecision;
 import dev.strataproxy.api.PlayerIdentity;
 import dev.strataproxy.api.PlayerView;
@@ -43,6 +44,7 @@ import java.util.jar.JarEntry;
 import java.util.jar.JarOutputStream;

 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -52,6 +54,104 @@ class PluginHostTest {
    private static final PlayerView PLAYER = new PlayerView(
            new PlayerIdentity(UUID.randomUUID(), 1), "TestPlayer", Optional.empty());

+    @Test
+    void commandsDispatchOffCallerThreadAndReleaseTheirNames() throws Exception {
+        var plugin = new CapturingPlugin("unused", false);
+        var host = new PluginHost(new InMemoryBackendCatalog(), players(), Duration.ofSeconds(1));
+        try {
+            host.load(List.of(plugin));
+            var observed = new CompletableFuture<CommandInvocation>();
+            var worker = new CompletableFuture<Thread>();
+            var reply = new CompletableFuture<String>();
+            var registration = plugin.context.commands().register("Ping", invocation -> {
+                worker.complete(Thread.currentThread());
+                observed.complete(invocation);
+                invocation.reply("pong");
+            });
+            assertThrows(IllegalArgumentException.class,
+                    () -> plugin.context.commands().register("PING", invocation -> { }));
+            host.enable();
+            assertFalse(host.dispatchCommand(PLAYER, "/other", reply::complete));
+            assertFalse(host.dispatchCommand(PLAYER, "/other", reply::complete,
+                    () -> { throw new AssertionError("unknown command must not consume admission"); }));
+            assertTrue(host.dispatchCommand(PLAYER, "/PING blocked", reply::complete, () -> false));
+            assertFalse(worker.isDone());
+            assertTrue(host.dispatchCommand(PLAYER, "/PING hello world", reply::complete));
+            assertNotEquals(Thread.currentThread(), worker.get(5, TimeUnit.SECONDS));
+            assertEquals("ping", observed.get(5, TimeUnit.SECONDS).name());
+            assertEquals("hello world", observed.get(5, TimeUnit.SECONDS).arguments());
+            assertEquals("pong", reply.get(5, TimeUnit.SECONDS));
+            registration.unregister();
+            registration.unregister();
+            assertFalse(host.dispatchCommand(PLAYER, "/ping", text -> { }));
+        } finally {
+            host.close();
+        }
+        assertThrows(IllegalStateException.class,
+                () -> plugin.context.commands().register("late", invocation -> { }));
+    }
+
+    @Test
+    void pluginShutdownHasOneBoundedDeadlineEvenIfDisableIgnoresInterrupts() throws Exception {
+        var entered = new CountDownLatch(1);
+        var release = new CountDownLatch(1);
+        var context = new java.util.concurrent.atomic.AtomicReference<PluginContext>();
+        var plugin = new Plugin() {
+            @Override public void onLoad(PluginContext value) { context.set(value); }
+            @Override public void onDisable() {
+                entered.countDown();
+                while (release.getCount() != 0) {
+                    try { release.await(100, TimeUnit.MILLISECONDS); }
+                    catch (InterruptedException ignored) { /* Simulate a broken plugin. */ }
+                }
+            }
+        };
+        var host = new PluginHost(new InMemoryBackendCatalog(), players(), Duration.ofSeconds(1),
+                1, 1, Duration.ofMillis(100));
+        try {
+            host.load(List.of(plugin));
+            host.enable();
+            long start = System.nanoTime();
+            host.close();
+            assertTrue(entered.await(1, TimeUnit.SECONDS));
+            assertTrue(System.nanoTime() - start < TimeUnit.SECONDS.toNanos(2));
+            assertThrows(IllegalStateException.class,
+                    () -> context.get().commands().register("late", invocation -> { }));
+        } finally {
+            release.countDown();
+            host.close();
+        }
+    }
+
+    @Test
+    void saturatedCommandWorkersRejectWithoutRunningPluginCodeOnCaller() throws Exception {
+        var plugin = new CapturingPlugin("unused", false);
+        var entered = new CountDownLatch(1);
+        var release = new CountDownLatch(1);
+        var calls = new AtomicInteger();
+        var host = new PluginHost(new InMemoryBackendCatalog(), players(), Duration.ofSeconds(1), 1, 1);
+        try {
+            host.load(List.of(plugin));
+            plugin.context.commands().register("slow", invocation -> {
+                calls.incrementAndGet();
+                entered.countDown();
+                release.await(5, TimeUnit.SECONDS);
+            });
+            host.enable();
+            assertTrue(host.dispatchCommand(PLAYER, "/slow one", text -> { }));
+            assertTrue(entered.await(5, TimeUnit.SECONDS));
+            assertTrue(host.dispatchCommand(PLAYER, "/slow two", text -> { }));
+            var rejectedReply = new CompletableFuture<String>();
+            assertTrue(host.dispatchCommand(PLAYER, "/slow three", rejectedReply::complete));
+            assertEquals("Proxy command service is busy. Please try again.",
+                    rejectedReply.get(1, TimeUnit.SECONDS));
+            assertEquals(1, calls.get());
+        } finally {
+            release.countDown();
+            host.close();
+        }
+    }
+
    @Test
    void disabledBrokenJarCannotPreventConfiguredPluginFromLoading(@TempDir Path directory) throws Exception {
        serviceJar(directory.resolve("a-selected.jar"), ConfiguredProvider.class.getName());
@@ -10,6 +10,33 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;

 final class Minecraft1710PlayPacketsTest {
+    @Test
+    void readsOnlyChatFramesWithoutChangingTheirReaderIndex() {
+        ByteBuf chatBody = Unpooled.buffer();
+        ByteBuf ordinaryBody = Unpooled.buffer();
+        ByteBuf chat = null;
+        ByteBuf ordinary = null;
+        try {
+            ProtocolVarInt.write(chatBody, Minecraft1710PlayPackets.CLIENT_CHAT);
+            ProtocolStrings.write(chatBody, "/where 岛屿", 100);
+            chat = Minecraft1710PlayPackets.frame(UnpooledByteBufAllocator.DEFAULT, chatBody);
+            int start = chat.readerIndex();
+            assertEquals("/where 岛屿", Minecraft1710PlayPackets.playerChat(chat).orElseThrow());
+            assertEquals(start, chat.readerIndex());
+
+            ProtocolVarInt.write(ordinaryBody, 0x03);
+            ordinaryBody.writeByte(0x55);
+            ordinary = Minecraft1710PlayPackets.frame(UnpooledByteBufAllocator.DEFAULT, ordinaryBody);
+            start = ordinary.readerIndex();
+            assertTrue(Minecraft1710PlayPackets.playerChat(ordinary).isEmpty());
+            assertEquals(start, ordinary.readerIndex());
+        } finally {
+            chatBody.release();
+            ordinaryBody.release();
+            if (chat != null) chat.release();
+            if (ordinary != null) ordinary.release();
+        }
+    }
    @Test
    void readsStockJoinGameAndForgeDimensionOverride() {
        ByteBuf join = Unpooled.buffer();
@@ -2,6 +2,9 @@ package dev.strataproxy.core.session;

 import com.fasterxml.jackson.databind.ObjectMapper;
 import dev.strataproxy.api.PlacementDecision;
+import dev.strataproxy.api.Plugin;
+import dev.strataproxy.api.PluginContext;
+import dev.strataproxy.core.plugin.PluginHost;
 import dev.strataproxy.core.auth.MinecraftEncryptionRequest;
 import dev.strataproxy.core.auth.MinecraftEncryptionResponse;
 import dev.strataproxy.core.auth.ProfileProperty;
@@ -58,6 +61,126 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;

 final class ProxySessionListenerTest {
+    @Test
+    void connectionLimitRejectsExcessIncompleteLoginsAndReleasesCapacity() throws Exception {
+        var listener = new ProxySessionListener(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0),
+                new InMemoryBackendCatalog());
+        listener.setMaxConnections(1);
+        listener.setPlacement(player -> CompletableFuture.completedFuture(Optional.empty()));
+        try {
+            int port = ((InetSocketAddress) listener.start().toCompletableFuture()
+                    .get(5, TimeUnit.SECONDS).localAddress()).getPort();
+            try (Socket first = new Socket(InetAddress.getLoopbackAddress(), port)) {
+                long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+                while (listener.allSessions().size() != 1 && System.nanoTime() < deadline) Thread.sleep(5);
+                assertEquals(1, listener.allSessions().size());
+                try (Socket rejected = new Socket(InetAddress.getLoopbackAddress(), port)) {
+                    rejected.setSoTimeout(5000);
+                    try { assertEquals(-1, rejected.getInputStream().read()); }
+                    catch (SocketException reset) { /* Windows may reset a socket closed before its first read. */ }
+                }
+                assertEquals(1, listener.allSessions().size());
+            }
+            long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+            while (!listener.allSessions().isEmpty() && System.nanoTime() < deadline) Thread.sleep(5);
+            assertTrue(listener.allSessions().isEmpty());
+            try (Socket admitted = new Socket(InetAddress.getLoopbackAddress(), port)) {
+                deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+                while (listener.allSessions().size() != 1 && System.nanoTime() < deadline) Thread.sleep(5);
+                assertEquals(1, listener.allSessions().size());
+            }
+        } finally {
+            listener.close().toCompletableFuture().get(5, TimeUnit.SECONDS);
+        }
+    }
+    @Test
+    void registeredPlayerCommandRepliesWithoutReachingBackendAndOtherChatPassesThrough() throws Exception {
+        String username = "CommandUser";
+        UUID uuid = UUID.nameUUIDFromBytes(("OfflinePlayer:" + username).getBytes(StandardCharsets.UTF_8));
+        var catalog = new InMemoryBackendCatalog();
+        var handlerThread = new CompletableFuture<String>();
+        try (ServerSocket backendServer = new ServerSocket(0, 8, InetAddress.getLoopbackAddress())) {
+            var received = new CompletableFuture<List<byte[]>>();
+            Thread backendThread = new Thread(() -> {
+                try (Socket socket = backendServer.accept()) {
+                    socket.setSoTimeout(5000);
+                    var input = new DataInputStream(socket.getInputStream());
+                    var output = new DataOutputStream(socket.getOutputStream());
+                    readFrame(input);
+                    readFrame(input);
+                    var success = new ByteArrayOutputStream();
+                    writeVarInt(success, 2);
+                    writeString(success, uuid.toString());
+                    writeString(success, username);
+                    writeFrame(output, success.toByteArray());
+                    writeFrame(output, new byte[]{0x01, 0, 0, 0, 42, 0, 0, 1, 20,
+                            7, 'd', 'e', 'f', 'a', 'u', 'l', 't'});
+                    var position = new ByteArrayOutputStream();
+                    var data = new DataOutputStream(position);
+                    data.writeByte(0x08);
+                    data.writeDouble(0);
+                    data.writeDouble(64);
+                    data.writeDouble(0);
+                    data.writeFloat(0);
+                    data.writeFloat(0);
+                    data.writeByte(0);
+                    writeFrame(output, position.toByteArray());
+                    received.complete(List.of(readFrame(input), readFrame(input)));
+                } catch (Throwable failure) {
+                    received.completeExceptionally(failure);
+                }
+            }, "fake-command-backend");
+            backendThread.setDaemon(true);
+            backendThread.start();
+            catalog.register(new BackendRegistration(new BackendId("lobby"), new BackendOwner("static", 0),
+                    URI.create("tcp://127.0.0.1:" + backendServer.getLocalPort()), Map.of(), Map.of()));
+            var listener = new ProxySessionListener(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), catalog);
+            try (var plugins = new PluginHost(catalog, listener, Duration.ofSeconds(5))) {
+                plugins.load(List.of(new Plugin() {
+                    @Override public void onLoad(PluginContext context) {
+                        context.commands().register("where", invocation -> {
+                            handlerThread.complete(Thread.currentThread().getName());
+                            invocation.reply(invocation.player().username() + ":" + invocation.arguments());
+                        });
+                    }
+                }));
+                plugins.enable();
+                listener.setPlacement(plugins::placeInitial);
+                listener.setCommandDispatcher(plugins::dispatchCommand);
+                int port = ((InetSocketAddress) listener.start().toCompletableFuture()
+                        .get(5, TimeUnit.SECONDS).localAddress()).getPort();
+                try (Socket client = new Socket(InetAddress.getLoopbackAddress(), port)) {
+                    client.setSoTimeout(5000);
+                    sendLogin(client, username);
+                    var input = new DataInputStream(client.getInputStream());
+                    var output = new DataOutputStream(client.getOutputStream());
+                    assertEquals(2, readVarInt(readFrame(input)));
+                    assertEquals(1, readVarInt(readFrame(input)));
+                    assertEquals(8, readVarInt(readFrame(input)));
+                    writeFrame(output, chatPacket("/where lobby"));
+                    var reply = new java.io.ByteArrayInputStream(readFrame(input));
+                    assertEquals(2, readVarInt(reply));
+                    assertEquals("CommandUser:lobby",
+                            new ObjectMapper().readTree(readString(reply, 32767)).path("text").asText());
+                    assertTrue(handlerThread.get(5, TimeUnit.SECONDS).startsWith("strataproxy-plugin-command-"));
+                    writeFrame(output, chatPacket("hello"));
+                    writeFrame(output, chatPacket("/unknown arg"));
+                    var backendChat = received.get(5, TimeUnit.SECONDS);
+                    assertArrayEquals(chatPacket("hello"), backendChat.get(0));
+                    assertArrayEquals(chatPacket("/unknown arg"), backendChat.get(1));
+                }
+            } finally {
+                listener.close().toCompletableFuture().get(5, TimeUnit.SECONDS);
+            }
+        }
+    }
+
+    private static byte[] chatPacket(String text) throws Exception {
+        var packet = new ByteArrayOutputStream();
+        writeVarInt(packet, 1);
+        writeString(packet, text);
+        return packet.toByteArray();
+    }
    @Test
    void shutdownWhileAcceptingClientsClosesEveryRegisteredSession() throws Exception {
        var listener = new ProxySessionListener(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0),
@@ -793,7 +916,7 @@ final class ProxySessionListenerTest {
                    output.write(framed(success.toByteArray()));
                    output.write(framed(new byte[]{0x02, 0x55}));
                    output.flush();
-                    assertArrayEquals(new byte[]{0x01, 0x33}, readFrame(input));
+                    assertArrayEquals(new byte[]{0x03, 0x33}, readFrame(input));
                    writeFrame(output, new byte[]{0x03, 0x44});
                    backendDone.complete(null);
                } catch (Throwable failure) {
@@ -813,6 +936,12 @@ final class ProxySessionListenerTest {
                        return CompletableFuture.completedFuture(Optional.of(new VerifiedProfile(uuid, "Alice", java.util.List.of())));
                    });
            listener.setPlacement(player -> CompletableFuture.completedFuture(Optional.of(PlacementDecision.select("lobby"))));
+            listener.setCommandDispatcher((player, message, reply, admission) -> {
+                if (!message.equals("/ping")) return false;
+                if (!admission.getAsBoolean()) return true;
+                reply.accept("pong");
+                return true;
+            });
            try {
                var bound = listener.start().toCompletableFuture().get(5, TimeUnit.SECONDS);
                try (Socket client = new Socket(InetAddress.getLoopbackAddress(),
@@ -853,7 +982,14 @@ final class ProxySessionListenerTest {
                    assertEquals(uuid.toString(), readString(success, 36));
                    assertEquals("Alice", readString(success, 16));
                    assertArrayEquals(new byte[]{0x02, 0x55}, readFrame(encryptedInput));
-                    byte[] encryptedPlay = aes(secret, Cipher.ENCRYPT_MODE).update(framed(new byte[]{0x01, 0x33}));
+                    Cipher encryptor = aes(secret, Cipher.ENCRYPT_MODE);
+                    clearOutput.write(encryptor.update(framed(chatPacket("/ping"))));
+                    clearOutput.flush();
+                    var commandReply = new java.io.ByteArrayInputStream(readFrame(encryptedInput));
+                    assertEquals(2, readVarInt(commandReply));
+                    assertEquals("pong",
+                            new ObjectMapper().readTree(readString(commandReply, 32767)).path("text").asText());
+                    byte[] encryptedPlay = encryptor.update(framed(new byte[]{0x03, 0x33}));
                    assertEquals(3, encryptedPlay.length);
                    clearOutput.write(encryptedPlay);
                    clearOutput.flush();
@@ -0,0 +1,7 @@
+package dev.strataproxy.api;
+
+/** Runs on a bounded plugin worker, never on a player's network event loop. */
+@FunctionalInterface
+public interface CommandHandler {
+    void execute(CommandInvocation invocation) throws Exception;
+}
@@ -0,0 +1,28 @@
+package dev.strataproxy.api;
+
+import java.util.Objects;
+import java.util.function.Consumer;
+
+/** One player command. The raw arguments exclude the leading command name and separator. */
+public final class CommandInvocation {
+    private final PlayerView player;
+    private final String name;
+    private final String arguments;
+    private final Consumer<String> reply;
+
+    public CommandInvocation(PlayerView player, String name, String arguments, Consumer<String> reply) {
+        this.player = Objects.requireNonNull(player, "player");
+        this.name = Objects.requireNonNull(name, "name");
+        this.arguments = Objects.requireNonNull(arguments, "arguments");
+        this.reply = Objects.requireNonNull(reply, "reply");
+    }
+
+    public PlayerView player() { return player; }
+    public String name() { return name; }
+    public String arguments() { return arguments; }
+
+    /** Queues plain text chat if the session is connected and writable; excess replies may be dropped. */
+    public void reply(String message) {
+        reply.accept(Objects.requireNonNull(message, "message"));
+    }
+}
@@ -0,0 +1,10 @@
+package dev.strataproxy.api;
+
+/** A command registration owned by one plugin. Unregistration is idempotent. */
+public interface CommandRegistration extends AutoCloseable {
+    void unregister();
+
+    @Override default void close() {
+        unregister();
+    }
+}
@@ -0,0 +1,10 @@
+package dev.strataproxy.api;
+
+/** Registers player commands handled by this plugin at the proxy. */
+public interface Commands {
+    /**
+     * Reserves a root command name. Names are case insensitive and unique across plugins.
+     * An unregistered command continues to the backend unchanged.
+     */
+    CommandRegistration register(String name, CommandHandler handler);
+}
@@ -11,7 +11,8 @@ public interface Plugin {

    /**
     * Called after pending placement requests are ended and callback workers are asked to stop.
-     * A callback that ignores interruption may still be running during this method.
+     * A callback that ignores interruption may still be running during this method. The proxy
+     * allows a total of ten seconds for all plugins to finish their disable hooks.
     */
    default void onDisable() {
    }
@@ -9,6 +9,8 @@ public interface PluginContext {

    Servers servers();

+    Commands commands();
+
    Logger logger();

    /** Immutable configuration assigned to this plugin by the proxy. */
@@ -88,6 +88,7 @@ backends:
        @"
 listen: "0.0.0.0:25577"
 authentication: OFFLINE
+allowOfflinePublicAccess: true
 plugins:
  directory: "/opt/strataproxy/plugins"
 $enabled